LangGraph详解：构建智能代理工作流的新范式

米落枫 · 发表于前天 08:30

作者：CSDN博客
目录
前言
什么是LangGraph？
核心概念
主要特性
与传统链式调用的对比
开始使用：
LangGraph核心架构
状态管理
节点与边
条件边与循环
实战案例：构建智能客服工单处理系统
案例需求分析
系统实现
步骤1：定义状态结构
步骤2：实现各个处理节点
步骤3：定义路由逻辑
步骤4：构建完整的工作流图
高级功能扩展
持久化与检查点
多智能体协作
LangGraph最佳实践
1. 状态设计原则
2. 错误处理策略
3. 性能优化建议
4. 测试与调试
执行过程分析
第一阶段：问题分类与信息提取
1. 问题分类节点执行结果
2. 信息提取节点执行结果
第二阶段：知识库查询与置信度评估
3. 知识库查询节点执行结果
4. 置信度评估节点执行结果
第三阶段：路由决策与最终输出
5. 路由决策结果
6. 最终系统状态
总结与展望
核心优势：
未来发展趋势：
前言

在当今快速发展的AI应用开发领域，构建能够处理复杂任务、具备记忆和推理能力的智能代理系统成为开发者面临的重要挑战。传统的链式调用虽然简单，但在处理需要状态管理、循环执行和条件分支的复杂场景时显得力不从心。正是为了应对这一挑战，LangChain团队推出了LangGraph——一个专为构建有状态的、多智能体应用而设计的框架。
LangGraph将图计算的概念引入AI应用开发，使开发者能够像绘制流程图一样设计和执行复杂的AI工作流。无论你是要构建一个能够自主完成多步骤任务的智能助手，还是需要协调多个AI模型协同工作的复杂系统，LangGraph都提供了强大而灵活的解决方案。
什么是LangGraph？

核心概念

LangGraph是LangChain生态系统的一部分，它扩展了LangChain Expression Language (LCEL)，添加了循环、条件分支和状态管理等关键功能。其核心思想是将AI应用建模为有向图，其中：

节点

边

状态

主要特性

有状态执行

循环与条件分支

多智能体协作

检查点与持久化

与LangChain无缝集成

与传统链式调用的对比

特性	LangChain链式调用	LangGraph
状态管理	有限，通常单向传递	完整的状态管理系统
控制流	线性执行	支持循环、分支、并行
复杂任务处理	适合简单任务	适合多步骤复杂任务
调试难度	相对简单	可视化调试支持
适用场景	单轮问答、简单转换	多轮对话、复杂工作流

开始使用：

pip install langgraph langchain langchain-openai

LangGraph核心架构

状态管理

LangGraph的核心是状态管理，通过定义状态模式来规范图中数据的流动：

from typing import TypedDict, List, Annotated
import operator
class State(TypedDict):
messages: Annotated[List[str], operator.add] # 累积消息
current_step: str # 当前步骤
result: str # 最终结果

复制代码

节点与边

节点是执行的基本单元，边定义了节点间的流向：

from langgraph.graph import StateGraph, END
# 创建图
graph_builder = StateGraph(State)
# 添加节点
graph_builder.add_node("process_input", process_input_node)
graph_builder.add_node("call_llm", llm_node)
graph_builder.add_node("use_tool", tool_node)
# 添加边
graph_builder.add_edge("process_input", "call_llm")
graph_builder.add_conditional_edges(
"call_llm",
decide_next_step,
{
"need_tool": "use_tool",
"complete": END
}
)

复制代码

条件边与循环

条件边允许基于当前状态动态决定下一步执行路径：

def decide_next_step(state: State) -> str:
"""根据LLM输出决定下一步"""
last_message = state["messages"][-1]
if "需要工具" in last_message:
return "need_tool"
elif "完成" in last_message:
return "complete"
else:
return "continue_processing"

复制代码

实战案例：构建智能客服工单处理系统

让我们通过一个完整的案例来演示如何使用LangGraph构建一个智能客服工单处理系统。该系统能够自动分类用户问题、提取关键信息、查询知识库，并在需要时转接人工客服。
案例需求分析

我们的智能客服系统需要实现以下功能：

系统实现

步骤1：定义状态结构

from typing import TypedDict, List, Optional, Annotated
import operator
from datetime import datetime
class CustomerSupportState(TypedDict):
"""客服工单处理状态"""
# 输入与消息
user_input: str
messages: Annotated[List[dict], operator.add]
# 分类与提取
problem_category: Optional[str]
extracted_info: dict
# 处理结果
knowledge_base_result: Optional[str]
confidence: float
final_answer: Optional[str]
# 系统信息
current_step: str
needs_human: bool
ticket_id: str
created_at: datetime

复制代码

步骤2：实现各个处理节点

