跳到主内容
YP

LangGraph 中文教程:从入门到精通的 2026 完全指南

基于 LangGraph 0.6.11 版本编写的中文系统教程,涵盖状态图、Checkpoint 持久化、MiniMax 集成、多 Agent 编排、生产部署等核心概念,5800+ 字完整中文资料。

·5800·首发地址
LangGraph 中文教程LangGraph 入门LangChainAI Agent 编排MiniMax多Agent状态图Checkpoint

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 至少有三个摩擦点:

  1. 英文文档门槛:官方文档默认英文,interrupt / Command(resume=...) 这类 API 第一次接触时需要查多份资料。
  2. 中国 LLM 适配:默认模板以 OpenAI / Anthropic 为主,国内常用的 MiniMax / Qwen / DeepSeek 等需要自行包装。
  3. 本地化 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 TypedDictPydantic 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.7MiniMax-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 502docker 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 .

十一、延伸阅读


资源与声明

  • 原项目langchain-ai/langgraph
  • 原项目许可证:MIT License(本店交付包内 LICENSES/ORIGINAL_LICENSE 附完整原文)
  • 关于本店:本店提供「中文本地化增强包 / 打包整理服务」——包含中文文档(docs-zh/)、8 个国产模型集成模板、5 个中文场景 demo、32 条错误码中文库(errors-zh/)。增强包为本店原创整理工作,与原项目官方无关联;原项目本身可从其官方渠道免费获取。
  • 本店增强包内容基于 LangGraph v0.6.11(commit dc0ee40)整理,转载请注明原项目出处。

本文由裕普智汇 AI 智能体工厂整理发布。转载请注明出处。