1. LangGraph:用图思维重构Agent工作流
LangGraph是LangChain团队开源的图式Agent编排框架,它基于"有向图"模型,将Agent的运行流程抽象为"节点 + 状态流转"。其核心设计理念是用有向状态图(Directed State Graph)来管理复杂的工作流。
1.1 LangGraph核心架构
LangGraph的架构建立在四个核心组件上:
节点(Node):代表一个独立单元,可以是Agent节点(封装独立Agent能力)或Tool节点(调用具体工具)
边(Edge):标注状态流转的决策路径,决定下一步跳转到哪个节点
状态(State):贯穿整个流程,记录数据或交互状态,驱动节点间的流转
图(Graph):LangGraph的核心容器,所有节点、边和状态都定义在其中
1.2 LangGraph与传统工作流框架的比较
特性 | 传统单一Agent | LangGraph多Agent |
---|---|---|
控制流 | 线性、易失控 | 图结构、循环/分支原生支持 |
状态管理 | 会话级记忆 | 持久化快照、可回溯 |
人审干预 | 事后补救 | 任意节点可插人审 |
调试体验 | 黑盒日志 | 节点级可视化、时间旅行 |
2. LangGraph的核心优势
LangGraph相比于传统开发方式具有显著优势:
循环状态管理:支持多轮迭代任务(如RAG结果优化)
持久化引擎:自动保存检查点,宕机后可从断点恢复
人工干预接口:实现关键节点人工审核
可视化调试:降低维护成本,支持时间旅行调试
最重要的是,LangGraph原生支持三个关键功能:
✅ Cyclic Graphs(循环图):支持智能体在不同节点之间循环跳转,方便实现多轮思考与行动
✅ Persistence(状态持久化):图中状态可以随时保存、恢复,支持断点续跑与回溯
✅ Human-in-the-loop(人类参与环节):支持在流程中插入人工确认、反馈或干预节点,增强系统可控性
3. LangGraph实战:构建智能客服工单系统
下面通过一个智能客服工单系统的例子,展示如何使用LangGraph构建实际应用。
3.1 定义状态结构
python
from typing import TypedDict, List from langchain_core.messages import BaseMessage, HumanMessage, AIMessageclass AgentState(TypedDict):messages: List[BaseMessage]intent: str
3.2 创建节点逻辑
python
from langgraph.graph import StateGraph# 创建工单处理工作流 workflow = StateGraph(AgentState)# 意图识别节点 def identify_intent(state):last_message = state["messages"][-1].contentif "账号" in last_message or "登录" in last_message:intent = "technical"else:intent = "general"return {"intent": intent}# 专家处理节点 def call_expert(state):return {"messages": [AIMessage(content="专家处理结果:您的问题需要专业技术支持")]}# 人工审核节点 def human_review(state):return {"messages": [AIMessage(content="人工审核完成:您的问题已处理")]}# 添加三个Agent节点 workflow.add_node("intent_agent", identify_intent) workflow.add_node("expert_agent", call_expert) workflow.add_node("review_agent", human_review)
3.3 构建工作流逻辑
python
# 设置入口点 workflow.set_entry_point("intent_agent")# 条件边:根据意图选择路径 def router(state):if state["intent"] == "technical":return "expert_agent"else:return "review_agent"workflow.add_conditional_edges("intent_agent", router) workflow.add_edge("expert_agent", "review_agent") workflow.set_finish_point("review_agent")# 编译工作流 app = workflow.compile()
3.4 执行工作流
python
# 执行工作流 output = app.invoke({"messages": [HumanMessage(content="我的账号无法登录")]}) print(output["messages"][-1].content)
执行结果:"人工审核完成:您的问题已处理"
程序执行流程为:用户输入 → 意图识别 → 技术问题 → 专家Agent → 人工审核 → 返回结果
4. 生产环境最佳实践
4.1 可视化调试方案
通过graph.get_graph().draw_png()
生成拓扑图,快速验证循环逻辑
4.2 状态持久化实战
python
# 保存状态 checkpoint = graph.get_state(message_id)# 故障恢复 graph.recover_state(checkpoint)
4.3 人工干预设计
在关键节点插入审批机制:
python
def human_approve(state):if state["risk_level"] > 0.8:return "human_review" # 转人工审核return "auto_process"