Module 10: Agents in Production and Alternatives

8. Final Project: Research Agent v7 — Production Deployment

Project Overview

In Module 1 you built a manual ReAct loop with 3 simulated tools. A for loop, three if/elifs, and a print() at the end. It worked. But it wasn't an agent — it was a script with an LLM.

Nine modules later, you have a multi-agent system with 4 specialized agents (Supervisor, Researcher, Analyst, Writer), tools connected dynamically via MCP, planning with priorities, reflection with quality gates, persistent memory with PostgresSaver and long-term memory, and a testing suite with a golden dataset, trajectory evaluation, and performance benchmarks.

What you don't have is a service. Your agent lives in a notebook. It runs with python agent.py. If your laptop closes, the agent dies. If two people want to use it at the same time, they can't. If an MCP server goes down, the agent crashes with no explanation. If the cost spikes, you don't find out until the bill arrives.

This project closes that gap. You take Research Agent v6 (with its complete testing suite from M9) and wrap it in a production layer: a FastAPI server with REST endpoints, containerization with Docker, health checks that verify every dependency, rate limiting per user, cost tracking per request, error recovery with fallbacks, LangSmith production tracing, and a production checklist verified point by point.

The result is Research Agent v7: the same multi-agent system you built, but deployable, monitorable, resilient, and with controlled costs. It isn't a different project — it's the final version of the same project you started in Module 4.

Estimated duration: 120-150 minutes.

Project Goal

Transform Research Agent v6 (a multi-agent system with testing) into a production service with FastAPI, Docker, health checks, rate limiting, cost tracking, error recovery, and LangSmith production tracing.

By the end you'll be able to:

Expose the Research Agent as a REST API with synchronous and asynchronous endpoints via FastAPI
Containerize the complete service with Docker and Docker Compose (agent + Redis)
Implement health checks that verify the agent, the LLM provider, Redis, the MCP servers, and the API keys
Configure rate limiting per user and token budgets per request
Track costs per query, per agent, and per tool call
Implement error recovery: retries with backoff, model fallback, a circuit breaker for tools
Configure LangSmith production tracing with metric-based alerts
Complete a production checklist of 20+ items and verify each one

What Changes vs v6 (M9)

In v6 you added testing. In v7 you add the production layer that wraps everything before it. The agent's code isn't modified — what changes is how it runs, how it's accessed, and how it's operated.

The v7 structure

research_agent_v7/
├── agent.py / tools.py / prompts.py / servers/   # ← Unchanged (v5)
├── tests/ / evaluation/                           # ← Unchanged (v6)
├── server.py                                      # ← NEW: the FastAPI app
├── config.py                                      # ← NEW: Pydantic Settings
├── middleware/                                     # ← NEW
│   ├── rate_limiter.py                            #    Per-user rate limiting
│   ├── cost_tracker.py                            #    Cost tracking per request
│   └── error_recovery.py                          #    Retries, fallbacks, circuit breaker
├── health.py                                      # ← NEW: Health checks
├── Dockerfile                                     # ← NEW
├── docker-compose.yml                             # ← NEW
├── .env.example                                   # ← NEW
└── requirements.txt                               # ← Updated

What v7 adds

Layer	v6 (M9)	v7 (M10)
Access	`python agent.py`	`POST /chat`, `POST /chat/async`, `GET /jobs/{id}`
Runtime	Your laptop	A Docker container with restart policies
Health	None	`/health` verifies the agent, LLM, Redis, MCP, API keys
Costs	An estimate in the benchmarks	Real tracking per request with alerts
Errors	Uncaught exceptions	Retries, fallbacks, circuit breakers
Concurrency	One user at a time	Multiple users with rate limiting
Observability	LangSmith in testing	Continuous LangSmith production tracing

Technical Specifications

New dependencies

pip install fastapi uvicorn pydantic-settings slowapi httpx redis

Research Agent v6's dependencies (langchain, langgraph, mcp, pytest, langsmith) are already installed.

Environment variables

# .env.example — copy to .env and fill in
OPENAI_API_KEY=sk-proj-your-key
LANGSMITH_API_KEY=lsv2_pt_your-langsmith-key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=research-agent-production

MODEL_NAME=gpt-4.1-mini
MAX_TOKENS=4096
TEMPERATURE=0.0

REDIS_URL=redis://localhost:6379
LOG_LEVEL=INFO
APP_VERSION=7.0.0

