post · 2026-05-30

搭 Agent 的脚手架：主流开发框架

作者：toy ---

作者：toy

一、为什么需要 Agent 框架

从"手写循环"到框架

构建一个最简单的 ReAct Agent，大概需要五十行 Python：一个 while 循环，调 LLM，解析输出，调工具，把结果塞回上下文，再循环。写完第一个 Agent 的人通常会觉得，原来如此简单。

但第二个 Agent 出现时，问题来了。你需要注册新工具，要让 Agent 知道每个工具的签名和调用时机。状态怎么传，是函数参数、全局变量，还是 dataclass？多轮对话的历史消息怎么存、怎么截断？调用失败时重试几次、用什么退避策略？每次模型返回时如何验证输出格式？这些问题，每个都需要代码。

到第三个 Agent，你开始复制粘贴。到第五个，你发现自己在维护一套私有的 Agent 基础设施，而这套东西还没有测试。从这个角度看，框架要解决的不是"能不能运行"，而是"能不能在六个月后还读得懂、改得动、扩得了"。

工具注册、状态管理、错误处理，这些都要自己写吗

Agent Harness 的概念来自 LangChain 的工程师 Vivek Trivedy："如果你不是模型，你就是 Harness。" 框架提供的正是这个 Harness，把 Agent 跑起来所需的一切基础设施。

一个生产级 Agent 至少需要处理以下内容：工具注册与描述生成（让 LLM 知道有哪些工具、每个工具做什么、参数是什么类型）；调用历史的序列化与截断（上下文窗口有限，不能无限追加）；状态管理（多步骤任务中，每步的输入输出如何传递）；错误处理与重试（工具调用失败、LLM 输出格式错误、网络超时）；跨会话持久化（任务被中断后如何从断点恢复）；并发控制（多个 Agent 同时跑时如何隔离状态）。

如果全部手写，每个项目都重来一遍，这是工程债务，不是工程能力。框架的价值在于把这些问题的解法标准化，让开发者把注意力放在业务逻辑而非基础设施。

框架的价值：标准化抽象 vs 灵活性损失的权衡

框架是一把双刃剑。标准化带来了协作成本的下降，新成员读过文档就能理解代码结构，不需要阅读三千行私有基础设施。它也带来了调试上的便利，框架通常有成熟的日志、追踪和回放机制。

但代价也是真实的。框架的抽象层使得错误消息变得晦涩。当 LangChain 抛出一个三层嵌套的 ChainCallbackHandler 错误时，调试所需的心智模型比手写循环复杂得多。框架还会引入版本锁定问题，LangChain 从 0.0.x 到 0.1.x 到 0.2.x 到 0.3.x，每次主版本都有破坏性变更，维护成本不可忽视。

选框架的核心判断是：你的 Agent 是一次性原型，还是需要长期迭代的生产系统？原型阶段用框架快速验证想法合理；生产阶段需要评估框架的稳定性、社区支持和长期维护承诺。

二、LangChain：最广泛使用的拼装工具箱

核心概念：Chain/Agent/Tool/Memory/Callback 的关系

LangChain 发布于 2022 年末，赶上了 GPT-4 引发的 Agent 热潮，成为这个领域事实上的标准参考实现。理解 LangChain 需要先理清它的几个核心抽象。

Chain 是最基础的抽象单元，把一系列操作串联起来：接收输入 → 调 LLM → 后处理输出。LangChain 提供了数十种预制 Chain，从简单的 LLMChain 到复杂的 ConversationalRetrievalChain。Agent 在 Chain 之上加了一层决策逻辑，LLM 根据工具列表和当前状态自己决定下一步调哪个工具。Tool 是 Agent 可以调用的函数，配有名称、描述和输入 Schema。Memory 管理对话历史，控制哪些内容进入上下文、如何压缩旧消息。Callback 是贯穿整个调用链的钩子系统，用于日志、追踪、流式输出。

这五个概念的关系可以这样理解：Tool 是扳手，Memory 是工作台，Callback 是摄像头，Chain 是流水线，Agent 是操作工人。框架把这些零件组装在一起，让你专注于定义业务规则。

LangChain Expression Language（LCEL）的管道式写法

LCEL 是 LangChain 0.1 版引入的新 API，用管道操作符 | 把各个组件串联起来，使得构建 Chain 的代码更接近函数式风格：

# 传统方式：命令式组装
chain = LLMChain(llm=llm, prompt=prompt, output_parser=parser)
result = chain.run(input="hello")

# LCEL 方式：声明式管道
chain = prompt | llm | parser
result = chain.invoke({"input": "hello"})

LCEL 的优势不仅在于写法简洁。它让所有组件自动支持流式输出（.stream()）、异步调用（.ainvoke()）和批量处理（.batch()），不需要每个组件单独实现。每一步的输入输出类型通过 Pydantic 验证，调试时可以打印中间状态。

但 LCEL 也引入了一层新的认知成本。RunnablePassthrough、RunnableLambda、RunnableParallel 这些类型在初学者看来并不直观，管道中的隐式类型转换偶尔会导致难以定位的运行时错误。

实际使用中的槽点：过度抽象、版本不稳定、调试困难

在真实项目里使用 LangChain，常见的痛点集中在三个地方。第一是过度抽象。LangChain 试图统一所有 LLM 提供商的接口，导致很多简单操作被包裹在复杂的类层级中。想要给某次 API 调用设置自定义 header，可能需要继承 BaseChatModel 并覆写三个方法。第二是版本不稳定。该库的迭代速度极快，半年前的教程代码在新版本下经常无法运行，langchain、langchain-core、langchain-community 三个包之间的依赖关系也经常出现冲突。第三是调试困难。当 Chain 调用链嵌套超过三层时，错误堆栈会包含大量框架内部代码，定位实际出错的业务逻辑需要相当经验。

这些问题不是 LangChain 独有的，任何快速演进的框架都会经历类似阶段。但它提醒开发者，在采用之前，值得评估自己的 Agent 是否真的需要 LangChain 提供的全部能力，还是只需要其中一小部分。

什么场景适合 LangChain：快速原型 vs 生产使用的边界

LangChain 最适合的场景是快速原型验证。它内置了对几十个 LLM 提供商的适配、几百种向量数据库的连接器、常见的 RAG 模式实现，很多东西开箱即用。如果你需要在一周内跑通一个新想法，LangChain 是合理的起点。

