Capstone: Production AI Agent System

The Capstone: ResearchPro Agent System

In this capstone, we build ResearchPro, a production multi-agent research and report system that applies every concept from the course:

flowchart LR
    subgraph API["FastAPI · Auth · Rate Limit · SSE"]
        EP[Endpoint]
    end
    subgraph LG["LangGraph Supervisor · Budget · Checkpoints"]
        SUP[Supervisor]
    end
    subgraph Workers["Parallel Workers via Send"]
        RES["Research Agent\nweb + RAG"]
        ANA["Analysis Agent\ncode execution"]
        WRI["Writer Agent\nstructured output"]
    end
    subgraph Infra["Infrastructure"]
        QD[("Qdrant\nmemory")]
        PG[("Postgres\nstate")]
        RD[("Redis\ncache")]
        LS["LangSmith\ntraces"]
    end
    EP --> SUP
    SUP --> RES & ANA & WRI
    RES & ANA & WRI --> SUP
    LG -.-> Infra
    API -.-> Infra
    style SUP fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style EP fill:#EEF0F7,stroke:#818CF8,color:#0F172A
    style RES fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style ANA fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style WRI fill:#f0fdf9,stroke:#0D9488,color:#0F172A

System State Design

The state schema is the single source of truth for everything that flows between nodes, all parallel workers read from and write to this shared structure.

# src/researchpro/state.py
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
import operator

class ResearchProState(TypedDict):
    # Input
    topic: str
    report_format: str  # "brief" | "detailed" | "executive"
    
    # Agent communication: add_messages reducer appends instead of replacing
    messages: Annotated[list[BaseMessage], add_messages]
    
    # Research phase
    research_queries: list[str]
    # operator.add reducer accumulates findings from all parallel workers into a single list
    research_findings: Annotated[list[dict], operator.add]
    
    # Analysis phase: stores the code-execution agent's inputs and outputs
    data_points: dict
    analysis_code: str | None
    analysis_output: str | None
    
    # Report phase
    report_sections: dict  # section_name → content, built incrementally by the writer node
    final_report: str | None
    
    # Control fields: used by routing functions to decide which node runs next
    current_phase: str  # "research" | "analysis" | "writing" | "complete"
    iteration_count: int
    budget_spent_usd: float  # tracked per-node so the supervisor can halt on cost overrun
    error: str | None

The Full Supervisor Graph

The graph topology below defines the exact execution order: planning, parallel research, aggregation, analysis, writing, and quality review, with conditional edges for revision loops.

# src/researchpro/graph.py
from langgraph.graph import StateGraph, END
from langgraph.types import Send
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def build_researchpro_graph(checkpointer):
    builder = StateGraph(ResearchProState)
    
    # Phase 1: Query Planning: one LLM call that produces the list of research queries
    builder.add_node("plan_research", plan_research_queries)
    
    # Phase 2: Parallel Research (map-reduce): fan_out_research_queries dispatches one worker per query
    builder.add_node("research_worker", run_research_worker)
    builder.add_node("aggregate_research", aggregate_research_findings)  # waits for all workers to finish
    
    # Phase 3: Data Analysis: runs the E2B code-execution agent on the aggregated findings
    builder.add_node("analyze_data", run_data_analysis)
    
    # Phase 4: Report Writing: Pydantic AI structured output produces a validated FullReport object
    builder.add_node("write_report", write_structured_report)
    
    # Phase 5: Quality Check: optional revision loop before returning to the caller
    builder.add_node("quality_check", check_report_quality)
    
    # Routing: fan_out_research_queries returns a list of Send() objects, one per query
    builder.set_entry_point("plan_research")
    builder.add_conditional_edges("plan_research", fan_out_research_queries)
    builder.add_edge("research_worker", "aggregate_research")
    builder.add_edge("aggregate_research", "analyze_data")
    builder.add_edge("analyze_data", "write_report")
    builder.add_conditional_edges("write_report", route_quality_check, {
        "quality_check": "quality_check",
        END: END,  # skip quality check if the report already meets the threshold
    })
    builder.add_conditional_edges("quality_check", route_after_quality, {
        "write_report": "write_report",  # loop back for a revision pass
        END: END,                         # accepted, return to caller
    })
    
    # Postgres checkpointer persists state after every node: runs can be resumed after failures
    return builder.compile(checkpointer=checkpointer)

