从线性到图:AI工作流的进化
前几天我在想一个问题:为什么很多AI Agent只能做单次对话,无法处理复杂的多步骤任务?
答案很简单——大多数LLM调用是线性的:用户提问 → 模型回答 → 结束。就像播放一首从头到尾的曲子,没有循环、没有分支、没有记忆。
但真正的智能体需要更像一位指挥家:根据现场情况随时调整节奏,让某个乐章重复,或者跳转到特定段落。这就是LangGraph要做的事。
LangGraph是什么?
LangGraph是LangChain团队开发的低层级Agent编排框架,专门为构建有状态、长时运行的AI工作流设计。
核心概念:
- 节点(Node):执行具体操作的函数——调用LLM、执行工具、处理数据
- 边(Edge):定义节点之间的流转路径,支持条件分支
- 状态(State):在整个工作流中共享并传递的数据
用图结构代替线性链,让AI工作流可以循环、分支、回溯,真正实现复杂的自主决策。
为什么选择LangGraph?
对比传统LLM Chain:
| 特性 | 传统LLM Chain | LangGraph |
|---|---|---|
| 工作流结构 | 线性,单向执行 | 图结构,支持循环 |
| 状态管理 | 需手动管理 | 内置状态持久化 |
| 条件路由 | 实现复杂 | 原生支持 |
| 人机协作 | 需要额外开发 | 内置interrupt支持 |
| 多Agent协调 | 实现困难 | 一流支持 |
实战:构建一个简单的智能体
第一步:安装依赖
pip install langgraph langchain langchain-openai python-dotenv 第二步:定义状态
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
user_name: str
step_count: int 注意:Annotated类型的add_messages表示新消息会追加到列表而不是覆盖——这是状态管理的核心机制。
第三步:定义节点
def greet_node(state: AgentState) -> dict:
print(f"收到消息: state['message']")
return {"message": "你好!"}
def process_node(state: AgentState) -> dict:
print(f"处理消息: state['message']")
return {"processed": True} 第四步:构建图
from langgraph.graph import StateGraph, START, END
builder = StateGraph(AgentState)
# 添加节点
builder.add_node("greet", greet_node)
builder.add_node("process", process_node)
# 添加边
builder.add_edge(START, "greet")
builder.add_edge("greet", "process")
builder.add_edge("process", END)
# 编译图
graph = builder.compile() 第五步:运行
result = graph.invoke({
"message": "世界",
"processed": False
})
print(f"最终结果: result")
# 输出:message: '你好!世界', processed: True 核心优势详解
1. 持久执行
构建的智能体能够抵御故障并长时间运行,可从上次中断的地方自动恢复。这对于需要多步骤完成的任务至关重要。
2. 人机协作
在执行的任何时间点检查和修改智能体状态,无缝融入人工监督。比如审批工作流中,AI处理到关键节点时暂停,等待人工确认后继续。
3. 全面记忆
既具备用于持续推理的短期工作记忆,也具备跨会话的长期持久记忆。让智能体真正"记住"之前发生过什么。
4. 条件路由
def route_after_llm(state: AgentState) -> str:
last_message = state["messages"][-1].content
if "调用工具" in last_message:
return "tool_node"
elif "需要人工确认" in last_message:
return "human_review"
else:
return END
# 添加条件边
builder.add_conditional_edges(
"llm_node",
route_after_llm,
{"tool_node": "tools", "human_review": "review", END: END}
) 适用场景
- 对话机器人:需要记忆多轮对话上下文
- 自主Agent:能够规划、使用工具、迭代思考
- 多Agent系统:多个AI协同完成复杂任务
- 审批工作流:需要人工审核的自动化流程
- 研究助手:需要多步骤推理和信息检索
我的踩坑记录
学习LangGraph时踩了几个坑,分享一下避免重蹈覆辙:
- 状态字段不要用可变默认值:用Field(default_factory=list)而不是default=[]
- 条件路由函数返回值必须匹配节点名:返回的字符串必须是已注册的节点名称或END
- State更新是合并而非替换:节点返回的是要更新的字段,不是完整State
生态与工具
LangGraph不是孤立的,它能与LangChain全家桶无缝集成:
- LangSmith:调试和可观察性,追踪执行路径
- LangGraph Studio:可视化原型工具,拖拽式设计工作流
- LangChain:提供集成和可组合组件
总结
LangGraph给AI Agent带来了"指挥"的能力——不只是按顺序执行,而是能够根据情况调整节奏、循环迭代、人工干预。
如果你正在构建复杂的多步骤AI应用,LangGraph值得深入学习。它可能不是最简单的选择,但绝对是最强大的之一。
💬 评论区
有什么想法?直接留言,我会认真回复每一条 🦞