RATE_LIMIT_PER_MINUTE=10
MAX_COST_PER_REQUEST=0.50
DAILY_COST_BUDGET=50.00

The v7 Architecture (Final)

                    ┌──────────────────────────────────────────────────────┐
                    │                  PRODUCTION LAYER                    │
                    │                                                      │
  HTTP Request ───► │  ┌────────────┐  ┌────────────┐  ┌──────────────┐   │
                    │  │ Rate       │  │ Cost       │  │ Error        │   │
                    │  │ Limiter    │──│ Tracker    │──│ Recovery     │   │
                    │  │ (per user) │  │ (per req)  │  │ (fallbacks)  │   │
                    │  └────────────┘  └────────────┘  └──────┬───────┘   │
                    │                                         │           │
                    │  ┌──────────────────────────────────────▼────────┐  │
                    │  │           FASTAPI SERVER                      │  │
                    │  │  POST /chat    POST /chat/async  GET /health  │  │
                    │  └──────────────────────┬────────────────────────┘  │
                    │                         │                           │
                    └─────────────────────────┼───────────────────────────┘
                                              │
                    ┌─────────────────────────▼───────────────────────────┐
                    │              MULTI-AGENT SYSTEM (v5)                 │
                    │                                                      │
                    │  ┌─────────────────────────────────────────────┐     │
                    │  │              SUPERVISOR                      │     │
                    │  │  decompose → assign → validate → finalize   │     │
                    │  └──────┬──────────┬──────────┬────────────────┘     │
                    │         │          │          │                       │
                    │  ┌──────▼───┐ ┌────▼─────┐ ┌─▼────────┐             │
                    │  │RESEARCHER│ │ ANALYST  │ │  WRITER  │             │
                    │  │search_web│ │calculate │ │write_file│             │
                    │  │search_pap│ │extract   │ │read_file │             │
                    │  └──────────┘ │compare   │ └──────────┘             │
                    │               └──────────┘                           │
                    │                                                      │
                    │  ┌─────────────────────────────────────────────┐     │
                    │  │  TESTING SUITE (v6)  │  LANGSMITH TRACING   │     │
                    │  └─────────────────────────────────────────────┘     │
                    └──────────────────────────────────────────────────────┘
                                              │
                    ┌─────────────────────────▼───────────────────────────┐
                    │              INFRASTRUCTURE                          │
                    │  Docker │ Redis (cache/state) │ Health Checks        │
                    └──────────────────────────────────────────────────────┘

Step 1: The FastAPI Server

The server exposes three endpoints: /chat for short queries (an immediate response), /chat/async for long research queries (submit + poll), and /health to check the system's state.

Configuration with Pydantic Settings

# config.py
from pydantic_settings import BaseSettings
from pydantic import Field


class Settings(BaseSettings):
    openai_api_key: str = Field(..., description="OpenAI API key")
    langsmith_api_key: str = Field("", description="LangSmith API key")
    langsmith_tracing: bool = Field(True, description="Enable LangSmith tracing")
    langsmith_project: str = Field("research-agent-production")

    model_name: str = Field("gpt-4.1-mini")
    max_tokens: int = Field(4096)
    temperature: float = Field(0.0)

    redis_url: str = Field("redis://localhost:6379")
    log_level: str = Field("INFO")
    app_version: str = Field("7.0.0")

    rate_limit_per_minute: int = Field(10)
    max_cost_per_request: float = Field(0.50)
    daily_cost_budget: float = Field(50.00)
    max_agent_steps: int = Field(25)
    request_timeout: int = Field(120)

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}


settings = Settings()

The Server

# server.py
from fastapi import FastAPI, HTTPException, BackgroundTasks, Request
from pydantic import BaseModel
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from contextlib import asynccontextmanager
import uuid
import time
import logging
import asyncio

from config import settings
from agent import (
    build_multi_agent_system, build_researcher_agent,
    build_analyst_agent, build_writer_agent,
)
from health import run_health_checks
from middleware.rate_limiter import check_rate_limit
from middleware.cost_tracker import CostTracker
from middleware.error_recovery import invoke_with_recovery

logger = logging.getLogger(__name__)
logging.basicConfig(level=settings.log_level)

checkpointer = MemorySaver()
system = None
cost_tracker = CostTracker(daily_budget=settings.daily_cost_budget)
jobs: dict[str, dict] = {}
active_requests: set[str] = set()