def fan_out_research_queries(state: ResearchProState) -> list[Send]:
    """Fan out to parallel research workers, one per query."""
    # Send() fans out to one research_worker node per query: they all run in parallel
    return [
        Send("research_worker", {"query": q, "topic": state["topic"]})
        for q in state["research_queries"]
    ]

Key Node Implementations

Each node below is a pure async function that receives state and returns a partial state update, LangGraph merges the returned dict into the shared state before calling the next node.

# src/researchpro/nodes.py

async def plan_research_queries(state: ResearchProState) -> dict:
    """Generate 3-5 targeted research queries for the topic."""
    from pydantic_ai import Agent
    
    class QueryPlan(BaseModel):
        queries: list[str] = Field(min_length=3, max_length=5)  # enforce 3–5 queries via Pydantic validation
        reasoning: str
    
    # result_type=QueryPlan forces Pydantic AI to validate the LLM output before returning it
    planner = Agent("openai:gpt-5.6-sol", result_type=QueryPlan)
    result = await planner.run(
        f"Generate research queries for: {state['topic']}\nFormat: {state['report_format']}"
    )
    
    return {
        "research_queries": result.data.queries,
        "current_phase": "research",
        "messages": [AIMessage(content=f"Planning: {result.data.reasoning}")],
    }

async def run_research_worker(state: dict) -> dict:
    """Individual research worker, searches web + knowledge base."""
    query = state["query"]
    
    # asyncio.gather runs web search and vector retrieval concurrently: halves latency vs sequential
    web_results, kb_results = await asyncio.gather(
        safe_search_and_process(query),   # uses dual LLM pattern from lesson 29 to sanitize content
        vector_store.search(query, top_k=5),
    )
    
    finding = {
        "query": query,
        "web_findings": web_results[:2000],                        # truncate to keep token cost bounded
        "kb_findings": [r["content"] for r in kb_results],
    }
    
    # Return a single-item list: the operator.add reducer in state will append this to research_findings
    return {"research_findings": [finding]}

async def write_structured_report(state: ResearchProState) -> dict:
    """Write the final report using Pydantic AI for structured output."""
    
    class ReportSection(BaseModel):
        title: str
        content: str = Field(min_length=200)            # reject sections that are too thin
        key_points: list[str] = Field(min_length=3, max_length=7)
    
    class FullReport(BaseModel):
        title: str
        executive_summary: str = Field(min_length=100, max_length=500)
        sections: list[ReportSection] = Field(min_length=3)  # at least 3 sections required
        conclusion: str
        data_sources: list[str]
    
    # Pydantic AI validates the model output against FullReport before this await resolves
    writer = Agent("anthropic:claude-sonnet-5", result_type=FullReport)
    
    # Cap at 5 findings × 1000 chars each to stay within a reasonable context budget
    context = "\n\n---\n\n".join([
        f"Research: {f['web_findings'][:1000]}"
        for f in state["research_findings"][:5]
    ])
    
    result = await writer.run(
        f"Write a {state['report_format']} report on '{state['topic']}'\n\nContext:\n{context}"
    )
    
    # Convert the validated Pydantic object to a markdown string for storage and delivery
    report = result.data
    markdown = f"# {report.title}\n\n## Executive Summary\n{report.executive_summary}\n\n"
    for section in report.sections:
        markdown += f"## {section.title}\n{section.content}\n\n"
    markdown += f"## Conclusion\n{report.conclusion}"
    
    return {"final_report": markdown, "current_phase": "complete"}

FastAPI Service with Full Observability

The API endpoint returns immediately with a run_id while the graph executes in the background, callers poll /research/{run_id} for the result rather than holding a long HTTP connection open.

# src/researchpro/api.py
from fastapi import FastAPI, Depends, BackgroundTasks
from langsmith import trace

