Phil IPS

Posted on Oct 12

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

#langgraph #aiagents #workflow #tutorial

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

Complete guide to building production-ready AI agents using LangGraph — state management, workflows, human-in-the-loop, and multi-agent orchestration.

Introduction

LangGraph extends LangChain with stateful, cyclic graphs for building agents. Unlike linear chains, LangGraph supports:

Cycles & loops for iterative reasoning
State persistence across steps
Human-in-the-loop checkpoints
Multi-agent coordination

# langgraph/basics.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List[str], operator.add]  # Auto-append
    query: str
    answer: str
    iterations: int

def agent_node(state: AgentState) -> AgentState:
    # Process and return updated state
    return {"answer": "processed", "iterations": state["iterations"] + 1}

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.set_entry_point("agent")
graph.add_edge("agent", END)

app = graph.compile()
result = app.invoke({"query": "Hello", "messages": [], "iterations": 0})

Part 1: State Management

TypedDict State Schema

# langgraph/state.py
from typing import TypedDict, Annotated, List, Optional, Dict, Any
import operator
from pydantic import BaseModel, Field

class AgentState(TypedDict):
    # Core fields
    query: str
    messages: Annotated[List[Dict], operator.add]  # Auto-merge lists

    # Retrieval
    retrieved_docs: Annotated[List[Dict], operator.add]
    retrieval_scores: Annotated[List[float], operator.add]

    # Generation
    answer: str
    citations: Annotated[List[str], operator.add]

    # Control
    iterations: int
    max_iterations: int
    should_continue: bool

    # Human-in-the-loop
    human_feedback: Optional[str]
    requires_human: bool

    # Multi-agent
    active_agent: str
    agent_outputs: Annotated[Dict[str, Any], operator.add]

# Pydantic for validation
class ValidatedState(BaseModel):
    query: str = Field(min_length=1)
    messages: List[Dict] = []
    iterations: int = Field(default=0, ge=0, le=10)

    class Config:
        arbitrary_types_allowed = True

State Reducers

# langgraph/reducers.py
from typing import Annotated, List, Dict, Any
import operator
from functools import reduce

# Custom reducers for complex merging
def merge_dicts(existing: Dict, new: Dict) -> Dict:
    """Deep merge dictionaries."""
    result = existing.copy()
    for k, v in new.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = merge_dicts(result[k], v)
        else:
            result[k] = v
    return result

def merge_lists_unique(existing: List, new: List) -> List:
    """Merge lists, deduplicate by ID if dicts."""
    seen = set()
    result = []
    for item in existing + new:
        key = item.get("id") if isinstance(item, dict) else item
        if key not in seen:
            seen.add(key)
            result.append(item)
    return result

# Usage in state
class StateWithReducers(TypedDict):
    config: Annotated[Dict, merge_dicts]
    documents: Annotated[List[Dict], merge_lists_unique]
    metrics: Annotated[Dict[str, float], lambda old, new: {**old, **new}]

Part 2: Core Graph Patterns

Basic Agent Loop

# langgraph/agent_loop.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    query: str
    answer: str
    steps: int

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def reason(state: AgentState) -> AgentState:
    """LLM reasoning step."""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {
        "messages": [AIMessage(content=response.content)],
        "steps": state["steps"] + 1
    }

def should_continue(state: AgentState) -> str:
    """Decide whether to continue or end."""
    if state["steps"] >= 5:
        return "end"
    last_msg = state["messages"][-1]
    if "FINAL ANSWER" in last_msg.content:
        return "end"
    return "continue"

graph = StateGraph(AgentState)
graph.add_node("reason", reason)
graph.set_entry_point("reason")
graph.add_conditional_edges(
    "reason",
    should_continue,
    {"continue": "reason", "end": END}
)

# With checkpointing
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# Run with thread ID for persistence
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(
    {"query": "Research AI agents", "messages": [HumanMessage(content="Research AI agents")], "steps": 0},
    config=config
)

Tool-Using Agent

