Orchestrating Agentic Workflows with LangGraph

Most AI workflows today are built as linear chains: embed a query, retrieve documents, generate a response. This works - until it does not. Real-world problems require loops, branching, error recovery, and human judgement. When your agent needs to search, evaluate, refine, and search again, a simple chain breaks.

LangGraph solves this by modelling workflows as stateful graphs instead of linear pipelines. This post walks through the core concepts, a complete multi-agent example, and production patterns you can use today.

What you will learn: Why graphs beat chains for agentic workflows · The three core LangGraph concepts (nodes, edges, state schema) · A complete multi-agent researcher with self-correction · How checkpointing enables time travel and human-in-the-loop · Production patterns for error handling and state management

1. Why Graphs Beat Chains

In a traditional LangChain-style pipeline, every step feeds into the next in a fixed sequence. A RAG query flows: embed → search → retrieve → generate. If the search returns low-quality results, there is no way to loop back and try again with a better query.

A graph, by contrast, allows cycles. A reviewer node can reject a researcher's output and send it back for another attempt. A planner node can split a goal into parallel branches. The orchestrator can pause, wait for human input, and resume from the same state.

LangGraph makes cyclic workflows a first-class citizen. Each node is a function. Each edge is a transition. The state persists across every step. This small shift unlocks capabilities that are impractical in linear chains: self-correcting agents, parallel sub-tasks, human approval gates, and workflows that can run for hours across multiple API calls.

2. Core Concepts: Nodes, Edges, State

LangGraph has three building blocks. Understand these, and you understand the entire framework.

State Schema

Every graph has a typed state - the memory that flows between nodes. You define it as a Python TypedDict or Pydantic model:

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list, add]  # append-only
    next_agent: str                  # routing key
    retries: int                     # attempt counter
    final_report: str                # output

The Annotated reducer tells LangGraph how to merge updates from multiple nodes. The add reducer appends to a list - perfect for accumulating messages across a conversation. Without a reducer, the last write wins.

Nodes

A node is a function that receives the state and returns an update. That is it. No framework magic, no base class:

def research_node(state: AgentState) -> dict:
    """Run a search and return results."""
    results = search_web(state["messages"][-1])
    return {"messages": [f"Research result: {results}"]}

Edges

Edges define the flow. A normal edge goes from one node to the next. Aconditional edge inspects the state and decides where to go:

def should_continue(state: AgentState) -> str:
    """Route to reviewer if research is done, otherwise back to research."""
    if state["retries"] >= 3:
        return "summarizer"
    if quality_check(state["messages"][-1]):
        return "reviewer"
    return "researcher"  # loop back

3. Building a Multi-Agent Researcher

Let us build a complete, working example. This system has four agents (nodes) connected in a graph with a self-correction loop:

• Planner - breaks a question into sub-queries
• Researcher - executes searches for each sub-query
• Reviewer - scores result quality and requests refinement
• Summarizer - writes the final report

Define the State

from typing import TypedDict, Annotated
from operator import add
from dataclasses import dataclass

@dataclass
class SearchResult:
    query: str
    content: str
    quality_score: float = 0.0

class ResearchState(TypedDict):
    question: str
    sub_queries: Annotated[list[str], add]
    results: Annotated[list[SearchResult], add]
    current_query: str
    retries: int
    report: str

Define the Nodes

import random  # simulate LLM calls

def planner_node(state: ResearchState) -> dict:
    """Decompose question into sub-queries."""
    question = state["question"]
    sub_queries = [
        f"{question} key statistics",
        f"{question} recent developments",
        f"{question} expert opinions",
    ]
    return {"sub_queries": sub_queries, "current_query": sub_queries[0]}

def researcher_node(state: ResearchState) -> dict:
    """Execute search for the current sub-query."""
    query = state["current_query"]
    # In production: call search API
    content = f"Simulated result for: {query}"
    quality = random.uniform(0.3, 0.95)
    return {
        "results": [SearchResult(query, content, quality)],
        "retries": state.get("retries", 0) + 1,
    }

def reviewer_node(state: ResearchState) -> dict:
    """Score results and decide next action."""
    last = state["results"][-1]
    if last.quality_score >= 0.7:
        return {"retries": 0}  # good enough
    return {"current_query": f"REFINED: {last.query}"}

def summarizer_node(state: ResearchState) -> dict:
    """Synthesize all results into a final report."""
    lines = []
    for r in state["results"]:
        lines.append(f"- {r.query}: {r.content}")
    report = "## Research Report\n\n" + "\n".join(lines)
    return {"report": report}

Wire the Graph

from langgraph.graph import StateGraph, END

graph = StateGraph(ResearchState)

# Add nodes
graph.add_node("planner", planner_node)
graph.add_node("researcher", researcher_node)
graph.add_node("reviewer", reviewer_node)
graph.add_node("summarizer", summarizer_node)

# Set entry point
graph.set_entry_point("planner")

# Define edges
graph.add_edge("planner", "researcher")

# Conditional edge: reviewer routes to researcher or summarizer
def route_from_reviewer(state: ResearchState) -> str:
    if state["retries"] >= 3 or state["results"][-1].quality_score >= 0.7:
        if all(q in state.get("sub_queries", [])
               for q in state.get("sub_queries", [])
               if any(r.query.startswith(q) for r in state["results"])):
            return "summarizer"
        # More sub-queries to process
        return "researcher"
    # Refine and retry
    return "researcher"

graph.add_conditional_edges("reviewer", route_from_reviewer, {
    "researcher": "researcher",
    "summarizer": "summarizer",
})

graph.add_edge("reviewer", "researcher")
graph.add_edge("summarizer", END)

# Compile
app = graph.compile()

Run It