进入生产阶段后，判断是否继续使用 LangChain 的标准变了：你的 Agent 逻辑中有多少是框架提供的，有多少是你自己写的？如果框架贡献的是 70% 的核心逻辑，继续使用是合理的。如果你大量使用 RunnableLambda 包装自定义代码，框架只是提供了一个调用入口，那么直接手写可能反而更清晰。

有一个实际的数据点值得关注：LangChain 官方工程师曾通过优化 LangChain 的 Harness 配置（不改变底层模型），让测试系统在 TerminalBench 基准上的排名从 30 名开外跳到第 5 名，跨越了 20 多个位置。这说明框架层的工程质量对 Agent 的最终表现影响很大。但框架层的优化空间需要深入理解框架内部才能挖掘，对于只是"用用"框架的团队，这个红利很难拿到。

对于需要复杂工作流控制、状态持久化和人机协同的场景，LangChain 的 Agent 模块能力有限，这时候应该考虑 LangGraph。LangChain 和 LangGraph 虽然是同一个生态（都由 LangChain 公司维护），但代码库完全独立，选择 LangGraph 不要求必须使用 LangChain 的其他部分。

三、LangGraph：用状态机管理 Agent 工作流

从 LangChain 到 LangGraph 的进化：为什么需要图结构

LangChain 的 Chain 和 Agent 都是线性的。Chain 是顺序执行的管道，Agent 是固定结构的 ReAct 循环。这对简单任务够用，但对于需要条件分支、循环、并行执行和状态回溯的复杂工作流，线性结构力不从心。

LangGraph 用有向图（Directed Graph）解决这个问题。图中的节点（Node）是函数，边（Edge）是控制流，状态（State）是贯穿整个图执行过程的共享数据结构。这个模型的好处是表达能力强：你可以描述任意复杂的控制流，包括条件跳转（"如果工具返回错误，走重试节点"）、循环（"继续执行直到满足某个条件"）、并行子图（"同时调用多个工具，等全部完成后汇总"）。

从工程角度看，图结构还带来了另一个优势：可视化。LangGraph 可以把图的结构渲染为 Mermaid 流程图，让开发者在代码之外看到工作流的全貌，在团队协作和代码审查时价值明显。

核心概念：StateGraph、Node（节点函数）、Edge（条件路由）

StateGraph 是 LangGraph 的核心类。构建 Agent 的第一步是定义状态 Schema，用 TypedDict 或 Pydantic BaseModel 描述整个工作流共享的数据结构：

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

# 状态 Schema：每个字段的类型和 Reducer 函数
class ChatState(TypedDict):
    # add_messages Reducer：新消息追加到列表，而非覆盖；并处理消息 ID 去重
    messages: Annotated[list[BaseMessage], add_messages]
    # 默认 Reducer：完全覆盖，新值替换旧值
    user_name: str
    retry_count: int

Reducer 函数是 LangGraph 状态管理的关键设计。每个字段可以附加一个 Reducer，定义"当节点返回新值时，如何将新值合并到现有状态"。默认的 Reducer 是完全覆盖（新值替换旧值），add_messages 这个特殊 Reducer 则会把新消息追加到列表末尾，同时去重（基于消息 ID）并处理反序列化。

Node 是普通的 Python 函数，接收当前 State，返回一个字典（包含要更新的字段）：

def tool_calling_node(state: ChatState):
    """节点函数：接收完整 state，只返回需要更新的字段"""
    messages = state["messages"]
    # 调用 LLM，获取工具调用决策
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}  # add_messages reducer 会追加到现有列表

