图工作流
图工作流是用有向图来组织 Agent 执行流程的编排模式。节点是处理步骤，边是转换规则，支持条件分支、循环和并行。LangGraph 是其典型实现。
#type / concept
#status / evergreen
#tech / ai
#tech / architecture
[!info] related notes
- 所属 MOC: AI Agent Application MOC
- 相关概念: 工作流编排, 状态机
- 框架: LangGraph
图工作流
一句话定义
图工作流是用有向图（Directed Graph）来组织 Agent 执行流程的编排模式。节点（Node）是处理步骤，边（Edge）是转换规则，支持条件分支、循环、并行和子图嵌套。
它解决什么问题
线性流水线（A → B → C）无法表达：
- 条件分支: 根据中间结果选择不同路径
- 循环: 生成 → 评估 → 修改 → 再评估
- 并行: 同时执行多个独立任务
- 回跳: 某个步骤失败后回到之前的步骤重试
图工作流提供了比线性流水线更灵活、比完全自主 Agent 更可控的编排方式。
核心原理
图的基本元素
节点 (Node): 一个处理步骤（LLM 调用、工具执行、条件判断）
边 (Edge): 节点之间的连接（表示执行顺序）
条件边 (Conditional Edge): 根据运行时状态选择下一个节点
入口 (Entry): 图的起始节点
出口 (END): 图的终止标记有向图 vs 有向无环图 (DAG)
| DAG | 有向图（可含环） | |
|---|---|---|
| 循环 | 不允许 | 允许 |
| 适用场景 | 数据管线、ETL | Agent、交互式任务 |
| 典型框架 | Airflow、Prefect | LangGraph |
Agent 需要循环（“搜索 → 发现不够 → 换关键词再搜”），所以 LangGraph 使用有向图而非 DAG。
图 vs 状态机
图工作流和状态机非常接近，但有微妙区别：
| 状态机 | 图工作流 | |
|---|---|---|
| 关注点 | 状态 + 转换 | 节点 + 边 |
| 状态管理 | 显式状态变量 | 图的全局状态 |
| 典型用法 | UI 状态管理、协议 | Agent 编排 |
LangGraph 的 StateGraph 本质上是两者的融合：用图结构定义流程，用共享状态管理数据。
典型工程实现
LangGraph 的核心 API
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
# 1. 定义状态
class WorkflowState(TypedDict):
query: str
results: list[str]
iteration: int
final_answer: str
# 2. 创建图
graph = StateGraph(WorkflowState)
# 3. 添加节点
graph.add_node("search", search_node)
graph.add_node("evaluate", evaluate_node)
graph.add_node("refine", refine_node)
graph.add_node("answer", answer_node)
# 4. 定义边
graph.set_entry_point("search")
graph.add_edge("search", "evaluate")
# 5. 条件边
graph.add_conditional_edges(
"evaluate", # 源节点
lambda state: "answer" if state["iteration"] >= 3 or state["quality"] > 0.8 else "refine",
{
"answer": "answer",
"refine": "refine",
}
)
graph.add_edge("refine", "search") # 回环
graph.add_edge("answer", END)
# 6. 编译
app = graph.compile()并行节点
# 并行执行多个搜索
graph.add_node("search_web", search_web_node)
graph.add_node("search_db", search_db_node)
graph.add_node("search_docs", search_docs_node)
# 并行执行后汇总
graph.add_node("merge_results", merge_node)
# LangGraph 通过 Fan-out/Fan-in 支持并行子图嵌套
# 子图：RAG 子流程
rag_subgraph = StateGraph(RAGState)
rag_subgraph.add_node("retrieve", retrieve_node)
rag_subgraph.add_node("rerank", rerank_node)
rag_subgraph.add_edge("retrieve", "rerank")
# 在主图中使用子图
main_graph.add_node("rag", rag_subgraph.compile())常见设计模式
1. Search-Evaluate-Refine 循环
搜索 → 评估质量 → 不够好就优化查询 → 再搜索。
2. Router 模式
一个路由器节点根据输入类型选择不同的子图。
3. Evaluator-Optimizer 循环
生成器节点 → 评估器节点 → 不满意就回到生成器。
4. Multi-Agent 图
每个 Agent 是一个节点，通过图定义协作关系。
常见坑
- 死循环: 没有最大迭代次数，图可能无限循环
- 状态膨胀: 每个节点都往状态里加数据，状态越来越大
- 条件函数副作用: 条件函数应该只读状态，不应该修改状态
- 不做 checkpoint: 长时间运行的图中断后无法恢复
- 节点粒度太细: 每个 LLM 调用都做成节点，图过于复杂
和其他概念的关系
- vs 状态机: 图工作流是状态机的泛化
- vs 工作流编排: 图工作流是编排的底层模型
- vs LangGraph: LangGraph 是图工作流的 Python 实现
- vs Agent Engine: Engine 的 Agent Loop 可以用图来实现
总结
图工作流是 Agent 编排的最灵活的底层模型。它既能表达确定性的流水线，也能表达带循环的自主 Agent 行为。