Beyond Linear Workflows
Real agent workflows branch. A customer support agent might route to different handling logic based on issue type. A research agent might fan out to process multiple documents in parallel. A pipeline agent might have completely different sub-workflows for different task types.
LangGraph handles all of this with conditional edges, Send(), and subgraphs.
Multi-way Conditional Routing
The basic tools_condition is binary (tools vs END). Real production routing is more nuanced:
from langgraph.graph import StateGraph, END
from typing import Literal
class RouterState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# intent is set by classify_intent and read by route_by_intent
intent: str | None
priority: Literal["low", "medium", "high"] | None
requires_human: bool
def classify_intent(state: RouterState) -> dict:
"""LLM classifies the intent of the user's request."""
prompt = f"""
Classify this customer request:
Message: {state['messages'][-1].content}
Return JSON with:
- intent: one of ["billing", "technical", "account", "general"]
- priority: one of ["low", "medium", "high"]
- requires_human: true if the issue is complex or the customer is upset
"""
result = llm.invoke([HumanMessage(content=prompt)])
# json.loads parses the LLM's structured JSON output into a Python dict
classification = json.loads(result.content)
# partial state update: only these three fields are changed
return {
"intent": classification["intent"],
"priority": classification["priority"],
"requires_human": classification["requires_human"],
}
def route_by_intent(state: RouterState) -> str:
"""Route to the appropriate handler based on intent and priority."""
# requires_human overrides everything: escalate before checking intent
if state["requires_human"]:
return "human_handoff"
# high priority issues go to a dedicated escalated path regardless of intent type
if state["priority"] == "high":
return "escalated_handler"
# intent_map translates the LLM's classification into a graph node name
intent_map = {
"billing": "billing_agent",
"technical": "tech_support_agent",
"account": "account_agent",
"general": "general_agent",
}
# dict.get with a default prevents KeyError if the LLM returns an unexpected intent
return intent_map.get(state["intent"], "general_agent")
# Wire it up
builder.add_node("classify", classify_intent)
# The routing dict maps every possible return value of route_by_intent to a node name
# LangGraph validates at compile time that all returned values are in this dict
builder.add_conditional_edges(
"classify",
route_by_intent,
{
"billing_agent": "billing_agent",
"tech_support_agent": "tech_support_agent",
"account_agent": "account_agent",
"general_agent": "general_agent",
"escalated_handler": "escalated_handler",
"human_handoff": "human_handoff",
}
)Dynamic Fan-out with Send()
Send() enables true parallelism over dynamic lists, the core of the map-reduce pattern:
from langgraph.types import Send
from typing import TypedDict, Annotated
import operator
class MapReduceState(TypedDict):
documents: list[str] # input documents
# operator.add as reducer means each worker's list appends to the shared list
# (map-reduce aggregation: all parallel summaries accumulate here)
summaries: Annotated[list[str], operator.add]
final_report: str
class DocumentState(TypedDict):
# each parallel branch gets its own isolated DocumentState
doc: str
summary: str | None
# Map: fan out one branch per document
def fan_out_documents(state: MapReduceState) -> list[Send]:
"""Return one Send per document to process them in parallel."""
# Send(node_name, state_for_that_branch): LangGraph spawns one execution per Send
return [
Send("summarize_document", {"doc": doc, "summary": None})
for doc in state["documents"]
]
# Worker: each branch processes one document independently
def summarize_document(state: DocumentState) -> dict:
"""Summarize a single document."""
# truncate to 3000 chars to avoid token overflow in long documents
prompt = f"Summarize this in 2-3 sentences:\n\n{state['doc'][:3000]}"
result = llm.invoke([HumanMessage(content=prompt)])
# returning {"summary": ...}: the operator.add reducer in MapReduceState
# will append this to the parent's summaries list when the branch completes
return {"summary": result.content} # goes back to parent via operator.add reducer
# Reduce: aggregate all summaries after all parallel branches complete
def create_final_report(state: MapReduceState) -> dict:
"""Combine all summaries into a final report."""
# state["summaries"] contains one entry per document, in completion order
combined = "\n\n".join([f"Document {i+1}: {s}" for i, s in enumerate(state["summaries"])])
prompt = f"Create a comprehensive report from these document summaries:\n\n{combined}"
result = llm.invoke([HumanMessage(content=prompt)])
return {"final_report": result.content}
# Build the map-reduce graph
builder = StateGraph(MapReduceState)
builder.add_node("summarize_document", summarize_document)
builder.add_node("create_report", create_final_report)
builder.set_entry_point("fan_out")
# passthrough node: exists only to trigger the conditional edge for fan-out
builder.add_node("fan_out", lambda s: s)
# fan_out_documents returns a list of Send objects: LangGraph runs them in parallel
builder.add_conditional_edges("fan_out", fan_out_documents)
# LangGraph waits for ALL parallel summarize_document branches before advancing to create_report
builder.add_edge("summarize_document", "create_report")
builder.add_edge("create_report", END)Subgraphs: Composable Agent Workflows
Subgraphs let you encapsulate a complete agent workflow and use it as a single node in a larger graph:
# Define a reusable research subgraph: it has its own isolated state schema
class ResearchState(TypedDict):
query: str
search_results: list[dict]
summary: str
def search_phase(state: ResearchState) -> dict:
# search_tool.invoke() runs the tool synchronously with the given query
results = search_tool.invoke(state["query"])
return {"search_results": results}
def summarize_phase(state: ResearchState) -> dict:
# join all result content fields into a single string for the LLM to summarize
docs = "\n\n".join([r["content"] for r in state["search_results"]])
result = llm.invoke([HumanMessage(content=f"Summarize:\n{docs}")])
return {"summary": result.content}
# Build and compile the subgraph: compile() makes it runnable as a standalone graph
research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_phase)
research_builder.add_node("summarize", summarize_phase)
research_builder.set_entry_point("search")
research_builder.add_edge("search", "summarize")
research_builder.add_edge("summarize", END)
# research_subgraph is now a compiled graph: it can be invoked independently or used as a node
research_subgraph = research_builder.compile()
# Use the subgraph as a node in a parent graph
class MainState(TypedDict):
task: str
research_query: str
research_summary: str # populated by subgraph
final_report: str
def prepare_research_query(state: MainState) -> dict:
"""Extract a search query from the task."""
result = llm.invoke([HumanMessage(content=f"Turn this task into a search query: {state['task']}")])
return {"research_query": result.content.strip()}
def invoke_research_subgraph(state: MainState) -> dict:
"""Run the research subgraph and extract the summary."""
# invoke the subgraph with its own state schema: it runs its full internal graph
result = research_subgraph.invoke({"query": state["research_query"]})
# extract only the field the parent graph needs from the subgraph's output
return {"research_summary": result["summary"]}
def generate_report(state: MainState) -> dict:
result = llm.invoke([HumanMessage(
content=f"Write a report for '{state['task']}' based on: {state['research_summary']}"
)])
return {"final_report": result.content}
# Parent graph: invoke_research_subgraph wraps the entire subgraph as a single node
main_builder = StateGraph(MainState)
main_builder.add_node("prepare_query", prepare_research_query)
main_builder.add_node("research", invoke_research_subgraph)
main_builder.add_node("report", generate_report)
main_builder.set_entry_point("prepare_query")
main_builder.add_edge("prepare_query", "research")
main_builder.add_edge("research", "report")
main_builder.add_edge("report", END)
main_agent = main_builder.compile()The Recursive Retry Pattern
A common pattern: retry a node with different parameters if it fails validation:
class ValidationState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
draft: str | None
validation_errors: list[str] # errors from the previous attempt
retry_count: int
def generate_draft(state: ValidationState) -> dict:
# inject previous validation errors into the prompt so the model can self-correct
context = ""
if state["validation_errors"]:
context = f"\n\nPrevious attempt had errors:\n" + "\n".join(state["validation_errors"])
result = llm.invoke(state["messages"] + [HumanMessage(content=context)])
# increment retry_count so the routing function knows how many attempts have been made
return {"draft": result.content, "retry_count": state["retry_count"] + 1}
def validate_draft(state: ValidationState) -> dict:
# run_validation returns a list of error strings, or empty list on success
errors = run_validation(state["draft"])
return {"validation_errors": errors}
def should_retry_or_finish(state: ValidationState) -> str:
# empty validation_errors means the draft passed all checks
if not state["validation_errors"]:
return "approved"
# cap retries to avoid an infinite loop: force END even with errors
if state["retry_count"] >= 3:
return "failed"
# errors remain and retry budget allows: loop back to regenerate
return "retry"
builder.add_conditional_edges(
"validate",
should_retry_or_finish,
# "retry" routes back to "generate": creating a controlled retry loop
{"approved": END, "failed": END, "retry": "generate"}
)These patterns: multi-way routing, fan-out with Send(), subgraph composition, and retry loops, are the building blocks of every complex production agent workflow you’ll encounter.