使用 LangGraph 构建生产级有状态 AI Agent 工作流
解决复杂 Agent 难以控制的问题:通过将 AI 工作流定义为图(Graph)结构,实现精准的状态管理、条件分支路由以及可中断的人机协作模式。
为什么需要这个技能
传统的线性 Chain 或简单的 ReAct Agent 在面对复杂业务逻辑时,往往会出现难以调试、无法精准控制流程跳转或丢失历史状态的问题。
LangGraph 将 Agent 建模为状态机,将每一步操作定义为节点(Node),将跳转逻辑定义为边(Edge)。这种方式让 AI 的执行路径变得可见且可预测,支持在工作流中引入循环(Cycles)以实现自我修正,并能通过 Checkpointer 实现对话状态的持久化,让 Agent 能够像人类一样在长时间任务中“记得”之前的进度。
适用场景
- 复杂多步骤任务:需要根据上一步的结果决定下一步去向的动态路由场景。
- 高可靠性 Agent:需要人工审核(Human-in-the-loop)才能执行敏感操作(如发送邮件、删除数据库)的场景。
- 长程对话系统:需要将对话状态存储在数据库(SQLite/Postgres)中并支持随时恢复的系统。
- 并行处理任务:利用 Map-Reduce 模式同时启动多个研究分支,最后汇总结果。
核心工作流
- 定义状态(State):使用
TypedDict定义 Agent 在节点间传递的全局状态,通过reducer(如add_messages)决定状态如何更新。 - 构建图拓扑(Graph Topology):
- 节点(Nodes):定义具体的处理函数(如调用 LLM、执行工具)。
- 边(Edges):定义节点间的跳转逻辑。
- 条件边(Conditional Edges):根据当前状态决定流向哪个节点。
- 配置持久化(Persistence):集成
SqliteSaver或PostgresSaver,通过thread_id区分不同用户的会话状态。 - 引入人工干预(Interrupts):在关键节点设置
interrupt_before,使程序暂停等待人类确认后再继续执行。
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# 1. 定义状态
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
# 2. 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
# 3. 设置路由逻辑
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")
# 4. 编译并运行
app = graph.compile()
下载和安装
解压后将目录放入你的 AI 工具 skills 文件夹,重启工具后即可使用。具体路径参考内附的 USAGE.zh.md。
你可能还需要
暂无推荐