LangGraph 中文教程:从入门到精通的 2026 完全指南
基于 LangGraph 0.6.11 版本编写的中文系统教程,涵盖状态图、Checkpoint 持久化、MiniMax 集成、多 Agent 编排、生产部署等核心概念,5800+ 字完整中文资料。
LangGraph 中文教程:从入门到精通的 2026 完全指南
本文基于 LangGraph v0.6.11 版本编写,内容基于开源项目 langchain-ai/langgraph(MIT License)翻译整理。 中文译注 + 原创 demo 版权归裕普网络有限公司所有。
一、LangGraph 在 2026 年 AI 生态中的定位
进入 2026 年,AI Agent 已经从"概念验证"走向"生产落地"。在这一波浪潮中,LangGraph 扮演着一个非常特殊且关键的角色。
1.1 LangGraph 与 LangChain 的关系
LangChain 是 2023 年最早将大语言模型(LLM)能力产品化的开源框架,它提供了 Chain(链式调用)、Agent(智能体)、Memory(记忆)等核心抽象。然而,随着用户需求的深入,LangChain 的 chain 抽象暴露出一个根本性限制:它不适合表达复杂的长流程状态转移。
LangGraph 正是 LangChain 团队对这一限制的回应。它的核心哲学是:把任何多步流程——对话、审批、检索增强生成(RAG)——都建模成一张可暂停、可恢复、可观察的「状态图」。
两者的关系可以这样理解:
- LangChain Chain:适合「单步 LLM 调用 + 工具」的链式编排,像流水线。
- LangGraph StateGraph:适合「多步长流程 + 状态持久化 + 人在回路」的复杂应用,像状态机。
在 LangGraph 内部,完全兼容所有 LangChain Runnable。你可以把任意 LangChain 链作为一个 Node 加入 StateGraph,享受两者的优势。
1.2 2026 年 LangGraph 生态新特性
2026 年,LangGraph 生态有几个值得关注的新变化:
- Runnable 新特性:LangChain 0.3.x 系列对 Runnable 协议做了大幅优化,
invoke/stream/batch三种调用模式在 LangGraph Node 中可以无缝切换。 - Human-in-the-loop 标准化:
interrupt(value)+Command(resume=value)已经成为多步骤审批流程的事实标准 API。 - LangGraph Platform:官方推出了托管的 LangGraph Platform(langchain.com),支持可视化调试图执行过程,极大降低了排查问题的门槛。
- 中文社区活跃度:截至 2026 年 9 月,LangGraph GitHub Stars 已突破 15k,国内掘金/知乎/CSDN 相关文章数量年增长率超过 300%。
1.3 为什么国内开发者需要 LangGraph
对国内个人开发者而言,直接使用 LangGraph 至少有三个摩擦点:
- 英文文档门槛:官方文档默认英文,
interrupt/Command(resume=...)这类 API 第一次接触时需要查多份资料。 - 中国 LLM 适配:默认模板以 OpenAI / Anthropic 为主,国内常用的 MiniMax / Qwen / DeepSeek 等需要自行包装。
- 本地化 demo 缺失:官方 examples 多是通用业务场景,缺少中文业务场景(采购审批、中文地址识别、工单分派)做参考。
裕普智汇中文本地化交付包就是为了消除这三点摩擦,在保持 LangGraph v0.6.11 原汁原味的前提下,提供完整的中文文档、真实场景 demo、MiniMax 集成模板。
二、5 分钟快速开始
2.1 环境准备
# 1. 创建虚拟环境(推荐 Python 3.11 / 3.12)
python3 -m venv langgraph-zh-env
source langgraph-zh/bin/activate
# 2. 安装 LangGraph(锁定版本)
pip install langgraph==0.6.11
# 3. 验证安装
python3 -c "import langgraph; print(langgraph.__version__)"
注意:LangGraph 0.6.11 需要
langchain-core>=0.3.0,requires-python ≥ 3.9。Python 3.8 及以下不兼容。
2.2 第一个 StateGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict
# 定义状态
class State(TypedDict):
messages: list[str]
# 定义节点函数
def say_hello(state: State) -> dict:
return {"messages": state["messages"] + ["你好,LangGraph!"]}
# 构建图
graph = StateGraph(State)
graph.add_node("hello", say_hello)
graph.add_edge("hello", END)
app = graph.compile()
result = app.invoke({"messages": []})
print(result["messages"]) # ['你好,LangGraph!']
2.3 运行交付包内的 demo
解压 yupu-localization-langgraph-zh-v0.1.0.zip 后,可以直接运行中文 demo:
cd yupu-localization-langgraph-zh/
# 运行状态机 demo(mock 模式,无需 API key)
python3 examples-zh/02-state-machine/main.py
# 期望输出:状态机执行完成:state → A → B → END
三、8 大核心能力详解
3.1 State(状态)
State 是 LangGraph 的「整张图的当前内存」。开发者用 Python TypedDict 或 Pydantic BaseModel 声明 State 的字段,节点函数读写 State 的字段,边函数依据 State 的值决定路由。
核心要点:
- State 是不可变更新:节点函数返回 dict,LangGraph 会把返回的字段合并进当前 state(merge 而非 replace)。
- 支持
Annotation定义 reducer:比如messages: Annotated[list, add_messages]可以让多节点追加消息而不互相覆盖。 - State 类型一旦确定,后续所有节点的输入输出都必须符合该 schema。
3.2 Node(节点)
Node 是 StateGraph 的处理单元。本质是一个 Python 函数 (state: State) -> dict,读取 state 的字段,返回要更新的字段。
Node 的设计原则:
- 单一职责:一个 Node 只做一件事(如分类、查询、生成回复)。
- 纯函数倾向:Node 不应该有副作用(如直接写数据库),副作用应该放在 Tool 或外部服务中。
- 可测试性:Node 是普通 Python 函数,可以直接 unit test。
5 个典型 Node 代码示例:
# 1. 简单转换节点
def normalize_input(state: State) -> dict:
return {"text": state["raw_input"].strip().lower()}
# 2. LLM 调用节点
from langchain_core.messages import HumanMessage
def call_llm(state: State) -> dict:
response = llm.invoke([HumanMessage(content=state["prompt"])])
return {"response": response.content}
# 3. 条件判断节点
def should_continue(state: State) -> dict:
if state["step_count"] > 5:
return {"next": "end"}
return {"next": "continue"}
# 4. Human-in-the-loop 审批节点
def request_approval(state: State) -> dict:
from langgraph.types import interrupt
approval = interrupt("请审批以下内容:" + state["draft"])
return {"approved": approval}
# 5. 工具调用节点
from langgraph.prebuilt import ToolNode
tools = [search_tool, calculator_tool]
tool_node = ToolNode(tools)
3.3 Edge(边)
Edge 声明节点间的转移路径。LangGraph 支持三种边:
- 普通边
add_edge("A", "B"):执行完 A 后一定执行 B。 - 条件边
add_conditional_edges("A", routing_fn):基于 routing_fn 返回值选择下一个节点。 - 入口/出口
set_entry_point("A")/set_finish_point("END")。
# 条件边示例:根据意图路由到不同 Agent
def route_by_intent(state: State) -> str:
intent = state.get("intent", "unknown")
if intent == "technical":
return "tech_agent"
elif intent == "billing":
return "billing_agent"
return "fallback_agent"
graph.add_conditional_edges("classifier", route_by_intent)
3.4 Checkpoint(检查点)
Checkpoint 是 State 的持久化快照。LangGraph 支持 Memory / Sqlite / Postgres 三种 checkpointer。
为什么 Checkpoint 重要:
- 断点续传:图执行到
interrupt暂停后,可以从 checkpoint 恢复,不需要重新跑前面节点。 - 多用户并发:配合
thread_id,同一张 graph 可以同时为多个用户服务,数据不串扰。 - 调试回溯:生产环境出问题时,可以回溯任意一次执行的完整状态快照。
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=memory)
# 带 thread_id 执行
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(input, config=config)
3.5 Tool(工具)
Tool 是 LLM 可调用的函数。LangGraph 提供 ToolNode(prebuilt 节点)执行 LLM 决策的工具调用,配合 bind_tools() API 让 LLM 知道有哪些工具可用。
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索网页获取实时信息。"""
# 实际项目中对接搜索 API
return f"搜索结果:{query}"
llm_with_tools = llm.bind_tools([search_web])
3.6 Human-in-the-Loop(人在回路)
这是 LangGraph 的核心差异化能力。interrupt(value) 让当前节点暂停,把 value 暴露给上层;Command(resume=value) 从暂停点恢复。
from langgraph.types import interrupt, Command
def approval_node(state: State) -> Command:
# 暂停,等待人工审批
human_decision = interrupt(state["proposal"])
if human_decision["approved"]:
return Command(goto="execute", update={"status": "approved"})
return Command(goto="reject", update={"status": "rejected"})
3.7 Streaming(流式输出)
LangGraph 一等公民支持流式输出。stream_mode 参数支持多种模式:
# 逐 step 流式输出完整 state
for event in app.stream(input, stream_mode="values"):
print(event)
# 只输出每一步的 diff
for event in app.stream(input, stream_mode="updates"):
print(event)
# LLM token 级流式(打字机效果)
for event in app.stream(input, stream_mode="messages"):
print(event, end="", flush=True)
配合 SSE / WebSocket 可做出「打字机效果」的实时 UI。
3.8 Multi-Agent(多 Agent)
LangGraph 的 StateGraph 加上 add_conditional_edges + supervisor 模式可以轻松建模多 Agent 协作:
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
task: str
result: str
next_agent: str
def supervisor(state: AgentState) -> dict:
# 决定下一步让哪个 Agent 处理
if "code" in state["task"]:
return {"next_agent": "coder"}
elif "doc" in state["task"]:
return {"next_agent": "writer"}
return {"next_agent": "END"}
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("coder", code_agent)
graph.add_node("writer", doc_agent)
graph.add_conditional_edges("supervisor", lambda s: s["next_agent"], {
"coder": "coder",
"writer": "writer",
"END": END
})
graph.set_entry_point("supervisor")
四、5 个中文 Demo 缩略代码
4.1 工单分派(01-multi-agent)
from langgraph.graph import StateGraph, END
class TicketState(TypedDict):
ticket: str
category: str
assignee: str
reply: str
graph = StateGraph(TicketState)
graph.add_node("classify", classify_ticket)
graph.add_node("dispatch", dispatch_ticket)
graph.add_node("reply", generate_reply)
graph.add_edge("classify", "dispatch")
graph.add_conditional_edges("dispatch", lambda s: s["assignee"], {
"tech": "tech_reply",
"biz": "biz_reply",
"END": END
})
4.2 状态机(02-state-machine)
class StateMachine(TypedDict):
state: str
input: str
def step_a(s: StateMachine) -> dict:
return {"state": "B"}
def step_b(s: StateMachine) -> dict:
return {"state": "END"}
graph = StateGraph(StateMachine)
graph.add_sequence([step_a, step_b])
graph.set_entry_point("step_a")
4.3 Human-in-the-Loop 审批(03-human-in-loop)
class ApprovalState(TypedDict):
request: str
approved: bool
def submit_for_approval(state: ApprovalState):
from langgraph.types import interrupt
decision = interrupt(state["request"])
return {"approved": decision.get("approved", False)}
graph = StateGraph(ApprovalState)
graph.add_node("approval", submit_for_approval)
graph.add_edge("approval", END)
4.4 RAG 检索增强生成(04-rag)
class RAGState(TypedDict):
question: str
context: list[str]
answer: str
def retrieve(state: RAGState) -> dict:
docs = vector_store.similarity_search(state["question"], k=3)
return {"context": [d.page_content for d in docs]}
def generate(state: RAGState) -> dict:
prompt = f"Context: {state['context']}\nQuestion: {state['question']}"
return {"answer": llm.invoke(prompt).content}
graph = StateGraph(RAGState)
graph.add_sequence([retrieve, generate])
4.5 中文地址识别(05-chinese-ner)
class AddressState(TypedDict):
text: str
province: str
city: str
district: str
def extract_province(state: AddressState) -> dict:
# 简单规则:前两个中文字符可能是省
text = state["text"]
return {"province": text[:2] if len(text) >= 2 else "未知"}
def extract_city(state: AddressState) -> dict:
return {"city": "北京市" if state["province"] == "北京" else "未知"}
graph = StateGraph(AddressState)
graph.add_sequence([extract_province, extract_city])
五、MiniMax 集成模板配置示例
5.1 chat-stream 模板配置
以 integrations-zh/chat_stream.py 为例,展示如何在 LangGraph Node 中使用 MiniMax 流式 API:
import os
from typing import Any, AsyncIterator
import aiohttp
DEFAULT_API_BASE = "https://api.minimaxi.com/v1"
DEFAULT_MODEL = "MiniMax-M2.7"
async def chat_stream(prompt: str, *, model: str = DEFAULT_MODEL,
api_base: str = DEFAULT_API_BASE,
temperature: float = 0.7, **extra: Any) -> AsyncIterator[dict]:
api_key = os.getenv("MINIMAX_API_KEY")
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"stream": True,
}
payload.update(extra)
url = f"{api_base}/chat/completions"
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload) as resp:
async for raw_line in resp.content:
line = raw_line.decode("utf-8").strip()
if not line or not line.startswith("data:"):
continue
data = line[len("data:"):].strip()
if data == "[DONE]":
yield {"_done": True}
return
yield json.loads(data)
5.2 在 StateGraph 中调用
from langgraph.graph import StateGraph, END
class ChatState(TypedDict):
user_input: str
reply: str
async def minimax_node(state: ChatState) -> dict:
chunks = []
async for chunk in chat_stream(state["user_input"]):
if not chunk.get("_done"):
delta = chunk.get("choices", [{}])[0].get("delta", {})
content = delta.get("content") or ""
chunks.append(content)
return {"reply": "".join(chunks)}
graph = StateGraph(ChatState)
graph.add_node("chat", minimax_node)
graph.add_edge("chat", END)
配置提示:运行前设置环境变量
export MINIMAX_API_KEY=<你的 key>。MiniMax 提供MiniMax-M2.7和MiniMax-M1-80k两个推荐模型,前者适合对话,后者适合长文本处理。
六、错误码库典型场景引用
在实际使用 LangGraph 过程中,以下高频错误最容易遇到:
6.1 GraphRecursionError
现象:
GraphRecursionError: Recursion limit reached根因:图的执行步数超过了recursion_limit(默认 25 步),通常是条件边形成了环路。 解决方案:检查条件边逻辑,确保每一步都有向前的方向;必要时提高 limit(不推荐,治标不治本)。 详见../errors-zh/GraphRecursionError.md
6.2 InvalidUpdateError
现象:
InvalidUpdateError: Invalid update at node xxx根因:节点返回的 dict 包含了 State schema 中未声明的字段,或字段类型不匹配。 解决方案:检查 TypedDict/Pydantic 定义,确保 Node 返回的字段名和类型完全匹配。 详见../errors-zh/InvalidUpdateError.md
6.3 NodeTimeoutError
现象:
NodeTimeoutError: Node timed out after 30s根因:节点执行时间超过配置的 timeout(默认 30 秒),常见于调用外部 API 或 LLM。 解决方案:增加 timeout 配置,或改用异步节点 + streaming 模式。 详见../errors-zh/NodeTimeoutError.md
6.4 CheckpointNotFound
现象:
CheckpointNotFound: thread_id xxx not found根因:指定的thread_id在 checkpointer 存储中不存在(可能已过期或被清理)。 解决方案:确认 thread_id 是否正确;检查 checkpointer 的 TTL 配置。 详见../errors-zh/CheckpointNotFound.md
6.5 CheckpointerConnectionError
现象:
CheckpointerConnectionError: could not connect to postgres根因:Postgres checkpointer 连接失败(网络/密码/服务未启动)。 解决方案:检查 Postgres 服务状态;验证连接字符串中的 host/port/password。 详见../errors-zh/CheckpointerConnectionError.md
完整错误码库(32 个错误码)见 ../errors-zh/INDEX.md。
七、Docker Compose 部署与验证
7.1 一键启动
cd deploy/
docker compose up -d
7.2 验证步骤
# 1. 检查容器状态
docker compose ps
# 期望:所有服务 status 为 "Up"
# 2. 健康检查
curl -s http://localhost:8888/health | jq .
# 期望输出包含:{"status": "healthy", "database": true, "redis": true}
# 3. 检查 LangGraph 服务响应
curl -s http://localhost:8888/api/agent/rankings | jq .
# 期望返回 JSON 数组(可能为空,但不应报错)
# 4. 检查 nginx TLS(如已配置)
curl -I https://localhost/health
# 期望 HTTP/2 200
7.3 常见部署问题
| 问题 | 排查命令 | 解决方案 |
|---|---|---|
| 容器启动失败 | docker compose logs | 检查端口冲突、镜像版本 |
| 健康检查失败 | curl localhost:8888/health | 确认 database + redis 可用 |
| 中文乱码 | 安装 Noto Sans SC 字体 | 见本目录「中文字体说明」 |
| nginx 502 | docker compose logs nginx | 确认 upstream 8888 可达 |
详细部署文档见 ../deploy/README.md。
八、合规与开源声明
本教程基于 LangGraph(MIT License)翻译整理。
- 原始项目版权归 LangChain Team 所有。
- 中文译注、原创 demo、集成模板版权归 裕普网络有限公司 所有。
- 本教程不提供任何商业授权许可,仅供学习交流使用。
完整合规信息见 ../LICENSES/ATTRIBUTION.md。
九、常见问题(FAQ)
Q1: LangGraph 适合什么场景?
LangGraph 最适合需要多步状态流转 + 人在回路 + 状态持久化的场景:客服工单分派、审批流、复杂 RAG 管道、多 Agent 协作系统。如果你的需求只是「单次 LLM 调用 + 工具」,LangChain Chain 就够了。
Q2: LangGraph 和 CrewAI / AutoGen 怎么选?
- LangGraph:适合需要精细状态控制 + Human-in-the-loop 的场景。生态成熟,与 LangChain 生态深度集成。
- CrewAI:适合快速原型,Pythonic 抽象,学习曲线平缓。适合不需要复杂状态管理的多 Agent 场景。
- AutoGen:适合微软生态(Azure OpenAI / Semantic Kernel),对话式 Agent 协作是其强项。
Q3: 中文 LLM(MiniMax / Qwen / DeepSeek)怎么接入?
交付包提供了 8 个 MiniMax 集成模板(integrations-zh/),覆盖 chat / chat-stream / embeddings / vision / tool-use / function-call / multimodal / json-mode。其他国内厂商(Qwen / DeepSeek)可通过 OpenAI 兼容协议自行包装。
Q4: 学习曲线 steep 吗?
LangGraph 的核心概念(State + Node + Edge + Checkpoint)可以在 1-2 天内掌握。难点在于 Human-in-the-loop 的 interrupt / Command 模式和多 Agent supervisor 设计,建议先跑通 5 个 demo 再上手实战。
Q5: 生产环境稳定性如何?
LangGraph 0.6.x 已经是生产就绪版本。Checkpointer 支持 Postgres 持久化,配合 thread_id 可以实现多租户隔离。LangGraph Platform(托管版)提供了可视化调试和监控能力。
Q6: 如何调试图执行过程?
使用 graph.stream(input, stream_mode="updates") 可以逐步观察每个节点的输入输出。LangGraph Platform 提供了可视化的图执行时间线,推荐在生产环境使用。
十、验证命令速查
# 1) 验证 LangGraph 安装版本
python3 -c "import langgraph; print(f'LangGraph {langgraph.__version__}')"
# 2) 验证 StateGraph 基本功能
python3 -c "
from langgraph.graph import StateGraph, END
from typing import TypedDict
class S(TypedDict): x: int
g = StateGraph(S); g.add_node('a', lambda s: {'x': s['x']+1}); g.add_edge('a', END); g.set_entry_point('a')
print('StateGraph OK:', g.compile().invoke({'x': 0}))
"
# 3) 验证 MiniMax 流式模板语法
python3 -m py_compile integrations-zh/chat_stream.py && echo 'MiniMax template syntax OK'
# 4) 验证错误码库条目数
grep -c '^### ' errors-zh/INDEX.md && echo 'error entries >= 30 required'
# 5) Docker compose 启动后健康检查
curl -s http://localhost:8888/health | jq .
十一、延伸阅读
- LangGraph 官方文档
- LangGraph GitHub 仓库
- LangChain 中文社区
- 交付包内完整文档:
docs-zh/QUICKSTART.md/docs-zh/ARCHITECTURE.md/docs-zh/STATE.md
资源与声明
- 原项目:langchain-ai/langgraph
- 原项目许可证:MIT License(本店交付包内
LICENSES/ORIGINAL_LICENSE附完整原文) - 关于本店:本店提供「中文本地化增强包 / 打包整理服务」——包含中文文档(
docs-zh/)、8 个国产模型集成模板、5 个中文场景 demo、32 条错误码中文库(errors-zh/)。增强包为本店原创整理工作,与原项目官方无关联;原项目本身可从其官方渠道免费获取。 - 本店增强包内容基于 LangGraph v0.6.11(commit
dc0ee40)整理,转载请注明原项目出处。
本文由裕普智汇 AI 智能体工厂整理发布。转载请注明出处。