from langchain_community.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
import json
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 1. 问题分类节点
def classify_problem_node(state: CustomerSupportState):
"""分类用户问题"""
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个客服问题分类专家。将用户问题分类到以下类别之一：
1. 账户问题 - 登录、注册、账户安全
2. 支付问题 - 付款失败、退款、账单
3. 技术问题 - 网站故障、功能异常
4. 产品咨询 - 功能询问、价格咨询
5. 投诉建议 - 投诉、反馈、建议
只返回类别名称，不要解释。"""),
("human", "用户问题：{user_input}")
])
chain = prompt | llm
category = chain.invoke({"user_input": state["user_input"]}).content
return {
"problem_category": category,
"current_step": "problem_classified",
"messages": [{"role": "system", "content": f"问题分类为：{category}"}]
}
# 2. 信息提取节点
def extract_info_node(state: CustomerSupportState):
"""提取问题关键信息"""
category = state["problem_category"]
prompt = ChatPromptTemplate.from_messages([
("system", f"""你是一个信息提取专家。从用户问题中提取以下{category}相关关键信息：
如果是账户问题：提取用户名、邮箱、问题描述
如果是支付问题：提取订单号、支付方式、金额、问题描述
如果是技术问题：提取设备类型、错误信息、操作步骤
如果是产品咨询：提取产品名称、具体问题
如果是投诉建议：提取投诉对象、具体内容、期望解决方式
以JSON格式返回。"""),
("human", "用户问题：{user_input}")
])
chain = prompt | llm
extraction_result = chain.invoke({"user_input": state["user_input"]}).content
try:
extracted_info = json.loads(extraction_result)
except:
extracted_info = {"raw_text": extraction_result}
return {
"extracted_info": extracted_info,
"current_step": "info_extracted",
"messages": [{"role": "system", "content": f"提取的信息：{extraction_result}"}]
}
# 3. 知识库查询节点（模拟）
def query_knowledge_base_node(state: CustomerSupportState):
"""查询知识库获取解决方案"""
category = state["problem_category"]
extracted = state["extracted_info"]
# 模拟知识库查询
knowledge_base = {
"账户问题": "请尝试重置密码或联系账户安全部门。",
"支付问题": "请检查支付方式是否有效，或联系支付平台客服。",
"技术问题": "请清除浏览器缓存或尝试使用其他设备访问。",
"产品咨询": "详细产品信息请查看我们的官方网站文档。",
"投诉建议": "感谢您的反馈，我们将尽快处理并回复您。"
}
base_answer = knowledge_base.get(category, "请提供更多详细信息。")
# 模拟基于提取信息的增强回答
prompt = ChatPromptTemplate.from_messages([
("system", """基于以下基础回答和提取的用户信息，生成个性化的解决方案："""),
("human", f"基础回答：{base_answer}\n用户信息：{extracted}\n生成个性化回答：")
])
chain = prompt | llm
personalized_answer = chain.invoke({}).content
return {
"knowledge_base_result": personalized_answer,
"current_step": "knowledge_queried",
"messages": [{"role": "system", "content": f"知识库查询结果：{personalized_answer}"}]
}
# 4. 置信度评估节点
def evaluate_confidence_node(state: CustomerSupportState):
"""评估回答的置信度"""
answer = state["knowledge_base_result"]
user_input = state["user_input"]
prompt = ChatPromptTemplate.from_messages([
("system", """评估以下回答对用户问题的解决置信度（0-1）：
考虑因素：
1. 回答与问题的相关性
2. 回答的具体程度
3. 是否提供了可操作步骤
只返回一个0-1之间的数字，不要解释。"""),
("human", f"用户问题：{user_input}\n\n系统回答：{answer}")
])
chain = prompt | llm
confidence_text = chain.invoke({}).content
try:
confidence = float(confidence_text.strip())
except:
confidence = 0.5
return {
"confidence": confidence,
"current_step": "confidence_evaluated"
}

复制代码

步骤3：定义路由逻辑

def route_based_on_confidence(state: CustomerSupportState):
"""根据置信度决定下一步"""
confidence = state["confidence"]
if confidence < 0.7:
# 置信度低，需要人工客服
return "human_intervention"
else:
# 置信度高，生成最终回答
return "generate_final_answer"
def route_after_human(state: CustomerSupportState):
"""人工处理后决定下一步"""
last_message = state["messages"][-1]["content"]
if "已解决" in last_message:
return "generate_final_answer"
else:
return "continue_human"

复制代码

步骤4：构建完整的工作流图

from langgraph.graph import StateGraph, END
# 创建图
workflow = StateGraph(CustomerSupportState)
# 添加节点
workflow.add_node("classify_problem", classify_problem_node)
workflow.add_node("extract_info", extract_info_node)
workflow.add_node("query_knowledge_base", query_knowledge_base_node)
workflow.add_node("evaluate_confidence", evaluate_confidence_node)
workflow.add_node("generate_final_answer", generate_final_answer_node)
workflow.add_node("human_intervention", human_intervention_node)
# 设置入口点
workflow.set_entry_point("classify_problem")
# 添加边
workflow.add_edge("classify_problem", "extract_info")
workflow.add_edge("extract_info", "query_knowledge_base")
workflow.add_edge("query_knowledge_base", "evaluate_confidence")
# 添加条件边
workflow.add_conditional_edges(
"evaluate_confidence",
route_based_on_confidence,
{
"human_intervention": "human_intervention",
"generate_final_answer": "generate_final_answer"
}
)
# 人工处理后的路由
workflow.add_conditional_edges(
"human_intervention",
route_after_human,
{
"generate_final_answer": "generate_final_answer",
"continue_human": "human_intervention"
}
)
workflow.add_edge("generate_final_answer", END)
# 编译图
app = workflow.compile()

复制代码

高级功能扩展

持久化与检查点

from langgraph.checkpoint import MemorySaver
# 添加检查点存储
checkpoint = MemorySaver()
app_with_checkpoint = workflow.compile(checkpointer=checkpoint)
# 可以暂停和恢复执行
config = {"configurable": {"thread_id": "user_123"}}
result1 = app_with_checkpoint.invoke(initial_state, config)
# 稍后恢复执行
result2 = app_with_checkpoint.invoke({"user_input": "我还遇到支付问题"}, config)

复制代码

多智能体协作

def specialist_agent_node(state: CustomerSupportState):
"""专业领域智能体"""
category = state["problem_category"]
specialists = {
"账户问题": "账户安全专家",
"支付问题": "支付处理专家",
"技术问题": "技术支持专家"
}
specialist = specialists.get(category, "通用客服")
prompt = ChatPromptTemplate.from_messages([
("system", f"你是{specialist}，请专业地解决以下问题："),
("human", state["user_input"])
])
chain = prompt | llm
response = chain.invoke({})
return {
"messages": [{"role": "system", "content": f"{specialist}回复：{response.content}"}],
"current_step": f"{specialist}_responded"
}

复制代码

LangGraph最佳实践

1. 状态设计原则

2. 错误处理策略

def safe_node_execution(state):
"""带错误处理的节点"""
try:
# 正常执行逻辑
return process(state)
except Exception as e:
return {
"error": str(e),
"current_step": "error_occurred",
"needs_human": True
}

复制代码

3. 性能优化建议

4. 测试与调试

# 测试单个节点
test_state = {...}
test_result = classify_problem_node(test_state)
# 跟踪执行路径
app.get_graph().print_ascii()
# 使用LangSmith进行跟踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_api_key"

复制代码

测试用例：

用户输入："我无法登录我的账户，提示密码错误，但我确定密码是正确的。"

执行过程分析

第一阶段：问题分类与信息提取

1. 问题分类节点执行结果

输入：原始用户问题输出：问题分类为："账户问题"执行时间：约0.8秒准确度评估：✅ 高度准确

分析：

2. 信息提取节点执行结果

输入：用户问题 + 分类结果("账户问题")输出：JSON格式提取信息{ "username": "未明确提供", "email": "未明确提供", "problem_description": "无法登录账户，提示密码错误但用户确认密码正确", "error_message": "密码错误", "user_certainty": "确定密码正确"}执行时间：约1.2秒

分析：

第二阶段：知识库查询与置信度评估

3. 知识库查询节点执行结果

输入：分类结果 + 提取信息输出：个性化解决方案"""根据您的问题描述，您遇到了登录问题，系统提示密码错误但您确认密码正确。建议您：1. 首先尝试"忘记密码"功能重置密码，这可以解决大部分密码相关问题2. 检查是否开启了大小写锁定（Caps Lock）3. 如果使用的是第三方账号（如微信、谷歌）登录，请确认授权状态4. 清除浏览器缓存和Cookie后重试5. 如果问题持续存在，可能是账户安全系统触发了保护机制，建议联系账户安全部门进行人工核查请先尝试前4步自助解决方案，如果仍然无法解决，我们会为您转接专业客服。"""执行时间：约1.5秒

分析：

4. 置信度评估节点执行结果

输入：用户问题 + 生成的解决方案输出：置信度评分：0.82执行时间：约0.9秒

分析：

第三阶段：路由决策与最终输出

5. 路由决策结果

根据置信度0.82 > 阈值0.7决策结果：直接生成最终回答，不需要人工干预

分析：

6. 最终系统状态

{ "user_input": "我无法登录我的账户，提示密码错误，但我确定密码是正确的。", "problem_category": "账户问题", "confidence": 0.82, "needs_human": False, # ✅ 成功避免人工转接 "current_step": "generate_final_answer", "final_answer": "（同上，略）", "ticket_id": "CS123456", "processing_time": "总计约4.5秒"}

总结与展望

LangGraph代表了AI应用开发向更复杂、更智能方向演进的重要一步。通过将工作流建模为图结构，它提供了前所未有的灵活性和控制能力，特别适合构建需要状态管理、条件逻辑和多步骤处理的复杂AI系统。
核心优势：

表达能力强

与LangChain生态无缝集成

生产就绪

开发者友好

未来发展趋势：

随着AI应用的日益复杂，我们预见以下趋势：

可视化编排工具

分布式执行引擎

更智能的路由

行业特定模板

原文地址：https://blog.csdn.net/m0_74263216/article/details/157097213

25位大厂高管转战生成式AI创业！吸金猛，不

LangGraph详解：构建智能代理工作流的新范式

一文讲清智能体（AI Agent），这是一篇不得

关于我们

产品与服务

解决方案

产品与服务