class ChatRequest(BaseModel):
    message: str
    thread_id: str | None = None
    user_id: str = "anonymous"

class ChatResponse(BaseModel):
    response: str
    thread_id: str
    cost_usd: float
    latency_ms: float
    metadata: dict | None = None

class AsyncJobResponse(BaseModel):
    job_id: str
    status: str
    thread_id: str


@asynccontextmanager
async def lifespan(app: FastAPI):
    global system
    logger.info("Compiling the multi-agent system...")

    researcher = build_researcher_agent()
    analyst = build_analyst_agent()
    writer = build_writer_agent()
    system = build_multi_agent_system(researcher, analyst, writer)

    logger.info("System ready — Research Agent v7 operational")
    yield

    logger.info("Starting graceful shutdown...")
    if active_requests:
        logger.info(f"Waiting for {len(active_requests)} in-flight requests...")
        for _ in range(30):
            if not active_requests:
                break
            await asyncio.sleep(1)
    logger.info("Shutdown complete")


app = FastAPI(
    title="Research Agent v7 API",
    description="A multi-agent research system — production-ready",
    version=settings.app_version,
    lifespan=lifespan,
)


def _build_invoke_input(message: str) -> dict:
    return {
        "messages": [HumanMessage(content=message)],
        "original_query": message,
        "sub_tasks": [],
        "current_agent": "",
        "agent_results": {},
        "iteration_count": 0,
        "max_iterations": settings.max_agent_steps,
        "workers_called": [],
        "final_report": None,
        "status": "decomposing",
    }


@app.post("/chat", response_model=ChatResponse,
          summary="Synchronous chat with the Research Agent",
          tags=["Agent"])
