mastering-langgraph

LangGraph 开发指南

通过定义由边（转换）连接的节点（步骤）图，构建有状态的 AI Agent 和工作流。

目录

• 快速开始
• 常见构建场景
• 核心原则
• 开发工作流
• 常见问题
• 环境设置
• 快速验证
• API 要点
• 后续步骤

快速开始

带记忆功能的最简聊天机器人：

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

# 1. Define state
class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # Append mode

# 2. Define node
llm = ChatOpenAI(model="gpt-4")

def chat(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 3. Build graph
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# 4. Compile with memory
chain = graph.compile(checkpointer=InMemorySaver())

# 5. Invoke with thread_id for persistence
result = chain.invoke(
    {"messages": [HumanMessage(content="Hello!")]},
    config={"configurable": {"thread_id": "user-123"}}
)
print(result["messages"][-1].content)

关键模式：

• Annotated[list, operator.add] — 追加到列表，而不是替换
• InMemorySaver() — 启用跨调用记忆
• thread_id — 标识用于持久化的对话

常见构建场景

简单聊天机器人 / 问答

上面的快速开始已涵盖此场景。可根据需要添加更多节点进行预处理或后处理。

使用工具的 Agent

循环调用外部工具（API、计算器、搜索），直到任务完成的 Agent。 → 参见 references/tool-agent-pattern.md

结构化工作流

包含条件分支、并行执行或提示词链的多步骤流水线。 → 参见 references/workflow-patterns.md

具有长期记忆的 Agent

跨会话持久化对话、启用时间回溯调试，并在崩溃后恢复。 → 参见 references/persistence-memory.md

人机协作

在工作流中暂停，以便获取人工审批、纠正或补充输入。 → 参见 references/hitl-patterns.md

调试 / 生产监控

对节点进行单元测试、可视化图，并使用 LangSmith 进行追踪。 → 参见 references/debugging-monitoring.md

多 Agent 系统

使用交接工具构建基于监督者或群集的多 Agent 工作流。 → 参见 references/multi-agent-patterns.md

生产部署

部署到 LangGraph Platform（云端 / 可自托管）或自定义基础设施。 → 参见 references/production-deployment.md

初次接触 LangGraph？

学习核心概念：状态、节点、边和图 API。 → 参见 references/core-api.md

核心原则

1. 保持状态原始

存储事实，而不是格式化后的提示词。每个节点都可以根据需要格式化数据。

# ✓ Good: raw data
class State(TypedDict):
    user_question: str
    retrieved_docs: list[str]
    intent: str

# ✗ Bad: pre-formatted
class State(TypedDict):
    full_prompt: str  # Mixes data with formatting

2. 节点各司其职

每个节点只做一件事，并使用描述性名称。

# ✓ Good: clear responsibilities
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_knowledge", search_knowledge)
graph.add_node("generate_response", generate_response)

3. 显式路由

使用条件边进行决策。不要将路由逻辑隐藏在节点内部。

def route_by_intent(state) -> str:
    if state["intent"] == "billing":
        return "billing_handler"
    return "general_handler"

graph.add_conditional_edges("classify", route_by_intent, 
    ["billing_handler", "general_handler"])

4. 为列表使用聚合器

任何会累积值的列表字段都需要使用 operator.add：

class State(TypedDict):
    messages: Annotated[list, operator.add]      # ✓ Appends
    current_step: str                             # Replaces (no annotation)

5. 有计划地处理错误

错误类型	策略
暂时性（网络）	在节点上使用 RetryPolicy
大模型可恢复（解析失败）	通过状态将错误传给大模型，再循环返回
用户可修复（缺少信息）	使用 interrupt() 暂停并询问
意外错误（Bug）	让错误向上抛出以便调试

开发工作流

1. 定义步骤 — 将任务拆分为独立操作（每个操作对应一个节点）
2. 分类步骤 — 大模型调用？数据检索？执行操作？用户输入？
3. 设计状态 — 使用包含所有必要字段的 TypedDict；保持原始
4. 实现节点 — 为每个步骤定义 def node(state) -> dict
5. 连接图 — 使用 add_node()、add_edge()、add_conditional_edges()
6. 编译并测试 — 使用 graph.compile()，并通过示例输入进行测试

常见问题

1. 忘记在列表上使用 operator.add

表现： 消息消失，只保留最后一条消息。

# ✗ Wrong: messages: list[AnyMessage]
# ✓ Fix: messages: Annotated[list[AnyMessage], operator.add]

2. 记忆功能缺少 thread_id

表现： Agent 忘记之前的对话轮次。

# ✓ Fix: Always pass config with thread_id
chain.invoke(input, config={"configurable": {"thread_id": "unique-id"}})

3. 调用前未编译

表现： 图对象上出现 AttributeError。

# ✗ Wrong: graph.invoke(input)
# ✓ Fix: chain = graph.compile(); chain.invoke(input)

4. 非确定性节点未使用 @task

表现： 从检查点恢复后结果不同。

from langgraph.func import task

@task  # Wrap for durable execution
def fetch_data(state):
    return {"data": requests.get(url).json()}

5. 类型提示中的循环导入

症状： 定义状态类时出现 ImportError。

# ✓ Fix: Use string annotations
from __future__ import annotations

环境设置

# Core
pip install -U langgraph

# LLM providers (pick one or more)
pip install langchain-openai
pip install langchain-anthropic

# Production persistence
pip install langgraph-checkpoint-postgres

# Observability
pip install langsmith

环境变量：

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export LANGSMITH_API_KEY="ls-..."
export LANGSMITH_TRACING=true

快速验证

构建前

• python -c "import langgraph; print(langgraph.__version__)" 可正常运行
• 已设置 LLM API 密钥（OPENAI_API_KEY 或 ANTHROPIC_API_KEY）
• 可选：用于追踪的 LANGSMITH_API_KEY

构建后

• 图编译成功，无错误：chain = graph.compile()
• 可视化渲染正常：print(chain.get_graph().draw_mermaid())
• 使用示例输入调用成功：chain.invoke({...})
• 列表可正确累积（检查 operator.add 注解）
• 内存可在多次调用间保留（使用相同的 thread_id 测试两次）
• 条件路由按预期工作（测试每个分支）

API 基础知识

# Imports
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, Annotated
import operator

# State with append-mode list
class State(TypedDict):
    messages: Annotated[list, operator.add]

# Node signature
def node(state: State) -> dict:
    return {"messages": [new_message]}

# Graph construction
graph = StateGraph(State)
graph.add_node("name", node_fn)
graph.add_edge(START, "name")
graph.add_edge("name", END)

# Conditional routing
graph.add_conditional_edges("from", router_fn, ["option1", "option2", END])

# Compile and run
chain = graph.compile(checkpointer=InMemorySaver())
result = chain.invoke(input, config={"configurable": {"thread_id": "id"}})

# Visualization
print(chain.get_graph().draw_mermaid())

如需详细的 API 参考 → 请参阅 references/core-api.md

后续步骤

• 工具 Agent：references/tool-agent-pattern.md
• 工作流：references/workflow-patterns.md
• 持久化：references/persistence-memory.md
• 人工参与环节：references/hitl-patterns.md
• 测试/监控：references/debugging-monitoring.md
• 多 Agent：references/multi-agent-patterns.md
• 生产环境：references/production-deployment.md
• 核心概念：references/core-api.md
• 官方文档：references/official-resources.md