# langgraph/tool_agent.py
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
import requests

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Implement with SerpAPI, Tavily, etc.
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate mathematical expression."""
    try:
        return str(eval(expression))
    except:
        return "Invalid expression"

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # Implement with weather API
    return f"Weather in {city}: Sunny, 22°C"

tools = [search_web, calculate, get_weather]
tool_node = ToolNode(tools)

llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]

def agent(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_use_tools(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_use_tools, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")

app = graph.compile()

# Run
result = app.invoke({
    "messages": [HumanMessage(content="What's the weather in Tokyo?")]
})

Part 3: RAG with LangGraph

Adaptive RAG Flow

# langgraph/rag_flow.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator

class RAGState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    retrieved_docs: Annotated[List[Dict], operator.add]
    answer: str
    grade: Literal["relevant", "irrelevant", "hallucinated"]
    iterations: int

def retrieve(state: RAGState) -> RAGState:
    """Retrieve documents for query."""
    docs = retriever.invoke(state["query"])
    return {"retrieved_docs": docs}

def grade_documents(state: RAGState) -> RAGState:
    """Grade retrieved document relevance."""
    query = state["query"]
    docs = state["retrieved_docs"]

    grades = []
    for doc in docs:
        score = grader.invoke({"query": query, "document": doc})
        grades.append(score)

    relevant_docs = [d for d, g in zip(docs, grades) if g == "relevant"]
    return {
        "retrieved_docs": relevant_docs,
        "grade": "relevant" if relevant_docs else "irrelevant"
    }

def generate(state: RAGState) -> RAGState:
    """Generate answer from retrieved docs."""
    context = "\n".join([d["content"] for d in state["retrieved_docs"]])
    answer = generator.invoke({"query": state["query"], "context": context})
    return {"answer": answer}

def check_hallucination(state: RAGState) -> RAGState:
    """Verify answer is grounded in context."""
    score = hallucination_checker.invoke({
        "answer": state["answer"],
        "context": state["retrieved_docs"]
    })
    return {"grade": score}

def rewrite_query(state: RAGState) -> RAGState:
    """Rewrite query for better retrieval."""
    new_query = query_rewriter.invoke({"query": state["query"]})
    return {"query": new_query, "iterations": state["iterations"] + 1}

graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("grade_docs", grade_documents)
graph.add_node("generate", generate)
graph.add_node("check_hallucination", check_hallucination)
graph.add_node("rewrite_query", rewrite_query)

graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "grade_docs")

graph.add_conditional_edges(
    "grade_docs",
    lambda s: s["grade"],
    {"relevant": "generate", "irrelevant": "rewrite_query"}
)

graph.add_edge("generate", "check_hallucination")
graph.add_conditional_edges(
    "check_hallucination",
    lambda s: s["grade"] if s["iterations"] < 3 else "max_retries",
    {"relevant": END, "hallucinated": "generate", "max_retries": END}
)

graph.add_edge("rewrite_query", "retrieve")

app = graph.compile()

Part 4: Human-in-the-Loop

Checkpoint & Resume

# langgraph/human_in_loop.py
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Optional
import operator
import uuid

class HILState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    answer: str
    requires_approval: bool
    human_feedback: Optional[str]
    approved: bool

def generate_draft(state: HILState) -> HILState:
    """Generate initial answer."""
    answer = llm.invoke(f"Answer: {state['query']}")
    return {
        "answer": answer.content,
        "requires_approval": True
    }

def human_review(state: HILState) -> HILState:
    """Wait for human input (pauses graph)."""
    # This node pauses execution
    # Human provides feedback via interrupt
    return state

def apply_feedback(state: HILState) -> HILState:
    """Incorporate human feedback."""
    if state["approved"]:
        return {"messages": [AIMessage(content=state["answer"])]}
    else:
        # Regenerate with feedback
        new_answer = llm.invoke(f"Revise: {state['answer']}\nFeedback: {state['human_feedback']}")
        return {"answer": new_answer.content, "requires_approval": True}

graph = StateGraph(HILState)
graph.add_node("generate", generate_draft)
graph.add_node("human_review", human_review)
graph.add_node("apply_feedback", apply_feedback)

graph.set_entry_point("generate")
graph.add_edge("generate", "human_review")
graph.add_edge("human_review", "apply_feedback")
graph.add_conditional_edges(
    "apply_feedback",
    lambda s: "approved" if s["approved"] else "revise",
    {"approved": END, "revise": "human_review"}
)

# Compile with checkpointer
checkpointer = SqliteSaver.from_conn_string("hil.db")
app = graph.compile(checkpointer=checkpointer, interrupt_before=["human_review"])

# Run - will pause at human_review
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = app.invoke({"query": "Write a report", "messages": []}, config=config)

# Check if paused
state = app.get_state(config)
print(f"Next node: {state.next}")  # ('human_review',)

# Resume with human input
app.update_state(config, {"human_feedback": "Make it shorter", "approved": False})
result = app.invoke(None, config=config)  # Continues from interrupt

Streaming Human Feedback

# langgraph/streaming_hil.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class StreamingState(TypedDict):
    query: str
    chunks: Annotated[List[str], operator.add]
    complete: bool

def stream_generate(state: StreamingState, config) -> StreamingState:
    """Stream tokens, allow interruption."""
    full_answer = ""
    for chunk in llm.stream(state["query"]):
        # Check for interrupt signal
        if config.get("interrupt"):
            return {"chunks": [chunk], "complete": False}
        full_answer += chunk
        yield {"chunks": [chunk], "complete": False}

    return {"chunks": [full_answer], "complete": True}

# Client-side interrupt
async def run_with_interrupt(query: str):
    config = {"configurable": {"thread_id": "stream-1"}, "interrupt": False}

    async for chunk in app.astream({"query": query}, config=config):
        print(chunk, end="", flush=True)
        # User can press Ctrl+C to interrupt
        # Set config["interrupt"] = True to pause

    # Resume later with updated config

Part 5: Multi-Agent Systems

Supervisor Pattern

# langgraph/multi_agent.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator
from langchain_openai import ChatOpenAI

class MultiAgentState(TypedDict):
    query: str
    messages: Annotated[List, operator.add]
    active_agent: str
    agent_outputs: Annotated[Dict[str, str], operator.add]
    final_answer: str

llm = ChatOpenAI(model="gpt-4o")

# Specialist agents
def researcher(state: MultiAgentState) -> MultiAgentState:
    result = llm.invoke(f"Research: {state['query']}")
    return {
        "agent_outputs": {"researcher": result.content},
        "active_agent": "analyst"
    }

def analyst(state: MultiAgentState) -> MultiAgentState:
    research = state["agent_outputs"].get("researcher", "")
    result = llm.invoke(f"Analyze: {research}")
    return {
        "agent_outputs": {"analyst": result.content},
        "active_agent": "writer"
    }

def writer(state: MultiAgentState) -> MultiAgentState:
    research = state["agent_outputs"].get("researcher", "")
    analysis = state["agent_outputs"].get("analyst", "")
    result = llm.invoke(f"Write report based on:\nResearch: {research}\nAnalysis: {analysis}")
    return {
        "final_answer": result.content,
        "active_agent": "supervisor"
    }

def supervisor(state: MultiAgentState) -> MultiAgentState:
    """Route to next agent or end."""
    if state["active_agent"] == "supervisor":
        return {"active_agent": "end"}
    return {"active_agent": state["active_agent"]}

graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_node("supervisor", supervisor)

graph.set_entry_point("supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda s: s["active_agent"],
    {"researcher": "researcher", "analyst": "analyst", "writer": "writer", "end": END}
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("analyst", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

Collaborative Agents with Shared State

# langgraph/collaborative.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Dict
import operator

class CollaborativeState(TypedDict):
    task: str
    messages: Annotated[List[Dict], operator.add]
    shared_context: Annotated[Dict, operator.add]  # Deep merge
    agent_notes: Annotated[Dict[str, List[str]], operator.add]
    consensus: bool

def agent_a(state: CollaborativeState) -> CollaborativeState:
    note = f"Agent A: {llm.invoke(f'A perspective on {state[\"task\"]}')}"
    return {
        "agent_notes": {"A": [note]},
        "shared_context": {"perspective_a": note}
    }

def agent_b(state: CollaborativeState) -> CollaborativeState:
    note = f"Agent B: {llm.invoke(f'B perspective on {state[\"task\"]}')}"
    return {
        "agent_notes": {"B": [note]},
        "shared_context": {"perspective_b": note}
    }

def synthesizer(state: CollaborativeState) -> CollaborativeState:
    notes = state["agent_notes"]
    consensus = llm.invoke(f"Synthesize: {notes}")
    return {
        "shared_context": {"consensus": consensus.content},
        "consensus": True
    }

graph = StateGraph(CollaborativeState)
graph.add_node("agent_a", agent_a)
graph.add_node("agent_b", agent_b)
graph.add_node("synthesizer", synthesizer)

graph.set_entry_point("agent_a")
graph.add_edge("agent_a", "agent_b")
graph.add_edge("agent_b", "synthesizer")
graph.add_edge("synthesizer", END)

# Run in parallel using subgraphs
from langgraph.graph import StateGraph

parallel_graph = StateGraph(CollaborativeState)
parallel_graph.add_node("agent_a", agent_a)
parallel_graph.add_node("agent_b", agent_b)
parallel_graph.set_entry_point("agent_a")
parallel_graph.add_edge("agent_a", "agent_b")  # Sequential for demo

# For true parallel, use subgraph with fan-out

Part 6: Advanced Patterns

Reflection & Self-Correction

# langgraph/reflection.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
import operator

class ReflectionState(TypedDict):
    query: str
    draft: str
    critique: str
    revised: str
    iteration: int
    max_iterations: int
    status: Literal["drafting", "critiquing", "revising", "done"]

def generate_draft(state: ReflectionState) -> ReflectionState:
    draft = llm.invoke(f"Write: {state['query']}")
    return {"draft": draft.content, "status": "critiquing", "iteration": 1}

def critique(state: ReflectionState) -> ReflectionState:
    critique = llm.invoke(f"Critique this draft:\n{state['draft']}")
    return {"critique": critique.content, "status": "revising"}

def revise(state: ReflectionState) -> ReflectionState:
    revised = llm.invoke(f"Revise based on critique:\nDraft: {state['draft']}\nCritique: {state['critique']}")
    return {
        "revised": revised.content,
        "draft": revised.content,  # Update draft for next iteration
        "iteration": state["iteration"] + 1,
        "status": "critiquing" if state["iteration"] < state["max_iterations"] else "done"
    }

graph = StateGraph(ReflectionState)
graph.add_node("draft", generate_draft)
graph.add_node("critique", critique)
graph.add_node("revise", revise)

graph.set_entry_point("draft")
graph.add_edge("draft", "critique")
graph.add_edge("critique", "revise")
graph.add_conditional_edges(
    "revise",
    lambda s: s["status"],
    {"critiquing": "critique", "done": END}
)

app = graph.compile()

Planning & Execution

# langgraph/planning.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Dict
import operator
import json

class PlanState(TypedDict):
    goal: str
    plan: List[Dict]
    current_step: int
    results: Annotated[Dict[str, str], operator.add]
    completed: bool

def planner(state: PlanState) -> PlanState:
    """Create execution plan."""
    plan_prompt = f"""Create a step-by-step plan for: {state['goal']}
    Return JSON: [{{"step": 1, "action": "...", "tool": "..."}}]"""
    plan = json.loads(llm.invoke(plan_prompt).content)
    return {"plan": plan, "current_step": 0}

def executor(state: PlanState) -> PlanState:
    """Execute current step."""
    step = state["plan"][state["current_step"]]
    result = execute_step(step)
    return {
        "results": {f"step_{state['current_step']}": result},
        "current_step": state["current_step"] + 1
    }

def should_continue(state: PlanState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "done"
    return "continue"

graph = StateGraph(PlanState)
graph.add_node("plan", planner)
graph.add_node("execute", executor)
graph.set_entry_point("plan")
graph.add_edge("plan", "execute")
graph.add_conditional_edges("execute", should_continue, {"continue": "execute", "done": END})

app = graph.compile()

Part 7: Persistence & Scaling

Postgres Checkpointing

# langgraph/postgres_checkpoint.py
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

# Connection pool
pool = ConnectionPool(
    conninfo="postgresql://user:pass@localhost/langgraph",
    max_size=20,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Creates tables

app = graph.compile(checkpointer=checkpointer)

# List all threads
threads = checkpointer.list(None)  # All threads
for thread in threads:
    print(f"Thread: {thread.config['configurable']['thread_id']}")
    print(f"  Steps: {len(thread.checkpoint['channel_values'].get('messages', []))}")

# Get specific thread state
state = app.get_state({"configurable": {"thread_id": "user-123"}})

Redis for High Throughput

# langgraph/redis_checkpoint.py
from langgraph.checkpoint.redis import RedisSaver
import redis

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
checkpointer = RedisSaver(redis_client)

app = graph.compile(checkpointer=checkpointer)

# TTL for auto-cleanup
checkpointer.ttl = 86400  # 24 hours

Part 8: Testing & Debugging

Unit Testing Graphs

# langgraph/testing.py
import pytest
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class TestState(TypedDict):
    value: int
    history: Annotated[List[int], operator.add]

def increment(state: TestState) -> TestState:
    return {"value": state["value"] + 1}

def test_graph():
    graph = StateGraph(TestState)
    graph.add_node("inc", increment)
    graph.set_entry_point("inc")
    graph.add_edge("inc", END)
    app = graph.compile()

    result = app.invoke({"value": 0, "history": []})
    assert result["value"] == 1
    assert result["history"] == [1]

def test_graph_with_mock(monkeypatch):
    """Mock external dependencies."""
    def mock_llm(*args, **kwargs):
        return "Mocked response"

    monkeypatch.setattr("langgraph.testing.llm", mock_llm)
    # Test with mocked LLM

Visualization & Debugging

# langgraph/visualization.py
from langgraph.graph import StateGraph
import mermaid

def visualize_graph(app):
    """Generate Mermaid diagram of graph."""
    graph = app.get_graph()

    mermaid_code = "graph TD\n"
    for node_id, node in graph.nodes.items():
        mermaid_code += f"  {node_id}[{node_id}]\n"

    for edge in graph.edges:
        mermaid_code += f"  {edge.source} --> {edge.target}\n"

    return mermaid_code

# Print graph structure
print(app.get_graph().draw_ascii())

# State inspection
state = app.get_state(config)
print(f"Current node: {state.next}")
print(f"State values: {state.values}")
print(f"Pending tasks: {state.tasks}")

Part 9: Production Deployment

FastAPI Server

# langgraph/api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langgraph.graph import StateGraph
import uvicorn
import uuid

app = FastAPI(title="LangGraph Agent API")

class QueryRequest(BaseModel):
    query: str
    thread_id: str = None
    config: dict = {}

class QueryResponse(BaseModel):
    thread_id: str
    answer: str
    state: dict

@app.post("/invoke", response_model=QueryResponse)
async def invoke_agent(request: QueryRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}, **request.config}

    try:
        result = await agent_app.ainvoke(
            {"query": request.query, "messages": []},
            config=config
        )
        return QueryResponse(
            thread_id=thread_id,
            answer=result.get("answer", ""),
            state=result
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/stream")
async def stream_agent(request: QueryRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    async def generate():
        async for chunk in agent_app.astream(
            {"query": request.query, "messages": []},
            config=config
        ):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

@app.get("/state/{thread_id}")
async def get_state(thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    state = agent_app.get_state(config)
    return state.values

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Docker Deployment

# langgraph/Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

EXPOSE 8000

CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# langgraph/docker-compose.yml
version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - POSTGRES_URL=postgresql://postgres:pass@db:5432/langgraph
    depends_on:
      - db
    deploy:
      replicas: 3

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=langgraph
      - POSTGRES_PASSWORD=pass
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Part 10: Monitoring & Observability

LangSmith Integration

# langgraph/monitoring.py
from langsmith import Client
from langgraph.graph import StateGraph
from langgraph.callbacks import LangGraphTracer

client = Client()

# Trace runs
tracer = LangGraphTracer(project_name="langgraph-agents")

app = graph.compile()
app = app.with_config({"callbacks": [tracer]})

# Custom metrics
def log_metrics(state, config):
    client.create_run(
        name="agent_step",
        run_type="chain",
        inputs={"state": state},
        outputs={},
        extra={
            "metrics": {
                "tokens": state.get("token_count", 0),
                "latency_ms": state.get("latency_ms", 0),
            }
        }
    )

Resources

LangGraph transforms agents from linear chains into stateful, controllable systems. Master the graph, master the agent.

DEV Community

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

Building AI Agents with LangGraph: State Machines, Workflows & Multi-Agent Systems

Introduction

Part 1: State Management

TypedDict State Schema

State Reducers

Part 2: Core Graph Patterns

Basic Agent Loop

Tool-Using Agent

Part 3: RAG with LangGraph

Adaptive RAG Flow

Part 4: Human-in-the-Loop

Checkpoint & Resume

Streaming Human Feedback

Part 5: Multi-Agent Systems

Supervisor Pattern

Collaborative Agents with Shared State

Part 6: Advanced Patterns

Reflection & Self-Correction

Planning & Execution

Part 7: Persistence & Scaling

Postgres Checkpointing

Redis for High Throughput

Part 8: Testing & Debugging

Unit Testing Graphs

Visualization & Debugging

Part 9: Production Deployment

FastAPI Server

Docker Deployment

Part 10: Monitoring & Observability

LangSmith Integration

Resources

Top comments (0)