别再只盯着 LangChain 了：从执行模型源码看 LangChain/LangGraph、LlamaIndex、CrewAI、AutoGen 的真实分水岭
引言
先说一个我去年踩过的坑。
团队要做一个"合同审阅 Agent"：解析 PDF → 抽取条款 → 对照法规库检索 → 逐条给出风险评级 → 生成审阅报告。第一版我们用 LangChain 的 AgentExecutor + 一堆 Tool 拼出来，Demo 跑得飞快，老板很满意。上线两周后问题来了：
- 审阅到第 3 条时模型突然"忘记"前面已经查过哪条法规，重复检索，token 烧得飞快；
- 某次法规库 API 超时，整个流程直接抛出
MaxIterationsExceeded，前面 20 分钟的解析全白干； - 我们想让"抽取 Agent"和"评级 Agent"并行跑，发现
AgentExecutor的 ReAct 循环是串行文本驱动的，根本插不进去。
痛定思痛，我把四个框架的源码翻了一遍，最后用 LangGraph 重写，问题 2、3 迎刃而解。那次之后我形成一个观点：选 Agent 框架，别看它有多少 Star、多少 Tool 集成，要看它的"执行模型"——也就是"控制流由谁来编排"。这个差异，比任何 API 表面差异都致命。
这篇文章就从这个角度切入，带你从源码层面看清四个框架的分水岭。
核心概念：控制流到底归谁管？
先上一个生活化类比。
想象你开一家餐厅，厨房里有一群厨师（LLM + Tool）：
- LangChain 的 AgentExecutor：像一个"传菜员"，他拿着顾客的原始需求，跑到后厨问大厨"下一步干啥？"，大厨说"切菜"，传菜员就去切菜窗口；切完再跑回来问"下一步？"，大厨说"下锅"…… 每一步都要回到大厨那里问一次，而且传菜员脑子里只有"大厨刚说的那句话"，没有全局菜单。这就是 ReAct 循环的本质：文本驱动的单步决策。
- LangGraph：像一张"厨房流程图"贴在墙上，每个工位（Node）负责一道工序，工序之间用箭头（Edge）连起来，还有"如果汤咸了就回锅"这种条件分支（Conditional Edge）。控制流被显式地画出来了，状态放在中央备餐台（State）上，谁都能看。
- LlamaIndex：像一本"索引目录 + 标准菜谱"。它最擅长的是"根据关键词快速翻到对应菜谱页"（检索），然后按菜谱步骤做。它的 Workflow 是事件驱动的，比 LangGraph 更"松耦合"，适合以"数据/文档"为中心的流程。
- CrewAI：像一个"餐厅管理团队"，有店长（Manager）、服务员（Role）、后厨（Task），店长负责分派任务、检查成果。它把"角色 + 任务 + 协作"作为一等公民，本质是"多 Agent 编排层"。
- AutoGen：像一个"开放式厨房"，几个厨师围着一口锅，谁有想法谁说话（GroupChat），或者两两结对（Two-Agent）。它把"对话"作为协作原语，控制流是"涌现"出来的，而不是预先画好的。
用一张表先钉住四者的执行模型差异：
| 框架 | 控制流归属 | 状态管理 | 协作原语 | 典型心智模型 |
|---|---|---|---|---|
| LangChain (AgentExecutor) | 框架内部循环 | 隐式（在 scratchpad 文本里） | Tool 调用 | 单 Agent + 工具 |
| LangGraph | 开发者显式编排 | 显式 State（TypedDict/Pydantic） | 图节点/边 | 状态机 |
| LlamaIndex (Workflow) | 事件触发 | Context（可持久化） | Event | 事件驱动流水线 |
| CrewAI | Manager/流程 | Task 上下文 | Role + Task | 组织架构 |
| AutoGen | 对话协议 | 消息历史 | 消息/GroupChat | 多智能体对话 |
下面进入源码深水区。
源码/原理深度分析
1. LangChain AgentExecutor：为什么它天生"健忘"且"难并行"
我们直接看 langchain 包里 AgentExecutor 的核心循环（简化自 libs/langchain/langchain/agents/agent.py）：
# 简化版 AgentExecutor._call 核心逻辑
async def _call(self, inputs, run_manager=None):
# 1. 初始化中间步骤记录（scratchpad）
intermediate_steps = []
# 2. 进入 ReAct 循环，最多 max_iterations 次
for i in range(self.max_iterations):
# 2.1 用 agent 的 plan 方法决定下一步动作
# 注意：这里把 intermediate_steps 序列化成文本喂给 LLM
action = await self._agent.plan(
intermediate_steps, callbacks, **inputs
)
# 2.2 如果 agent 认为可以结束了，就返回
if isinstance(action, AgentFinish):
return action.return_values
# 2.3 否则执行工具，把结果 append 回 intermediate_steps
observation = await self._tool_run(action, ...)
intermediate_steps.append((action, observation))
raise MaxIterationsExceeded(...) # 循环耗尽直接抛异常关键点有三个，全是从这段代码里能读出来的：
第一，"状态"其实就是 intermediate_steps 这个 list。 而它被 _agent.plan 序列化成一段文本（Thought/Action/Observation 的格式）塞进 prompt。这意味着：所谓"记忆"，本质是把历史对话拼成字符串。上下文窗口一满，早期的 Observation 就被截断——这就是我遇到的"忘记查过哪条法规"的根因。它不是 bug，是设计使然。
第二，循环是串行且"黑盒"的。 for i in range(max_iterations) 里每一步都强依赖上一步的 observation。你想让"抽取"和"检索"并行？在这个循环里没有位置。max_iterations 耗尽直接抛异常，中间产物不落盘，这就是我"20 分钟白干"的原因。
第三，plan 的决策权完全在 LLM 手里。 框架不知道"下一步该干什么"，它只知道"问 LLM"。这种文本驱动的架构优点是灵活（LLM 想怎么走怎么走），缺点是不可预测、不可测试、不可恢复。
一句话总结：
AgentExecutor是一个"while 循环 + LLM 做 if-else"的朴素实现。它能跑通 Demo，但撑不起生产级的可观测性和可恢复性。
2. LangGraph：把"隐式循环"变成"显式状态机"
LangGraph 的核心抽象就三个：StateGraph、Node、Edge。看它的编译和执行逻辑（简化自 langgraph/graph/state.py 和 pregel 执行器）：
# 简化版 StateGraph 定义与编译
class StateGraph:
def __init__(self, state_schema):
self.schema = state_schema # 状态结构，如 TypedDict
self.nodes = {} # 节点：name -> callable
self.edges = [] # 普通边
self.conditional_edges = {} # 条件边
def add_node(self, name, action):
self.nodes[name] = action
def add_edge(self, start, end):
self.edges.append((start, end))
def add_conditional_edges(self, source, path_fn, mapping):
# path_fn(state) 返回一个 key，mapping[key] 决定下一个节点
self.conditional_edges[source] = (path_fn, mapping)
def compile(self):
# 编译成 Pregel 执行器（超步/BSP 模型）
return PregelExecutor(self.nodes, self.edges, self.conditional_edges, self.schema)执行器用的是 Pregel/BSP（Bulk Synchronous Parallel）模型：每一"超步"里，所有活跃节点并行执行，产出对 State 的增量更新，然后用 reducer 合并。这是它和 AgentExecutor 最本质的区别——状态是显式的、可合并的、可持久化的。
# 状态合并的 reducer 语义（示意）
from typing import Annotated
from operator import add
class ReviewState(TypedDict):
clauses: list[str]
# Annotated + reducer：每次节点返回的新 list 会与旧值 add 合并
findings: Annotated[list[dict], add]
# 不带 reducer 的字段：默认"后写覆盖"
report: str这个 reducer 机制是精髓。你在一个超步里并行跑"抽取条款""检索法规""风控校验"三个节点，它们各自只返回自己那部分增量，执行器负责合并。并发天然被支持，因为节点之间通过 State 解耦，而不是通过 observation 文本串行传递。
再看它的 checkpointer（langgraph/checkpoint/）：每个超步结束，State 被序列化落盘（SQLite/Postgres/内存）。这意味着流程可以断点续跑——我那个"API 超时"的痛点，在 LangGraph 里只需在失败节点加重试策略，或从上一个 checkpoint 恢复，前面的解析结果都还在。这就是"显式状态"带来的工程红利。
用图直观对比两种执行模型：
3. LlamaIndex Workflow：事件驱动的"松耦合流水线"
LlamaIndex 早期以 RAG 闻名，但它的 Workflow（llama_index/core/workflow/）是一个被低估的编排引擎。核心是 @step 装饰器 + Event：
# 简化自 llama_index/core/workflow/workflow.py
class Workflow:
def __init__(self):
self._steps = {} # event_type -> [step_fn]
self._queues = {} # 每个 step 的输入队列
def _get_steps(self, event):
# 根据 Event 类型分发到对应的 step
return self._steps.get(type(event), [])
async def run(self, ctx=None):
# 事件循环：从队列取 Event，分发给 step，step 产出新 Event 入队
while not self._done():
event = await self._queue.get()
for step in self._get_steps(event):
new_events = await step(event) # step 返回 Event 或 list[Event]
for e in new_events:
await self._queue.put(e)和 LangGraph 的差异在于：LangGraph 的边是显式画出来的（add_edge），而 LlamaIndex 的 step 之间是通过 Event 类型隐式连接的——step A 产出 EventX，step B 声明 accept EventX，两者就自动接上了。这是发布/订阅模型。
优点：解耦彻底，加一个 step 不用改图；step 可以并发（多个 step 订阅同一 Event）。缺点：控制流不直观，调试时你得"顺着 Event 类型"去追，图不在你脑子里。所以它更适合以数据处理为中心的流水线（文档解析 → 分块 → 嵌入 → 检索 → 生成），而不是复杂的多轮决策。
4. CrewAI 与 AutoGen：多 Agent 协作的两种哲学
CrewAI（crewai/crew.py）的核心是 Crew + Agent + Task。它有两种 Process：sequential（按 Task 顺序）和 hierarchical（有个 Manager Agent 分派）。看它的 sequential 执行：
# 简化自 crewai/crew.py 的 _run_sequential_process
def _run_sequential_process(self):
for task in self.tasks:
# 每个 task 绑定一个 agent，agent 执行 task
output = task.execute_sync(agent=task.agent, context=...)
# 上一个 task 的输出作为下一个 task 的 context
self._task_outputs.append(output)它的价值在抽象层次高：你写的是"角色、目标、任务"，而不是"节点、边"。对业务同学友好。但代价是控制流灵活性受限——你要做复杂分支、循环、并行，就得下沉到它的底层或换框架。它本质是"在 LangChain 等底座上加了一层组织编排 DSL"。
AutoGen（autogen/agentchat/）的核心是 ConversableAgent + GroupChat。它的协作原语是消息：
# 简化自 autogen 的 GroupChat 发言选择逻辑
class GroupChat:
def select_speaker(self, last_speaker, selector):
# 由 selector（可以是 LLM，也可以是规则）决定下一个发言者
return selector.select(self.agents, self.messages)
async def run_chat(self):
while not self._is_termination_msg(self.messages[-1]):
speaker = self.select_speaker(...)
reply = await speaker.generate_reply(self.messages)
self.messages.append(reply)它的控制流是涌现式的：谁发言由 selector 决定（可以是 LLM 投票、可以是轮询、可以是自定义规则）。优点是极其灵活，能模拟"头脑风暴""辩论""结对编程"；缺点是不可预测、容易跑飞、成本不可控（多 Agent 互相聊，token 爆炸）。生产上必须配 max_round 和严格的终止条件。
实战代码：三个完整可运行示例
依赖：
pip install langgraph langchain-openai llama-index-core crewai，并设置OPENAI_API_KEY。以下代码均为可运行的最小完整示例。
示例 1：LangGraph 实现"可恢复 + 并行"的合同审阅
这是我把引言里那个坑填上的方案。核心：State 显式化、并行节点、checkpointer 断点续跑。
# contract_review_graph.py
import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ---------- 1. 显式定义状态，用 reducer 支持并行合并 ----------
class ReviewState(TypedDict):
contract_text: str
# 三个并行节点各自追加 findings，用 add 合并
findings: Annotated[list[dict], operator.add]
retry_count: int
report: str
# ---------- 2. 定义节点（每个节点只做一件事） ----------
def extract_clauses(state: ReviewState) -> dict:
"""抽取条款：真实场景可换成 LLM 结构化抽取"""
clauses = [c.strip() for c in state["contract_text"].split("；") if c.strip()]
return {"findings": [{"type": "clause", "value": c} for c in clauses]}
def check_regulation(state: ReviewState) -> dict:
"""法规检索：这里 mock，真实场景接向量库"""
# 模拟一次可能失败的检索
if state.get("retry_count", 0) < 1 and "违约" in state["contract_text"]:
# 故意制造一次失败，演示条件边重试
raise RuntimeError("法规库超时")
return {"findings": [{"type": "regulation", "value": "民法典第577条"}]}
def risk_rating(state: ReviewState) -> dict:
"""风险评级：汇总 findings 交给 LLM"""
prompt = f"根据以下发现给出风险评级：{state['findings']}"
rating = llm.invoke(prompt).content
return {"findings": [{"type": "risk", "value": rating}]}
def generate_report(state: ReviewState) -> dict:
return {"report": f"审阅完成，共 {len(state['findings'])} 条发现。"}
# ---------- 3. 构建图 ----------
builder = StateGraph(ReviewState)
builder.add_node("extract", extract_clauses)
builder.add_node("regulation", check_regulation)
builder.add_node("risk", risk_rating)
builder.add_node("report", generate_report)
# 并行：START 同时扇出到三个节点
builder.add_edge(START, "extract")
builder.add_edge(START, "regulation")
builder.add_edge(START, "risk")
# 三个节点都完成后汇聚到 report（LangGraph 会等全部入边就绪）
builder.add_edge(["extract", "regulation", "risk"], "report")
builder.add_edge("report", END)
# ---------- 4. 编译时挂 checkpointer，支持断点恢复 ----------
graph = builder.compile(checkpointer=MemorySaver())
# ---------- 5. 运行 ----------
config = {"configurable": {"thread_id": "contract-001"}}
text = "甲方应于30日内付款；乙方违约需支付违约金；合同有效期一年"
try:
result = graph.invoke({"contract_text": text, "findings": [], "retry_count": 0}, config)
print(result["report"])
except RuntimeError as e:
# 法规库失败：从 checkpoint 恢复，递增 retry_count 后重跑
print(f"节点失败：{e}，准备从 checkpoint 恢复")
snapshot = graph.get_state(config) # 拿到失败前的状态
print("已恢复状态：", snapshot.values)
result = graph.invoke(
{"contract_text": text, "findings": [], "retry_count": 1}, config
)
print(result["report"])要点：Annotated[..., operator.add] 让三个并行节点的 findings 自动合并，无需手工协调；checkpointer 让失败后可 get_state 恢复。这就是"显式状态"的工程价值。
示例 2：LlamaIndex Workflow 做"事件驱动"的文档处理流水线
适合"解析 → 分块 → 嵌入 → 入库"这类松耦合数据处理。
# doc_pipeline_workflow.py
from llama_index.core.workflow import (
Workflow, step, Event, StartEvent, StopEvent, Context
)
# ---------- 定义事件类型（这就是 step 之间的"接线端子"） ----------
class ParsedEvent(Event):
text: str
class ChunkedEvent(Event):
chunks: list[str]
class EmbeddedEvent(Event):
vectors: list[list[float]]
class DocPipeline(Workflow):
@step
async def parse(self, ctx: Context, ev: StartEvent) -> ParsedEvent:
# 真实场景：PDF 解析
raw = ev.get("raw_text", "")
return ParsedEvent(text=raw)
@step
async def chunk(self, ctx: Context, ev: ParsedEvent) -> ChunkedEvent:
# 按固定长度切块
size = 200
chunks = [ev.text[i:i+size] for i in range(0, len(ev.text), size)]
return ChunkedEvent(chunks=chunks)
@step
async def embed(self, ctx: Context, ev: ChunkedEvent) -> EmbeddedEvent:
# 真实场景：调用 embedding 模型；这里 mock
vectors = [[float(len(c))] for c in ev.chunks]
return EmbeddedEvent(vectors=vectors)
@step
async def store(self, ctx: Context, ev: EmbeddedEvent) -> StopEvent:
# 真实场景：写入向量库
return StopEvent(result={"stored": len(ev.vectors)})
# ---------- 运行 ----------
import asyncio
async def main():
wf = DocPipeline(timeout=30)
result = await wf.run(raw_text="这是一段很长的合同文本。" * 50)
print(result) # {'stored': N}
asyncio.run(main())要点：step 之间靠 Event 类型自动接续，加一个 step 只需定义新 Event 并 @step，不用改已有代码。这是发布/订阅的威力，但也意味着你必须"顺着 Event 追控制流"。
示例 3：CrewAI 多 Agent 协作 + AutoGen 对话式协作对照
先看 CrewAI（角色/任务 DSL）：
# crew_review.py
from crewai import Agent, Task, Crew, Process
# ---------- 定义角色 ----------
extractor = Agent(
role="条款抽取专家",
goal="从合同中抽取所有关键条款",
backstory="你是一名资深法务，擅长快速定位合同风险点",
verbose=True,
)
rater = Agent(
role="风险评估专家",
goal="对抽取的条款做风险评级",
backstory="你熟悉民法典，能给出专业风险判断",
verbose=True,
)
# ---------- 定义任务，注意 context 传递 ----------
task1 = Task(
description="抽取合同中的付款、违约、有效期条款",
expected_output="条款列表",
agent=extractor,
)
task2 = Task(
description="对以下条款做风险评级：{task1_output}",
expected_output="风险评级报告",
agent=rater,
context=[task1], # 显式声明依赖，task1 输出注入 task2
)
# ---------- 顺序执行 ----------
crew = Crew(
agents=[extractor, rater],
tasks=[task1, task2],
process=Process.sequential, # 也可 Process.hierarchical 加 Manager
)
result = crew.kickoff(inputs={"contract": "..."})
print(result)再看 AutoGen（对话式协作，同一问题的另一种解法）：
# autogen_review.py
from autogen import ConversableAgent, GroupChat, GroupChatManager
llm_config = {"model": "gpt-4o-mini", "api_key": "..."}
extractor = ConversableAgent(
name="Extractor",
system_message="你负责抽取合同条款，完成后说 TERMINATE。",
llm_config=llm_config,
is_termination_msg=lambda m: "TERMINATE" in (m.get("content") or ""),
)
rater = ConversableAgent(
name="Rater",
system_message="你负责对条款做风险评级，完成后说 TERMINATE。",
llm_config=llm_config,
is_termination_msg=lambda m: "TERMINATE" in (m.get("content") or ""),
)
# GroupChat：谁发言由 selector 决定（默认 LLM 自动选择）
groupchat = GroupChat(agents=[extractor, rater], messages=[], max_round=6)
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
# 启动对话
extractor.initiate_chat(manager, message="请审阅这份合同：...")对照：CrewAI 用 context=[task1] 显式声明依赖，控制流可预测；AutoGen 靠 GroupChat + max_round 隐式协商，灵活但容易跑飞。生产上我通常给 AutoGen 配严格的 is_termination_msg 和 max_round。
方案对比：一张决策表
| 维度 | LangChain AgentExecutor | LangGraph | LlamaIndex Workflow | CrewAI | AutoGen |
|---|---|---|---|---|---|
| 控制流 | LLM 文本驱动 | 显式图/状态机 | 事件驱动 | Task 顺序/Manager | 对话涌现 |
| 状态可持久化 | ❌（文本 scratchpad） | ✅ Checkpointer | ✅ Context | 部分 | 消息历史 |
| 并行能力 | ❌ | ✅ 超步并行 | ✅ Event 并发 | 有限 | 有限 |
| 可测试性 | 差 | 好（节点单测） | 好（step 单测） | 中 | 差 |
| 学习曲线 | 低 | 中高 | 中 | 低 | 中 |
| 适合场景 | Demo/单 Agent | 复杂有状态流程 | 数据/RAG 流水线 | 角色化多 Agent | 开放式多 Agent 讨论 |
| 生产可用性 | 低 | 高 | 高 | 中高 | 中（需强约束） |
选型建议（我的经验法则）：
- 流程有明确分支、循环、需要断点续跑 → LangGraph；
- 流程是文档/数据流水线、要频繁加处理环节 → LlamaIndex Workflow；
- 业务方参与设计、强调角色分工、流程相对线性 → CrewAI；
- 需要开放式多 Agent 讨论/辩论/结对、能接受一定不确定性 → AutoGen；
- 只是做个单 Agent 调几个 Tool 的 Demo → LangChain AgentExecutor 够用，但别上生产。
最佳实践与避坑指南
- 状态别塞进 prompt。
AgentExecutor的intermediate_steps序列化是反模式。用 LangGraph 的显式 State，或至少自己维护结构化记忆，别指望 LLM 记住 20 轮前的 Observation。
- 给每个 Agent 设"预算"。 AutoGen 的
max_round、LangGraph 的recursion_limit、CrewAI 的max_iter，不设就是给账单开盲盒。我见过 AutoGen 两个 Agent 互相客套聊了 40 轮烧掉 3 美元的。
- 节点/step 要幂等。 断点恢复意味着节点可能被重跑，写库、发消息这类副作用必须幂等（用 upsert 而非 insert，用幂等 key）。
- 并行节点别共享可变对象。 LangGraph 的 reducer 只合并"返回值增量"，如果你在节点里
state["findings"].append(...)原地改，会踩并发坑。永远返回新对象。
- 可观测性从第一天就上。 用 LangSmith / Langfuse 之类追踪每个节点的输入输出。多 Agent 系统出问题时，"谁在哪一步说了什么"是唯一救命稻草。
- 别为了"多 Agent"而多 Agent。 很多需求单 Agent + 好工具就能解决。多 Agent 的通信开销和不确定性是指数级的，CrewAI/AutoGen 不是银弹。
- MCP 作为工具层解耦。 2025 年后 MCP（Model Context Protocol）逐渐成为工具接入的事实标准。把 Tool 实现成 MCP Server，四个框架都能接，避免工具层和编排层绑死——这是我今年最推荐的架构习惯。
总结
回到开头那个坑。四个框架的分水岭，不在 API 多花哨，而在"控制流归谁管"：
AgentExecutor把控制流交给 LLM 的文本决策，简单但不可控、不可恢复；
- LangGraph 把控制流显式画成状态机，用 reducer 支持并行、用 checkpointer 支持恢复，是复杂生产流程的首选；
- LlamaIndex Workflow 用事件解耦，适合数据流水线；
- CrewAI 和 AutoGen 把协作抽象成角色/任务和对话，前者可预测、后者更灵活，都需强约束才能上生产。
一个延伸思考：这四种执行模型其实对应了分布式系统里四种经典范式——命令式循环、状态机、发布订阅、Actor 对话。理解了这层映射，你换任何新框架都能快速定位它的"性格"。下次再有人问你"选哪个 Agent 框架"，你可以反问他一句："你的控制流，是想画出来，还是想聊出来？"