LangGraph 第2章:适用场景
LangGraph 作为一个灵活的图计算框架,适用于多种 LLM 应用场景。本章详细介绍各个典型场景及其对应的图结构设计。
对话机器人
对话机器人是最常见的 LLM 应用之一。使用 LangGraph 构建对话机器人的优势在于:
- 状态持久化:通过 Checkpointer 保存对话历史
- 多轮上下文:MesssageState 自动管理消息列表
- 流式输出:支持逐 token 流式响应
from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI
# 使用内置的 MessagesState 管理对话历史
class ChatState(MessagesState):
pass # messages 字段由 MessagesState 自动管理
def chat_node(state: ChatState):
llm = ChatOpenAI(model="gpt-4")
# 自动使用 messages 作为上下文
response = llm.invoke(state["messages"])
return {"messages": [response]}
# 构建对话图
builder = StateGraph(ChatState)
builder.add_node("chat", chat_node)
builder.set_entry_point("chat")
builder.add_edge("chat", "__end__")
对话机器人的典型图结构为单节点循环模式:
用户输入 -> [Chat Node] -> 输出回答
^
| (消息历史自动累积)
自主 Agent
自主 Agent(Autonomous Agent)能够自主推理、调用工具、观察结果并决定下一步行动。这是 LangGraph 最擅长的领域。
典型场景包括:
- 代码生成 Agent:理解需求 -> 编写代码 -> 运行测试 -> 修复问题
- 研究助手:分析问题 -> 搜索资料 -> 整理报告 -> 输出结论
- 数据分析 Agent:连接数据库 -> 编写 SQL -> 执行查询 -> 生成可视化
def reasoning_node(state: AgentState):
"""推理节点:决定下一步行动"""
decision = llm.invoke(f"当前状态: {state}. 下一步做什么?")
if "search" in decision:
return {"action": "search"}
elif "code" in decision:
return {"action": "write_code"}
else:
return {"action": "respond"}
自主 Agent 的图结构通常包含一个循环:
+----------+
| Reasoning|<----+
+-----+----+ |
| |
+--------+--------+|
| | ||
+-----v--+ +--v---+ +--v-+---+
| Search | | Code | | Respond |
+-----+--+ +--+---+ +----+----+
| | |
+-------+-----------+
|
[END]
多 Agent 协作
将复杂任务拆解为多个子任务,每个子任务由专门的 Agent 负责。
典型场景:
| 场景 | 涉及的 Agent | 协作模式 |
|---|---|---|
| 文章写作 | 研究员 + 写手 + 评审 | Supervisor 调度 |
| 客服系统 | 意图识别 + 知识库 + 工单 | 路由分发 |
| 代码审查 | 静态分析 + 安全检查 + 风格检查 | 并行执行 |
| 数据分析 | 数据清洗 + 分析 + 可视化 | 流水线式 |
# Supervisor 模式
def supervisor(state):
# 决定下一步调用哪个 Agent
decision = llm.invoke(f"下一步任务: {state}")
return Command(goto=decision["next_agent"])
复杂工作流
企业级应用经常需要复杂的工作流,LangGraph 的图结构天然适合这种需求:
审批流程
# 审批工作流示例
def review_node(state):
"""自动审核节点"""
if state["amount"] > 10000:
return {"needs_approval": True} # 需要人工审批
else:
return {"needs_approval": False, "approved": True}
def auto_approve(state):
"""自动通过"""
return {"status": "approved", "message": "自动审批通过"}
# 条件路由
graph.add_conditional_edges(
"review",
lambda s: "human_approval" if s["needs_approval"] else "auto_approve"
)
代码生成管道
# 代码生成管道
def generate_code(state): ...
def run_tests(state): ...
def fix_bugs(state): ...
def deploy(state): ...
# 构建管道
graph.add_edge("generate_code", "run_tests")
graph.add_conditional_edges(
"run_tests",
lambda s: "fix_bugs" if s["tests_failed"] else "deploy"
)
graph.add_edge("fix_bugs", "run_tests") # 修复后重新测试
流程引擎与规则系统
LangGraph 还可以用作流程引擎,执行业务规则和流程编排:
- 客户 onboarding 流程:注册 -> 验证 -> 配置 -> 激活
- 订单处理流程:下单 -> 支付 -> 库存 -> 发货
- 内容审核流程:提交 -> 自动审核 -> 人工审核 -> 发布
场景选择指南
选择合适的架构模式可以参考以下决策树:
| 场景复杂度 | 推荐架构 | 示例 |
|---|---|---|
| 简单问答 | 单节点 | 基础 QA Bot |
| 多轮对话 | 单节点 + Memory | 客服 Chatbot |
| 需要工具调用 | ReAct Agent | 代码生成 Agent |
| 需要人工审批 | Human-in-the-Loop | 审批工作流 |
| 复杂多步骤 | 多节点流水线 | 数据处理管道 |
| 多 Agent | Supervisor 模式 | 文章写作系统 |
| 超复杂系统 | 分层 Subgraph | 企业级 AI 平台 |
根据实际需求选择合适的架构,不要过度设计。简单的场景用简单的结构,复杂的场景再引入 LangGraph 的高级特性。