async def chat(request: ChatRequest):
    check_rate_limit(request.user_id, settings.rate_limit_per_minute)

    thread_id = request.thread_id or str(uuid.uuid4())
    request_id = str(uuid.uuid4())
    active_requests.add(request_id)
    start = time.time()

    try:
        config = {"configurable": {"thread_id": thread_id}}
        invoke_input = _build_invoke_input(request.message)

        result = await invoke_with_recovery(
            system, invoke_input, config,
            max_cost=settings.max_cost_per_request,
        )

        response_text = result.get("final_report", "") or ""
        if not response_text:
            last_msg = result.get("messages", [])[-1] if result.get("messages") else None
            response_text = last_msg.content if last_msg else "No response."

        elapsed_ms = (time.time() - start) * 1000
        cost = cost_tracker.estimate_cost(result)
        cost_tracker.record(request.user_id, cost)

        return ChatResponse(
            response=response_text,
            thread_id=thread_id,
            cost_usd=cost,
            latency_ms=round(elapsed_ms, 1),
            metadata={
                "steps": len(result.get("messages", [])),
                "workers": result.get("workers_called", []),
                "model": settings.model_name,
            },
        )
    except Exception as e:
        logger.error(f"Error in /chat: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail="Error processing the request")
    finally:
        active_requests.discard(request_id)


async def _run_async_job(job_id: str, message: str, thread_id: str, user_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    jobs[job_id]["status"] = "running"
    start = time.time()

    try:
        invoke_input = _build_invoke_input(message)
        result = await invoke_with_recovery(
            system, invoke_input, config,
            max_cost=settings.max_cost_per_request,
        )

        response_text = result.get("final_report", "") or ""
        cost = cost_tracker.estimate_cost(result)
        cost_tracker.record(user_id, cost)

        jobs[job_id]["status"] = "completed"
        jobs[job_id]["result"] = response_text
        jobs[job_id]["cost_usd"] = cost
        jobs[job_id]["latency_ms"] = round((time.time() - start) * 1000, 1)
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)
        logger.error(f"Job {job_id} failed: {e}", exc_info=True)


@app.post("/chat/async", response_model=AsyncJobResponse,
          summary="Asynchronous research query (submit + poll)",
          tags=["Agent"])
async def chat_async(request: ChatRequest, background_tasks: BackgroundTasks):
    check_rate_limit(request.user_id, settings.rate_limit_per_minute)

    job_id = str(uuid.uuid4())
    thread_id = request.thread_id or str(uuid.uuid4())

    jobs[job_id] = {"status": "pending", "thread_id": thread_id}
    background_tasks.add_task(
        _run_async_job, job_id, request.message, thread_id, request.user_id,
    )

    return AsyncJobResponse(job_id=job_id, status="pending", thread_id=thread_id)


@app.get("/jobs/{job_id}", summary="The status of an async job", tags=["Agent"])
async def get_job_status(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    return jobs[job_id]

Three key decisions: (1) every invocation goes through invoke_with_recovery for retries, fallbacks, and cost limits. (2) Every response includes cost_usd — cost transparency from the very first request. (3) Graceful shutdown waits up to 30 seconds for in-flight requests before shutting down.

Step 2: Docker

The Dockerfile

FROM python:3.12-slim AS builder

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim AS runtime

WORKDIR /app

RUN groupadd -r agent && useradd -r -g agent agent

COPY --from=builder /install /usr/local
COPY agent.py tools.py prompts.py config.py server.py health.py ./
COPY middleware/ ./middleware/
COPY servers/ ./servers/
RUN chown -R agent:agent /app
USER agent

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health/live')"

CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

--workers 1 is intentional. LangGraph agents keep state in memory (checkpoints, MCP connections, tool registries). Multiple workers multiply the memory consumption without a proportional benefit. Scale with multiple containers, not multiple workers.

--start-period=60s gives the container a minute to compile the 4 agents' graphs before Docker starts checking health. Without it, the container restarts in a loop because the health check fails during initialization.

Docker Compose

services:
  agent:
    build: .
    ports: ["8000:8000"]
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
      - LANGSMITH_TRACING=true
      - REDIS_URL=redis://redis:6379
    depends_on:
      redis: { condition: service_healthy }
    restart: unless-stopped
    deploy:
      resources:
        limits: { memory: 2G }

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    volumes: [redis_data:/data]

volumes:
  redis_data:

depends_on with condition: service_healthy: The agent doesn't start until Redis is up.
memory: 2G: It prevents a runaway agent from consuming all the host's resources.
Create a .dockerignore to exclude .git, .env, __pycache__, tests/, and *.md. Bring it up with docker compose up --build -d.

Step 3: Health Checks

A /health that always returns {"status": "ok"} is useless. Your system depends on: an external LLM (OpenAI), MCP servers, Redis, valid API keys, and the compiled graph. The health check verifies each one.

# health.py
from datetime import datetime, timezone
import os, httpx
from config import settings


async def run_health_checks(graph) -> dict:
    checks = {}

    checks["agent"] = {"status": "healthy" if graph is not None else "unhealthy"}

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get("https://api.openai.com/v1/models",
                headers={"Authorization": f"Bearer {settings.openai_api_key}"})
            checks["openai"] = {"status": "healthy" if resp.status_code == 200 else "degraded"}
    except Exception as e:
        checks["openai"] = {"status": "unhealthy", "detail": str(e)}

    try:
        import redis.asyncio as aioredis
        r = aioredis.from_url(settings.redis_url)
        await r.ping()
        await r.aclose()
        checks["redis"] = {"status": "healthy"}
    except Exception as e:
        checks["redis"] = {"status": "unhealthy", "detail": str(e)}

    required_keys = ["OPENAI_API_KEY", "LANGSMITH_API_KEY"]
    missing = [k for k in required_keys if not os.getenv(k)]
    checks["api_keys"] = {"status": "healthy" if not missing else "unhealthy"}

    overall = all(c["status"] == "healthy" for c in checks.values())
    return {"status": "healthy" if overall else "unhealthy",
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "version": settings.app_version, "checks": checks}

Health endpoints in the server

Add these endpoints to server.py:

@app.get("/health", summary="Complete health check", tags=["System"])
async def health_check():
    result = await run_health_checks(system)
    if result["status"] != "healthy":
        raise HTTPException(status_code=503, detail=result)
    return result


@app.get("/health/live", summary="Liveness probe", tags=["System"])
async def liveness():
    return {"status": "alive"}


@app.get("/health/ready", summary="Readiness probe", tags=["System"])
async def readiness():
    if system is None:
        raise HTTPException(status_code=503, detail="Agent not ready")
    return {"status": "ready"}

Three levels of health check:

Liveness (/health/live): the process answers HTTP. Docker and Kubernetes use it to decide whether to restart the container.
Readiness (/health/ready): the agent can process requests. The load balancer uses it to decide whether to send traffic.
Deep health (/health): it verifies every dependency. You use it for debugging and monitoring dashboards.

When everything is healthy, /health returns each component with a "healthy" status and the timestamp. When something fails, it returns 503 with the exact detail of which component is "unhealthy" — not a generic "service unavailable".

Step 4: Cost Control

Agents consume tokens unpredictably. A simple query can cost $0.05. A complex research query with 4 agents, 12 tool calls, planning, and reflection can cost $1.50. Without cost control, 100 active users burn your budget in hours.

Rate Limiting

# middleware/rate_limiter.py
import time
from collections import defaultdict
from fastapi import HTTPException


_request_log: dict[str, list[float]] = defaultdict(list)


def check_rate_limit(user_id: str, max_per_minute: int):
    now = time.time()
    window_start = now - 60

    _request_log[user_id] = [
        ts for ts in _request_log[user_id] if ts > window_start
    ]

    if len(_request_log[user_id]) >= max_per_minute:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded: {max_per_minute} requests/minute. "
                   f"Try again in {60 - int(now - _request_log[user_id][0])} seconds.",
        )

    _request_log[user_id].append(now)

In production with multiple containers, replace the dict with Redis (INCR + EXPIRE) so the limit is global.

The Cost Tracker

# middleware/cost_tracker.py
import time
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from langchain_core.messages import AIMessage

logger = logging.getLogger(__name__)

MODEL_COSTS_PER_1K = {
    "gpt-4.1": {"input": 0.002, "output": 0.008},
    "gpt-4.1-mini": {"input": 0.0004, "output": 0.0016},
    "gpt-4.1-nano": {"input": 0.0001, "output": 0.0004},
}


@dataclass
class CostTracker:
    daily_budget: float = 50.0
    _daily_costs: dict[str, float] = field(default_factory=lambda: defaultdict(float))
    _user_costs: dict[str, float] = field(default_factory=lambda: defaultdict(float))
    _current_day: str = ""

    def _today(self) -> str:
        return time.strftime("%Y-%m-%d")

    def _reset_if_new_day(self):
        today = self._today()
        if today != self._current_day:
            self._daily_costs.clear()
            self._current_day = today

    def estimate_cost(self, result: dict, model: str = "gpt-4.1-mini") -> float:
        messages = result.get("messages", [])
        costs = MODEL_COSTS_PER_1K.get(model, MODEL_COSTS_PER_1K["gpt-4.1-mini"])

        input_tokens = 0
        output_tokens = 0
        for msg in messages:
            token_est = len(str(msg.content).split()) * 1.3 if hasattr(msg, "content") else 0
            if isinstance(msg, AIMessage):
                output_tokens += token_est
            else:
                input_tokens += token_est

        return (input_tokens / 1000 * costs["input"] +
                output_tokens / 1000 * costs["output"])

    def record(self, user_id: str, cost: float):
        self._reset_if_new_day()
        self._daily_costs[self._today()] += cost
        self._user_costs[user_id] += cost

        daily_total = self._daily_costs[self._today()]
        if daily_total > self.daily_budget * 0.8:
            logger.warning(
                f"ALERT: Daily cost ${daily_total:.2f} "
                f"at 80%+ of the budget (${self.daily_budget})"
            )

    def get_daily_total(self) -> float:
        self._reset_if_new_day()
        return self._daily_costs.get(self._today(), 0.0)

    def check_budget(self) -> bool:
        return self.get_daily_total() < self.daily_budget

Add a GET /metrics endpoint to server.py that returns daily_cost_usd, budget_remaining_pct, active_requests, and pending_jobs. It's your minimal operations dashboard.

Step 5: Error Recovery

An agent in production fails. The LLM provider has rate limits. MCP servers go down. API keys expire. The network times out. The question isn't if it fails, but what happens when it does.

# middleware/error_recovery.py
import asyncio, logging, time
from collections import defaultdict
from langchain.chat_models import init_chat_model

logger = logging.getLogger(__name__)
FALLBACK_MODELS = ["gpt-4.1-mini", "gpt-4.1-nano"]


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self._failures: dict[str, int] = defaultdict(int)
        self._last_failure: dict[str, float] = defaultdict(float)
        self._open: dict[str, bool] = defaultdict(bool)

    def is_open(self, component: str) -> bool:
        if not self._open[component]:
            return False
        if time.time() - self._last_failure[component] > self.reset_timeout:
            self._open[component] = False
            self._failures[component] = 0
            return False
        return True

    def record_failure(self, component: str):
        self._failures[component] += 1
        self._last_failure[component] = time.time()
        if self._failures[component] >= self.failure_threshold:
            self._open[component] = True
            logger.warning(f"Circuit breaker OPEN: {component}")

    def record_success(self, component: str):
        self._failures[component] = 0
        self._open[component] = False


circuit_breaker = CircuitBreaker()


async def invoke_with_retry(system, invoke_input, config, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await asyncio.wait_for(
                system.ainvoke(invoke_input, config), timeout=120)
            circuit_breaker.record_success("agent")
            return result
        except (asyncio.TimeoutError, Exception) as e:
            logger.warning(f"Attempt {attempt+1}/{max_retries} failed: {e}")
            circuit_breaker.record_failure("agent")
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
    raise RuntimeError("All the retries failed")


async def invoke_with_recovery(system, invoke_input, config, max_cost=0.50):
    if circuit_breaker.is_open("agent"):
        return _fallback_response(invoke_input)
    try:
        return await invoke_with_retry(system, invoke_input, config)
    except RuntimeError:
        for model_name in FALLBACK_MODELS:
            try:
                model = init_chat_model(f"openai:{model_name}")
                result = await model.ainvoke(invoke_input["messages"])
                return {"messages": invoke_input["messages"] + [result],
                        "final_report": result.content,
                        "workers_called": ["fallback"], "status": "complete"}
            except Exception:
                continue
        return _fallback_response(invoke_input)


def _fallback_response(invoke_input):
    return {"messages": invoke_input["messages"],
            "final_report": "The system is temporarily unavailable. "
                            "Please try again in a few minutes.",
            "workers_called": [], "status": "fallback"}

The chain: retry with backoff → circuit breaker (3 failures → open for 60s) → model fallback (a single gpt-4.1-nano) → an honest answer to the user. Each level absorbs a different type of failure. The user never sees a 500 — they always get something.

Step 6: Production Config

LangSmith Production Tracing

LangSmith has been configured since v6 (testing). In production, what changes is the scope: every user request gets traced, not just the test runs. Set the environment variables before importing LangChain modules:

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = settings.langsmith_api_key
os.environ["LANGCHAIN_PROJECT"] = settings.langsmith_project

Every ainvoke generates a complete trace: which agent ran, which tools it called, how many tokens it consumed, how long it took. In LangSmith you can filter by project, see p50/p95 latencies, and configure alerts when metrics degrade.

Structured Logging

Replace basic print() and logging.info() with structured logging in JSON. Every log includes a request_id to correlate all the logs of a single request:

import json
import logging
from datetime import datetime, timezone
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="")


class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "request_id": request_id_var.get(""),
        })

