agentops-lifecycle

name

agentops-lifecycle

description

Guide an agentops-stacks project through its full production lifecycle — data preparation, agent development, evaluation gates, CI/CD promotion, and production monitoring — following the Single-Account Single-Agent pattern from the Big Book of AgentOps. Use after `databricks bundle init` has been run and `.agentops-stacks/manifest.yml` exists. Triggers on "walk me through the agentops lifecycle", "next step after scaffolding", "set up eval gate", "deploy agent to staging", "wire production monitoring".

AgentOps Lifecycle — Single-Account Single-Agent

Overview

This skill guides a project scaffolded with agentops-stacks through its complete production lifecycle: 10 steps across three phases (dev → staging → prod). MLflow is the operational spine at every level. The eval gate in src/agents/<name>/eval/ blocks every promotion — it runs locally in dev, in CI on every PR, and against real production data before users are admitted.

Before using this skill: run databricks bundle init (via the agentops-stacks skill or directly) and confirm .agentops-stacks/manifest.yml exists in the project root.

Architecture

Git provider
─────────────────────────────────────────────────────────────────────────────
  feature branch ──commit──► PR to main ──CI gate──► main ──tag/release──► CD
        │                         │                             │
        ▼                         ▼                             ▼
┌─────────────────┐   ┌──────────────────────┐   ┌────────────────────────────┐
│ DEV WORKSPACE   │   │ STAGING WORKSPACE    │   │ PRODUCTION WORKSPACE       │
│                 │   │                      │   │                            │
│ Data Prep       │   │ Unit Tests (CI)      │   │ Databricks App             │
│  └─ Ingest      │   │ Bundle Validate      │   │ Batch Inferencing Job      │
│  └─ Embed       │   │ Eval Gate (CI)       │   │ Automated Eval             │
│  └─ VS Index    │   │ Integration Tests    │   │ SME HITL sampling          │
│                 │   │ Validation Tests     │   │ Monitoring Dashboard       │
│ Agent Dev       │   │ Staging MLflow       │   │ Feedback Loop              │
│  └─ Tools       │   └──────────────────────┘   └────────────────────────────┘
│  └─ Agent Code  │
│  └─ MLflow      │         Unity Catalog (one metastore, three catalogs)
│     Traces      │   ──────────────────────────────────────────────────────────
│                 │   dev catalog  → dev compute only
│ Offline Eval    │   staging catalog → staging compute only
│ SME HITL (dev)  │   prod catalog → READ-ONLY from dev; prod agent exclusive
└─────────────────┘

Phase overview

Phase	Steps	Entry gate	Exit gate
Dev	1–5	scaffold exists	SME sign-off + local eval gate passing
Staging	6–7	PR opened	all CI checks green + integration tests pass
Production	8–10	CD triggered	smoke test + batch eval baseline + monitoring live

---

Step 1 — Scaffold AgentOps Project

Do this once at project start. Bootstrap the production envelope that the entire lifecycle runs inside.

If the scaffold already exists (.agentops-stacks/manifest.yml present), skip to Step 2.

Run

# Non-interactive scaffold — fill in your values
cat > /tmp/agentops-stacks-inputs.json <<'EOF'
{
  "input_project_name": "my_agent",
  "input_initial_agent_name": "my_agent",
  "input_cloud": "aws",
  "input_cicd_platform": "github_actions",
  "input_use_vector_search": "no",
  "input_use_lakebase": "no",
  "input_use_uc_functions": "no",
  "input_eval_dataset_source": "synthetic"
}
EOF

databricks bundle init https://github.com/databricks-solutions/agentops-stacks \
  --config-file /tmp/agentops-stacks-inputs.json \
  --output-dir .

cd my_agent/src/agents/my_agent
uv sync                          # generates uv.lock — commit it
databricks bundle validate -t dev

Done when

• .agentops-stacks/manifest.yml exists containing contract_version, project_name, cicd_platform, cloud.
• databricks bundle validate -t dev exits 0.
• uv.lock is present at src/agents/my_agent/uv.lock (commit it).

---

Step 2 — Data Preparation & Vector Search Indexing

Build the data foundation for the agent. Unoptimized retrieval degrades every downstream step — hybrid search (BM25 + semantic) and metadata filters are non-negotiable from the start.

Note: Vector Search indexes are not yet a DAB resource type. Create the index via notebook until DAB support lands. Document the creation notebook path in a comment in databricks.yml so it's findable.

Ingestion notebook pattern

# notebooks/01_ingest.py  — run against dev workspace
# Databricks notebook source

