agentops-lifecycle

# 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 ```bash # 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 ```python # 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 ```python # 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: ```python # 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-5") 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 ```bash 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 ```bash 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 3.5 — Add a Supervisor (only when you have >1 agent) If the project has more than one agent, add a supervisor to route user queries across them. This is a post-scaffold pattern — use the `/add-supervisor` skill. Skip if you have a single agent. First check that a supervisor is warranted — if a deterministic router or a sequential chain would do, prefer that (the Big Book names premature multi-agent orchestration an anti-pattern). Add one only when routing genuinely depends on the request across ≥2 specialists. The supervisor is a **custom LangGraph** agent (GA): a hand-written graph served as a Databricks App, fully declared in `databricks.yml`. Scaffolded as an agent App, it gets its own `eval/gates.yml`, which CI's `detect_patterns → eval_gate` picks up with no workflow change. See `docs/supervisor-patterns.md`. (A managed "Supervisor API" pattern was removed — that API is deprecated, EOL 2026-09-30; Databricks recommends custom agents on Apps.) ```bash # In your coding assistant: /add-supervisor # or: "add a supervisor that routes between my rag and support agents" ``` The supervisor is recorded in `.agentops-stacks/manifest.yml` under `supervisor:`. From here, the rest of the lifecycle (eval gate, CI, staging, prod) applies to the supervisor agent exactly as it does to any agent. --- ## 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: ```bash 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): ```yaml 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 ```bash 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): ```python # 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 ...