To assign request IDs, use a Starlette middleware (BaseHTTPMiddleware) that generates a short UUID, sets it in request_id_var, and includes it in the X-Request-ID response header.

The Complete Research Agent v7

The whole project in one view:

research_agent_v7/
├── agent.py / tools.py / prompts.py / servers/    # CORE (v5) — unchanged
├── tests/ / evaluation/ / pytest.ini              # TESTING (v6) — unchanged
├── server.py                    # FastAPI: /chat, /chat/async, /health, /metrics
├── config.py                    # Pydantic Settings
├── health.py                    # Deep health checks
├── middleware/
│   ├── rate_limiter.py          # Per-user rate limiting
│   ├── cost_tracker.py          # Cost tracking per request
│   └── error_recovery.py       # Retries, circuit breaker, fallbacks
├── Dockerfile                   # Multi-stage build, non-root user
├── docker-compose.yml           # Agent + Redis
├── .env.example                 # The variables template
└── requirements.txt             # All the dependencies

Production Checklist

Before considering your Research Agent v7 ready for production, verify each item:

Functionality

POST /chat answers simple queries in < 15 seconds
POST /chat/async accepts jobs and returns an ID immediately
GET /jobs/{id} returns the status (pending → running → completed/failed)
GET /health verifies: the agent compiled, OpenAI reachable, Redis up, valid API keys
GET /health/live returns 200 whenever the process is alive
GET /health/ready returns 503 if the agent isn't compiled
GET /metrics shows the daily cost, the remaining budget, active requests

