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 文件：

1
2
3

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

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

1 2	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 只有两个核心装饰器：

1	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 的工作原理：

工作流执行到 interrupt() 时，立即暂停，状态保存到 Checkpointer。
调用方获得中断信息，展示给人类。
人类做出决策后，调用方通过 Command(resume=决策内容) 恢复工作流。
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 存储（单机生产）

1	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 存储（生产推荐）

1	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                           # 迭代计数

原则：

State 越精简越好——每个字段都会被持久化，臃肿的 State = 高存储成本 + 低性能。
用 Annotated[type, reducer] 管理需要追加/累加的字段，避免覆盖。
临时变量不要放 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

原则：

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

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 到生产之间的鸿沟：

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

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

下一步建议：

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

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

参考资源

LangGraph 官方文档

LangGraph GitHub 仓库

LangGraph Functional API 博客

LangGraph Platform GA 公告

LangGraph 中文文档

阅读导航

速览：LangGraph 能力地图

1. 为什么你需要 LangGraph？

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

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

3.1 创建项目与虚拟环境

3.2 安装依赖

3.3 配置 API Key

3.4 验证安装

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

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

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

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

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

4.5 一张图理解全貌

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

5.1 Hello World：最简单的图

5.2 实战：完整的 ReAct Agent

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

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

6.1 核心装饰器

6.2 Hello World：Functional 风格

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

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

6.5 在 Functional API 中调用 Graph API

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

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

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

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

8.1 完整实现

9. 持久化与状态管理

9.1 为什么需要持久化？

9.2 内存存储（开发用）

9.3 SQLite 存储（单机生产）

9.4 PostgreSQL 存储（生产推荐）

9.5 状态查询与回放

10. 生产部署：LangGraph Platform

10.1 LangGraph Platform 的核心能力

10.2 部署步骤

10.3 LangGraph Studio：可视化调试利器

11. 避坑指南与最佳实践

11.1 State 设计原则

11.2 循环必须有出口

11.3 Tool 设计原则

11.4 流式输出 ≠ 简单 print

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

11.6 Token 消耗控制

12. 速查表：Graph API vs Functional API

选择决策树

写在最后

站点导航中心

文章选项

联系我