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

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"

复制代码

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

复制代码

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"
}

复制代码

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"

复制代码

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"
}

复制代码

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

复制代码

# 测试单个节点
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"

复制代码

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