app = FastAPI(title="ResearchPro API", version="1.0.0")

@app.post("/research/start")
async def start_research(
    request: ResearchRequest,
    background_tasks: BackgroundTasks,
    api_meta: dict = Depends(check_rate_limit),  # rate limiter runs before the handler executes
):
    run_id = str(uuid.uuid4())
    
    async def execute():
        # LangSmith trace context wraps the entire run: every node call appears as a child span
        with trace(
            name="ResearchPro Run",
            project_name="researchpro-production",
            tags=[f"format:{request.format}", f"user:{api_meta['user_id']}"],
            metadata={"topic": request.topic[:100], "user_id": api_meta["user_id"]},
        ):
            checkpointer = await create_postgres_checkpointer()
            graph = await build_researchpro_graph(checkpointer)
            
            config = {
                "configurable": {"thread_id": run_id},  # thread_id links this run to its checkpointed state
                "recursion_limit": 50,                  # LangGraph raises an error if the graph cycles more than 50 times
            }
            
            # All state fields must be explicitly initialized: there are no defaults in TypedDict
            initial_state: ResearchProState = {
                "topic": request.topic,
                "report_format": request.format,
                "messages": [],
                "research_queries": [],
                "research_findings": [],
                "data_points": {},
                "analysis_code": None,
                "analysis_output": None,
                "report_sections": {},
                "final_report": None,
                "current_phase": "planning",
                "iteration_count": 0,
                "budget_spent_usd": 0.0,
                "error": None,
            }
            
            result = await graph.ainvoke(initial_state, config=config)
            
            # Write the completed result to the in-memory task store (replace with Redis in production)
            task_store[run_id] = {
                "status": "completed",
                "report": result.get("final_report"),
                "completed_at": datetime.utcnow().isoformat(),
            }
    
    # Mark the run as started before adding it to the background: callers may poll immediately
    task_store[run_id] = {"status": "running", "created_at": datetime.utcnow().isoformat()}
    background_tasks.add_task(execute)
    
    # Return immediately: the 2–5 minute graph execution happens in the background
    return {"run_id": run_id, "status": "running", "poll_url": f"/research/{run_id}"}

The Production Readiness Checklist

Before going live with any agent system, verify:

Reliability

• Iteration limits (max 20 steps per phase)
• Cost limits ($2 per run maximum)
• Timeout (5 minutes wall clock)
• LLM fallback: primary → secondary → cached response
• All tool failures return structured errors (no uncaught exceptions)

Observability

• Every run creates a LangSmith trace with user ID and run ID
• Cost per run is tracked and logged
• Error rate dashboard is set up
• Latency percentiles (p50, p95, p99) are monitored

Security

• Input classification before routing to agent
• External content sandboxed with dual LLM
• Tool access controlled by user tier
• Output scanned for PII leakage
• API rate limiting per key

Quality

• Golden dataset of ≥50 test cases
• Evaluator suite runs on every deployment
• Regression threshold: quality must not drop >5%

Operations

• Dockerfile with multi-stage build
• Kubernetes deployment with HPA
• CI/CD pipeline with automated tests
• Rollback procedure documented and tested
• On-call runbook for common failure modes

What Comes Next

You’ve now built every component of a production AI agent system:

• Tool calling and schema design
• LangGraph state machines with checkpointing
• Vector memory and agentic RAG
• Multi-agent orchestration and handoffs
• Parallel execution and streaming
• Observability, cost control, and evaluation
• Docker, Kubernetes, and CI/CD
• MCP, code execution, and browser agents
• Security, guardrails, and injection defense

The next step is building for your specific domain. The patterns are general; the implementations are yours to adapt. Ship something. Observe it in production. Iterate on what breaks.

That’s what production AI engineering looks like.

Project: Take a repetitive, multi-step task from your own work. Map it to one of the architectural patterns from Module 1. Build the agent using the stack from this course. Instrument it with LangSmith from day one. Deploy it to Cloud Run or a Kubernetes cluster. Leave it running for a week. Review the traces. What surprised you? What broke? What do you fix first?

The answer to those questions is your next project.