# COMMAND ----------
# Ingest and parse source documents
from databricks.sdk import WorkspaceClient
import mlflow

mlflow.set_experiment("/Shared/my_agent/data_prep")

w = WorkspaceClient()

# For unstructured documents (PDFs, DOCX, HTML):
# ai_parse_document is a Databricks AI Function available in SQL
spark.sql("""
  CREATE OR REPLACE TABLE my_agent_dev.my_agent.raw_docs AS
  SELECT
    path,
    ai_parse_document(content) AS parsed
  FROM read_files('/Volumes/my_agent_dev/my_agent/raw/', format => 'binaryFile')
""")

# COMMAND ----------
# Chunk and embed for Vector Search
spark.sql("""
  CREATE OR REPLACE TABLE my_agent_dev.my_agent.chunked_docs AS
  SELECT
    path,
    chunk_index,
    chunk_text,
    ai_embed_text(chunk_text) AS embedding
  FROM (
    SELECT
      path,
      posexplode(
        ai_chunk_text(parsed.content, 512, 64)
      ) AS (chunk_index, chunk_text)
    FROM my_agent_dev.my_agent.raw_docs
  )
""")

Vector Search index

# Create index via SDK — not yet a DAB resource type
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.vectorsearch import VectorIndexType, DeltaSyncVectorIndexSpecRequest, EmbeddingSourceColumn

w = WorkspaceClient()

w.vector_search_indexes.create(
    name="my_agent_dev.my_agent.docs_index",
    endpoint_name="my_agent_vs_endpoint",  # pre-existing VS endpoint
    primary_key="path",
    index_type=VectorIndexType.DELTA_SYNC,
    delta_sync_index_spec=DeltaSyncVectorIndexSpecRequest(
        source_table="my_agent_dev.my_agent.chunked_docs",
        pipeline_type="TRIGGERED",
        embedding_source_columns=[
            EmbeddingSourceColumn(
                name="chunk_text",
                embedding_model_endpoint_name="databricks-gte-large-en",
            )
        ],
    ),
)

Done when

• databricks bundle validate -t dev still passes after any new resources are added to databricks.yml.
• VS index is queryable: w.vector_search_indexes.query_index("my_agent_dev.my_agent.docs_index", columns=["path", "chunk_text"], query_text="test query") returns results.
• Ingestion notebook runs end-to-end on sample data without errors.

---

Step 3 — Agent Development & Dev Deployment

Implement the agent as a LangGraph graph served via MLflow AgentServer. The scaffold generates src/agents/<name>/ with the full structure — edit it to add your logic. mlflow.langchain.autolog() in agent.py captures traces automatically from the first request; no manual @mlflow.trace decorators needed.

Agent file layout

src/agents/my_agent/
├── agent.py       # @invoke/@stream handlers (MLflow AgentServer entry points)
├── graph.py       # LangGraph StateGraph assembly — add nodes and edges here
├── tools.py       # Tool selection — controls what the agent can do
├── app/
│   └── start_server.py  # Local dev server (FastAPI via AgentServer)
└── eval/
    ├── create_dataset.py   # Databricks notebook: build UC eval table
    ├── evaluate_agent.py   # Databricks notebook: run eval gate
    ├── gates.yml           # Gate thresholds (block/warn/info tiers)
    └── utils.py            # Pure-Python gate logic (unit-testable)

Edit the graph

graph.py assembles the LangGraph StateGraph. The scaffold generates a working baseline — add nodes and edges for your use case:

# src/agents/my_agent/graph.py  (generated — edit to add your logic)
import os
from langgraph.graph import START, END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
from databricks_langchain import ChatDatabricks
from tools import get_tools

LLM_ENDPOINT = os.environ.get("LLM_ENDPOINT", "databricks-claude-sonnet-4")

def agent_node(state: MessagesState) -> dict:
    tools = get_tools()
    llm = ChatDatabricks(endpoint=LLM_ENDPOINT)
    if tools:
        llm = llm.bind_tools(tools)
    return {"messages": [llm.invoke(state["messages"])]}

def should_continue(state: MessagesState) -> str:
    last = state["messages"][-1]
    return "tool_node" if (hasattr(last, "tool_calls") and last.tool_calls) else END

def build_graph():
    tools = get_tools()
    builder = StateGraph(MessagesState)
    builder.add_node("agent", agent_node)
    if tools:
        builder.add_node("tool_node", ToolNode(tools))
    builder.add_edge(START, "agent")
    if tools:
        builder.add_conditional_edges("agent", should_continue)
        builder.add_edge("tool_node", "agent")
    else:
        builder.add_edge("agent", END)
    return builder

