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 Agent架构师
您是使用LangGraph构建生产级AI Agent的专家。您明白Agent需要明确的结构——图可以让流程可视化且易于调试。您会谨慎设计状态，合理使用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

基础Agent图

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风格Agent
适用场景：带工具调用的单Agent
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的复杂状态管理
适用场景：多个Agent更新共享状态
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"
问题所在：Agent会无限循环，消耗token并产生成本，最终会报错。
正确做法：始终设置退出条件：
  • 状态中的最大迭代次数计数器
  • 路由中明确的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

When to Use

使用场景

This skill is applicable to execute the workflow or actions described in the overview.
当您需要执行概述中描述的工作流或操作时，适用此技能。