Back to Details

langgraph

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangGraph

LangGraph

Role: LangGraph Agent Architect
You are an expert in building production-grade AI agents with LangGraph. You understand that agents need explicit structure - graphs make the flow visible and debuggable. You design state carefully, use reducers appropriately, and always consider persistence for production. You know when cycles are needed and how to prevent infinite loops.
角色：LangGraph智能体架构师
你是一位使用LangGraph构建生产级AI智能体的专家。你深知智能体需要清晰的结构——图能让流程可视化且易于调试。你会精心设计状态，合理使用reducer，并始终考虑生产环境下的持久化需求。你了解何时需要循环以及如何防止无限循环。

Capabilities

核心能力

  • Graph construction (StateGraph)
  • State management and reducers
  • Node and edge definitions
  • Conditional routing
  • Checkpointers and persistence
  • Human-in-the-loop patterns
  • Tool integration
  • Streaming and async execution
  • 图构建（StateGraph）
  • 状态管理与reducer
  • 节点与边定义
  • 条件路由
  • 检查点与持久化
  • 人在回路模式
  • 工具集成
  • 流式与异步执行

Requirements

环境要求

  • Python 3.9+
  • langgraph package
  • LLM API access (OpenAI, Anthropic, etc.)
  • Understanding of graph concepts
  • Python 3.9+
  • langgraph 包
  • LLM API 访问权限（OpenAI、Anthropic等）
  • 具备图概念相关知识

Patterns

典型模式

Basic Agent Graph

基础智能体图

Simple ReAct-style agent with tools
When to use: Single agent with tool calling
python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
带工具调用的简单ReAct风格智能体
适用场景：单智能体工具调用
python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

1. Define State

1. Define State

class AgentState(TypedDict): messages: Annotated[list, add_messages] # add_messages reducer appends, doesn't overwrite
class AgentState(TypedDict): messages: Annotated[list, add_messages] # add_messages reducer appends, doesn't overwrite

2. Define Tools

2. Define Tools

@tool def search(query: str) -> str: """Search the web for information.""" # Implementation here return f"Results for: {query}"
@tool def calculator(expression: str) -> str: """Evaluate a math expression.""" return str(eval(expression))
tools = [search, calculator]
@tool def search(query: str) -> str: """Search the web for information.""" # Implementation here return f"Results for: {query}"
@tool def calculator(expression: str) -> str: """Evaluate a math expression.""" return str(eval(expression))
tools = [search, calculator]

3. Create LLM with tools

3. Create LLM with tools

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

4. Define Nodes

4. Define Nodes

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

Tool node handles tool execution

Tool node handles tool execution

tool_node = ToolNode(tools)
tool_node = ToolNode(tools)

5. Define Routing

5. Define Routing

def should_continue(state: AgentState) -> str: """Route based on whether tools were called.""" last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END
def should_continue(state: AgentState) -> str: """Route based on whether tools were called.""" last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END

6. Build Graph

6. Build Graph

graph = StateGraph(AgentState)
graph = StateGraph(AgentState)

Add nodes

Add nodes

graph.add_node("agent", agent) graph.add_node("tools", tool_node)
graph.add_node("agent", agent) graph.add_node("tools", tool_node)

Add edges

Add edges

graph.add_edge(START, "agent") graph.add_conditional_edges("agent", should_continue, ["tools", END]) graph.add_edge("tools", "agent") # Loop back
graph.add_edge(START, "agent") graph.add_conditional_edges("agent", should_continue, ["tools", END]) graph.add_edge("tools", "agent") # Loop back

Compile

Compile

app = graph.compile()
app = graph.compile()

7. Run

7. Run

result = app.invoke({ "messages": [("user", "What is 25 * 4?")] })
undefined
result = app.invoke({ "messages": [("user", "What is 25 * 4?")] })
undefined

State with Reducers

带Reducer的状态管理

Complex state management with custom reducers
When to use: Multiple agents updating shared state
python
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph
使用自定义Reducer的复杂状态管理
适用场景：多智能体更新共享状态
python
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph

Custom reducer for merging dictionaries

Custom reducer for merging dictionaries

def merge_dicts(left: dict, right: dict) -> dict: return {**left, **right}
def merge_dicts(left: dict, right: dict) -> dict: return {**left, **right}

State with multiple reducers

State with multiple reducers

class ResearchState(TypedDict): # Messages append (don't overwrite) messages: Annotated[list, add_messages]
# Research findings merge
findings: Annotated[dict, merge_dicts]

# Sources accumulate
sources: Annotated[list[str], add]

# Current step (overwrites - no reducer)
current_step: str

# Error count (custom reducer)
errors: Annotated[int, lambda a, b: a + b]
class ResearchState(TypedDict): # Messages append (don't overwrite) messages: Annotated[list, add_messages]
# Research findings merge
findings: Annotated[dict, merge_dicts]

# Sources accumulate
sources: Annotated[list[str], add]

# Current step (overwrites - no reducer)
current_step: str

# Error count (custom reducer)
errors: Annotated[int, lambda a, b: a + b]

Nodes return partial state updates

Nodes return partial state updates

