004|LangGraph 入门:多 Agent 工作流框架
LangGraph 是 LangChain 团队出的多 Agent 工作流框架,它把状态、节点、边三件事封装好,让复杂 Agent 系统变得清晰。
核心结论:LangGraph(读作:兰格·格拉夫,/læŋˈɡræf/)是 LangChain(读作:兰·切恩)团队出的框架,专门用来把多个 LLM 调用或 Agent 组织成一个有状态、有分支、可循环的图。它解决的是复杂多 Agent 系统的流程编排问题。
一、LangGraph 是什么
LangGraph = Lang + Graph。
- Lang 来自 LangChain,代表 LLM 应用。
- Graph 代表图结构,用节点和边组织流程。
一句话
LangGraph 让你用**图(Graph)**的方式,定义一个多步骤、多 Agent 的工作流。
二、LangGraph 和 LangChain 的关系
| LangChain(兰·切恩) | LangGraph(兰格·格拉夫) |
|---|---|
| 做单次 LLM 调用 | 做多个 LLM 调用组成的流程 |
| 像函数 | 像状态机 |
| 链式调用(Linear) | 图式调用(Graph),有分支、循环、并行 |
| 适合简单任务 | 适合复杂多 Agent 系统 |
LangChain 像乐高积木块,LangGraph 像用这些积木搭一个有复杂结构的模型。
三、为什么需要 LangGraph
实际做 Agent 时,流程往往不是直线。
比如数据分析 Agent:
用户问题
↓
意图识别
↓
生成 SQL
↓
执行 SQL
↓
执行成功?
├─ 是 → 归因分析
└─ 否 → 修改 SQL → 重新执行
↓
超过3次 → 返回失败
↓
风控检查
↓
输出
这个流程里有:
- 分支(if / else)
- 循环(重试)
- 多个 Agent 节点
普通 LangChain 的链式调用很难表达这种结构。LangGraph 用图的方式,天然适合。
四、LangGraph 三个核心概念
1. State(斯泰特)—— 状态
State 是全局共享数据,所有节点都能读写。
class AgentState(TypedDict):
question: str # 用户问题
intent: str # 意图
sql: str # 生成的 SQL
result: any # 查询结果
hypotheses: list # 归因假设
risk: dict # 风控结果
final_output: str # 最终输出
retry_count: int # 重试次数
2. Node(诺德)—— 节点
每个节点是一个函数或一个 Agent。
def 数据查询节点(state):
sql = 生成SQL(state["question"])
result = 执行SQL(sql)
return {"sql": sql, "result": result}
节点返回的字典会合并到 State 里。
3. Edge(艾吉)—— 边
边控制流程走向。
workflow.add_edge("意图识别", "数据查询")
workflow.add_conditional_edge("数据查询", 检查SQL状态)
普通边:固定流程
条件边:根据状态决定下一步
五、LangGraph 核心 API
| API(读法) | 作用 |
|---|---|
StateGraph(斯泰特·格拉夫) | 创建图 |
add_node(艾德·诺德) | 添加节点 |
add_edge(艾德·艾吉) | 添加普通边 |
add_conditional_edge(艾德·肯·迪什·艾吉) | 添加条件边 |
set_entry_point(赛特·恩·吹·泡·恩特) | 设置入口节点 |
compile(康·派尔) | 编译图 |
invoke(因·沃克) | 执行图 |
stream(斯·特里姆) | 流式查看中间状态 |
六、完整伪代码示例
from typing import TypedDict
from langgraph.graph import StateGraph, END
# 1. 定义状态
class AgentState(TypedDict):
question: str
intent: str
sql: str
result: any
hypotheses: list
risk_level: str
final_output: str
retry_count: int
# 2. 定义节点函数
def 意图识别节点(state):
intent = 判断意图(state["question"])
return {"intent": intent}
def 数据查询节点(state):
sql = 生成SQL(state["question"])
result = 执行SQL(sql)
return {"sql": sql, "result": result}
def 重试节点(state):
return {"retry_count": state["retry_count"] + 1}
def 归因节点(state):
hypotheses = 生成假设(state["result"])
return {"hypotheses": hypotheses}
def 风控节点(state):
risk_level = 检查风险(state["result"], state["hypotheses"])
return {"risk_level": risk_level}
def 输出节点(state):
output = 生成回复(state)
return {"final_output": output}
# 3. 路由函数
def 检查SQL后路由(state):
if state["result"]["status"] == "成功":
return "归因节点"
elif state["retry_count"] < 3:
return "重试节点"
else:
return "输出节点"
# 4. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("意图识别节点", 意图识别节点)
workflow.add_node("数据查询节点", 数据查询节点)
workflow.add_node("重试节点", 重试节点)
workflow.add_node("归因节点", 归因节点)
workflow.add_node("风控节点", 风控节点)
workflow.add_node("输出节点", 输出节点)
workflow.set_entry_point("意图识别节点")
workflow.add_edge("意图识别节点", "数据查询节点")
workflow.add_conditional_edge("数据查询节点", 检查SQL后路由)
workflow.add_edge("重试节点", "数据查询节点")
workflow.add_edge("归因节点", "风控节点")
workflow.add_edge("风控节点", "输出节点")
workflow.add_edge("输出节点", END)
# 5. 编译并运行
graph = workflow.compile()
result = graph.invoke({
"question": "为什么广东销售额下降?",
"retry_count": 0
})
print(result["final_output"])
七、LangGraph 的循环怎么实现
SQL 执行失败后的重试:
def 检查SQL后路由(state):
if state["result"]["status"] == "成功":
return "归因节点"
if state["retry_count"] < 3:
return "重试节点"
return "输出节点"
workflow.add_conditional_edge("数据查询节点", 检查SQL后路由)
workflow.add_edge("重试节点", "数据查询节点")
流程:
数据查询节点 → 失败 → 重试节点 → 重试次数 +1 → 回到数据查询节点
→ 成功 → 归因节点
→ 重试 3 次还失败 → 输出节点
八、LangGraph 的并行怎么实现
workflow.add_edge("意图识别", "数据查询")
workflow.add_edge("意图识别", "归因分析")
workflow.add_edge("数据查询", "汇总")
workflow.add_edge("归因分析", "汇总")
数据查询和归因分析同时执行,汇总节点等两个都完成后再执行。
九、LangGraph 的调试
LangGraph 可以拿到每个节点的中间状态:
for event in graph.stream({"question": "为什么广东销售额下降?"}):
print(event)
输出每个节点执行后的 state 变化,方便调试。
十、LangGraph 的优缺点
| 优点 | 缺点 |
|---|---|
| 流程可控 | 依赖 LangChain 生态 |
| 调试能力强 | 学习成本较高 |
| 生态成熟 | 节点太多时图会复杂 |
| 适合复杂系统 | 异步支持有限 |
十一、LangGraph 的生态系统
| 工具 | 读法 | 作用 |
|---|---|---|
| LangChain | 兰·切恩 | 基础 LLM、工具、RAG |
| LangGraph | 兰格·格拉夫 | 多 Agent 工作流 |
| LangSmith | 兰·斯密斯 | 监控、调试、评估 |
| LangServe | 兰·瑟夫 | 部署 API 服务 |
这四个通常一起用。
十二、一句话总结
LangGraph 把多个 LLM 调用或 Agent 组织成一个有状态、有分支、可循环的图。它让复杂 Agent 系统的流程变得清晰、可调试、可扩展。
下一步可以读
- 005|主流多 Agent 框架对比:LangGraph / CrewAI / AutoGen / LlamaIndex(已更新)
- 006|Agent 项目工程化:代码组织、测试、部署(待更新)