如果说 LangChain 是”函数调用”，那 LangGraph 就是”状态机”。当你的 AI 工作流需要处理分支、循环、回滚、历史记忆时，LangGraph 就是你需要的武器。
背景：为什么需要 LangGraph？
先看一个真实场景——智能客服工作流：
用户：我要退订套餐
AI：
1. 验证用户身份（身份验证）
├─ 验证成功 → 进入套餐查询
└─ 验证失败 → 转人工
2. 查询当前套餐（查询API）
├─ 是合约套餐 → 提示违约金
└─ 非合约套餐 → 继续
3. 发起退订（退订API）
├─ 退订成功 → 发送确认短信
└─ 退订失败 → 记录工单
4. 更新CRM（CRM API）
5. 发送满意度调查
这个流程：
– 有条件分支（合约套餐 vs 非合约）
– 有失败处理（退订失败要转人工）
– 有状态持久化（每一步的结果要传给下一步）
– 有历史记录（方便客服接管时查看）
用普通代码写，你会发现状态管理越来越乱。LangGraph 就是来解决这个问题的——用状态机的思路来管理 AI 工作流。
LangGraph 核心概念
与 LangChain 的关键区别
| 特性 | LangChain | LangGraph |
|---|---|---|
| 执行模型 | 线性 Chain（A→B→C） | 图结构（可分支、可循环） |
| 状态管理 | 无内置状态 | 内置状态机 + 检查点 |
| 错误处理 | 需手动 try/catch | 内置 conditional 路由 |
| 可视化 | 无 | 内置 Mermaid 可视化 |
| 适用场景 | 简单问答、单一工具调用 | 复杂多步骤对话、Agent |
LangGraph 四件套
| 概念 | 说明 | 类比 |
|---|---|---|
| StateGraph | 图结构容器，定义所有节点和边 | 流程图编辑器 |
| State | 在节点间流转的共享数据对象 | 流程表单 |
| Node | 图中的一个个处理函数 | 流程图中的方框 |
| Reducer | 状态更新规则 | 表单填写规则 |
第1步：搭建基础状态机（智能客服核心）
# smart客服_workflow.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
import operator
# ========== 1. 定义 State（流转的数据结构）==========
class CustomerServiceState(TypedDict):
"""客服工作流的状态结构"""
# messages 是消息历史（append模式的 reducer）
messages: Annotated[list, add_messages]
# 用户信息
user_id: str
phone: str
# 业务状态
identity_verified: bool
current_plan: str | None
is_contract_plan: bool
cancellation_fee: float
# 处理结果
cancellation_success: bool
ticket_id: str | None
error_message: str | None
# 流程控制
next_step: str # 标记下一步去哪
escalation_needed: bool # 是否需要转人工
# ========== 2. 定义 Node（处理函数）==========
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
def verify_identity(state: CustomerServiceState) -> CustomerServiceState:
"""节点1：身份验证"""
print(f"🔐 正在验证用户身份: {state.get('user_id')}")
# 模拟调用身份验证 API
# 实际项目中这里应该是: api.verify_user(state['phone'], state['user_id'])
verified = len(state.get('user_id', '')) > 3
return {
"identity_verified": verified,
"next_step": "end_session" if not verified else "query_plan",
"messages": [AIMessage(content="身份验证" + ("通过" if verified else "失败"))],
}
def query_plan(state: CustomerServiceState) -> CustomerServiceState:
"""节点2：查询当前套餐"""
print(f"📋 查询用户套餐: {state.get('user_id')}")
# 模拟查询套餐
# 实际: plan = api.get_user_plan(state['user_id'])
current_plan = "企业版-年套餐"
is_contract = "年" in current_plan # 年套餐 = 合约套餐
fee = 299.0 if is_contract else 0.0
return {
"current_plan": current_plan,
"is_contract_plan": is_contract,
"cancellation_fee": fee,
"next_step": "handle_cancellation",
"messages": [AIMessage(
content=f"当前套餐：{current_plan}，退订手续费：¥{fee}"
)],
}
def handle_cancellation(state: CustomerServiceState) -> CustomerServiceState:
"""节点3：处理退订逻辑"""
print(f"🚪 发起退订...")
if state.get('is_contract_plan'):
return {
"next_step": "end_session",
"messages": [AIMessage(
content="您当前是合约套餐，退订需要支付违约金 ¥299。"
"如需继续，请回复「确认退订」。"
)],
}
# 模拟退订 API 调用
success = True
if success:
return {
"cancellation_success": True,
"next_step": "update_crm",
"messages": [AIMessage(content="退订成功！正在为您办理后续手续...")],
}
else:
return {
"cancellation_success": False,
"escalation_needed": True,
"next_step": "create_ticket",
"error_message": "退订接口超时",
"messages": [AIMessage(content="系统繁忙，退订失败，已为您创建工单。")],
}
def update_crm(state: CustomerServiceState) -> CustomerServiceState:
"""节点4：更新 CRM 系统"""
print(f"📝 更新 CRM...")
# 模拟 CRM 更新
# api.update_user_status(user_id, "cancelled")
return {
"next_step": "send_confirmation",
"messages": [AIMessage(content="CRM 系统已更新。")],
}
def send_confirmation(state: CustomerServiceState) -> CustomerServiceState:
"""节点5：发送确认短信"""
print(f"📱 发送退订确认短信...")
# 模拟发送短信
# sms.send(state['phone'], "您的套餐已退订成功...")
return {
"next_step": "end_session",
"messages": [AIMessage(
content="退订完成！确认短信已发送至您手机，请注意查收。"
"如有疑问，请回复「人工」转接客服。"
)],
}
def create_ticket(state: CustomerServiceState) -> CustomerServiceState:
"""失败处理：创建工单"""
print(f"🎫 创建转接工单...")
import uuid
ticket_id = f"TKT-{uuid.uuid4().hex[:8].upper()}"
return {
"ticket_id": ticket_id,
"next_step": "end_session",
"messages": [AIMessage(
content=f"工单已创建：{ticket_id}，人工客服将在24小时内联系您。"
)],
}
def end_session(state: CustomerServiceState) -> CustomerServiceState:
"""结束会话"""
print(f"✅ 会话结束")
return {"next_step": END}
# ========== 3. 构建 StateGraph ==========
def build_customer_service_graph():
"""构建客服工作流图"""
# 创建图
graph = StateGraph(CustomerServiceState)
# 注册节点
graph.add_node("verify_identity", verify_identity)
graph.add_node("query_plan", query_plan)
graph.add_node("handle_cancellation", handle_cancellation)
graph.add_node("update_crm", update_crm)
graph.add_node("send_confirmation", send_confirmation)
graph.add_node("create_ticket", create_ticket)
graph.add_node("end_session", end_session)
# 设置入口点
graph.set_entry_point("verify_identity")
# 定义边（条件路由）
graph.add_conditional_edges(
"verify_identity",
lambda state: state["next_step"],
{
"query_plan": "query_plan",
"end_session": "end_session",
}
)
graph.add_conditional_edges(
"query_plan",
lambda state: state["next_step"],
{"handle_cancellation": "handle_cancellation"}
)
# 退订结果分支
def cancellation_router(state: CustomerServiceState) -> str:
if state.get("escalation_needed"):
return "create_ticket"
elif state.get("cancellation_success"):
return "update_crm"
return "end_session"
graph.add_conditional_edges(
"handle_cancellation",
cancellation_router,
{
"update_crm": "update_crm",
"create_ticket": "create_ticket",
"end_session": "end_session",
}
)
graph.add_conditional_edges(
"update_crm",
lambda state: state["next_step"],
{"send_confirmation": "send_confirmation"}
)
graph.add_conditional_edges(
"send_confirmation",
lambda state: state["next_step"],
{"end_session": "end_session"}
)
graph.add_conditional_edges(
"create_ticket",
lambda state: state["next_step"],
{"end_session": "end_session"}
)
# 结束节点
graph.add_edge("end_session", END)
return graph.compile()
# ========== 4. 运行工作流 ==========
if __name__ == "__main__":
# 编译图
workflow = build_customer_service_graph()
# 可视化（Mermaid 格式）
print("📊 工作流图结构：")
print(workflow.get_graph().draw_mermaid())
# 初始状态
initial_state = CustomerServiceState(
user_id="user123",
phone="138****8888",
messages=[
HumanMessage(content="我要退订套餐")
],
identity_verified=False,
current_plan=None,
is_contract_plan=False,
cancellation_fee=0.0,
cancellation_success=False,
ticket_id=None,
error_message=None,
next_step="",
escalation_needed=False,
)
# 运行
print("\n🚀 开始执行工作流：\n")
result = workflow.invoke(initial_state)
print("\n📜 执行结果：")
print(f"- 退订成功: {result['cancellation_success']}")
print(f"- 当前套餐: {result['current_plan']}")
print(f"- 工单号: {result['ticket_id']}")
第2步：添加检查点（状态持久化与恢复）
上面那个例子，如果用户说到一半退出了，下次进来要从头开始，体验很差。LangGraph 的 Checkpoint 功能可以持久化状态：
# checkpoint_workflow.py
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用 SQLite 作为检查点存储（生产环境用 PostgreSQL）
checkpointer = SqliteSaver.from_conn_string(":memory:") # 测试用内存
# 生产环境: checkpointer = SqliteSaver("/data/checkpoints/customerservice.db")
workflow = build_customer_service_graph().checkpointer(checkpointer)
# 首次运行（新建会话）
config = {"configurable": {"thread_id": "session-001"}}
# 第一轮对话
result1 = workflow.invoke(
{
**initial_state,
"messages": [HumanMessage(content="我要退订套餐")]
},
config=config
)
print(result1["messages"][-1].content)
# 输出: 退订成功！正在为您办理后续手续...
# 模拟用户退出后重新进入
print("\n--- 用户重新进入，查看历史状态 ---\n")
# 只传入新消息，保留历史
result2 = workflow.invoke(
{"messages": [HumanMessage(content="好的，谢谢")], "user_id": "user123"},
config=config # 同一个 thread_id！
)
# LangGraph 会自动恢复之前的完整状态
print(result2["messages"][-1].content)
# 输出: 退订完成！确认短信已发送至您手机...
# 查看完整消息历史（包括之前的）
for msg in result2["messages"]:
role = "用户" if isinstance(msg, HumanMessage) else "AI"
print(f"{role}: {msg.content}")
支持多种检查点存储
# PostgreSQL（生产环境推荐）
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@host:5432/langgraph"
)
# Redis（高并发场景）
from langgraph.checkpoint.redis import RedisSaver
checkpointer = RedisSaver.from_url("redis://localhost:6379")
# Memory（单进程测试）
checkpointer = MemorySaver()
第3步：构建 ReAct Agent（带工具调用的状态机）
最强大的用法——把状态机变成 ReAct Agent：
# react_agent_workflow.py
from typing import Literal
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
# ========== 1. 定义工具 ==========
@tool
def get_weather(city: str) -> str:
"""查询城市天气"""
weather_data = {
"北京": "晴，26°C",
"上海": "多云，28°C",
"深圳": "雷阵雨，30°C",
}
return weather_data.get(city, f"{city}的天气数据暂不可用")
@tool
def search_news(keyword: str) -> list[dict]:
"""搜索新闻"""
# 实际项目中调用新闻 API
return [
{"title": f"{keyword}的最新进展", "source": "科技日报", "time": "2小时前"},
{"title": f"{keyword}引发行业讨论", "source": "财经周刊", "time": "5小时前"},
]
@tool
def send_email(to: str, subject: str, body: str) -> dict:
"""发送邮件"""
print(f"📧 发送邮件: {to}")
print(f" 主题: {subject}")
print(f" 内容: {body[:50]}...")
return {"status": "sent", "message_id": "msg-123"}
# 工具列表
tools = [get_weather, search_news, send_email]
# ========== 2. 创建 ReAct Agent ==========
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)
def create_agentic_workflow():
"""创建带 Agent 能力的工作流"""
# 使用预建的 ReAct Agent
agent = create_react_agent(llm_with_tools, tools)
# 包装成 StateGraph
graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("human", human_feedback_node) # 人工审核节点
graph.set_entry_point("agent")
# 条件边：Agent 判断是否需要人工
def should_escalate(state: AgentState) -> str:
if state.get("needs_human"):
return "human"
return END
graph.add_conditional_edges(
"agent",
should_escalate,
{"human": "human", END: END}
)
graph.add_edge("human", "agent") # 人工审核后继续
return graph.compile()
# ========== 3. 完整状态定义 ==========
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
needs_human: bool
final_response: str
def human_feedback_node(state: AgentState) -> AgentState:
"""人工审核节点（实际项目中这里会暂停等待人工确认）"""
print("\n" + "="*50)
print("📋 人工审核请求")
print("="*50)
print(f"最后一条消息: {state['messages'][-1].content}")
print("\n[自动通过 - 演示模式]")
return {
"needs_human": False,
"messages": state["messages"] + [
AIMessage(content="[人工审核已通过]")
]
}
# ========== 4. 运行 ==========
if __name__ == "__main__":
workflow = create_agentic_workflow()
print("📊 Agent 工作流图：")
print(workflow.get_graph().draw_mermaid())
# 复杂任务：多工具协同
result = workflow.invoke({
"messages": [HumanMessage(content=(
"帮我查一下深圳的天气，然后搜索一下深圳最近有什么科技新闻，"
"最后把结果发到 [email protected]"
))],
"needs_human": False,
})
print("\n📜 最终响应：")
print(result["messages"][-1].content)
可视化：查看工作流结构
LangGraph 自带 Mermaid 可视化，运行后输出：
print(workflow.get_graph().draw_mermaid())
输出：
graph TD
A[verify_identity] --> B{验证结果?}
B -->|通过| C[query_plan]
B -->|失败| Z[end_session]
C --> D{合约套餐?}
D -->|是| E[handle_cancellation]
D -->|否| F[update_crm]
E --> G[create_ticket]
F --> H[send_confirmation]
H --> Z
G --> Z
你可以把这个 Mermaid 代码粘贴到 https://mermaid.live 生成可视化图。
性能对比
| 指标 | 原始方案（if-else） | LangGraph |
|---|---|---|
| 代码行数 | 300+ 行 | 150 行 |
| 新增步骤耗时 | 30 分钟 | 5 分钟 |
| 状态调试难度 | 高（要打印变量） | 低（可视化追踪） |
| 支持回滚 | 不支持 | 检查点恢复 |
| 并发处理 | 需自己实现 | thread_id 自动隔离 |
| 测试难度 | 高（mock 复杂） | 节点独立可测试 |
总结
LangGraph 的核心价值：
- 状态机思维：把复杂的 AI 工作流拆成节点和边，可维护性大增
- 内置检查点：用户会话中断没关系，从断点恢复，体验流畅
- 条件路由：复杂的分支逻辑不再是噩梦，路由函数一目了然
- 可视化调试：Mermaid 图直接看出整个流程哪里有问题
适用场景：智能客服、AI 审批流、多步骤数据处理、长期记忆的对话 Agent。
👤 作者简介
一枚在大中原腹地（河南）卖公有云的从业者，主营腾讯云/阿里云/火山云，曾踩坑无数，现专注AI大模型应用落地。关注公众号「AI热点日报」，围观AI前沿动态~
博客：y1cloud.cn
你的项目有没有遇到复杂 AI 工作流难以管理的问题？是用什么方案解决的？欢迎评论区交流！