使用 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 模式同时启动多个研究分支,最后汇总结果。

核心工作流

  1. 定义状态(State):使用 TypedDict 定义 Agent 在节点间传递的全局状态,通过 reducer(如 add_messages)决定状态如何更新。
  2. 构建图拓扑(Graph Topology)
    • 节点(Nodes):定义具体的处理函数(如调用 LLM、执行工具)。
    • 边(Edges):定义节点间的跳转逻辑。
    • 条件边(Conditional Edges):根据当前状态决定流向哪个节点。
  3. 配置持久化(Persistence):集成 SqliteSaverPostgresSaver,通过 thread_id 区分不同用户的会话状态。
  4. 引入人工干预(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()

下载和安装

下载 langgraph 中文版 Skill ZIP

解压后将目录放入你的 AI 工具 skills 文件夹,重启工具后即可使用。具体路径参考内附的 USAGE.zh.md

你可能还需要

暂无推荐