Case 6 (Capstone) — Production Agent Microservice

Executive Briefing: Transform an experimental LangGraph agent into an enterprise-grade cloud microservice. The service exposes asynchronous streaming endpoints via FastAPI (/v1/agent/stream, /v1/agent/interrupt/resume), persists state to PostgreSQL with connection pooling, logs every execution trace to LangSmith with metadata tags, and integrates an automated CI evaluation harness benchmarking 30+ golden dataset test cases before deployment.

1. Enterprise Scenario & Requirements

Building an agent in a Jupyter notebook is 20% of the battle; shipping to production is the remaining 80%:

1. Asynchronous FastAPI Service: Build a production FastAPI application that exposes Server-Sent Events (SSE) streaming (StreamingResponse) emitting token chunks and intermediate node states.
2. PostgreSQL Checkpoint Pooling: Use psycopg_pool.AsyncConnectionPool to back AsyncPostgresSaver without thread starvations.
3. Observability & LangSmith Tracing: Instrument the service with LANGSMITH_TRACING=true. Ensure every run passes metadata={"user_id": ..., "environment": "production", "release": "v1.4.0"} for span attribution.
4. CI/CD Evaluation Suite: Create an offline evaluation script using langsmith.evaluate() with:
   • A golden evaluation dataset of 30+ prompt-reference pairs.
   • Deterministic custom code evaluators (token count limits, schema conformity).
   • An LLM-as-a-judge evaluator grading response factual correctness and tone.
5. Containerization: Write a multi-stage Dockerfile and docker-compose.yml orchestrating the FastAPI agent and PostgreSQL database.

2. Explicit Deliverables

1. app/: FastAPI application code (main.py, agent.py, config.py, schemas.py).
2. evals/: Evaluation suite (golden_dataset.json, run_evals.py, evaluators.py).
3. Dockerfile & docker-compose.yml: Containerized deployment orchestration.
4. eval_report.json / screenshot: LangSmith experiment dashboard showing benchmark scores.
5. README.md: Architecture runbook with endpoint documentation and deployment steps.

3. Starter Code & Scaffold

# app/main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from contextlib import asynccontextmanager
import json

app = FastAPI(title="Production LangGraph Agent Service", version="1.0.0")

class QueryRequest(BaseModel):
    query: str
    thread_id: str
    user_id: str

async def stream_agent_events(query: str, thread_id: str, user_id: str):
    config = {
        "configurable": {"thread_id": thread_id},
        "metadata": {"user_id": user_id, "environment": "production"}
    }
    # Stream events from compiled LangGraph instance
    async for event in agent_app.astream_events({"messages": [query]}, config=config, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                yield f"data: {json.dumps({'type': 'token', 'content': content})}

"
        elif kind == "on_tool_start":
            yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}

"

@app.post("/v1/agent/stream")
async def handle_stream(req: QueryRequest):
    return StreamingResponse(
        stream_agent_events(req.query, req.thread_id, req.user_id),
        media_type="text/event-stream"
    )

4. 100-Point Grading Rubric