def should_continue(state: ChatState) -> str:
    """条件边函数：返回下一个节点名称"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 有工具调用，走 tools 节点
    return "end"        # 无工具调用，结束

Edge 分为普通边（始终走这条路）和条件边（根据函数返回值决定走哪条路）。条件边是实现分支逻辑的核心机制：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(ChatState)
builder.add_node("agent", tool_calling_node)
builder.add_node("tools", tool_execution_node)

builder.add_edge(START, "agent")
# 条件边：根据 should_continue 的返回值决定走 tools 还是 END
builder.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END
})
builder.add_edge("tools", "agent")  # 工具执行完再回到 agent

Checkpoint 机制：状态如何序列化/反序列化，跨会话恢复

LangGraph 的 Checkpoint 机制是它区别于其他框架的核心特性。每当图执行完一个"super-step"（节点执行完毕、准备进入下一条边），LangGraph 会生成一个 Checkpoint 快照，记录当前完整的状态。

每个 Checkpoint 包含：thread_id（标识一个独立的对话命名空间）、checkpoint_id（UUID，唯一标识这个快照）、parent_checkpoint_id（父快照的 ID，构成快照链）、序列化的状态 blob 和 metadata 字典。默认序列化器是 JsonPlusSerializer，用 ormsgpack 做主序列化，失败时降级到扩展 JSON。它支持的特殊类型包括 set/frozenset/deque、datetime 系列、UUID、Enum、Pydantic v1/v2 模型、dataclass、NumPy 数组（保留 shape/dtype）以及 LangGraph 内部的 Send 协议对象。

一个重要的安全提示：JsonPlusSerializer 存在 RCE 风险（CVE-2025-64439），生产环境需要设置环境变量 LANGGRAPH_STRICT_MSGPACK=true。

以下是一个完整的 PostgreSQL 持久化示例：

# 安装: pip install langgraph langgraph-checkpoint-postgres asyncpg

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import asyncpg

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_name: str

def chatbot_node(state: ChatState):
    last_msg = state["messages"][-1]
    reply = f"Hello {state['user_name']}, you said: {last_msg.content}"
    return {"messages": [AIMessage(content=reply)]}

builder = StateGraph(ChatState)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

async def main():
    DB_URI = "postgresql://user:password@localhost:5432/agentdb"

    # 创建连接池，建议 min_size=2, max_size=10
    pool = await asyncpg.create_pool(DB_URI, min_size=2, max_size=10)
    checkpointer = AsyncPostgresSaver(pool)

    # setup() 自动创建 4 张表：
    # checkpoints（主表，复合主键 thread_id+checkpoint_ns+checkpoint_id）
    # checkpoint_blobs（大对象，BYTEA 二进制存储）
    # checkpoint_writes（节点中间写入暂存区）
    # checkpoint_migrations（schema 版本追踪）
    await checkpointer.setup()

    graph = builder.compile(checkpointer=checkpointer)

    # thread_id 隔离每个用户的会话命名空间
    config = {"configurable": {"thread_id": "user-alice-session-001"}}

    # 第一次交互
    result1 = await graph.ainvoke(
        {"messages": [HumanMessage(content="Hi!")], "user_name": "Alice"},
        config=config
    )

    # 第二次交互：graph 自动从 PostgreSQL 恢复 alice 的最新 checkpoint
    result2 = await graph.ainvoke(
        {"messages": [HumanMessage(content="What did I say before?")]},
        config=config  # 同一 thread_id 触发状态恢复
    )

    # 查看所有历史 checkpoint（用于调试或审计）
    history = list(graph.aget_state_history(config))
    print(f"Total checkpoints: {len(history)}")

    await pool.close()

PostgresSaver 的写延迟是：直连 5–15 ms，连接池 3–8 ms。对于大多数 Web 服务，这个开销可以接受。

Human-in-the-loop：interrupt_before/after 的实现方式

生产 Agent 的一个核心需求是人工审批，在执行某些高风险操作之前暂停、等待人工确认。LangGraph 通过 interrupt() 函数实现这个能力。

interrupt() 在节点内部调用时，会暂停图执行并保存当前 Checkpoint。调用方收到 NodeInterrupt 异常，可以从中提取暂停时传递的 payload。当人工决策完成后，用 Command(resume=value) 重新 invoke 同一个 thread_id，图从断点继续执行，resume 的值成为 interrupt() 的返回值。

from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver

class ApprovalState(TypedDict):
    action: str
    approved: bool
    result: str

def propose_action(state: ApprovalState):
    return {"action": "DELETE user_data WHERE age < 18"}

def human_approval(state: ApprovalState):
    # interrupt() 在此处暂停执行，payload 传递给调用方
    decision = interrupt({
        "question": "Approve this action?",
        "action": state["action"],
        "risk_level": "HIGH"
    })
    # 恢复执行时 decision = Command(resume=True/False) 的 resume 值
    return {"approved": decision}

def execute_or_reject(state: ApprovalState):
    if state["approved"]:
        return {"result": f"Executed: {state['action']}"}
    return {"result": "Action rejected by human reviewer"}

builder = StateGraph(ApprovalState)
builder.add_node("propose", propose_action)
builder.add_node("approve", human_approval)
builder.add_node("execute", execute_or_reject)
builder.add_edge(START, "propose")
builder.add_edge("propose", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

# interrupt 必须配合 checkpointer 使用
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "approval-001"}}

# 第一次 invoke：执行到 interrupt() 处暂停
events = list(graph.stream({}, config=config, stream_mode="values"))
# 此时 graph 在 human_approval 节点暂停

# 人工审核后，用 Command(resume=True) 恢复执行
final = graph.invoke(Command(resume=True), config=config)
print(final["result"])  # "Executed: DELETE user_data WHERE age < 18"

除了动态 interrupt()，LangGraph 还支持在 compile() 时静态配置断点：

# 静态断点：进入 execute 前暂停，propose 完成后暂停
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["execute"],
    interrupt_after=["propose"],
)

四、LlamaIndex：文档型 Agent 的首选

与 LangChain 的定位差异

LlamaIndex 和 LangChain 经常被放在一起比较，但它们的定位从一开始就不同。LangChain 是通用 Agent 框架，目标是统一各类 LLM 任务；LlamaIndex 从文档问答出发，深度优化的场景是"如何让 LLM 高质量地处理大量外部文档"。

在 RAG 场景，LlamaIndex 提供了更丰富的开箱即用能力：文档加载器（PDF、Word、Notion、Confluence、数据库……），多种分块策略（固定大小、句子、语义分块），检索增强（混合检索、关键词 + 向量、父子文档检索），以及查询后处理（重排序、相关度过滤）。如果你的 Agent 的核心能力是"对大量文档进行问答和推理"，LlamaIndex 在这个方向上省去的工程工作量，远超 LangChain。

QueryEngine / ChatEngine / AgentRunner 三层架构

LlamaIndex 的 Agent 能力通过三层架构暴露：

QueryEngine 是最基础的一层，接收一个问题，从索引中检索相关文档，生成回答。它是无状态的，每次调用都是独立的，适合单轮问答场景。

ChatEngine 在 QueryEngine 之上加了对话历史，让 LLM 能够参考上下文理解指代关系（"它" 指的是什么），适合多轮对话场景。

AgentRunner 是完整的 Agent 抽象，可以配置工具（包括 QueryEngine 作为工具），让 LLM 自主决定何时检索文档、何时调用其他工具。这一层让 LlamaIndex 真正进入了 Agent 框架的范畴：

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.tools import QueryEngineTool
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

# 构建文档索引
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)

# 把 QueryEngine 封装成 Tool
query_engine = index.as_query_engine()
query_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="doc_search",
    description="搜索技术文档，回答关于 API 和实现细节的问题"
)

# AgentRunner：LLM 决定何时调用 query_tool
llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools([query_tool], llm=llm, verbose=True)
response = agent.chat("LangGraph 的 Checkpoint 存储格式是什么？")

SubQuestion QueryEngine：复杂问题分解的内置支持

SubQuestion QueryEngine 是 LlamaIndex 的一个特色功能：当用户提出一个复杂问题时，系统自动将其分解为多个子问题，分别检索和回答，最后合并生成综合答案。

这个能力对于需要跨文档推理的场景特别有价值。例如，"比较 LangGraph 和 AutoGen 在状态持久化上的设计差异"这类问题，单次检索很难同时找到两个框架的相关内容，SubQuestion 机制会自动拆分为"LangGraph 状态持久化如何实现"和"AutoGen 状态持久化如何实现"两个子查询，分别检索，然后让 LLM 综合对比。

from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool

# 多个 QueryEngine，每个对应一个文档集
tools = [
    QueryEngineTool.from_defaults(langgraph_engine, name="langgraph_docs"),
    QueryEngineTool.from_defaults(autogen_engine, name="autogen_docs"),
]

# SubQuestion 自动分解复杂问题
sub_question_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    use_async=True  # 并行执行子问题，降低延迟
)
response = sub_question_engine.query(
    "比较 LangGraph 和 AutoGen 在状态持久化上的设计差异"
)

与向量数据库的深度集成

LlamaIndex 与向量数据库的集成是其最成熟的部分。它支持 Chroma、Pinecone、Weaviate、Qdrant、pgvector 等主流向量数据库，通过统一的 VectorStore 接口访问，切换后端只需修改一行配置。

生产场景中，LlamaIndex 还支持混合检索（BM25 关键词检索 + 向量检索结果融合）、多索引路由（根据问题类型自动选择合适的索引）和增量更新（文档变更时只重索引发生变化的部分），这些能力把 RAG 系统从原型推向了生产可用。

一个常见的组合是：LlamaIndex 负责文档摄入、分块、索引和检索，LangGraph 负责把检索结果注入到 Agent 的工作流中，管理多步推理的状态。这两个框架的职责边界清晰：LlamaIndex 是数据层，LangGraph 是控制流层，拼接它们只需要把 LlamaIndex 的 QueryEngine 包装成 LangGraph 节点的一个函数调用。

五、AutoGen：多 Agent 对话框架

ConversableAgent 抽象：任何实体都可以是 Agent

AutoGen 来自微软研究院，它的核心设计哲学与 LangGraph 不同：AutoGen 认为，复杂任务最好通过多个 Agent 之间的自然语言对话来解决，而不是通过预先定义好的控制流图。

ConversableAgent 是 AutoGen 的基础抽象。任何能够接收消息、生成响应的实体都可以是 ConversableAgent，可以是 LLM、人类、代码执行器，甚至是另一个 AutoGen 团队。这个抽象让 Agent 的定义非常宽泛，也让 AutoGen 的代码执行能力成为其区别于其他框架的关键特性。

AutoGen 0.4（重命名为 AgentChat）对架构做了重大重构，把多 Agent 协作的单元称为 Team，提供四种预设：

• RoundRobinGroupChat：轮询顺序，每个 Agent 按顺序发言
• SelectorGroupChat：LLM 动态选择下一个发言人，最灵活
• Swarm：基于 HandoffMessage 的转移，适合流水线式分工
• MagenticOneGroupChat：微软内部研究成果，有 Orchestrator 统一协调

AssistantAgent + UserProxyAgent 的经典组合

AutoGen 0.2 中最经典的模式是 AssistantAgent + UserProxyAgent 的配对。AssistantAgent 是 LLM 驱动的对话代理，负责生成方案和代码；UserProxyAgent 代表人类，负责执行代码并把结果反馈给 AssistantAgent。

这个设计的精妙之处在于代码执行的分离：LLM 不直接执行代码，而是生成代码交给 UserProxyAgent 执行，执行结果再回传给 LLM。这种分离让代码执行安全可控：在隔离的 Docker 容器中运行，限制文件访问权限，或者要求人工确认后才执行。

# AutoGen 0.2 经典模式
import autogen

config_list = [{"model": "gpt-4o", "api_key": "sk-..."}]

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list}
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",    # 不需要人工输入，自动执行
    max_consecutive_auto_reply=10,
    code_execution_config={
        "work_dir": "./workspace",
        "use_docker": True,  # 在 Docker 中执行代码
    }
)

user_proxy.initiate_chat(
    assistant,
    message="写一个 Python 函数，计算斐波那契数列的第 n 项，并测试 n=10 的结果"
)

SelectorGroupChat：多个 Agent 的协调机制

AutoGen 0.4 的 SelectorGroupChat 用 LLM 动态决定每轮对话后谁来发言。这比轮询更智能，但也更贵，每次选择发言人都需要一次 LLM 调用。

在一个 4-Agent、5-round 的 GroupChat 中，AutoGen 0.2 的 auto 模式至少产生 20 次 LLM 调用，每次携带完整对话历史，token 成本随轮次线性增长。这是使用 AutoGen 时必须正视的成本问题。

以下是一个使用 AutoGen 0.4 SelectorGroupChat 的完整示例：

# 安装: pip install autogen-agentchat autogen-ext[openai]

import asyncio
import json
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    researcher = AssistantAgent(
        name="Researcher",
        description="搜索和分析信息",
        model_client=model_client,
        system_message="你负责搜集信息并进行分析。"
    )

    writer = AssistantAgent(
        name="Writer",
        description="撰写和格式化内容",
        model_client=model_client,
        system_message="你负责基于研究结果撰写清晰、结构良好的内容。"
    )

    critic = AssistantAgent(
        name="Critic",
        description="审阅和改进内容质量",
        model_client=model_client,
        system_message="你负责审阅内容并提出改进建议。对满意的内容说 APPROVE。"
    )

    # 自定义 selector_prompt 控制 LLM 的选择逻辑
    selector_prompt = """
    参与者: {participants}
    角色: {roles}
    历史: {history}

    根据对话流程选择最合适的下一个发言人。
    信息收集优先选 Researcher，内容撰写优先选 Writer，质量审阅优先选 Critic。
    只返回 agent 名称。
    """

    termination = TextMentionTermination("APPROVE") | MaxMessageTermination(20)

    team = SelectorGroupChat(
        participants=[researcher, writer, critic],
        model_client=model_client,
        selector_prompt=selector_prompt,
        allow_repeated_speaker=False,  # 防止同一 agent 连续发言
        termination_condition=termination
    )

    result = await team.run(task="写一篇关于 LangGraph checkpoint 机制的技术文章")

    # 保存 team 状态以跨会话恢复
    team_state = await team.save_state()
    # 格式: {type: TeamState, agent_states: {Researcher: {type: AssistantAgentState,
    #         version: '1.0.0', llm_messages: [...]}, ...}}
    with open("team_state.json", "w") as f:
        json.dump(team_state, f)

asyncio.run(main())

AutoGen 的跨会话持久化相对简单：team.save_state() 返回可序列化的字典（包含所有 Agent 的 llm_messages 历史），new_team.load_state(saved_state) 恢复状态。不依赖外部数据库，序列化格式是普通 JSON，便于调试。代价是状态需要开发者自行管理存储和加载时机。

代码执行安全：UserProxyAgent 的代码沙箱配置

AutoGen 的代码执行沙箱有几个关键配置项需要注意。use_docker=True 在 Docker 容器中执行代码，最高隔离级别；use_docker=False 直接在本地进程执行，有安全风险，仅适合受信任的代码。timeout 参数控制单次执行的最长时间，防止无限循环。work_dir 定义工作目录，沙箱内的文件读写都限制在这个路径下。

在生产场景中，如果允许 LLM 生成的代码直接执行，沙箱配置是安全的第一道防线。建议至少启用 Docker 隔离，并配置网络访问限制（防止代码对外发送请求）。

六、CrewAI：面向任务的 Agent 团队编排

三层抽象：Agent（角色）/ Task（任务）/ Crew（团队）

CrewAI 的设计比喻是项目管理：你有一支团队（Crew），团队中有不同角色的成员（Agent），每个成员负责特定任务（Task）。这个比喻对非技术背景的使用者非常友好，也让 CrewAI 在内容创作、报告生成、市场分析这类"角色清晰、分工明确"的场景中特别流行。

Agent 定义一个角色，包括 role（角色名）、goal（目标）、backstory（背景故事，作为 system prompt 的一部分）。Task 定义一个任务，包括 description（任务描述）、expected_output（期望输出）、agent（负责的 Agent）。Crew 把 Agent 和 Task 组合起来，配置执行方式。

# 安装: pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process

# 定义角色
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find accurate technical information about {topic}",
    backstory="专注于技术信息收集和综合的资深分析师，擅长从多个来源提炼关键洞察。",
    allow_delegation=False,
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Write clear documentation for {topic}",
    backstory="有丰富技术写作经验，能将复杂概念转化为清晰易懂的文档。",
    allow_delegation=False
)

# 定义任务
research_task = Task(
    description="研究 {topic} 的最新进展，重点关注实际应用场景。",
    expected_output="包含关键发现、代码示例和引用来源的结构化报告。",
    agent=researcher
)

writing_task = Task(
    description="基于研究报告撰写技术文章。",
    expected_output="一篇结构清晰的 1000 字技术文章。",
    agent=writer,
    context=[research_task]  # 显式声明上下文依赖
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={"topic": "LangGraph checkpointing"})

顺序执行 vs 分层执行

CrewAI 提供两种执行模式，区别在于任务如何分配给 Agent。

Process.sequential 是顺序执行模式。任务在定义时预分配给指定 Agent，按照 tasks 列表的顺序依次执行。每个任务的输出自动作为下一个任务的上下文（通过 context 参数显式声明依赖）。这个模式简单、可预测，适合步骤清晰的流水线场景。

Process.hierarchical 是分层执行模式。这里的"分层"是指由一个 Manager Agent 动态协调其他 Worker Agent。任务不预分配，Manager 在运行时评估每个 Worker 的能力描述（role + backstory），决定把任务委派给谁，验证完成质量，并在质量不达标时重新委派。Manager Agent 可以通过 manager_llm 参数自动创建，也可以通过 manager_agent 参数使用自定义 Agent。

hierarchical_crew = Crew(
    agents=[researcher, writer],    # 只列 worker agents
    tasks=[research_task, writing_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o",           # 自动创建 manager agent
    memory=True,                    # 开启记忆系统，manager 节省约 30% token
    verbose=True
)

memory=True 是一个重要的生产建议。不开启时，Manager 每次委派决策都重读完整任务列表，开销随任务数量线性增长；开启后使用 LanceDB 持久化记忆，Manager 可以引用历史决策，约降低 30% token 消耗。

CrewAI 统一记忆系统

CrewAI 新版本用单一 Memory 类取代了旧的四分类（short-term/long-term/entity/contextual）。后端默认使用 LanceDB，持久化到 ./.crewai/memory。

记忆检索使用复合评分公式：

composite = semantic_weight × similarity + recency_weight × decay + importance_weight × importance
decay = 0.5^(age_days / half_life_days)
默认权重: semantic=0.5, recency=0.3, importance=0.2

这个公式的设计思路是：新的、相关的、重要的记忆优先被检索；随时间推移，旧记忆的权重指数衰减，half_life_days 控制衰减速度（默认 7 天）。可以根据场景调整：长期知识库项目可以设置更长的 half_life_days，实时性强的场景可以调大 recency_weight。

与 LangGraph 的核心区别

CrewAI 和 LangGraph 都能构建多 Agent 系统，但设计哲学根本不同。

CrewAI 的控制流是隐式的：Agent 和 Task 的组合描述了"谁做什么"，框架负责编排执行顺序。开发者描述意图，框架决定执行细节。这让 CrewAI 对初学者更友好，但也让精确控制执行流程变得困难。

LangGraph 的控制流是显式的：每条边、每个条件分支都由开发者用代码精确定义。这带来了更强的可预测性和可审计性，但需要开发者对工作流的每一步都有清晰的设计。

从成功率数据看：中等复杂度任务（3-5 次工具调用），LangGraph 76% > AutoGen 68% > CrewAI 71%；复杂任务（更多工具调用和分支），LangGraph 62% > AutoGen 58% > CrewAI 54%。但在 QA 等结构化场景，CrewAI 比 LangGraph 快 5.76 倍，因为它的角色编排在这类场景开销更低。

简单总结：CrewAI 像项目管理软件，适合结构化分工；LangGraph 像状态机，适合需要精确控制流程的复杂工作流。

七、状态持久化：让 Agent 不怕中断

为什么无状态 Agent 在生产中撑不住

一个简单的 Agent 任务可能需要调用 20 次工具，执行 10 分钟。在这个过程中，网络超时、服务器重启、用户关闭浏览器，任何一个事件都可能中断执行。无状态 Agent 遇到中断只能从头开始，这在生产环境中不可接受。

更深层的问题是：无状态 Agent 很难实现真正的"人机协同"。如果你想在 Agent 执行到某一步时，让人工介入做决策，然后 Agent 继续执行，这个流程要求 Agent 的状态必须能够在人工决策期间被持久保存。人工可能在 5 分钟后决策，也可能在两天后决策，这期间进程可能重启无数次。没有持久化，"等待人工确认"这个需求就无法实现，你只能把所有审批逻辑都放到 Agent 执行之前，而不是在执行过程中动态介入。

状态持久化不只解决"断点续传"的问题，它还是整个 Agent 可审计性的基础。当 Agent 产生错误决策时，你需要能够回溯到任意一个历史状态，查看当时的输入输出，找出问题根因。没有持久化的 Checkpoint，这种排查就无从进行。

另一个场景是多用户并发。Web 服务中，同时有 100 个用户在使用 Agent，每个用户的对话历史不能混在一起。thread_id 隔离机制需要持久化后端来支撑。

持久化方案对比

LangGraph 提供了四种 Checkpoint 后端，覆盖从开发到生产的不同需求：

PostgresSaver 在 PostgreSQL 中创建 4 张表：checkpoints（主表，状态存 JSONB）、checkpoint_blobs（大型对象存 BYTEA 二进制）、checkpoint_writes（节点中间写入暂存区）、checkpoint_migrations（schema 版本追踪）。

Redis Checkpoint（langgraph-checkpoint-redis）提供了三个实现：RedisSaver（线程级持久化）、ShallowRedisSaver（只保存最新 checkpoint，节省存储）和 RedisStore（跨线程长期记忆，支持向量搜索和元数据过滤）。需要注意的是：v0.1.0 引入了存储格式的破坏性变更，旧格式 checkpoint 无法直接读取，升级时需要迁移。

LangGraph Checkpoint 的具体结构：ThreadID 与会话隔离

每个 Checkpoint 对象包含以下字段：

# Checkpoint 的数据结构（简化版）
{
    "thread_id": "user-alice-session-001",      # 会话命名空间
    "checkpoint_id": "550e8400-e29b-41d4-a716", # UUID，唯一标识
    "parent_checkpoint_id": "3f6c9d2a-...",     # 父 checkpoint，构成链
    "checkpoint": {                              # 序列化的状态
        "v": 1,
        "ts": "2026-05-30T10:00:00Z",
        "channel_values": {                     # 各状态字段的值
            "messages": [...],
            "user_name": "Alice"
        }
    },
    "metadata": {                               # 附加信息（调试用）
        "source": "loop",
        "step": 3,
        "writes": {"chatbot": {"messages": [...]}}
    }
}

thread_id 是会话隔离的核心：同一个 thread_id 的所有 checkpoint 构成一条链，代表一个独立的用户会话。不同用户使用不同的 thread_id，数据完全隔离。

LangGraph 默认最多保存 1000 个 super-step 的 checkpoint（recursion_limit 参数控制），超出后图执行抛出 GraphRecursionError，防止无限循环。

断点续传：任务中断后如何从上次状态恢复

断点续传的实现机制是：图执行时每个节点完成后自动写入 Checkpoint；下次 invoke 传入相同的 thread_id，图会先读取最新 Checkpoint，从上次停止的地方继续，而非重头执行。

对于人机协同场景，这个机制支持"异步审批"：Agent 执行到需要审批的节点，暂停并保存状态；人工在几分钟、几小时甚至几天后做出决策；系统恢复执行，从暂停点继续。整个过程状态始终保存在外部存储中，不依赖任何内存状态。

多 Agent 共享状态的一致性问题

当多个 Agent 并行写入同一个 Checkpoint 时，会出现并发冲突。LangGraph 通过两个机制来处理这个问题。

第一是 Reducer 函数的幂等性设计：add_messages 这类 Reducer 通过消息 ID 去重，即使同一条消息被写入两次，最终状态里也只有一份。第二是 PostgreSQL 的事务保证：PostgresSaver 利用数据库的 ACID 事务，每次 Checkpoint 写入是原子操作，不会出现部分写入的中间状态。

对于更复杂的共享状态场景（多个独立的 Agent 需要读写同一个全局状态），建议使用 LangGraph 的 Store API（区别于 Checkpoint 的会话级存储，Store 提供跨线程的全局 key-value 存储）配合乐观锁或 RedisStore 的原子操作。

八、多 Agent 协作模式

三种经典拓扑

多 Agent 系统在架构层面有三种经典拓扑，每种适合不同的任务特征。

流水线（Pipeline）是最简单的拓扑：Agent A 的输出是 Agent B 的输入，B 的输出是 C 的输入，串行执行。这对步骤清晰、各步骤之间依赖关系单一的任务最有效。CrewAI 的 Process.sequential 就是流水线模式。延迟是这种模式的主要缺点，所有步骤串行，总时间是各步骤之和。选择流水线时，最值得设计的是每个步骤的输出格式：下游 Agent 能不能直接消费上游 Agent 的输出，还是需要一层格式转换？越早标准化接口，后期的维护成本越低。

主从（Orchestrator-Worker）是一个 Orchestrator Agent 协调多个 Worker Agent：Orchestrator 分析任务，拆分子任务，分派给合适的 Worker，收集结果，整合输出。CrewAI 的 Process.hierarchical 和 AutoGen 的 MagenticOneGroupChat 都是这种模式。主从模式支持并行（多个 Worker 同时处理不同子任务），但 Orchestrator 是单点瓶颈。如果 Orchestrator 决策失误，整个系统的输出都会偏差。

对等（Peer-to-Peer）是所有 Agent 地位平等，通过消息互相协作。AutoGen 的 RoundRobinGroupChat 和 SelectorGroupChat 是这种模式。灵活性最高，但也最难控制，没有明确的控制权，Agent 之间可能陷入无限对话循环，或者产生重复劳动。

Agent 间通信：消息队列 vs 共享黑板 vs 直接函数调用

直接函数调用是最简单的 Agent 间通信方式：Agent A 直接调用 Agent B 的方法，等待返回值。LangGraph 的 Send API 支持动态创建边，让 Agent 在运行时"调用"另一个子图。这种方式延迟最低，调用链清晰，但耦合度高，A 必须知道 B 的接口。

消息队列是 AutoGen 和 CrewAI 的主要通信方式：Agent A 发送消息到队列，Agent B 从队列消费。这种方式解耦发送方和接收方，支持异步处理，但引入了中间层的延迟和维护成本。AutoGen GroupChat 的 speaker_selection_method 本质上是消息队列加上一个路由器，LLM 决定消息发给谁。

共享黑板（Blackboard）是一种不同的设计：所有 Agent 读写同一个中央知识空间，不直接互通。一篇发表在 arxiv 的论文（arxiv 2507.01701）测试了这种架构，在 GPQA-Diamond 基准上达到 54.04%（对比 ChatEval 的 51.26%），MMLU 达 85.35%（对比 CoT 的 84.82%），平均超 ChatEval 1.59 个百分点。更重要的是 token 效率：4.72M tokens 完成任务，对比 ChatEval 的 5.45M 和 AFlow 的 16.69M（节约 72%）。

黑板架构的优势在于：Agent 数量易于增删而不影响其他组件，因为每个 Agent 只与黑板交互。控制单元（Controller）基于黑板当前状态动态决定激活哪些 Agent，K=4 轮共识检测保证输出质量。

协调冲突：当多个 Agent 争抢同一资源时怎么办

多 Agent 系统的一个常见问题是资源竞争：多个 Agent 同时想写入同一个文件、调用同一个有速率限制的 API、或者修改同一块共享状态。

处理资源竞争有三个层次：

序列化是最直接的方案，强制某类资源的访问串行化。实现简单，但在资源竞争激烈时会成为瓶颈，适合写操作频率低的场景。

乐观锁的做法是 Agent 读取资源时记录版本号，写入时检查版本号是否变化，如果版本号变了，说明有其他 Agent 已经修改，重试或放弃。这是 LangGraph Store API 的推荐用法，冲突低时性能接近无锁，冲突高时退化为频繁重试。

仲裁 Agent 是设置一个专门的 Agent 负责资源分配，其他 Agent 不能直接访问受保护资源，必须向仲裁 Agent 申请。这增加了一层间接，但把冲突处理逻辑集中在一处，最容易审计和调试。在 LangGraph 中，仲裁 Agent 可以实现为一个专门的节点，所有需要受保护资源的 Worker 节点通过条件边路由到这个仲裁节点，由它统一决策后再分发执行权。

实际案例：一个代码审查多 Agent 系统的架构

用一个具体的例子来说明多 Agent 协作如何在实践中落地。考虑一个自动化代码审查系统：给定一个 Pull Request，自动输出审查意见。

这个系统可以分解为以下角色：

Orchestrator Agent 接收 PR 信息，分析修改范围，把不同类型的审查任务分派给专职 Agent，维护整个审查的进度，合并各 Agent 的输出，生成最终报告。

Security Agent 专门扫描安全漏洞：SQL 注入、XSS、硬编码密钥、不安全的依赖项。这类检查有明确的规则，适合用工具调用（静态分析工具）加 LLM 解释的模式。

Logic Agent 分析业务逻辑正确性：函数边界条件、错误处理覆盖、数据一致性。这类检查需要理解上下文，适合纯 LLM 推理。

Style Agent 检查代码风格：命名规范、注释质量、函数长度、复杂度。这类检查主要用规则（Linter），LLM 负责解释规则违反的原因和改进建议。

# 使用 LangGraph 实现代码审查多 Agent 系统（架构示意）

from langgraph.graph import StateGraph, START, END
from typing import Annotated
from typing_extensions import TypedDict

class ReviewState(TypedDict):
    pr_diff: str
    security_findings: list[str]
    logic_findings: list[str]
    style_findings: list[str]
    final_report: str

def orchestrator(state: ReviewState):
    """分析 PR，决定激活哪些审查 Agent"""
    # 实际实现中，这里调用 LLM 分析 diff 的修改范围和风险
    return {}  # Orchestrator 通过条件边路由到各专职 Agent

def security_review(state: ReviewState):
    """安全审查节点"""
    diff = state["pr_diff"]
    # 调用静态分析工具 + LLM 解释
    findings = run_security_scanner(diff)
    return {"security_findings": findings}

def logic_review(state: ReviewState):
    """逻辑审查节点"""
    diff = state["pr_diff"]
    # 纯 LLM 推理
    findings = llm.invoke(f"分析以下代码修改的逻辑正确性：\n{diff}")
    return {"logic_findings": [findings.content]}

def style_review(state: ReviewState):
    """风格审查节点"""
    diff = state["pr_diff"]
    findings = run_linter(diff)
    return {"style_findings": findings}

def synthesize_report(state: ReviewState):
    """整合所有 Agent 的发现，生成最终报告"""
    all_findings = (
        state["security_findings"] +
        state["logic_findings"] +
        state["style_findings"]
    )
    report = llm.invoke(f"整合以下审查意见，生成结构化报告：\n{all_findings}")
    return {"final_report": report.content}

# 图结构：并行执行三个审查 Agent，最后合并
builder = StateGraph(ReviewState)
builder.add_node("orchestrator", orchestrator)
builder.add_node("security", security_review)
builder.add_node("logic", logic_review)
builder.add_node("style", style_review)
builder.add_node("synthesize", synthesize_report)

builder.add_edge(START, "orchestrator")
# 并行执行三个审查 Agent
builder.add_edge("orchestrator", "security")
builder.add_edge("orchestrator", "logic")
builder.add_edge("orchestrator", "style")
# 三个 Agent 完成后汇入 synthesize
builder.add_edge("security", "synthesize")
builder.add_edge("logic", "synthesize")
builder.add_edge("style", "synthesize")
builder.add_edge("synthesize", END)

这个架构的好处是：三个审查 Agent 并行执行，总时间接近最慢单个 Agent 的时间（而非三者之和）；每个 Agent 职责单一，便于独立调试和优化；Checkpoint 让审查过程可以中断和恢复，审计日志完整。

九、LangChain 生态下的调试与可观测性

为什么 Agent 比普通应用更难调试

传统 Web 应用的调试有明确的堆栈追踪：函数 A 调了函数 B，B 抛出异常，你看堆栈，找到问题行。Agent 的调试根本不同，问题往往不是代码崩溃，而是 LLM 在某个决策点做出了次优选择，或者工具返回了 Agent 没有预期到的格式，或者对话历史太长导致 LLM 遗忘了关键约束。

这类问题不会抛出异常，程序正常运行，但输出是错的。更难的是，同样的输入在不同运行中可能产生不同输出（LLM 的随机性），无法依赖单元测试完全覆盖。调试 Agent 需要的能力是：记录每次 LLM 调用的完整输入输出、工具调用的参数和返回值、状态在每个节点的变化，以及把这些数据关联起来，让你能够重放一次特定的执行过程。

LangSmith：LangChain 官方追踪工具

LangSmith 是 LangChain 官方提供的 Agent 可观测性平台。接入方式极简：设置两个环境变量，所有 LangChain/LangGraph 的调用都会自动上报追踪数据：

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."

# 之后所有 LangChain/LangGraph 调用都自动追踪
# 无需修改业务代码

在 LangSmith 的界面中，你可以看到每次调用的完整链路：从最外层的 graph.invoke() 到每个节点函数，到每次 LLM API 请求（包括完整的 messages 列表和 response）、每次工具调用（参数和返回值）。耗时数据按层级展示，能快速定位瓶颈。

LangSmith 还提供数据集和评估功能：把一组输入-输出对存为数据集，定义评估标准（正确性、相关性、格式等），批量运行 Agent 并评分。这把 Agent 的质量评估从主观判断变成了可度量的数字，是把 Agent 从原型推向生产的关键步骤。

本地调试：LangGraph 的 Checkpoint 时间线

不用 LangSmith 时，LangGraph 的 Checkpoint 历史本身就是一个强大的调试工具。graph.get_state_history(config) 返回这个 thread 的所有历史 Checkpoint，从最新到最旧排列：

# 获取并打印 checkpoint 历史，用于调试
config = {"configurable": {"thread_id": "debug-session-001"}}
history = list(graph.get_state_history(config))

for checkpoint in history:
    print(f"Step {checkpoint.metadata.get('step', '?')}")
    print(f"  Node: {checkpoint.metadata.get('source', '?')}")
    print(f"  Messages: {len(checkpoint.values.get('messages', []))}")
    print(f"  Checkpoint ID: {checkpoint.config['configurable']['checkpoint_id'][:8]}...")
    print()

更强大的是，你可以把图的状态"回退"到某个历史 Checkpoint，然后重新执行：

# 找到出问题的那步 checkpoint
target_checkpoint = history[3]  # 第 4 个 checkpoint（从新到旧）

# 从这个 checkpoint 的状态重新 invoke
result = graph.invoke(
    None,  # 输入为 None，使用 checkpoint 中保存的状态
    config={
        "configurable": {
            "thread_id": "debug-session-001",
            "checkpoint_id": target_checkpoint.config["configurable"]["checkpoint_id"]
        }
    }
)

这让"从断点重放"成为可能：在开发阶段，你可以精确定位问题发生的节点，修改代码后从那个点重新执行，而不需要从头触发整个工作流。

十、框架选型的决策逻辑

面对一个具体的 Agent 项目，选型决策可以沿着三个维度进行。

第一个维度是工作流的控制需求。控制流是否需要精确定义、可回溯、可审计？选 LangGraph。任务分工清晰、流程相对固定？选 CrewAI。需要 Agent 间自然语言协商推理？选 AutoGen。核心能力是文档处理和 RAG？选 LlamaIndex。需要快速拼接不同来源的组件做原型？选 LangChain。

第二个维度是生产化要求。需要跨会话状态持久化和审计轨迹？LangGraph 原生支持，其他框架需要自行集成。需要高并发？LangGraph + PostgresSaver 或 RedisSaver 是经过验证的组合。需要控制 token 成本？避免 AutoGen 的 GroupChat 模式（20+ 次 LLM 调用），优先 LangGraph 的图编排（O(1) 调度复杂度）或 CrewAI 的顺序模式。

第三个维度是团队能力。框架的抽象层有学习成本。LangGraph 的图结构需要状态机思维；AutoGen 的对话模型更接近自然语言描述；CrewAI 最接近非技术团队的心智模型。选择团队能够理解和维护的框架，比选择"最强"的框架更重要。

框架对比汇总：

一个可执行的起点：用 LlamaIndex 构建文档检索层，用 LangGraph 管理工作流状态和持久化，用 CrewAI 的角色抽象定义业务逻辑，三者可以混用，通过标准的 Python 接口集成。不需要从一开始就做全局选型，从最小可用版本跑起来，再根据实际痛点决定是否引入框架。

框架的边界：什么时候应该放弃框架

框架不是银弹。有几个信号提示你可能应该减少对框架的依赖。

当你花在理解框架内部机制上的时间，超过了花在业务逻辑上的时间，说明框架的抽象层对你的场景来说过重。当框架的某个决策（比如 LangChain 的某个默认行为）与你的需求冲突，而绕过它需要 monkey patch 或 fork 框架代码，手写一个精简的实现往往更合理。当团队里只有一个人理解框架，其他人完全绕开它写代码，说明框架没有真正被团队接受，维护风险很高。

"Harness 厚度"是 Anthropic 工程师用来描述这个权衡的术语：Harness 可以很厚（框架提供大量功能，开发者配置使用），也可以很薄（框架只提供最基础的连接层，开发者自己实现逻辑）。薄 Harness 意味着更多手工代码，但每一行代码你都完全掌控；厚 Harness 意味着更快上手，但当框架出问题或需要定制时，代价更大。

Anthropic 自己在设计 Claude Code 时赌的是薄 Harness：尽量少依赖第三方框架，把复杂度放在精心设计的业务逻辑里，而非框架魔法里。这个选择让调试和维护更直接，但要求团队对每一个基础设施决策都做过深入思考。

对于大多数团队，合理的路径是：用框架快速跑通第一个版本，在实际使用中识别哪些框架抽象真正有价值、哪些只是增加了复杂度，然后有针对性地保留有价值的部分，替换掉带来麻烦的部分。这比一开始就手写一切，或者一开始就全面押注某个框架，都更务实。

从现在开始的行动建议：选一个你手头最近的 Agent 需求，限定一天时间，用 LangGraph 实现一个带 Checkpoint 的最小版本，不超过 100 行代码，跑通主路径，然后看哪里的摩擦最大。这比读十篇框架对比文章更能帮你做出真实的选型判断。

十一、框架选型指南

下面这张表可以直接带到项目选型时用。行是任务类型，列是关键维度，最后一列给出推荐框架和备注。

两个使用这张表的注意事项。

第一，"推荐"不等于"唯一合法"。表里的推荐是在没有其他约束时的起点，团队已有的框架积累、现有的基础设施（比如已有 PostgreSQL 但没有 Redis）、以及对框架的熟悉程度，都可以合理地覆盖表中的建议。

第二，框架可以叠加。LlamaIndex + LangGraph 是最常见的组合：前者解决数据接入和检索，后者解决状态机和持久化，两者通过普通 Python 函数调用集成，接口边界清晰。CrewAI + LangGraph 也有实际案例：用 CrewAI 的角色定义描述业务分工，把每个 CrewAI 任务包装成 LangGraph 节点，获得图编排的可审计性。不同框架的职责是正交的，数据层、控制流层、角色描述层可以分别选型。

框架选型没有永久正确的答案。六个月后你的 Agent 规模和复杂度变了，选型的最优解也会变。保留切换的能力（避免深度绑定框架内部 API、保持业务逻辑与框架逻辑的分离）比第一次选对更重要。

作者：toy