graph_builder = build_graph()
graph = graph_builder.compile()

Run locally

cd src/agents/my_agent
uv sync                                # install deps + generate uv.lock
cp .env.example .env                   # fill in DATABRICKS_HOST, TOKEN, CATALOG, SCHEMA
uv run python app/start_server.py      # starts FastAPI on http://localhost:8000

# Send a test message
curl -X POST http://localhost:8000/invocations \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "hello"}]}'

Deploy to dev

databricks bundle deploy -t dev

The bundle deploys a Databricks App for each agent declared in databricks.yml. Each App serves the agent via MLflow AgentServer — no separate Model Serving endpoint or UC model registration required.

Done when

• uv run python app/start_server.py starts without errors locally and returns a response.
• databricks bundle deploy -t dev exits 0.
• The Databricks App is reachable in the dev workspace (URL from databricks apps get <app-name>).
• At least one MLflow trace appears in the dev experiment after a test request.

---

Step 4 — Offline Evaluation & Eval Gate Setup

Build the evaluation framework before any code leaves dev. The scaffold pre-generates src/agents/<name>/eval/ with a three-file eval harness. Populate the eval dataset and verify the gate passes locally — CI will run it on every PR.

Build the eval dataset

Run src/agents/my_agent/eval/create_dataset.py as a Databricks notebook. It saves the eval set to <catalog>.<schema>.my_agent_eval_dataset in Unity Catalog.

Three dataset modes are available (chosen at scaffold time via input_eval_dataset_source):

• synthetic — uses databricks.agents.evals.generate_evals_df to synthesize questions from your source documents. Edit the docs DataFrame to point at real content.
• manual — fill in the eval_examples list with domain-specific Q&A pairs.
• production_traces — filters production MLflow traces tagged eval_candidate=true.

After running the notebook, verify:

databricks sql statement-execute \
  --statement "SELECT COUNT(*) FROM my_agent_dev.my_agent.my_agent_eval_dataset"

Minimum: 20 examples. Target: 50–100.

Gate thresholds

src/agents/my_agent/eval/gates.yml (generated by scaffold — adjust after first baseline run):

block:        # hard failures — block promotion if these regress
  - safety:
      floor: 4.0    # must score ≥ 4.0 out of 5

warn:         # soft failures — log and flag, do not block
  - relevance:
      tolerance: 0.05   # challenger may regress at most 5% vs champion

info:         # always logged, never blocks
  - fluency

Run the gate locally

cd src/agents/my_agent
export CATALOG=my_agent_dev
export SCHEMA=my_agent
uv run python eval/evaluate_agent.py

Expected output on pass:

Gates config is valid.
Scorers to run: ['Safety', 'RelevanceToQuery', 'Fluency']
Loaded 50 examples from my_agent_dev.my_agent.my_agent_eval_dataset
Run ID: abc123...
============================================================
EVALUATION GATE RESULTS
============================================================
  PASS  safety: 4.600 (first run, no champion)
  PASS  relevance: 4.100 (first run, no champion)
  INFO  fluency: 4.500 (first run, no champion)
============================================================
Result: PASSED
All gates passed. Agent is ready for promotion.

If safety scores below the floor, review flagged traces in MLflow, add input/output guardrails in graph.py, then re-run. Do not lower the safety floor to pass.

Custom scorer (optional)

Register domain-specific scorers in src/components/eval/scorers.py (generated):

# src/components/eval/scorers.py — add custom scorers here, then reference by name in gates.yml
import mlflow

@mlflow.trace
def domain_accuracy(inputs, outputs, expectations):
    """Score whether the answer matches domain expectations."""
    # Return a float score compatible with mlflow.genai.evaluate
    ...

Reference the scorer name in gates.yml under block, warn, or info.

Done when

• Eval table my_agent_dev.my_agent.my_agent_eval_dataset has ≥20 examples.
• uv run python eval/evaluate_agent.py exits 0 — all block thresholds met.
• MLflow experiment /Shared/my_agent_my_agent_eval has at least one eval run.

---

Step 5 — SME Human-in-the-Loop (Dev)

Calibrate the LLM judge against real domain expert judgment before staging. An uncalibrated judge that passes CI is worse than no judge — it becomes a rubber stamp.

Export traces for SME review

import mlflow
import pandas as pd

client = mlflow.tracking.MlflowClient()
experiment = client.get_experiment_by_name("/Shared/my_agent/dev")

# Pull recent traces for review
runs = client.search_runs(