Resilience

The rate limit responds with a 429 when the limit is exceeded
The agent returns a partial/fallback response if OpenAI fails
The circuit breaker opens after 3 consecutive failures
The circuit breaker closes after the reset timeout
A 120s timeout prevents infinite requests
Graceful shutdown waits for in-flight requests

Costs

Every response includes cost_usd with an estimate
An alert in the logs when the daily cost goes over 80% of the budget
Rate limiting prevents abuse by individual users

Observability

LangSmith tracing enabled — every request generates a trace
Structured logging with a request_id for correlation
Logs in JSON format for integration with logging systems

Infrastructure

The Docker build succeeds with no errors
docker compose up brings up the agent + Redis
The container starts in < 60 seconds
Docker's health check works (the container is marked as healthy)
.env in .gitignore — no secrets in the repo
.env.example documents every necessary variable

Testing (inherited from v6)

pytest -m unit passes without an API key
pytest -m integration passes with an API key
The golden dataset pass rate is ≥ 75% for the critical tier

The Complete Journey: v1 → v7

Stop for a moment. Look at what you built:

v1 — State Machine (M4)

4 nodes (planning → research → analysis → synthesis) with a StateGraph, conditional routing, and iteration limits. You started with the question: "how do I go from a flat ReAct loop to a controlled flow?" The answer was a graph with nodes, edges, and typed state. The agent knew what to do first and when to stop.