result = app.invoke({
    "question": "What is the impact of AI on healthcare?",
})

print(result["report"])
# Research Report
# - What is the impact of AI on healthcare? key statistics: Simulated result for: ...
# - What is the impact of AI on healthcare? recent developments: ...

This is a minimal but complete multi-agent system. The planner decomposes the question. The researcher executes each sub-query. The reviewer scores the results and loops back if quality is below threshold. The summarizer produces the final output.

4. Checkpointing: Time Travel and Persistence

The most powerful feature of LangGraph is built-in checkpointing. Every time a node finishes, LangGraph can persist the entire state to a store. This unlocks capabilities that are difficult to build yourself:

Crash Recovery

If the server restarts mid-workflow, the graph picks up exactly where it left off - no lost state, no reprocessing:

from langgraph.checkpoint.sqlite import SqliteSaver

# Enable checkpointing with SQLite
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

# Run with a thread ID so state is persisted
config = {"configurable": {"thread_id": "research-001"}}
result = app.invoke({"question": "Impact of AI on healthcare?"}, config)

# If the server crashes and restarts:
result = app.invoke(None, config)  # resumes from last checkpoint

Time Travel

Because every state is saved, you can rewind to any previous checkpoint and branch off from there:

# List all checkpoints for a thread
checkpoints = list(app.checkpointer.list(config))
print(checkpoints)

# Replay from checkpoint 2
config["configurable"]["checkpoint_id"] = checkpoints[2].checkpoint_id
app.invoke(None, config)  # replays from that point

This is invaluable for debugging. Instead of adding print statements and re-running, you replay a failed workflow step by step, inspect every state transition, and fix the bug at the exact point it occurred.

5. Human-in-the-Loop

Some actions should never be taken autonomously. Sending an email, approving a financial trade, or updating a database requires human judgement. LangGraph supports this viainterrupts:

from langgraph.graph import interrupt

def approval_node(state: ResearchState) -> dict:
    """Pause and wait for human approval."""
    # This raises a Breakpoint - LangGraph saves state and returns
    approval = interrupt({
        "question": "Approve this research report?",
        "report": state.get("report", "No report yet"),
    })

    if approval.get("approved"):
        return {"report": state["report"]}

    # Human rejected - refine the question and restart
    return {"question": approval.get("refined_question", state["question"])}

When the graph hits interrupt, it saves the checkpoint and returns control to the caller. The caller (your backend, a UI, or an API endpoint) shows the user what the graph wants to do and collects their decision:

from langgraph.types import Command

# Resume the graph with human input
app.invoke(
    Command(resume={"approved": True}),
    {"configurable": {"thread_id": "research-001"}}
)

# Or reject with refinements
app.invoke(
    Command(resume={
        "approved": False,
        "refined_question": "Focus on European healthcare systems only"
    }),
    {"configurable": {"thread_id": "research-001"}}
)

This pattern is critical for production systems. It enables progressive autonomy: start with every action requiring approval, then relax as you gain confidence. A common approach is to require human approval only for actions above a risk threshold - financial transactions over a certain amount, external communications, or destructive database operations.

6. Production Patterns

Based on deploying LangGraph systems in enterprise environments, here are patterns that separate a prototype from a production service.

Error Handling at the Node Level

Every external call can fail. Wrap each node with retry logic that updates the state on failure:

def safe_researcher(state: ResearchState) -> dict:
    """Research node with exponential backoff."""
    query = state["current_query"]
    for attempt in range(3):
        try:
            content = search_api(query)
            return {"results": [SearchResult(query, content, 0.9)]}
        except TimeoutError:
            wait = 2 ** attempt
            time.sleep(wait)
        except RateLimitError:
            time.sleep(60)
    # All retries exhausted - update state with the failure
    return {
        "results": [SearchResult(query, "[SEARCH FAILED]", 0.0)],
        "retries": state["retries"],
    }

Minimal State Schema

Only store what you need. A bloated state schema makes debugging harder and increases checkpoint storage costs. Split large workflows into sub-graphs, each with its own focused state:

# Good: focused state per sub-graph
class SearchState(TypedDict):
    query: str
    results: list[str]

class ReportState(TypedDict):
    sections: list[str]
    format: str

# Compose sub-graphs into a parent graph
parent = StateGraph(ParentState)
parent.add_node("search", search_subgraph)
parent.add_node("report", report_subgraph)

Structured Logging for Observability

You cannot debug a graph without visibility into each transition. Use structured logging at every node:

import structlog

logger = structlog.get_logger()

def logged_researcher(state: ResearchState) -> dict:
    logger.info("research.started", query=state["current_query"])
    try:
        results = search_api(state["current_query"])
        logger.info("research.completed", result_count=len(results))
        return {"results": results}
    except Exception as e:
        logger.error("research.failed", error=str(e))
        raise

7. Summary: What to Remember

If you are building agentic systems today, LangGraph is the right foundation. Start with the multi-agent researcher from section 3. Add checkpointing. Add a human approval gate. Then run it against a real API and watch how the self-correction loop handles failures you did not anticipate. That is where the real learning happens.

# Your challenge: extend the multi-agent researcher
# 1. Replace the simulated search with a real API call
# 2. Add a human-in-the-loop gate before the summarizer
# 3. Add structured logging with request IDs
# 4. Wire it up to a FastAPI endpoint that returns thread_id

# Starting point:
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ResearchRequest(BaseModel):
    question: str

@app.post("/research")
async def start_research(req: ResearchRequest):
    config = {"configurable": {"thread_id": f"research-{uuid4()}"}}
    result = app.invoke({"question": req.question}, config)
    return {"thread_id": config["configurable"]["thread_id"]}

# Then poll /research/{thread_id}/status to check if
# human approval is needed, or if the report is ready.