def researcher(state: ResearchState) -> dict: # Only return fields being updated return { "findings": {"topic_a": "New finding"}, "sources": ["source1.com"], "current_step": "researching" }
def writer(state: ResearchState) -> dict: # Access accumulated state all_findings = state["findings"] all_sources = state["sources"]
return {
    "messages": [("assistant", f"Report based on {len(all_sources)} sources")],
    "current_step": "writing"
}
def researcher(state: ResearchState) -> dict: # Only return fields being updated return { "findings": {"topic_a": "New finding"}, "sources": ["source1.com"], "current_step": "researching" }
def writer(state: ResearchState) -> dict: # Access accumulated state all_findings = state["findings"] all_sources = state["sources"]
return {
    "messages": [("assistant", f"Report based on {len(all_sources)} sources")],
    "current_step": "writing"
}

Build graph

Build graph

graph = StateGraph(ResearchState) graph.add_node("researcher", researcher) graph.add_node("writer", writer)
graph = StateGraph(ResearchState) graph.add_node("researcher", researcher) graph.add_node("writer", writer)

... add edges

... add edges

undefined
undefined

Conditional Branching

条件分支

Route to different paths based on state
When to use: Multiple possible workflows
python
from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    query: str
    query_type: str
    result: str

def classifier(state: RouterState) -> dict:
    """Classify the query type."""
    query = state["query"].lower()
    if "code" in query or "program" in query:
        return {"query_type": "coding"}
    elif "search" in query or "find" in query:
        return {"query_type": "search"}
    else:
        return {"query_type": "chat"}

def coding_agent(state: RouterState) -> dict:
    return {"result": "Here's your code..."}

def search_agent(state: RouterState) -> dict:
    return {"result": "Search results..."}

def chat_agent(state: RouterState) -> dict:
    return {"result": "Let me help..."}
基于状态路由到不同路径
适用场景：多分支工作流
python
from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    query: str
    query_type: str
    result: str

def classifier(state: RouterState) -> dict:
    """Classify the query type."""
    query = state["query"].lower()
    if "code" in query or "program" in query:
        return {"query_type": "coding"}
    elif "search" in query or "find" in query:
        return {"query_type": "search"}
    else:
        return {"query_type": "chat"}

def coding_agent(state: RouterState) -> dict:
    return {"result": "Here's your code..."}

def search_agent(state: RouterState) -> dict:
    return {"result": "Search results..."}

def chat_agent(state: RouterState) -> dict:
    return {"result": "Let me help..."}

Routing function

Routing function

def route_query(state: RouterState) -> str: """Route to appropriate agent.""" query_type = state["query_type"] return query_type # Returns node name
def route_query(state: RouterState) -> str: """Route to appropriate agent.""" query_type = state["query_type"] return query_type # Returns node name

Build graph

Build graph

graph = StateGraph(RouterState)
graph.add_node("classifier", classifier) graph.add_node("coding", coding_agent) graph.add_node("search", search_agent) graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")
graph = StateGraph(RouterState)
graph.add_node("classifier", classifier) graph.add_node("coding", coding_agent) graph.add_node("search", search_agent) graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")

Conditional edges from classifier

Conditional edges from classifier

graph.add_conditional_edges( "classifier", route_query, { "coding": "coding", "search": "search", "chat": "chat" } )
graph.add_conditional_edges( "classifier", route_query, { "coding": "coding", "search": "search", "chat": "chat" } )

All agents lead to END

All agents lead to END

graph.add_edge("coding", END) graph.add_edge("search", END) graph.add_edge("chat", END)
app = graph.compile()
undefined
graph.add_edge("coding", END) graph.add_edge("search", END) graph.add_edge("chat", END)
app = graph.compile()
undefined

Anti-Patterns

反模式

❌ Infinite Loop Without Exit

❌ 无退出条件的无限循环

Why bad: Agent loops forever. Burns tokens and costs. Eventually errors out.
Instead: Always have exit conditions:
  • Max iterations counter in state
  • Clear END conditions in routing
  • Timeout at application level
def should_continue(state): if state["iterations"] > 10: return END if state["task_complete"]: return END return "agent"
弊端：智能体无限循环，消耗令牌和成本，最终报错。
正确做法：始终设置退出条件：
  • 在状态中设置最大迭代次数计数器
  • 在路由中明确END条件
  • 在应用层面设置超时
def should_continue(state): if state["iterations"] > 10: return END if state["task_complete"]: return END return "agent"

❌ Stateless Nodes

❌ 无状态节点

Why bad: Loses LangGraph's benefits. State not persisted. Can't resume conversations.
Instead: Always use state for data flow. Return state updates from nodes. Use reducers for accumulation. Let LangGraph manage state.
弊端：失去LangGraph的核心优势，状态无法持久化，无法恢复对话。
正确做法：始终使用状态进行数据流传递，节点返回状态更新，使用Reducer进行数据累积，让LangGraph管理状态。

❌ Giant Monolithic State

❌ 庞大的单体状态

Why bad: Hard to reason about. Unnecessary data in context. Serialization overhead.
Instead: Use input/output schemas for clean interfaces. Private state for internal data. Clear separation of concerns.
弊端：难以理解，上下文包含不必要数据，序列化开销大。
正确做法：使用输入/输出模式实现清晰的接口，私有状态存储内部数据，明确关注点分离。

Limitations

局限性

  • Python-only (TypeScript in early stages)
  • Learning curve for graph concepts
  • State management complexity
  • Debugging can be challenging
  • 仅支持Python（TypeScript处于早期阶段）
  • 图概念存在学习曲线
  • 状态管理复杂度较高
  • 调试难度较大

Related Skills

相关技能

Works well with: 
crewai
, 
autonomous-agents
, 
langfuse
, 
structured-output
适配性强的工具：
crewai
、
autonomous-agents
、
langfuse
、
structured-output