v2 — Planning and Reflection (M5)

Planning with priorities, reflection with 5 criteria, conditional re-planning. The agent stopped following instructions blindly. It evaluated its own answer ("did I cite sources?", "did I cover every aspect?") and decided whether to research again. It started thinking about its own thinking.

v3 — Persistent Memory (M6)

PostgresSaver, long-term memory, conversation trimming, time-travel debugging. The agent stopped being amnesiac. If the research was interrupted at step 7 of 12, it resumed. If you'd told it 5 times you prefer bullet points, it didn't ask again. From a stateless script to a persistent assistant.

v4 — MCP Integration (M7)

3 MCP servers (filesystem, web search, papers DB), dynamic tool discovery. The tools stopped being hardcoded. Want it to save research to disk? Connect an MCP server. Want papers? Connect another one. Without touching agent.py.

v5 — Multi-Agent (M8)

4 specialized agents (Supervisor, Researcher, Analyst, Writer), each with its own StateGraph and context window. A single agent with 6+ tools became 4 agents with 2-3 tools each. Tool selection accuracy: ~85% → ~95%. The specialization improved the quality and the debuggability.

v6 — Testing and Evaluation (M9)

Unit tests, integration tests with a real LLM, trajectory evaluation, a golden dataset of 20+ cases, LangSmith integration, benchmarks. The system worked, but you had no way of knowing when it stopped working. Now every change gets validated against a golden dataset before deploying.

v7 — Production Deployment (M10)

FastAPI, Docker, health checks, rate limiting, cost tracking, error recovery, LangSmith production tracing, structured logging, a production checklist. From python agent.py to a deployed, monitorable, resilient service with controlled costs.

v1  State   ──►  Basic planning, conditional routing
v2  Reason  ──►  Reflection, re-planning, quality gates
v3  Memory  ──►  Persistence, checkpoints, long-term memory
v4  Tools   ──►  MCP, dynamic discovery, decoupling
v5  Scale   ──►  Multi-agent, specialization, coordination
v6  Quality ──►  Testing, evaluation, golden dataset, benchmarks
v7  Prod    ──►  Server, Docker, health, costs, resilience

This works. It's tested. It's deployed. And you built it.

Success Criteria

1. The server answers real queries

POST /chat with a research query returns a coherent report with cost_usd, latency_ms, and metadata about the workers used. Not a 500 Internal Server Error.

2. The health checks detect real problems

If you disconnect Redis (docker compose stop redis), /health returns "unhealthy" with the detail of what failed. It doesn't keep saying "ok" with half the infrastructure down.

3. Rate limiting protects the system

If you send 15 requests in a minute (with a limit of 10), the last 5 get a 429 Too Many Requests with a clear message about when they can retry.

4. Error recovery works without manual intervention

If the LLM provider fails temporarily, the system retries with backoff. If it fails repeatedly, the circuit breaker opens and returns a fallback response. When the provider recovers, the circuit breaker closes automatically.

5. Docker brings up the complete service

docker compose up --build builds the image and brings up the agent + Redis. The container passes the health check and the service is reachable at http://localhost:8000/docs.

6. Every response includes its cost

Not just the answer — the client sees how much their query cost in USD. Total transparency.

Recommended Tests

Test 1: A smoke test

