LangGraph 开发实战：从零构建生产级 AI Agent 工作流

一句话概括：LangGraph 是 LangChain 生态中用于构建有状态、可持久化、可人机协作的 AI Agent 工作流框架——它不替代 LangChain，而是补上了 LangChain 在"复杂流程编排"上的关键一环。

这篇文章重点帮你解决三类问题：

• 什么时候该从 LangChain 升级到 LangGraph
• 如何把 Graph API / Functional API 用在真实业务流里
• 如何把 Agent 从 demo 推到可部署、可恢复、可观测的生产状态

阅读导航

• 想先建立全局认知：看 第 1～4 章
• 想快速写出可运行 Agent：看 第 5～7 章
• 想做生产落地：重点看 第 8～12 章

速览：LangGraph 能力地图

能力维度	你将收获什么
流程编排	条件分支、循环执行、多 Agent 协作
状态管理	Checkpointer 持久化、断点恢复、线程隔离
人机协作	interrupt + Command 的审批与接管机制
工程化上线	LangGraph Platform 部署、流式输出、可视化调试

---

1. 为什么你需要 LangGraph？

想象一下这个场景——

你用 LangChain 的 LCEL（LangChain Expression Chain）写了一个问答链，跑起来很顺。然后产品经理说：

“能不能加个审核环节？LLM 回答之后先暂停，等人工确认了再继续？”

“能不能让 Agent 循环调用工具，直到它觉得任务完成了才停？”

“能不能记住上一轮对话的状态，断点续跑？”

你试着用 LangChain 的 Chain 串一串……然后发现：Chain 是线性的，没有循环；Memory 是附加的，不是原生的；Human-in-the-loop？对不起，Chain 跑起来就停不下来。

这就是 LangGraph 诞生的原因——

痛点	LangChain	LangGraph
线性流程	✅ Chain 天然支持	✅ 也支持
循环/条件分支	❌ Chain 无原生支持	✅ 图结构天然支持
状态持久化	⚠️ Memory 是外挂模块	✅ Checkpointer 原生内建
断点续跑	❌ 不支持	✅ interrupt + resume
人机协作	❌ 不支持	✅ interrupt + Command
复杂多 Agent	⚠️ 需手动编排	✅ 原生子图/Supervisor

一句话：LangChain 解决"怎么调 LLM"，LangGraph 解决"怎么编排 LLM 的工作流"。

---

2. LangChain vs LangGraph：一张图看清边界

┌─────────────────────────────────────────────────┐
│                  你的 AI 应用                      │
│                                                   │
│  ┌───────────┐    ┌──────────────────────────┐   │
│  │ LangChain  │    │       LangGraph          │   │
│  │            │    │                          │   │
│  │ · Prompt   │    │ · StateGraph / Func API  │   │
│  │ · Model    │    │ · Node / Edge / Router   │   │
│  │ · Tool     │    │ · Checkpointer           │   │
│  │ · Retriever│    │ · interrupt / resume     │   │
│  │ · Output   │    │ · SubGraph / Supervisor  │   │
│  │   Parser   │    │ · Streaming              │   │
│  └─────┬──────┘    └──────────┬───────────────┘   │
│        │                      │                    │
│        └──────┬───────────────┘                    │
│               ▼                                    │
│        ┌──────────────┐                            │
│        │  LLM Provider │  (OpenAI / Anthropic /…)  │
│        └──────────────┘                            │
└─────────────────────────────────────────────────┘

核心区别：

• LangChain 是"积木箱"——提供 Prompt、Model、Tool、Retriever 等组件，让你拼装 LLM 调用链。
• LangGraph 是"流水线引擎"——在 LangChain 积木的基础上，定义谁先跑、谁后跑、什么条件下走哪条路、什么时候暂停等人工。

两者不是替代关系，是互补关系。LangGraph 内部大量使用 LangChain 的 Model 和 Tool 组件——你不会因为学了 LangGraph 就丢掉 LangChain。

---

3. 环境搭建：5 分钟从零开始

3.1 创建项目与虚拟环境

# 创建项目目录
mkdir my-langgraph-project && cd my-langgraph-project

# 创建虚拟环境（Python 3.10+）
python -m venv .venv

# 激活虚拟环境
# Windows:
.venv\Scripts\activate
# macOS/Linux:
source .venv/bin/activate

3.2 安装依赖

# 核心安装（包含 LangChain 集成）
pip install langgraph langchain-openai

# 如果你用 Anthropic 的模型
pip install langgraph langchain-anthropic

# 如果你想用开源模型（如 Ollama 本地部署）
pip install langgraph langchain-community

# 开发调试工具
pip install python-dotenv

3.3 配置 API Key

在项目根目录创建 .env 文件：

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxx
# 如果用 Anthropic
# ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxx

创建 app.py，加载环境变量：

from dotenv import load_dotenv
load_dotenv()  # 加载 .env 中的环境变量

3.4 验证安装

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
response = model.invoke("说一句话证明你是 AI")
print(response.content)

看到回复就说明环境 OK。

---

4. 核心概念全景：State / Node / Edge / Graph

在动手写代码之前，你需要理解 LangGraph 的四大核心概念。别慌，它们比听起来简单得多。

