Python LangGraph Multi-Agent Collaboration: 5 Practical Patterns from State Machines to Workflow Orchestration
AI与大数据
Your AI Agent Can Only Do Q&A, Complex Tasks Require Manual Decomposition
Users say "analyze competitors and generate a report," your Agent can only ask questions step by step; you need 3 Agents to collaborate on a "research → analysis → writing" pipeline, but there's no ready-made orchestration framework; the Agent needs human confirmation mid-execution, but you don't know how to pause and resume. The single-Agent era is over — in 2026, LangGraph makes multi-Agent collaboration go from manual stitching to declarative orchestration.
This article starts from LangGraph state graph basics and guides you through state machine → multi-agent orchestration → conditional routing → human-in-the-loop → persistent state with 5 practical patterns, from development to production.
LangGraph Core Concepts
| Concept | Description |
|---|---|
| StateGraph | State graph, defining workflow nodes and edges |
| State | State, shared data structure passed between workflow nodes |
| Node | Node, function executing specific logic, receives State and returns updates |
| Edge | Edge, defines transition relationships between nodes |
| Conditional Edge | Conditional edge, dynamically determines next node based on State |
| Checkpoint | Checkpoint, persists State, supports pause/resume |
| Interrupt | Interrupt, pauses workflow waiting for external input (human-in-the-loop) |
| Tool Node | Tool node, encapsulates external tool calls |
| Subgraph | Subgraph, encapsulates complex workflows as reusable modules |
| Command | Command object, supports state updates and routing control between nodes |
Workflow Execution Flow
1. Define State (TypedDict or Pydantic Model)
2. Create StateGraph(State)
3. Add nodes: graph.add_node("name", function)
4. Add edges: graph.add_edge("node_a", "node_b")
5. Add conditional edges: graph.add_conditional_edges("node_a", router)
6. Set entry point: graph.set_entry_point("start")
7. Compile graph: app = graph.compile(checkpointer=...)
8. Execute: app.invoke({"input": ...}, config={"configurable": {"thread_id": "..."}})
Problem Analysis: 5 Major Multi-Agent Collaboration Challenges
- Chaotic state management: Shared state across multiple Agents, manual passing easily leads to omissions and conflicts
- Complex workflow orchestration: Conditional branching, loops, parallel execution, hardcoded if-else is hard to maintain
- Difficult human-in-the-loop: When Agents need human confirmation, no elegant way to pause and resume
- Fragile error recovery: Long workflows failing midway, retrying from the start is costly
- Missing observability: Multi-Agent collaboration execution is a black box, debugging is difficult
Step-by-Step: 5 Practical Patterns
Pattern 1: Basic State Machine — Single Agent Workflow
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class ResearchState(TypedDict):
messages: Annotated[list, add_messages]
topic: str
research_notes: str
summary: str
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def research_node(state: ResearchState) -> dict:
messages = [
SystemMessage(content="You are a professional researcher. Conduct in-depth research on the given topic and output detailed research notes."),
HumanMessage(content=f"Please research the following topic in depth: {state['topic']}"),
]
response = llm.invoke(messages)
return {"research_notes": response.content}
def summarize_node(state: ResearchState) -> dict:
messages = [
SystemMessage(content="You are a professional editor. Summarize research notes into a concise summary."),
HumanMessage(content=f"Please summarize the following research notes:\n\n{state['research_notes']}"),
]
response = llm.invoke(messages)
return {"summary": response.content}
def format_node(state: ResearchState) -> dict:
formatted = f"""# Research Report: {state['topic']}
## Research Notes
{state['research_notes']}
## Summary
{state['summary']}
"""
return {"messages": [HumanMessage(content=formatted)]}
graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("summarize", summarize_node)
graph.add_node("format", format_node)
graph.add_edge(START, "research")
graph.add_edge("research", "summarize")
graph.add_edge("summarize", "format")
graph.add_edge("format", END)
app = graph.compile()
result = app.invoke({
"messages": [],
"topic": "2026 Large Language Model Technology Trends",
"research_notes": "",
"summary": "",
})
print(result["messages"][-1].content)
Pattern 2: Multi-Agent Collaboration — Supervisor Pattern
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class CollaborationState(TypedDict):
messages: Annotated[list, add_messages]
task: str
next_agent: str
research_result: str
analysis_result: str
writing_result: str
review_feedback: str
iteration_count: int
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def supervisor_node(state: CollaborationState) -> dict:
if state["iteration_count"] >= 3:
return {"next_agent": "end"}
if not state["research_result"]:
return {"next_agent": "researcher"}
if not state["analysis_result"]:
return {"next_agent": "analyst"}
if not state["writing_result"]:
return {"next_agent": "writer"}
if not state["review_feedback"]:
return {"next_agent": "reviewer"}
if "needs revision" in state["review_feedback"].lower():
return {
"next_agent": "writer",
"writing_result": "",
"review_feedback": "",
"iteration_count": state["iteration_count"] + 1,
}
return {"next_agent": "end"}
def researcher_node(state: CollaborationState) -> dict:
messages = [
SystemMessage(content="You are a Researcher Agent. Collect and organize information related to the task."),
HumanMessage(content=f"Research task: {state['task']}"),
]
response = llm.invoke(messages)
return {"research_result": response.content}
def analyst_node(state: CollaborationState) -> dict:
messages = [
SystemMessage(content="You are an Analyst Agent. Conduct in-depth analysis based on research results."),
HumanMessage(content=f"Analyze based on the following research results:\n\n{state['research_result']}"),
]
response = llm.invoke(messages)
return {"analysis_result": response.content}
def writer_node(state: CollaborationState) -> dict:
context = f"Research results: {state['research_result']}\n\nAnalysis results: {state['analysis_result']}"
if state["review_feedback"]:
context += f"\n\nRevision feedback: {state['review_feedback']}"
messages = [
SystemMessage(content="You are a Writer Agent. Write high-quality articles based on research and analysis results."),
HumanMessage(content=f"Write an article:\n\n{context}"),
]
response = llm.invoke(messages)
return {"writing_result": response.content}
def reviewer_node(state: CollaborationState) -> dict:
messages = [
SystemMessage(content="You are a Reviewer Agent. Review article quality. If revisions are needed, state specific feedback. If satisfactory, say 'approved'."),
HumanMessage(content=f"Review the following article:\n\n{state['writing_result']}"),
]
response = llm.invoke(messages)
return {"review_feedback": response.content}
def route_after_supervisor(state: CollaborationState) -> str:
next_agent = state["next_agent"]
if next_agent == "end":
return END
return next_agent
graph = StateGraph(CollaborationState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)
graph.add_node("reviewer", reviewer_node)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_after_supervisor, {
"researcher": "researcher",
"analyst": "analyst",
"writer": "writer",
"reviewer": "reviewer",
END: END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("analyst", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("reviewer", "supervisor")
app = graph.compile()
result = app.invoke({
"messages": [],
"task": "Analyze 2026 AI Agent technology trends and write a report",
"next_agent": "",
"research_result": "",
"analysis_result": "",
"writing_result": "",
"review_feedback": "",
"iteration_count": 0,
})
print(result["writing_result"])
Pattern 3: Conditional Routing — Dynamic Workflows
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import json
class SupportState(TypedDict):
messages: Annotated[list, add_messages]
user_input: str
intent: str
category: str
response: str
escalated: bool
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def classify_intent_node(state: SupportState) -> dict:
messages = [
SystemMessage(content="""You are a customer service intent classifier. Analyze user input and return JSON format:
{"intent": "technical|billing|general|complaint", "category": "specific category", "escalated": false}
If the user is emotional or the issue is serious, set escalated to true."""),
HumanMessage(content=state["user_input"]),
]
response = llm.invoke(messages)
try:
result = json.loads(response.content)
return {
"intent": result.get("intent", "general"),
"category": result.get("category", ""),
"escalated": result.get("escalated", False),
}
except json.JSONDecodeError:
return {"intent": "general", "category": "Uncategorized", "escalated": False}
def technical_support_node(state: SupportState) -> dict:
messages = [
SystemMessage(content="You are a Technical Support Agent. Provide professional technical problem solutions."),
HumanMessage(content=f"User issue: {state['user_input']}\nCategory: {state['category']}"),
]
response = llm.invoke(messages)
return {"response": response.content}
def billing_support_node(state: SupportState) -> dict:
messages = [
SystemMessage(content="You are a Billing Support Agent. Handle billing-related issues including refunds and fee inquiries."),
HumanMessage(content=f"User issue: {state['user_input']}\nCategory: {state['category']}"),
]
response = llm.invoke(messages)
return {"response": response.content}
def general_support_node(state: SupportState) -> dict:
messages = [
SystemMessage(content="You are a General Support Agent. Handle general inquiry questions."),
HumanMessage(content=f"User issue: {state['user_input']}"),
]
response = llm.invoke(messages)
return {"response": response.content}
def escalation_node(state: SupportState) -> dict:
messages = [
SystemMessage(content="You are a Senior Support Agent. Handle complex or urgent issues requiring escalation."),
HumanMessage(content=f"Urgent issue: {state['user_input']}\nCategory: {state['category']}"),
]
response = llm.invoke(messages)
return {"response": response.content}
def route_by_intent(state: SupportState) -> str:
if state["escalated"]:
return "escalation"
intent_map = {
"technical": "technical",
"billing": "billing",
"general": "general",
"complaint": "escalation",
}
return intent_map.get(state["intent"], "general")
graph = StateGraph(SupportState)
graph.add_node("classify", classify_intent_node)
graph.add_node("technical", technical_support_node)
graph.add_node("billing", billing_support_node)
graph.add_node("general", general_support_node)
graph.add_node("escalation", escalation_node)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_intent, {
"technical": "technical",
"billing": "billing",
"general": "general",
"escalation": "escalation",
})
graph.add_edge("technical", END)
graph.add_edge("billing", END)
graph.add_edge("general", END)
graph.add_edge("escalation", END)
app = graph.compile()
result = app.invoke({
"messages": [],
"user_input": "My server suddenly became inaccessible, database connection timeout, very urgent!",
"intent": "",
"category": "",
"response": "",
"escalated": False,
})
print(result["response"])
Pattern 4: Human-in-the-Loop
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class ApprovalState(TypedDict):
messages: Annotated[list, add_messages]
task: str
draft: str
human_feedback: str
final_result: str
approved: bool
llm = ChatOpenAI(model="gpt-4o", temperature=0)
checkpointer = MemorySaver()
def generate_draft_node(state: ApprovalState) -> dict:
messages = [
SystemMessage(content="You are a Content Creation Agent. Generate a draft based on the task requirements."),
HumanMessage(content=f"Task: {state['task']}"),
]
response = llm.invoke(messages)
return {"draft": response.content}
def human_review_node(state: ApprovalState) -> dict:
return {}
def process_feedback_node(state: ApprovalState) -> dict:
if state["approved"]:
return {"final_result": state["draft"]}
messages = [
SystemMessage(content="You are a Content Revision Agent. Revise the draft based on feedback."),
HumanMessage(content=f"Original draft: {state['draft']}\n\nRevision feedback: {state['human_feedback']}"),
]
response = llm.invoke(messages)
return {"draft": response.content, "human_feedback": ""}
def should_continue(state: ApprovalState) -> str:
if state["approved"]:
return "end"
return "revise"
graph = StateGraph(ApprovalState)
graph.add_node("generate_draft", generate_draft_node)
graph.add_node("human_review", human_review_node)
graph.add_node("process_feedback", process_feedback_node)
graph.add_edge(START, "generate_draft")
graph.add_edge("generate_draft", "human_review")
graph.add_conditional_edges("human_review", should_continue, {
"revise": "process_feedback",
"end": END,
})
graph.add_edge("process_feedback", "human_review")
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"],
)
thread_id = "approval-001"
config = {"configurable": {"thread_id": thread_id}}
result = app.invoke({
"messages": [],
"task": "Write a 2026 AI industry trends report",
"draft": "",
"human_feedback": "",
"final_result": "",
"approved": False,
}, config=config)
current_state = app.get_state(config)
print("Draft content:", current_state.values.get("draft", ""))
app.update_state(config, {
"human_feedback": "Please add content about multimodal models",
"approved": False,
})
app.invoke(None, config=config)
current_state = app.get_state(config)
print("Revised draft:", current_state.values.get("draft", ""))
app.update_state(config, {"approved": True})
final_result = app.invoke(None, config=config)
print("Final result:", final_result["final_result"])
Pattern 5: Persistent State — PostgreSQL Checkpointer
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import asyncio
from psycopg_pool import AsyncConnectionPool
class LongRunningState(TypedDict):
messages: Annotated[list, add_messages]
task: str
step1_result: str
step2_result: str
step3_result: str
current_step: int
error: str
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def step1_node(state: LongRunningState) -> dict:
try:
messages = [
SystemMessage(content="You are a Data Processing Agent. Execute step 1: data collection and cleaning."),
HumanMessage(content=f"Processing task: {state['task']}"),
]
response = llm.invoke(messages)
return {"step1_result": response.content, "current_step": 1, "error": ""}
except Exception as e:
return {"error": str(e), "current_step": state["current_step"]}
def step2_node(state: LongRunningState) -> dict:
try:
messages = [
SystemMessage(content="You are an Analysis Agent. Execute step 2: data analysis and modeling."),
HumanMessage(content=f"Analyze based on step 1 results: {state['step1_result']}"),
]
response = llm.invoke(messages)
return {"step2_result": response.content, "current_step": 2, "error": ""}
except Exception as e:
return {"error": str(e), "current_step": state["current_step"]}
def step3_node(state: LongRunningState) -> dict:
try:
messages = [
SystemMessage(content="You are a Report Agent. Execute step 3: generate final report."),
HumanMessage(content=f"Generate report based on analysis results: {state['step2_result']}"),
]
response = llm.invoke(messages)
return {"step3_result": response.content, "current_step": 3, "error": ""}
except Exception as e:
return {"error": str(e), "current_step": state["current_step"]}
graph = StateGraph(LongRunningState)
graph.add_node("step1", step1_node)
graph.add_node("step2", step2_node)
graph.add_node("step3", step3_node)
graph.add_edge(START, "step1")
graph.add_edge("step1", "step2")
graph.add_edge("step2", "step3")
graph.add_edge("step3", END)
async def run_with_persistence():
connection_string = "postgresql://user:pass@localhost:5432/langgraph"
async with AsyncConnectionPool(connection_string) as pool:
checkpointer = AsyncPostgresSaver(pool)
await checkpointer.setup()
app = graph.compile(checkpointer=checkpointer)
thread_id = "long-running-task-001"
config = {"configurable": {"thread_id": thread_id}}
result = await app.ainvoke({
"messages": [],
"task": "Analyze Q1 sales data and generate forecast report",
"step1_result": "",
"step2_result": "",
"step3_result": "",
"current_step": 0,
"error": "",
}, config=config)
state = await app.aget_state(config)
print(f"Current step: {state.values['current_step']}")
print(f"Final result: {state.values.get('step3_result', 'Not completed')}")
if state.values.get("error"):
print(f"Resuming from step {state.values['current_step']}...")
result = await app.ainvoke(None, config=config)
asyncio.run(run_with_persistence())
Pitfall Guide
Pitfall 1: Directly Modifying Mutable Objects in State
# ❌ Wrong: directly modifying the list in state
def bad_node(state: MyState) -> dict:
state["items"].append("new_item") # Directly modifying original state
return state
# ✅ Correct: return new values, let LangGraph's reducer handle it
def good_node(state: MyState) -> dict:
return {"items": state["items"] + ["new_item"]}
# Or use Annotated + reducer
# items: Annotated[list, operator.add]
Pitfall 2: Conditional Edge Returns Non-existent Node Name
# ❌ Wrong: router function returns an unregistered node name
def bad_router(state: MyState) -> str:
return "non_existent_node"
graph.add_conditional_edges("start", bad_router)
# ✅ Correct: router only returns registered node names, list all possibilities in mapping
def good_router(state: MyState) -> str:
if state["intent"] == "tech":
return "technical"
return "general"
graph.add_conditional_edges("start", good_router, {
"technical": "technical",
"general": "general",
})
Pitfall 3: Forgetting to Set Checkpointer Prevents Resume
# ❌ Wrong: no checkpointer, interrupt_before won't work
app = graph.compile(interrupt_before=["human_review"])
# ✅ Correct: must provide checkpointer
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=["human_review"],
)
Pitfall 4: Node Function Returns Incomplete State Update
# ❌ Wrong: node returns None or empty dict, causing state loss
def bad_node(state: MyState) -> dict:
result = do_something()
# Forgot to return state update
return {}
# ✅ Correct: node must return state fields that need updating
def good_node(state: MyState) -> dict:
result = do_something()
return {"result": result, "status": "completed"}
Pitfall 5: Using Sync Checkpointer in Async Environment
# ❌ Wrong: using sync MemorySaver in async environment
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())
await app.ainvoke(input_data, config=config)
# ✅ Correct: use async checkpointer in async environment
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
app = graph.compile(checkpointer=AsyncPostgresSaver(pool))
await app.ainvoke(input_data, config=config)
Error Troubleshooting
| # | Error Message | Cause | Solution |
|---|---|---|---|
| 1 | KeyError: 'field_name' |
Missing required field in State | Ensure initial invoke includes all TypedDict fields |
| 2 | ValueError: Node 'xxx' not found |
Conditional edge references unregistered node | Check add_node and add_conditional_edges node names |
| 3 | GraphRecursionError |
Infinite loop in graph | Add loop counter or termination condition |
| 4 | Missing checkpointer |
Using interrupt without checkpointer | Pass checkpointer parameter when compiling |
| 5 | InvalidStateUpdate |
Node returns field not in State | Ensure returned keys match TypedDict definition |
| 6 | asyncio.run() cannot be called from a running event loop |
Calling asyncio.run in Jupyter | Use await or nest_asyncio |
| 7 | psycopg.OperationalError |
PostgreSQL connection failed | Check connection string, verify database is running |
| 8 | TypeError: 'NoneType' object is not subscriptable |
Node returns None | Ensure node function returns dict |
| 9 | LangGraphError: Cannot resume without thread_id |
Missing thread_id when resuming | Provide thread_id in config |
| 10 | RateLimitError from OpenAI |
API call rate exceeded | Add retry logic or reduce concurrency |
Advanced Optimization
1. Subgraph Encapsulation — Reusable Workflow Modules
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class ResearchSubState(TypedDict):
messages: Annotated[list, add_messages]
topic: str
research_output: str
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def deep_research_node(state: ResearchSubState) -> dict:
messages = [
SystemMessage(content="You are a Deep Researcher. Conduct comprehensive in-depth research."),
HumanMessage(content=f"Deep research: {state['topic']}"),
]
response = llm.invoke(messages)
return {"research_output": response.content}
def fact_check_node(state: ResearchSubState) -> dict:
messages = [
SystemMessage(content="You are a Fact Checker. Verify the accuracy of research results."),
HumanMessage(content=f"Fact check the following: {state['research_output']}"),
]
response = llm.invoke(messages)
return {"research_output": f"{state['research_output']}\n\nFact check: {response.content}"}
research_subgraph = StateGraph(ResearchSubState)
research_subgraph.add_node("deep_research", deep_research_node)
research_subgraph.add_node("fact_check", fact_check_node)
research_subgraph.add_edge(START, "deep_research")
research_subgraph.add_edge("deep_research", "fact_check")
research_subgraph.add_edge("fact_check", END)
research_app = research_subgraph.compile()
class MainState(TypedDict):
messages: Annotated[list, add_messages]
task: str
research_result: str
writing_result: str
def research_coordinator_node(state: MainState) -> dict:
result = research_app.invoke({
"messages": [],
"topic": state["task"],
"research_output": "",
})
return {"research_result": result["research_output"]}
def writing_node(state: MainState) -> dict:
messages = [
SystemMessage(content="You are a Writer Agent. Write an article based on research results."),
HumanMessage(content=f"Write based on research: {state['research_result']}"),
]
response = llm.invoke(messages)
return {"writing_result": response.content}
main_graph = StateGraph(MainState)
main_graph.add_node("research_coordinator", research_coordinator_node)
main_graph.add_node("writer", writing_node)
main_graph.add_edge(START, "research_coordinator")
main_graph.add_edge("research_coordinator", "writer")
main_graph.add_edge("writer", END)
main_app = main_graph.compile()
2. Parallel Node Execution
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
class ParallelState(TypedDict):
messages: Annotated[list, add_messages]
task: str
tech_analysis: str
market_analysis: str
competitor_analysis: str
final_report: str
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def tech_analysis_node(state: ParallelState) -> dict:
messages = [
SystemMessage(content="You are a Tech Analysis Agent."),
HumanMessage(content=f"Tech analysis: {state['task']}"),
]
response = llm.invoke(messages)
return {"tech_analysis": response.content}
def market_analysis_node(state: ParallelState) -> dict:
messages = [
SystemMessage(content="You are a Market Analysis Agent."),
HumanMessage(content=f"Market analysis: {state['task']}"),
]
response = llm.invoke(messages)
return {"market_analysis": response.content}
def competitor_analysis_node(state: ParallelState) -> dict:
messages = [
SystemMessage(content="You are a Competitor Analysis Agent."),
HumanMessage(content=f"Competitor analysis: {state['task']}"),
]
response = llm.invoke(messages)
return {"competitor_analysis": response.content}
def merge_node(state: ParallelState) -> dict:
combined = f"""# Comprehensive Analysis Report
## Tech Analysis
{state['tech_analysis']}
## Market Analysis
{state['market_analysis']}
## Competitor Analysis
{state['competitor_analysis']}
"""
return {"final_report": combined}
graph = StateGraph(ParallelState)
graph.add_node("tech", tech_analysis_node)
graph.add_node("market", market_analysis_node)
graph.add_node("competitor", competitor_analysis_node)
graph.add_node("merge", merge_node)
graph.add_edge(START, "tech")
graph.add_edge(START, "market")
graph.add_edge(START, "competitor")
graph.add_edge("tech", "merge")
graph.add_edge("market", "merge")
graph.add_edge("competitor", "merge")
graph.add_edge("merge", END)
app = graph.compile()
3. Tool Calling Integration
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
@tool
def search_database(query: str) -> str:
"""Search database for information"""
mock_results = {
"revenue": "Q1 2026 Revenue: $12M, YoY growth 35%",
"users": "Current active users: 5.8M, monthly growth 12%",
"products": "Product lines: 3 core products, 12 SKUs",
}
for key, value in mock_results.items():
if key in query.lower():
return value
return "No relevant data found"
@tool
def calculate_metrics(expression: str) -> str:
"""Calculate business metrics"""
try:
result = eval(expression, {"__builtins__": {}}, {})
return f"Calculation result: {result}"
except Exception as e:
return f"Calculation error: {e}"
tools = [search_database, calculate_metrics]
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
def agent_node(state: AgentState) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition, {
"tools": "tools",
END: END,
})
graph.add_edge("tools", "agent")
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage(content="Query Q1 revenue and calculate YoY growth rate (assuming last year Q1 was $8.9M)")],
})
for msg in result["messages"]:
if hasattr(msg, "content") and msg.content:
print(f"{msg.type}: {msg.content}")
Comparison Analysis
| Dimension | LangGraph | CrewAI | AutoGen | LangChain Agent | Dify |
|---|---|---|---|---|---|
| Workflow Orchestration | ✅ Declarative graph | ⚠️ Process definition | ⚠️ Conversation-driven | ❌ Linear chain | ✅ Visual |
| State Management | ✅ Built-in | ⚠️ Manual | ⚠️ Manual | ❌ None | ✅ Built-in |
| Human-in-the-Loop | ✅ interrupt | ❌ | ⚠️ Human proxy | ❌ | ✅ Visual |
| Persistence | ✅ Checkpointer | ❌ | ❌ | ❌ | ✅ Built-in |
| Conditional Routing | ✅ Conditional edges | ⚠️ Limited | ❌ | ❌ | ✅ Visual |
| Subgraph Reuse | ✅ Subgraph | ❌ | ❌ | ❌ | ⚠️ Templates |
| Parallel Execution | ✅ | ❌ | ✅ | ❌ | ⚠️ |
| Self-hosted | ✅ | ✅ | ✅ | ✅ | ⚠️ Docker |
| Learning Curve | Medium | Low | Medium | Low | Low |
| Production Ready | ✅ | ⚠️ | ⚠️ | ❌ | ✅ |
Summary: LangGraph isn't "yet another Agent framework" — it's "the operating system for AI workflows." Its core value lies in StateGraph — replacing imperative if-else with declarative graphs, replacing manual state management with Checkpointer, replacing hardcoded routing with conditional edges. The 2026 multi-Agent practice path: first run through the process with a single-Agent state machine → then orchestrate multiple Agents with the Supervisor pattern → finally add human-in-the-loop and persistence. The key is treating "state" as a first-class citizen — all inter-Agent communication goes through State, all flow control goes through graph topology, all interrupt/resume goes through Checkpoints.
Recommended Online Tools
- JSON Formatter: /en/json/format
- Base64 Encode/Decode: /en/encode/base64
- Hash Calculator: /en/encode/hash
- JWT Decode: /en/encode/jwt-decode
Try these browser-local tools — no sign-up required →
#Python#LangGraph#多Agent#AI工作流#状态机#2026#Agent编排