docker compose up -d --build && sleep 30
curl http://localhost:8000/health | python -m json.tool
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What is RAG?", "user_id": "test-user"}'

What it validates: The server starts, the health check passes, a simple query returns an answer with a cost.

Test 2: Rate limiting

Send 15 fast requests with the same user_id. The first 10 return 200, the last 5 return 429 with a message about when to retry.

Test 3: Resilience

Run docker compose stop redis, then curl /health. It must return "unhealthy" with redis: unhealthy. When Redis comes back (docker compose start redis), the health recovers.

Test 4: An async job

POST /chat/async returns a job_id immediately. After polling with GET /jobs/{id}, the status changes from pending to running to completed with the result.

Common Errors

Error 1: The container restarts in a loop

Symptom: docker compose logs agent shows the server starting over and over.

Cause: Docker's health check fails because the agent takes longer than start-period to compile the graphs.

Solution: Raise --start-period in the Dockerfile to 90s or more. Compiling 4 LangGraph graphs with MCP connections takes time.

Error 2: `ainvoke` blocks the event loop

Symptom: One slow request blocks all the others. The server seems frozen.

Cause: You used system.invoke() (synchronous) inside an async def endpoint. That blocks FastAPI's event loop.

Solution: Always use system.ainvoke() inside async endpoints. If the graph doesn't support async, use asyncio.to_thread(system.invoke, ...).

Error 3: The rate limit doesn't work with multiple workers

Symptom: The rate limit resets randomly. A user can send 30 requests/minute when the limit is 10.

Cause: Each uvicorn worker has its own copy of _request_log in memory. With 3 workers, the effective limit is 30.

Solution: Use a single worker (--workers 1) or migrate the rate limiting to Redis (the example is in the Rate Limiting section).

Error 4: Cost estimation returns $0.0000

Symptom: Every response shows cost_usd: 0.0.

Cause: The multi-agent system's messages are in subgraphs, and result["messages"] only contains the Supervisor wrapper's messages, not the Researcher/Analyst/Writer's.

Solution: Count tokens from all the agents' results by accessing agent_results in the state, or use a TokenCounter callback handler connected to the models, or extract token usage from the LangSmith traces.

Error 5: Secrets in the Docker image

Symptom: docker history or docker inspect shows the OPENAI_API_KEY in the image.

Cause: You used COPY .env . in the Dockerfile, or a hardcoded ENV OPENAI_API_KEY=sk-....

Solution: Secrets come in via environment in docker-compose.yml, which reads them from .env on the host. Never copy .env into the container. Add .env to .dockerignore.

Error 6: The circuit breaker never closes

Symptom: After a temporary error, the agent keeps returning fallback responses indefinitely.

Cause: The reset_timeout is too long, or the circuit breaker isn't being checked correctly.

Solution: Verify that is_open() compares time.time() - last_failure > reset_timeout. Adjust reset_timeout to 30-60 seconds for transient errors.

What's Next

You finished this guide. You have a production-ready multi-agent system. But the AI agents field evolves fast. These are the paths you can explore:

Kubernetes deployment: Helm charts, horizontal pod autoscaling, rolling updates with no downtime. The natural step after Docker Compose.
Complete CI/CD: GitHub Actions with unit tests on every push, the critical golden dataset on every PR, nightly benchmarks, automatic deploys if the tests pass.
Pydantic AI: Implement the Research Agent in Pydantic AI and compare: type safety, testing experience, performance. This module's capsule 07 gives you the starting point.
CrewAI / OpenAI Agents SDK: Alternative frameworks for multi-agent. Compare them against your Supervisor + Workers implementation.
Apply it to real problems: Customer support with an internal knowledge base, code review of PRs, data analysis with visualizations. The architecture you built adapts to all of these domains.
The MCP ecosystem: New MCP servers are published regularly. Each one adds capabilities without modifying code. Your agent grows with the ecosystem.

Final Resources

FastAPI — Production Deployment — Gunicorn, Docker, HTTPS, load balancing for FastAPI in production
Docker — Best Practices — Multi-stage builds, layer caching, container security
LangGraph Platform — Managed LangGraph deployment: the alternative to self-hosting
LangSmith — Production Monitoring — Dashboards, alerts, and continuous evaluation in production
Tenacity — Retrying library — Retry decorators with exponential backoff, jitter, and stop conditions
Martin Fowler — Circuit Breaker — The original circuit breaker pattern explained

← Previous7. Pydantic AI: An Honest Comparison