4.1 State（状态）—— 工作流的"共享内存"

State 是一个 TypedDict，定义了工作流中所有节点共享的数据结构。你可以把它理解为一个"共享白板"——每个节点都可以读取白板上的信息，也可以往白板上写新内容。

from typing import Annotated, TypedDict
from langgraph.graph import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 对话消息列表
    next_action: str                          # 下一步动作
    iteration: int                            # 迭代次数

关键点：

• Annotated[list, add_messages] 中的 add_messages 是一个 reducer 函数——它告诉 LangGraph：当多个节点往 messages 里写内容时，追加而不是覆盖。
• 如果不指定 reducer，后写入的值会直接覆盖先写入的值。
• LangGraph 内置了几个常用 reducer：add_messages（追加消息）、add（数值累加）等。

4.2 Node（节点）—— 工作流的"加工站"

Node 就是一个普通 Python 函数，接收当前 State，返回 State 的更新。

def think_node(state: AgentState) -> dict:
    """思考节点：调用 LLM 决定下一步"""
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}  # 返回部分更新，会通过 reducer 合并

def act_node(state: AgentState) -> dict:
    """行动节点：执行工具调用"""
    last_message = state["messages"][-1]
    # 解析工具调用并执行...
    return {"messages": [tool_result], "iteration": state["iteration"] + 1}

关键点：

• Node 函数接收完整的 State，但只返回需要更新的字段。
• 返回的字典会通过 reducer 与现有 State 合并——这就是为什么 add_messages 这么重要。
• Node 可以是同步函数，也可以是 async 函数。

4.3 Edge（边）—— 工作流的"路由规则"

Edge 定义了节点之间的执行顺序。有三种类型：

类型	作用	示例
普通边	A 执行完一定执行 B	graph.add_edge("think", "act")
条件边	A 执行完根据条件决定走 B 还是 C	graph.add_conditional_edges("think", router)
入口边	工作流从哪个节点开始	graph.set_entry_point("think")

条件边是 LangGraph 的精华所在——它让你的工作流可以根据 LLM 的输出动态决定走向：

def router(state: AgentState) -> str:
    """路由函数：根据最后一条消息决定走哪条边"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "act"        # LLM 要求调用工具 → 去执行
    return "end"            # LLM 没有工具调用 → 结束

graph.add_conditional_edges("think", router, {"act": "act", "end": END})

4.4 Graph（图）—— 把一切组装起来

把 State、Node、Edge 组装成一个可运行的图：

from langgraph.graph import StateGraph, END

# 1. 创建图构建器
builder = StateGraph(AgentState)

# 2. 添加节点
builder.add_node("think", think_node)
builder.add_node("act", act_node)

# 3. 设置入口
builder.set_entry_point("think")

# 4. 添加边
builder.add_conditional_edges("think", router, {"act": "act", "end": END})
builder.add_edge("act", "think")  # 执行完工具后回到思考节点（循环！）

# 5. 编译成可运行图
graph = builder.compile()

关键点：builder.compile() 会验证图的完整性（有没有孤立节点、有没有出口），并返回一个可调用的 CompiledGraph 对象。

4.5 一张图理解全貌

      ┌──────────────┐
      │   START      │
      └──────┬───────┘
             │
             ▼
      ┌──────────────┐
┌────▶│   think      │◀────────┐
│     │  (调用 LLM)   │          │
│     └──────┬───────┘          │
│            │                   │
│      ┌─────┴──────┐            │
│      │  条件路由    │            │
│      └─────┬──────┘            │
│       ┌────┴────┐              │
│       ▼         ▼              │
│  有工具调用   无工具调用          │
│       │         │              │
│       ▼         ▼              │
│ ┌──────────┐  ┌───────┐        │
│ │   act    │  │  END  │        │
│ │(执行工具) │  └───────┘        │
│ └────┬─────┘                   │
│      │                         │
└──────┘  循环回到 think           │

这就是经典的 ReAct（Reasoning + Acting）循环——LLM 思考 → 决定是否调工具 → 调完工具再思考 → 直到得出最终答案。

---

5. Graph API 实战：从 Hello World 到 ReAct Agent

5.1 Hello World：最简单的图

先来一个"Hello World"级别的例子，感受一下 Graph API 的基本用法：

from typing import TypedDict
from langgraph.graph import StateGraph, END

# 1. 定义 State
class HelloWorldState(TypedDict):
    greeting: str
    name: str
    result: str

# 2. 定义节点
def greet_node(state: HelloWorldState) -> dict:
    """生成问候语"""
    return {"greeting": f"你好，{state['name']}！"}

def format_node(state: HelloWorldState) -> dict:
    """格式化最终输出"""
    return {"result": f"🎉 {state['greeting']} 欢迎来到 LangGraph 的世界！"}

# 3. 构建图
builder = StateGraph(HelloWorldState)
builder.add_node("greet", greet_node)
builder.add_node("format", format_node)
builder.set_entry_point("greet")
builder.add_edge("greet", "format")
builder.add_edge("format", END)

graph = builder.compile()

# 4. 运行
result = graph.invoke({"name": "开发者", "greeting": "", "result": ""})
print(result["result"])
# 🎉 你好，开发者！欢迎来到 LangGraph 的世界！

解析：

• State 包含 3 个字段，初始值在 invoke 时传入。
• 每个节点只返回需要更新的字段，LangGraph 自动合并。
• 图的流向是线性的：greet → format → END。

5.2 实战：完整的 ReAct Agent

接下来构建一个真正有用的 Agent——一个能调用工具、循环思考的 ReAct Agent。这是 LangGraph 最经典的用例。

import json
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph import add_messages
from langgraph.prebuilt import ToolNode

# ── 1. 定义工具 ──

@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    # 实际项目中这里会调用天气 API
    weather_data = {
        "北京": "晴天，气温 18°C，空气质量良好",
        "上海": "多云转阴，气温 22°C，有轻度雾霾",
        "深圳": "阵雨，气温 28°C，湿度 85%",
    }
    return weather_data.get(city, f"抱歉，暂无{city}的天气数据")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式，例如 '2 + 3 * 4'"""
    try:
        result = eval(expression)  # 生产环境请用安全的方式！
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{e}"

