Learn how to build a production-grade LangGraph agent with typed state management, conditional routing, retry logic, and human-in-the-loop interrupts. This hands-on lesson walks through every component of a real multi-step research agent — from state schema design to graph assembly.
Imagine you're building an AI-powered data analysis assistant for a team of analysts. The workflow isn't simple: the user asks a question, the agent might need to query a database, evaluate whether the results are sufficient, optionally fetch supplementary data from an API, generate a summary, and then decide whether to ask the user a clarifying question or deliver a final answer. At each step, the agent needs to remember what it's already done, what it found, and what comes next.
A single LLM call can't do that. Even a chain of sequential calls falls apart when the logic branches — when sometimes you need step C after step B, and other times you need to loop back to step A. What you actually need is a graph: a structure where your AI workflow is explicitly modeled as nodes (units of work) connected by edges (transitions), with the ability to route dynamically based on what happened at each step.
That's exactly what LangGraph gives you. Built on top of the LangChain ecosystem but designed specifically for agentic, stateful workflows, LangGraph lets you express complex multi-step AI behavior as a directed graph with persistent state that flows through the entire execution. By the end of this lesson, you'll have built a fully functional multi-step research agent with conditional routing, error recovery, and persistent state that survives across turns.
What you'll learn:
You should be comfortable with:
Install the dependencies before we start:
pip install langgraph langchain-openai langchain-core python-dotenv
Before writing a single line of code, you need to get the mental model right. LangGraph is not a chain. It's not a sequence of steps. It's a state machine expressed as a directed graph, where:
The key insight that separates LangGraph from a simple pipeline: every node sees the same state object, and every node can update any part of it. The graph runtime merges those updates using reducers you define. This means a node at step 7 of your workflow can see what node 1 discovered — without you manually threading that data through function arguments.
Key insight
LangGraph's state isn't passed between nodes like function arguments — it's a shared document that all nodes read from and write updates to. The graph merges those updates using your reducer definitions. This distinction matters enormously when you're debugging unexpected state values.
This is philosophically similar to how agentic loop design works at an architectural level — but LangGraph gives you a concrete, executable implementation of that pattern.
The state schema is the backbone of your LangGraph application. Everything flows through it, so design it carefully. We'll use Python's TypedDict with LangGraph's Annotated reducers.
For our example, we're building a market research agent that:
Here's the state definition:
from typing import TypedDict, Annotated, Sequence, Optional, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class ResearchState(TypedDict):
# The conversation messages — add_messages is a built-in reducer
# that appends new messages rather than replacing the whole list
messages: Annotated[Sequence[BaseMessage], add_messages]
# The company being researched
company_name: str
# Raw data returned from tool calls
financial_data: Optional[dict]
news_data: Optional[list[dict]]
# Evaluation results
data_quality_score: Optional[float]
missing_data_flags: list[str]
# The final output
research_report: Optional[str]
# Routing control
retry_count: int
current_step: str
error_message: Optional[str]
Notice a few things here:
messages uses add_messages as a reducer. When a node returns {"messages": [new_message]}, LangGraph appends that message to the existing list rather than replacing it. Without a reducer, the last write wins — which is correct for most fields like company_name or research_report, but wrong for a conversation history you're building up.
retry_count and current_step are control fields. These aren't data fields — they're fields your conditional edge functions will read to make routing decisions. This is a common and important pattern: embed your routing logic into the state rather than trying to pass it through some side channel.
Most fields are Optional. At the start of execution, financial_data is None. Nodes check for None to understand what's already been done.
Tip
Keep your state schema flat. Deeply nested dicts inside your state become painful to update correctly (you end up replacing the entire nested object). If you need structured data inside a field, use a dataclass or Pydantic model for that field's type, but keep the top-level state as a flat TypedDict.
Each node is a Python function that takes the current state and returns a dict of updates. Let's build them one by one.
The planner's job is to look at the state, understand what needs to happen, and emit an LLM message that kicks off tool use.
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0)
SYSTEM_PROMPT = """You are a market research analyst. Your job is to gather and analyze
financial and news data for companies. When asked to research a company, you should:
1. Use the fetch_financial_data tool to get key metrics
2. Use the fetch_news tool to get recent news
3. Evaluate whether you have enough data to write a comprehensive report
Be systematic and thorough. If data is missing, say so explicitly."""
def planner_node(state: ResearchState) -> dict:
"""Decide what to research and emit the initial tool-calling message."""
company = state["company_name"]
retry_count = state.get("retry_count", 0)
if retry_count == 0:
user_message = HumanMessage(
content=f"Please research {company}. Start by fetching their financial data and recent news."
)
else:
missing = state.get("missing_data_flags", [])
user_message = HumanMessage(
content=f"The initial data for {company} was incomplete. Missing: {missing}. "
f"Please fetch the missing data now."
)
return {
"messages": [SystemMessage(content=SYSTEM_PROMPT), user_message],
"current_step": "planning"
}
This node calls the LLM with tools bound to it and executes whatever tool calls come back. For structured output from LLMs, the patterns from Structured Output: Getting JSON, Tables, and Code from LLMs apply directly here.
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage, AIMessage
import json
import random
# Mock tools — replace with real API calls in production
@tool
def fetch_financial_data(company_name: str) -> dict:
"""Fetch key financial metrics for a company."""
# In production, this would call a financial data API
return {
"ticker": company_name.upper()[:4],
"market_cap": f"${random.randint(1, 500)}B",
"revenue_ttm": f"${random.randint(1, 100)}B",
"pe_ratio": round(random.uniform(10, 50), 2),
"52w_high": round(random.uniform(100, 500), 2),
"52w_low": round(random.uniform(50, 200), 2),
}
@tool
def fetch_news(company_name: str, num_articles: int = 5) -> list[dict]:
"""Fetch recent news articles about a company."""
# In production, this would call a news API
return [
{
"title": f"{company_name} Reports Strong Q3 Earnings",
"date": "2024-11-15",
"sentiment": "positive",
"source": "Reuters"
},
{
"title": f"{company_name} Announces New Product Line",
"date": "2024-11-10",
"sentiment": "positive",
"source": "Bloomberg"
}
]
tools = [fetch_financial_data, fetch_news]
tools_by_name = {t.name: t for t in tools}
llm_with_tools = llm.bind_tools(tools)
def tool_executor_node(state: ResearchState) -> dict:
"""Call the LLM with tools and execute any tool calls that come back."""
messages = list(state["messages"])
# Get the LLM's response (which may include tool calls)
ai_response = llm_with_tools.invoke(messages)
updates = {"messages": [ai_response], "current_step": "tool_execution"}
# If the LLM requested tool calls, execute them
if ai_response.tool_calls:
tool_messages = []
for tool_call in ai_response.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
try:
result = tools_by_name[tool_name].invoke(tool_args)
# Store results in state directly for easy access by other nodes
if tool_name == "fetch_financial_data":
updates["financial_data"] = result
elif tool_name == "fetch_news":
updates["news_data"] = result
tool_messages.append(
ToolMessage(
content=json.dumps(result),
tool_call_id=tool_call["id"]
)
)
except Exception as e:
tool_messages.append(
ToolMessage(
content=f"Error executing {tool_name}: {str(e)}",
tool_call_id=tool_call["id"]
)
)
updates["error_message"] = str(e)
updates["messages"] = [ai_response] + tool_messages
return updates
This is where things get interesting. The evaluator looks at what was retrieved and decides whether it's good enough to write a report. This kind of meta-cognition — the agent evaluating the quality of its own work — is a core pattern in production agentic systems.
def evaluator_node(state: ResearchState) -> dict:
"""Evaluate the quality and completeness of collected data."""
missing_flags = []
score = 1.0
# Check financial data
financial = state.get("financial_data")
if not financial:
missing_flags.append("financial_data")
score -= 0.5
else:
required_fields = ["market_cap", "revenue_ttm", "pe_ratio"]
for field in required_fields:
if field not in financial or financial[field] is None:
missing_flags.append(f"financial.{field}")
score -= 0.1
# Check news data
news = state.get("news_data")
if not news:
missing_flags.append("news_data")
score -= 0.3
elif len(news) < 2:
missing_flags.append("news_data_insufficient")
score -= 0.1
score = max(0.0, score)
return {
"data_quality_score": score,
"missing_data_flags": missing_flags,
"current_step": "evaluation"
}
def report_writer_node(state: ResearchState) -> dict:
"""Generate the final research report from collected data."""
company = state["company_name"]
financial = state.get("financial_data", {})
news = state.get("news_data", [])
prompt = f"""Write a concise market research report for {company}.
Financial Data:
{json.dumps(financial, indent=2)}
Recent News:
{json.dumps(news, indent=2)}
Format the report with sections: Executive Summary, Financial Overview, News Sentiment, and Key Risks."""
response = llm.invoke([HumanMessage(content=prompt)])
return {
"research_report": response.content,
"current_step": "complete"
}
def error_handler_node(state: ResearchState) -> dict:
"""Handle errors gracefully and generate a partial report."""
company = state["company_name"]
error = state.get("error_message", "Unknown error")
partial_report = f"""# Research Report: {company} (Incomplete)
**Note:** This report is incomplete due to data retrieval issues.
**Error encountered:** {error}
**Data collected:**
- Financial data: {"Available" if state.get("financial_data") else "Unavailable"}
- News data: {"Available" if state.get("news_data") else "Unavailable"}
Please retry or consult an analyst for a complete assessment."""
return {
"research_report": partial_report,
"current_step": "error_handled"
}
Now we have nodes. The power comes from how we connect them.
A fixed edge always routes from one node to the next, unconditionally:
graph.add_edge("planner", "tool_executor")
graph.add_edge("tool_executor", "evaluator")
A conditional edge calls a function that inspects the current state and returns the name of the next node to route to. This is where your branching logic lives.
def route_after_evaluation(state: ResearchState) -> Literal["report_writer", "planner", "error_handler"]:
"""
After evaluation, decide what to do next:
- If data quality is high enough, write the report
- If data is missing but we haven't retried too many times, go back to planner
- If we've retried too many times or hit errors, handle gracefully
"""
score = state.get("data_quality_score", 0)
retry_count = state.get("retry_count", 0)
error = state.get("error_message")
# Hard error — go to error handler
if error and retry_count >= 1:
return "error_handler"
# Data is good enough
if score >= 0.7:
return "report_writer"
# Data is insufficient but we haven't retried yet
if retry_count < 2:
return "planner" # Loop back to gather more data
# We've tried twice and still can't get good data
return "error_handler"
def route_after_tools(state: ResearchState) -> Literal["evaluator", "tool_executor"]:
"""
After tool execution, check if the LLM wants to make more tool calls.
This handles the case where the LLM decides it needs to call multiple
rounds of tools before it's satisfied.
"""
messages = state["messages"]
last_message = messages[-1]
# If the last message is an AI message with tool calls pending, loop
# (This handles tool_executor not yet processing all calls)
# Otherwise, proceed to evaluation
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return "tool_executor"
return "evaluator"
Warning
Be careful with routing loops. In the example above, route_after_evaluation can send the graph back to planner, which goes to tool_executor, which goes to evaluator, which might route back to planner again. Always have a clear termination condition — in our case, retry_count >= 2. Without this, you can create an infinite loop that burns through API tokens and never terminates. For robust retry logic patterns, see Agentic Loop Design.
We also need a node that increments the retry counter when we loop back:
def increment_retry_node(state: ResearchState) -> dict:
"""Increment the retry counter before looping back to planner."""
return {
"retry_count": state.get("retry_count", 0) + 1,
"error_message": None # Clear previous error on retry
}
Now we put it all together using StateGraph:
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
def build_research_graph():
# Initialize the graph with our state schema
graph = StateGraph(ResearchState)
# Add all nodes
graph.add_node("planner", planner_node)
graph.add_node("tool_executor", tool_executor_node)
graph.add_node("evaluator", evaluator_node)
graph.add_node("increment_retry", increment_retry_node)
graph.add_node("report_writer", report_writer_node)
graph.add_node("error_handler", error_handler_node)
# Fixed edges: the happy path
graph.add_edge(START, "planner")
graph.add_edge("planner", "tool_executor")
# Conditional edge after tool execution
graph.add_conditional_edges(
"tool_executor",
route_after_tools,
{
"evaluator": "evaluator",
"tool_executor": "tool_executor" # Self-loop for multi-round tool use
}
)
# Conditional edge after evaluation — this is where branching gets interesting
graph.add_conditional_edges(
"evaluator",
route_after_evaluation,
{
"report_writer": "report_writer",
"planner": "increment_retry", # Route through retry counter first
"error_handler": "error_handler"
}
)
# After incrementing retry, go back to planner
graph.add_edge("increment_retry", "planner")
# Terminal nodes go to END
graph.add_edge("report_writer", END)
graph.add_edge("error_handler", END)
# Compile with a memory checkpointer for persistence
checkpointer = MemorySaver()
return graph.compile(checkpointer=checkpointer)
research_agent = build_research_graph()
The add_conditional_edges signature takes three arguments: the source node, the routing function, and a dict mapping the routing function's return values to destination node names. This explicit mapping makes it easy to see all possible routes from any given node — a huge debugging advantage over implicit routing.
Note
The MemorySaver checkpointer stores state in memory — perfect for development and single-process deployments. For production, LangGraph supports checkpointers backed by PostgreSQL and Redis, so state persists across restarts and scales across multiple workers.
The thread_id concept in LangGraph is what enables multi-turn conversations. Each thread is an independent conversation with its own persisted state. This maps naturally to implementing conversational memory — but LangGraph handles the persistence layer for you.
import asyncio
def run_research(company_name: str, thread_id: str = "default") -> str:
"""Run the research agent and return the final report."""
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"company_name": company_name,
"messages": [],
"financial_data": None,
"news_data": None,
"data_quality_score": None,
"missing_data_flags": [],
"research_report": None,
"retry_count": 0,
"current_step": "init",
"error_message": None
}
# invoke() runs the graph to completion
final_state = research_agent.invoke(initial_state, config=config)
return final_state["research_report"]
# Run it
if __name__ == "__main__":
report = run_research("Apple Inc", thread_id="research-apple-001")
print(report)
For a production UI, you typically want to stream updates as each node completes rather than waiting for the full graph to finish. LangGraph supports this natively:
def run_research_streaming(company_name: str, thread_id: str):
"""Stream node completions as the graph executes."""
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"company_name": company_name,
"messages": [],
"financial_data": None,
"news_data": None,
"data_quality_score": None,
"missing_data_flags": [],
"research_report": None,
"retry_count": 0,
"current_step": "init",
"error_message": None
}
for chunk in research_agent.stream(initial_state, config=config, stream_mode="updates"):
# chunk is a dict: {node_name: state_updates}
for node_name, updates in chunk.items():
print(f"\n[{node_name}] completed")
if "current_step" in updates:
print(f" Step: {updates['current_step']}")
if "data_quality_score" in updates and updates["data_quality_score"] is not None:
print(f" Data quality: {updates['data_quality_score']:.2f}")
if "research_report" in updates and updates["research_report"]:
print(f" Report generated ({len(updates['research_report'])} chars)")
This streaming model integrates naturally with building streaming AI interfaces — you'd pipe these node completion events to your frontend via Server-Sent Events.
One of LangGraph's most powerful production features is the ability to pause execution at a specific node and wait for human input before continuing. This is essential for workflows where automated decisions could have real consequences.
Let's add an interrupt before the report is published, allowing a human reviewer to approve it:
from langgraph.checkpoint.memory import MemorySaver
def build_research_graph_with_hitl():
graph = StateGraph(ResearchState)
# Add all nodes (same as before)
graph.add_node("planner", planner_node)
graph.add_node("tool_executor", tool_executor_node)
graph.add_node("evaluator", evaluator_node)
graph.add_node("increment_retry", increment_retry_node)
graph.add_node("report_writer", report_writer_node)
graph.add_node("error_handler", error_handler_node)
# Same edges as before
graph.add_edge(START, "planner")
graph.add_edge("planner", "tool_executor")
graph.add_conditional_edges("tool_executor", route_after_tools,
{"evaluator": "evaluator", "tool_executor": "tool_executor"})
graph.add_conditional_edges("evaluator", route_after_evaluation,
{"report_writer": "report_writer",
"planner": "increment_retry",
"error_handler": "error_handler"})
graph.add_edge("increment_retry", "planner")
graph.add_edge("report_writer", END)
graph.add_edge("error_handler", END)
checkpointer = MemorySaver()
# interrupt_before tells LangGraph to pause BEFORE executing this node
# The graph state is checkpointed, and execution resumes when you call invoke again
return graph.compile(
checkpointer=checkpointer,
interrupt_before=["report_writer"]
)
def run_with_approval(company_name: str, thread_id: str):
agent = build_research_graph_with_hitl()
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"company_name": company_name,
"messages": [],
"financial_data": None,
"news_data": None,
"data_quality_score": None,
"missing_data_flags": [],
"research_report": None,
"retry_count": 0,
"current_step": "init",
"error_message": None
}
# Phase 1: Run until the interrupt
print("Running research pipeline...")
agent.invoke(initial_state, config=config)
# Inspect the state at the interrupt point
snapshot = agent.get_state(config)
print(f"\nData quality score: {snapshot.values['data_quality_score']}")
print(f"Financial data collected: {bool(snapshot.values['financial_data'])}")
print(f"News articles: {len(snapshot.values.get('news_data') or [])}")
print(f"\nNext step would be: {snapshot.next}")
# Human decision point
approval = input("\nApprove report generation? (yes/no): ").strip().lower()
if approval == "yes":
# Resume from the checkpoint — LangGraph knows where it left off
final_state = agent.invoke(None, config=config)
return final_state["research_report"]
else:
print("Report generation cancelled by reviewer.")
return None
The critical thing to understand here: when you call agent.invoke(None, config=config) the second time, None as the input tells LangGraph "resume from the checkpoint, don't start fresh." The graph picks up exactly where it left off, with all state intact.
Key insight
Human-in-the-loop is not bolted on after the fact — it's a first-class feature of LangGraph's execution model. The checkpointer serializes the entire graph state (including execution position) so resumption is deterministic. This is what separates a production agent from a demo.
Now it's your turn to extend what we've built. Here's a realistic challenge:
Build an enhanced version of the research agent that includes a competitor analysis branch.
Specifically:
Add a new tool called fetch_competitor_data(company_name: str) -> list[dict] that returns a list of competitors and their market caps.
Add a new state field competitor_analysis: Optional[list[dict]] to ResearchState.
Add a new node competitor_analyzer_node that calls the LLM to summarize competitive positioning based on the data.
Add a routing decision after evaluation: if the data quality score is above 0.8 and the user asked for competitive analysis (you'll need to add a include_competitors: bool field to state), route to competitor_analyzer_node before report_writer. Otherwise, go straight to report_writer.
Update report_writer_node to include competitor information in the report when it's available.
Test the branching by running the graph twice — once with include_competitors=True and once with include_competitors=False — and verify that the correct path is taken by checking the current_step values in the streaming output.
This exercise will force you to touch every part of the system: state schema, node logic, routing functions, and graph assembly. The competitor analysis branch is a realistic feature that production research tools actually need.
Tip
When you're developing conditional routing, it helps to add a graph.get_graph().print_ascii() call during development to visualize the graph structure. It renders an ASCII art representation of your nodes and edges, which is invaluable for catching missing connections or accidental dead ends.
If your state has a list field (like missing_data_flags) and multiple nodes append to it, you need a reducer — otherwise the last write wins and you lose earlier entries.
Wrong:
class BadState(TypedDict):
errors: list[str] # Last write wins!
Right:
import operator
class GoodState(TypedDict):
errors: Annotated[list[str], operator.add] # Appends correctly
Nodes should return only the fields they're updating, not a copy of the entire state. Returning the full state causes subtle bugs where stale values from the node's snapshot overwrite updates made by other concurrent nodes.
Wrong:
def my_node(state: ResearchState) -> ResearchState:
new_state = dict(state) # Copy entire state
new_state["field"] = "value" # Update one field
return new_state # Return everything — BAD
Right:
def my_node(state: ResearchState) -> dict:
return {"field": "value"} # Return only what changed — GOOD
Your routing function's return type annotation (Literal["node_a", "node_b"]) must exactly match the keys in the dict you pass to add_conditional_edges. A mismatch causes a silent routing failure that's very confusing to debug.
# This will silently fail if routing function returns "report_writer"
# but the dict maps "writer" -> "report_writer"
graph.add_conditional_edges(
"evaluator",
route_after_evaluation,
{
"writer": "report_writer", # KEY MUST MATCH RETURN VALUE
}
)
Early in graph execution, many state fields will be None. Always use state.get("field", default) rather than state["field"] in nodes that might run before those fields are populated.
As mentioned in the routing section, any loop in your graph needs a clear termination condition embedded in state. The pattern we used — retry_count with a max threshold — is the standard approach. For more complex retry patterns including exponential backoff, see Handling LLM Errors, Timeouts, and Rate Limits.
# After invoke() or at an interrupt, inspect full state
snapshot = research_agent.get_state(config)
print("Current values:", snapshot.values)
print("Next node(s):", snapshot.next)
print("State history length:", len(list(research_agent.get_state_history(config))))
The get_state_history method lets you replay exactly what happened at each step — invaluable for debugging unexpected routing decisions. This kind of visibility is part of what good LLM observability looks like in practice.
Warning
In production, the MemorySaver checkpointer stores everything in process memory — which means state is lost on restart and doesn't work across multiple server instances. Before deploying, switch to a persistent checkpointer. LangGraph's official integrations include PostgreSQL and SQLite backends. This is covered in the deploying LLM applications lesson.
Token costs accumulate fast in multi-step graphs. Every node that calls the LLM adds to your bill. A few practices help:
report_writer_node in our example doesn't need to include all the intermediate messages — just the structured data.evaluator_node in our example doesn't actually call an LLM — it uses deterministic logic. When you do need LLM-based evaluation, consider using a cheaper model for that node. The LLM router pattern is designed exactly for this.stream_mode="updates" (node-level updates) is much cheaper than stream_mode="values" (full state at every step), because it doesn't serialize the entire state on every graph tick.You've now built a production-grade LangGraph agent with:
The mental model to carry forward: LangGraph forces you to be explicit about state, transitions, and routing. That explicitness is not bureaucracy — it's what makes complex agentic workflows debuggable, testable, and maintainable. When something goes wrong (and it will), you can inspect the state at every step and reason about exactly why the graph went the direction it did.
Where to go from here:
The pattern you've learned — state → nodes → conditional edges → persistence — scales to workflows far more complex than this example. Multi-agent systems, long-horizon planning, collaborative document editing, automated code review pipelines: they all use the same primitives, just with more nodes and more interesting branching logic.