Case 5 — Deploy a LangGraph Agent to Production with Observability

# Case 5 — Deploy a LangGraph Agent to Production with Observability > *"You have a working agent in a notebook. Deploy it as a production service: FastAPI endpoint, Postgres checkpointer, LangSmith tracing, structured logs, monitoring dashboards, CI/CD."* --- ## Step 1 — Project structure ``` agent-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI app │ ├── agent.py # LangGraph graph │ ├── tools.py # Tool definitions │ ├── state.py # State schemas │ ├── memory.py # Long-term memory │ └── monitoring.py # Structured logging ├── tests/ │ ├── test_agent.py │ └── eval_dataset.json # 30+ test cases ├── scripts/ │ ├── run_evals.py │ └── check_regressions.py ├── Dockerfile ├── requirements.txt ├── .env.example └── .github/workflows/ ├── ci.yml # tests + evals on PR └── deploy.yml # deploy on merge ``` ## Step 2 — Agent module `app/agent.py`: ```python import os from langgraph.checkpoint.postgres import PostgresSaver from langgraph.graph import StateGraph, END from langchain_anthropic import ChatAnthropic from .state import AgentState from .tools import TOOLS, TOOL_MAP from .monitoring import logger llm = ChatAnthropic(model="claude-sonnet-4-6") llm_with_tools = llm.bind_tools(TOOLS) def agent_node(state): logger.info("agent_decide", thread_id=state.get("thread_id"), iteration=state["iteration"]) response = llm_with_tools.invoke([SystemMessage(SYSTEM_PROMPT)] + state["messages"]) return {"messages": [response], "iteration": state["iteration"] + 1} def tool_node(state): # ... (as previous cases) pass def should_continue(state): # ... (as previous cases) pass def build_graph(): builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tools", tool_node) builder.set_entry_point("agent") builder.add_conditional_edges("agent", should_continue) builder.add_edge("tools", "agent") checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) checkpointer.setup() # creates tables on first run return builder.compile(checkpointer=checkpointer) GRAPH = build_graph() ``` ## Step 3 — FastAPI endpoint `app/main.py`: ```python import os from datetime import datetime from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from langchain_core.messages import HumanMessage from .agent import GRAPH from .monitoring import logger # LangSmith setup os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = os.environ["LANGSMITH_API_KEY"] os.environ["LANGCHAIN_PROJECT"] = os.environ.get("LANGSMITH_PROJECT", "agent-prod") app = FastAPI(title="Support Agent Service") class ChatRequest(BaseModel): user_id: str message: str conversation_id: str # client-managed class ChatResponse(BaseModel): response: str pending_approval: dict | None = None iteration: int @app.get("/health") def health(): return {"status": "ok"} @app.get("/ready") def ready(): # Verify DB connectivity, LangSmith reachable try: # Quick query return {"status": "ready"} except Exception as e: raise HTTPException(503, str(e)) @app.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest, user=Depends(authenticate)): if user.id != req.user_id: raise HTTPException(403, "user_id mismatch") thread_id = f"{req.user_id}:{req.conversation_id}" config = { "configurable": {"thread_id": thread_id, "user_id": req.user_id}, "metadata": { "user_id": req.user_id, "conversation_id": req.conversation_id, "version": os.environ.get("AGENT_VERSION", "dev"), }, } try: result = GRAPH.invoke({ "messages": [HumanMessage(req.message)], "user_id": req.user_id, "iteration": 0, }, config=config) last_msg = result["messages"][-1] return ChatResponse( response=last_msg.content, pending_approval=result.get("pending_approval"), iteration=result.get("iteration", 0), ) except Exception as e: logger.exception("chat_error", thread_id=thread_id, error=str(e)) raise HTTPException(500, "Agent error") @app.post("/approve") def approve(req: ApprovalRequest, user=Depends(authenticate_admin)): # ... HITL approval endpoint (Case 4) pass ``` ## Step 4 — Structured logging `app/monitoring.py`: ```python import structlog import logging structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.processors.add_log_level, structlog.processors.JSONRenderer(), ], wrapper_class=structlog.stdlib.BoundLogger, logger_factory=structlog.stdlib.LoggerFactory(), ) logger = structlog.get_logger() ``` Use throughout: ```python logger.info( "agent_re