@tool
def search_knowledge(query: str) -> str:
    """搜索知识库获取相关信息"""
    knowledge_base = {
        "Python": "Python 是一种解释型、面向对象的高级编程语言，由 Guido van Rossum 于 1991 年发布。",
        "LangGraph": "LangGraph 是 LangChain 生态中用于构建有状态 AI Agent 工作流的框架，支持循环、持久化和人机协作。",
        "RAG": "RAG（检索增强生成）是一种结合外部知识检索与 LLM 生成的技术，能有效减少幻觉。",
    }
    for key, value in knowledge_base.items():
        if key.lower() in query.lower():
            return value
    return f"未找到关于'{query}'的相关知识"

# 工具列表
tools = [search_weather, calculate, search_knowledge]

# ── 2. 定义 State ──

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# ── 3. 定义节点 ──

# 创建 LLM 并绑定工具
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)

def think(state: AgentState) -> dict:
    """思考节点：调用 LLM，决定是否使用工具"""
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 使用预构建的 ToolNode（自动处理工具调用）
tool_node = ToolNode(tools)

# ── 4. 定义路由 ──

def should_use_tool(state: AgentState) -> str:
    """判断 LLM 是否要求调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

# ── 5. 构建图 ──

builder = StateGraph(AgentState)

builder.add_node("think", think)
builder.add_node("tools", tool_node)

builder.set_entry_point("think")

# 条件边：think 之后判断是否需要调用工具
builder.add_conditional_edges(
    "think",
    should_use_tool,
    {"tools": "tools", "end": END}
)

# 工具执行完后回到 think，形成循环
builder.add_edge("tools", "think")

graph = builder.compile()

# ── 6. 运行 ──

from langchain_core.messages import HumanMessage

result = graph.invoke({
    "messages": [HumanMessage(content="北京今天天气怎么样？顺便帮我算一下 18 * 3 + 7")]
})

# 打印最终回答
for msg in result["messages"]:
    if hasattr(msg, "content") and msg.content:
        print(f"[{msg.type}] {msg.content[:200]}")

运行流程解析：

用户提问 → think（LLM 决定调用 search_weather 和 calculate）
         → tools（并行执行两个工具）
         → think（LLM 汇总工具结果，生成最终回答）
         → end（没有更多工具调用，结束）

这就是 ReAct 循环的威力——Agent 会自动循环思考→行动→再思考，直到任务完成。

5.3 流式输出：让用户不再干等

生产环境中，用户最讨厌的就是等。LangGraph 支持流式输出，让 LLM 的回答"打字机式"呈现：

# 流式输出 token
for chunk in graph.stream(
    {"messages": [HumanMessage(content="解释一下什么是 RAG")]},
    stream_mode="messages"  # 逐 token 流式
):
    if chunk and hasattr(chunk[0], "content"):
        print(chunk[0].content, end="", flush=True)

# 流式输出节点更新
for event in graph.stream(
    {"messages": [HumanMessage(content="深圳天气如何？")]},
    stream_mode="updates"  # 每个节点完成后立即输出
):
    print(f"\n节点更新: {list(event.keys())}")

stream_mode	输出内容	适用场景
"values"	每步之后的完整 State	调试
"updates"	每个节点的增量更新	监控进度
"messages"	逐 token 流式	前端打字机效果

---

6. Functional API 实战：装饰器风格的优雅写法

如果你觉得 Graph API 的 add_node / add_edge / add_conditional_edges 太"声明式"了，LangGraph 还提供了 Functional API——一种更接近普通 Python 函数的写法。

6.1 核心装饰器

Functional API 只有两个核心装饰器：

from langgraph.func import entrypoint, task

装饰器	作用	类比
@task	标记一个离散的工作单元	Graph API 中的 Node
@entrypoint	标记工作流的入口函数	Graph API 中的入口节点 + 编译

6.2 Hello World：Functional 风格

import uuid
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def is_even(number: int) -> bool:
    """判断数字是否为偶数"""
    return number % 2 == 0

@task
def format_message(even: bool) -> str:
    """格式化输出消息"""
    return "这是一个偶数 ✅" if even else "这是一个奇数 ❌"

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: dict) -> str:
    """工作流入口：串联多个任务"""
    result = is_even(inputs["number"]).result()
    return format_message(result).result()

# 运行
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
output = workflow.invoke({"number": 42}, config=config)
print(output)  # 这是一个偶数 ✅

关键点：

• @task 标记的函数返回的是 Future，需要调用 .result() 获取结果。
• 这种设计是为了支持并行执行——你可以先启动多个 task，再统一收集结果。
• @entrypoint 必须传入 checkpointer，这样工作流才具备持久化能力。

6.3 并行执行：Functional API 的杀手级特性

Functional API 最优雅的地方在于并行执行——不需要 Send 或多条边，Python 原生写法即可：

@task
def generate_paragraph(topic: str) -> str:
    """根据主题生成一段文字"""
    from langchain_openai import ChatOpenAI
    model = ChatOpenAI(model="gpt-4o-mini")
    response = model.invoke([
        {"role": "system", "content": "你是一个科普作家，写生动有趣的科普段落。"},
        {"role": "user", "content": f"写一段关于{topic}的科普。"}
    ])
    return response.content

@entrypoint(checkpointer=checkpointer)
def parallel_writer(topics: list[str]) -> str:
    """并行生成多个主题的科普段落"""
    # 同时启动所有任务（不阻塞）
    futures = [generate_paragraph(topic) for topic in topics]
    # 统一收集结果
    paragraphs = [f.result() for f in futures]
    return "\n\n---\n\n".join(paragraphs)

# 运行：5 个主题并行生成
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = parallel_writer.invoke(
    ["量子计算", "黑洞", "DNA 双螺旋", "深度学习", "核聚变"],
    config=config
)
print(result)

对比 Graph API 中实现并行需要 Send 对象和额外配置，Functional API 的写法自然得多。

6.4 聊天机器人：短期记忆 + 解耦保存

Functional API 中处理"记忆"的方式很巧妙——previous 参数自动获取上次保存的状态：

from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
checkpointer = InMemorySaver()

@task
def call_model(messages: list[BaseMessage]) -> BaseMessage:
    """调用 LLM"""
    return model.invoke(messages)

@entrypoint(checkpointer=checkpointer)
def chatbot(
    inputs: list[BaseMessage],
    *,                       # 以下为关键字参数
    previous: list[BaseMessage]  # 上次保存的对话历史
) -> BaseMessage:
    # 如果有历史消息，拼接起来
    if previous:
        inputs = add_messages(previous, inputs)

    # 调用 LLM
    response = call_model(inputs).result()

    # 关键：解耦返回值与保存值
    # value → 返回给调用者（只返回本次回复）
    # save  → 保存到检查点（保存完整对话历史，下次通过 previous 获取）
    return entrypoint.final(
        value=response,                          # 用户只看到本次回复
        save=add_messages(inputs, response)      # 检查点保存完整历史
    )

entrypoint.final(value=..., save=...) 的精妙之处：

• value 是用户看到的返回值——通常只返回本次 LLM 的回复。
• save 是保存到检查点的值——通常是完整对话历史，确保下次调用时 previous 能拿到完整上下文。
• 这种"解耦"设计让返回值和持久化值各司其职，避免混淆。

6.5 在 Functional API 中调用 Graph API

两种 API 可以混用！在 Functional API 的 entrypoint 内部，你可以调用用 Graph API 编译的图：

from langgraph.graph import StateGraph, END

# 用 Graph API 构建子图
class MathState(TypedDict):
    expression: str
    result: float

def compute(state: MathState) -> dict:
    return {"result": eval(state["expression"])}

sub_builder = StateGraph(MathState)
sub_builder.add_node("compute", compute)
sub_builder.set_entry_point("compute")
sub_builder.add_edge("compute", END)
math_graph = sub_builder.compile()

# 在 Functional API 中调用
@entrypoint(checkpointer=checkpointer)
def workflow(expr: str) -> dict:
    result = math_graph.invoke({"expression": expr, "result": 0.0})
    return {"expression": result["expression"], "result": result["result"]}

---

7. 进阶：条件路由与人机协作

7.1 条件路由：让工作流"会思考"

条件路由是 LangGraph 区别于普通 Chain 的核心能力。一个实际例子——智能客服路由：

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

class ServiceState(TypedDict):
    messages: Annotated[list, add_messages]
    category: str          # 问题分类
    resolved: bool         # 是否已解决

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def classify(state: ServiceState) -> dict:
    """分类节点：判断用户问题属于哪个类别"""
    last_msg = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="你是一个问题分类器。将用户问题分为：technical（技术问题）、billing（账单问题）、general（一般咨询）。只返回类别名称。"),
        HumanMessage(content=last_msg)
    ])
    return {"category": response.content.strip().lower()}

def handle_technical(state: ServiceState) -> dict:
    """处理技术问题"""
    response = model.invoke([
        SystemMessage(content="你是技术支持专家，请详细解答用户的技术问题。"),
        HumanMessage(content=state["messages"][-1].content)
    ])
    return {"messages": [response], "resolved": True}

def handle_billing(state: ServiceState) -> dict:
    """处理账单问题"""
    response = model.invoke([
        SystemMessage(content="你是账单支持专员，请帮助用户解决账单相关问题。如果涉及退款，请提示需要人工审核。"),
        HumanMessage(content=state["messages"][-1].content)
    ])
    return {"messages": [response], "resolved": True}

def handle_general(state: ServiceState) -> dict:
    """处理一般咨询"""
    response = model.invoke([
        SystemMessage(content="你是友好的客服代表，请回答用户的一般咨询。"),
        HumanMessage(content=state["messages"][-1].content)
    ])
    return {"messages": [response], "resolved": True}

# 路由函数
def route_by_category(state: ServiceState) -> str:
    return state["category"]

# 构建图
builder = StateGraph(ServiceState)
builder.add_node("classify", classify)
builder.add_node("technical", handle_technical)
builder.add_node("billing", handle_billing)
builder.add_node("general", handle_general)

builder.set_entry_point("classify")
builder.add_conditional_edges(
    "classify",
    route_by_category,
    {
        "technical": "technical",
        "billing": "billing",
        "general": "general"
    }
)
builder.add_edge("technical", END)
builder.add_edge("billing", END)
builder.add_edge("general", END)

service_graph = builder.compile()

# 测试
result = service_graph.invoke({
    "messages": [HumanMessage(content="我的订阅费用为什么多扣了 50 块？")],
    "category": "",
    "resolved": False
})
print(result["messages"][-1].content)

7.2 Human-in-the-Loop：让 AI 知道何时该"请教人类"

有些决策不能让 AI 自己做——比如退款审批、敏感操作、法律建议。LangGraph 的 interrupt 机制让工作流可以暂停执行，等待人类输入后再继续。

from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

class ApprovalState(TypedDict):
    messages: Annotated[list, add_messages]
    refund_amount: float
    approved: bool

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
checkpointer = InMemorySaver()

def assess_refund(state: ApprovalState) -> dict:
    """评估退款请求"""
    last_msg = state["messages"][-1].content
    response = model.invoke([
        SystemMessage(content="你是退款评估专员。根据用户描述判断退款金额。只返回数字。"),
        HumanMessage(content=last_msg)
    ])
    try:
        amount = float(response.content.strip())
    except ValueError:
        amount = 0.0
    return {"refund_amount": amount}

def human_approval(state: ApprovalState) -> dict:
    """人工审批环节——工作流在此暂停！"""
    # interrupt 会暂停工作流，将信息发送给人类
    decision = interrupt(
        f"⚠️ 退款审批请求\n"
        f"退款金额：¥{state['refund_amount']:.2f}\n"
        f"请确认是否批准此退款？（批准/拒绝）"
    )
    # 当人类通过 Command(resume=...) 恢复时，decision 就是人类的输入
    approved = "批准" in decision or "同意" in decision
    return {"approved": approved}

def process_refund(state: ApprovalState) -> dict:
    """处理退款结果"""
    if state["approved"]:
        msg = f"✅ 退款 ¥{state['refund_amount']:.2f} 已批准，预计 3-5 个工作日到账。"
    else:
        msg = "❌ 退款请求已被拒绝。如有疑问请联系客服主管。"
    return {"messages": [AIMessage(content=msg)]}

# 构建图
builder = StateGraph(ApprovalState)
builder.add_node("assess", assess_refund)
builder.add_node("approve", human_approval)
builder.add_node("process", process_refund)

builder.set_entry_point("assess")
builder.add_edge("assess", "approve")
builder.add_edge("approve", "process")
builder.add_edge("process", END)

approval_graph = builder.compile(checkpointer=checkpointer)

# ── 运行：第一阶段（执行到 interrupt 暂停）──

import uuid
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# 第一次调用：会执行到 human_approval 节点时暂停
for event in approval_graph.stream(
    {
        "messages": [HumanMessage(content="我购买的课程和描述不符，要求退款 299 元")],
        "refund_amount": 0.0,
        "approved": False
    },
    config=config,
    stream_mode="updates"
):
    print(event)
# 输出类似：{'approve': {'__interrupt__': (...)}}

# ── 运行：第二阶段（人类审批后恢复）──

# 人类做出决策后，通过 Command(resume=...) 恢复工作流
for event in approval_graph.stream(
    Command(resume="批准退款"),  # 人类的决策
    config=config,              # 同一个 thread_id
    stream_mode="updates"
):
    print(event)
# 输出：{'process': {'messages': [AIMessage(content='✅ 退款 ¥299.00 已批准...')]}}

interrupt 的工作原理：

1. 工作流执行到 interrupt() 时，立即暂停，状态保存到 Checkpointer。
2. 调用方获得中断信息，展示给人类。
3. 人类做出决策后，调用方通过 Command(resume=决策内容) 恢复工作流。
4. interrupt() 的返回值就是 resume 传入的内容——工作流从暂停处继续执行。

这就是 Human-in-the-Loop 的核心——AI 能自主完成大部分工作，但在关键决策点上，由人类拍板。

---

8. 进阶：多 Agent 协作——Supervisor 模式

当任务复杂到单个 Agent 难以应对时，你需要多个 Agent 协作。最常见的模式是 Supervisor（主管）模式：一个"主管 Agent"负责分配任务，多个"专家 Agent"各自处理自己擅长的领域。

             ┌─────────────────┐
             │   Supervisor    │
             │  (任务分配器)     │
             └───────┬─────────┘
                     │
       ┌─────────────┼─────────────┐
       │             │             │
       ▼             ▼             ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Researcher │ │   Coder    │ │  Reviewer  │
│  (研究员)   │ │  (程序员)   │ │  (审核员)   │
└────────────┘ └────────────┘ └────────────┘

8.1 完整实现

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI

# ── 1. 定义共享 State ──

class TeamState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str      # 下一个要执行的 Agent
    task_description: str  # 任务描述

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 2. 定义 Supervisor ──

def supervisor(state: TeamState) -> dict:
    """主管：决定把任务分配给谁"""
    response = model.invoke([
        SystemMessage(content="""你是一个项目主管。根据任务内容，决定下一步由谁处理：

- researcher：负责信息调研、资料收集、事实核查
- coder：负责编写代码、技术方案设计
- reviewer：负责审核成果、提出修改意见、质量把关

当任务已经完成时，返回 'FINISH'。

只返回 Agent 名称或 'FINISH'，不要返回其他内容。"""),
        HumanMessage(content=f"当前任务：{state['task_description']}\n\n对话历史：{state['messages'][-3:]}")
    ])

    next_agent = response.content.strip().lower()
    if next_agent not in ["researcher", "coder", "reviewer", "finish"]:
        next_agent = "researcher"  # 默认交给研究员

    return {"next_agent": next_agent}

# ── 3. 定义专家 Agent ──

def researcher(state: TeamState) -> dict:
    """研究员：负责调研"""
    response = model.invoke([
        SystemMessage(content="你是资深研究员。请针对任务进行深入调研，提供详实的背景信息和数据支撑。"),
        HumanMessage(content=state['task_description'])
    ])
    return {
        "messages": [AIMessage(content=f"📋 研究报告：\n{response.content}")],
        "next_agent": "supervisor"
    }

def coder(state: TeamState) -> dict:
    """程序员：负责编码"""
    response = model.invoke([
        SystemMessage(content="你是高级程序员。请根据需求和调研结果，编写高质量的代码实现。"),
        HumanMessage(content=f"任务：{state['task_description']}\n\n参考资料：{state['messages'][-1].content if state['messages'] else ''}")
    ])
    return {
        "messages": [AIMessage(content=f"💻 代码实现：\n{response.content}")],
        "next_agent": "supervisor"
    }

def reviewer(state: TeamState) -> dict:
    """审核员：负责审核"""
    response = model.invoke([
        SystemMessage(content="你是严格的审核员。请审核代码和调研内容，指出问题和改进建议。如果质量合格，请说'审核通过'。"),
        HumanMessage(content=f"请审核以下成果：\n{state['messages'][-1].content}")
    ])
    return {
        "messages": [AIMessage(content=f"🔍 审核意见：\n{response.content}")],
        "next_agent": "supervisor"
    }

# ── 4. 路由函数 ──

def route_to_agent(state: TeamState) -> str:
    """根据 supervisor 的决策路由到对应 Agent"""
    next_agent = state["next_agent"]
    if next_agent == "finish":
        return "end"
    if next_agent == "supervisor":
        return "supervisor"
    return next_agent

# ── 5. 构建图 ──

builder = StateGraph(TeamState)

builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_node("reviewer", reviewer)

builder.set_entry_point("supervisor")

# Supervisor → 各专家 Agent
builder.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "supervisor": "supervisor",
        "end": END
    }
)

# 各专家 Agent 完成后回到 Supervisor
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
builder.add_edge("reviewer", "supervisor")

team_graph = builder.compile()

# ── 6. 运行 ──

result = team_graph.invoke({
    "messages": [HumanMessage(content="帮我调研并实现一个基于 RAG 的文档问答系统")],
    "next_agent": "",
    "task_description": "调研并实现一个基于 RAG 的文档问答系统"
})

# 打印完整对话
for msg in result["messages"]:
    print(f"\n{'='*60}")
    print(msg.content[:500])

运行流程：

supervisor → researcher（调研 RAG 技术）
          → supervisor → coder（编写代码）
          → supervisor → reviewer（审核质量）
          → supervisor → FINISH

---

9. 持久化与状态管理

9.1 为什么需要持久化？

在生产环境中，Agent 的执行可能持续数分钟甚至数小时。如果中途中断（网络超时、服务重启），没有持久化就意味着一切从头来。LangGraph 的 Checkpointer 机制解决了这个问题——每一步的状态都被自动保存。

9.2 内存存储（开发用）

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

• 优点：零配置，开发调试极方便
• 缺点：重启即丢失，不支持多进程共享
• 适用：本地开发、单元测试

9.3 SQLite 存储（单机生产）

pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

# 同步版本
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke(inputs, config=config)

# 异步版本（推荐用于 Web 服务）
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    result = await graph.ainvoke(inputs, config=config)

9.4 PostgreSQL 存储（生产推荐）

pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

pool = ConnectionPool(conninfo="postgresql://user:pass@localhost:5432/mydb")
checkpointer = PostgresSaver(pool)
checkpointer.setup()  # 首次使用时创建表

graph = builder.compile(checkpointer=checkpointer)

• 优点：支持多进程/多机器共享、数据持久化、生产就绪
• 适用：正式部署环境

9.5 状态查询与回放

有了 Checkpointer，你可以查询历史执行记录、回放甚至"时光倒流"：

# 查询某个 thread 的所有检查点
checkpoints = list(checkpointer.list(config))

# 获取最新状态
latest = checkpointer.get(config)
print(latest.values)  # 当前 State 快照

# 从某个历史检查点重新执行（时光倒流！）
from langgraph.checkpoint import Checkpoint
# 指定 checkpoint_id 即可从任意历史点恢复
config_with_checkpoint = {
    "configurable": {
        "thread_id": thread_id,
        "checkpoint_id": checkpoint_id  # 某个历史检查点的 ID
    }
}
result = graph.invoke(None, config=config_with_checkpoint)

---

10. 生产部署：LangGraph Platform

当你把 Agent 开发完成，下一步就是部署。LangGraph Platform 是 LangChain 官方提供的托管服务，专为 LangGraph 应用设计。

10.1 LangGraph Platform 的核心能力

能力	说明
一键部署	langgraph deploy 命令即可上云
自动持久化	无需自己搭 PostgreSQL，平台内置
流式 API	原生支持 SSE/WebSocket 流式输出
Cron 调度	定时触发 Agent 执行
多租户	支持多用户隔离的 thread 管理
Studio 可视化	浏览器中实时查看 Agent 执行图

10.2 部署步骤

第一步：创建 langgraph.json 配置文件

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./app.py:graph"
  },
  "env": ".env"
}

这个配置告诉 LangGraph Platform：

• dependencies：依赖来源（当前目录的 pyproject.toml 或 requirements.txt）
• graphs：图的位置（文件路径:变量名）
• env：环境变量文件

第二步：本地测试

# 安装 CLI
pip install langgraph-cli

# 本地启动开发服务器
langgraph dev

# 测试通过后部署
langgraph deploy

第三步：调用部署后的 API

from langgraph_sdk import get_client

# 连接到部署的 LangGraph 应用
client = get_client(url="https://your-app.langgraph.com")

# 创建一个 thread
thread = await client.threads.create()

# 发送消息并流式获取响应
async for chunk in client.runs.stream(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "你好！"}]},
    stream_mode="messages"
):
    print(chunk.data)

10.3 LangGraph Studio：可视化调试利器

LangGraph Studio 是一个浏览器端的可视化工具，能让你看到 Agent 的执行过程：

• 实时显示当前执行到哪个节点
• 每个节点的输入/输出一目了然
• 支持在 interrupt 节点手动输入恢复数据
• 可以回放历史执行记录

对于调试复杂 Agent 工作流，Studio 的价值无法替代。

---

11. 避坑指南与最佳实践

踩过的坑不想让你再踩。以下是在生产环境中总结出的实战经验。

11.1 State 设计原则

❌ 错误做法：把所有东西都塞进 State

class BadState(TypedDict):
    messages: Annotated[list, add_messages]
    raw_html: str          # 几百 KB 的 HTML
    temp_counter: int      # 临时变量
    debug_log: list        # 调试日志

✅ 正确做法：State 只保留必要字段，用 reducer 管理聚合

class GoodState(TypedDict):
    messages: Annotated[list, add_messages]  # 有 reducer，追加而非覆盖
    query: str                               # 当前查询
    iteration: int                           # 迭代计数

原则：

1. State 越精简越好——每个字段都会被持久化，臃肿的 State = 高存储成本 + 低性能。
2. 用 Annotated[type, reducer] 管理需要追加/累加的字段，避免覆盖。
3. 临时变量不要放 State——用闭包或函数参数传递。

11.2 循环必须有出口

LangGraph 的循环图（如 ReAct Agent）最常见的一个 bug 是无限循环——Agent 不停调工具，永远不停下来。

✅ 加上迭代限制：

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration: int  # 迭代计数器

MAX_ITERATIONS = 10

def should_continue(state: AgentState) -> str:
    """路由函数：判断是否继续"""
    if state["iteration"] >= MAX_ITERATIONS:
        return "end"  # 达到上限，强制结束
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"
    return "end"

11.3 Tool 设计原则

❌ 错误做法：一个工具做所有事

@tool
def do_everything(query: str) -> str:
    """搜索天气、计算、发邮件、写代码……"""
    pass

✅ 正确做法：一个工具做一件事，描述清晰

@tool
def search_weather(city: str) -> str:
    """查询指定城市的当前天气信息。参数：city - 城市名称，如'北京'"""
    pass

@tool
def calculate(expression: str) -> str:
    """计算数学表达式的值。参数：expression - 数学表达式，如'2+3*4'"""
    pass

原则：

1. 工具的 docstring 就是 LLM 的使用说明——写清楚功能、参数含义、使用场景。
2. 单一职责——工具粒度越细，LLM 选择越准确。
3. 返回格式统一——每个工具都返回字符串，包含成功/失败信息。

11.4 流式输出 ≠ 简单 print

生产环境中，流式输出要配合前端框架使用：

# FastAPI + LangGraph 流式输出示例
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(message: str, thread_id: str):
    async def event_generator():
        async for event in graph.astream_events(
            {"messages": [HumanMessage(content=message)]},
            config={"configurable": {"thread_id": thread_id}},
            version="v2"
        ):
            kind = event["event"]
            if kind == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {token}\n\n"  # SSE 格式
    return StreamingResponse(event_generator(), media_type="text/event-stream")

11.5 错误处理：永远不要假设 LLM 输出是正确的

from langchain_core.messages import AIMessage

def safe_tool_execution(state: AgentState) -> dict:
    """安全的工具执行节点"""
    last_msg = state["messages"][-1]

    if not hasattr(last_msg, "tool_calls") or not last_msg.tool_calls:
        return {"messages": [AIMessage(content="没有检测到工具调用。")]}

    results = []
    for tool_call in last_msg.tool_calls:
        try:
            # 根据 tool_call.name 执行对应工具
            tool_fn = tool_map.get(tool_call["name"])
            if tool_fn is None:
                results.append(
                    ToolMessage(
                        content=f"未知工具：{tool_call['name']}",
                        tool_call_id=tool_call["id"]
                    )
                )
                continue
            result = tool_fn.invoke(tool_call["args"])
            results.append(
                ToolMessage(content=str(result), tool_call_id=tool_call["id"])
            )
        except Exception as e:
            # 工具执行失败时返回错误信息，而不是抛异常
            results.append(
                ToolMessage(
                    content=f"工具执行错误：{str(e)}",
                    tool_call_id=tool_call["id"]
                )
            )

    return {"messages": results}

11.6 Token 消耗控制

# 设置递减的 iteration_limit，防止 token 消耗失控
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    remaining_steps: int  # 剩余步数，每步减 1

def think(state: AgentState) -> dict:
    if state["remaining_steps"] <= 0:
        return {"messages": [AIMessage(content="已达最大步数限制，请简化您的问题。")]}
    # 正常逻辑...
    return {"messages": [response], "remaining_steps": state["remaining_steps"] - 1}

---

12. 速查表：Graph API vs Functional API

最后，一张表总结两种 API 的区别与选择策略：

维度	Graph API	Functional API
编程范式	声明式（定义节点和边）	命令式（写普通函数）
定义方式	StateGraph + add_node + add_edge	@entrypoint + @task
适合场景	复杂状态机、循环图、条件路由	线性/简单分支、快速改造现有代码
学习成本	较高，需理解图结构	低，接近普通 Python 函数
核心优势	灵活定义复杂拓扑	最小代码改动获得四大能力
并行执行	Send 对象或多条边	列表推导 + .result()
状态管理	显式 State TypedDict + reducer	previous 参数 + entrypoint.final()
人机协作	interrupt() + Command(resume=...)	同样支持
重试/缓存	节点级别配置	@task(retry_policy=...) / @task(cache_policy=...)
互通性	✅ 共享底层运行时	✅ 可在 entrypoint 内调用编译后的图
可视化	✅ Studio 可视化执行图	⚠️ 有限支持

选择决策树

你的工作流是线性的吗？
├── 是 → Functional API（简单、快速）
└── 否 → 需要条件路由或循环吗？
    ├── 否 → Functional API（简单分支用 if/else 即可）
    └── 是 → Graph API（条件路由是它的主场）

需要从现有代码快速改造吗？
├── 是 → Functional API（加几个装饰器就行）
└── 否 → Graph API（从零开始设计更灵活）

需要多 Agent 协作吗？
├── 是 → Graph API（SubGraph + Supervisor）
└── 否 → 看个人偏好，两者都可以

黄金法则：不确定就用 Graph API——它功能最全，能覆盖所有场景。Functional API 是"便捷通道"，适合已熟悉 Graph API 后简化代码。

---

写在最后

LangGraph 的核心价值不在于"又一个 LLM 框架"，而在于它解决了 LLM 应用从 demo 到生产之间的鸿沟：

1. 循环执行——Agent 不再是"跑一次就完"，而是可以持续思考、行动、反思。
2. 状态持久化——再也不怕中途崩溃，断点续跑是标配。
3. 人机协作——AI 能自主完成大部分工作，关键决策由人类拍板。
4. 生产就绪——从 Checkpointer 到 LangGraph Platform，部署链路完整。

从 Graph API 的 StateGraph + add_node + add_edge 出发，到 Functional API 的 @entrypoint + @task 简化写法，再到 interrupt 人机协作、Supervisor 多 Agent 协作、LangGraph Platform 一键部署——你已经掌握了 LangGraph 开发的完整路径。

下一步建议：

1. 用本文的 ReAct Agent 示例作为模板，替换成你自己的工具，跑通第一个 Agent。
2. 尝试加入 interrupt，体验 Human-in-the-Loop 的威力。
3. 用 langgraph dev 本地启动，配合 LangGraph Studio 可视化调试。
4. 准备好上线时，langgraph deploy 一键部署。

编程的快乐，在于创造。LangGraph 给了你创造 AI Agent 的积木——现在，搭你自己的吧。

---

参考资源

• LangGraph 官方文档
• LangGraph GitHub 仓库
• LangGraph Functional API 博客
• LangGraph Platform GA 公告
• LangGraph 中文文档