Agent 框架赛道 — 深度对比研究

调研时间：2026-06-22 基于 trpc-agent-go / trpc-agent-python / LangChain / LangGraph / CrewAI / Agno 源码阅读

总览：六大框架核心画像

1. LangChain — “Everything is a Runnable”

架构概览

LangChain 的核心抽象是 Runnable：任何可以接收输入、产出输出的组件（LLM、Tool、Retriever、Parser）都实现 Runnable 接口。通过 LCEL（LangChain Expression Language）的管道语法 | 把它们组合起来。

# LCEL 管道语法——一切皆 Runnable
chain = prompt | llm | output_parser
result = chain.invoke({"question": "什么是 Agent?"})

日常类比：Unix 命令的管道 cat file | grep error | wc -l。每个命令只做一件事，用 | 串起来完成复杂任务。LangChain 的 LCEL 就是 LLM 世界的管道。

仓库结构（Monorepo）

langchain/
├── libs/
│   ├── core/           # Runnable/LCEL/BaseMessage 等基础抽象
│   ├── langchain/      # 高层 Agent/Chain 实现
│   ├── community/      # 社区贡献的集成
│   └── partners/       # 官方 partner 包（openai/anthropic/google 等）
├── templates/          # 可复用的应用模板
└── docs/               # 文档站源码

关键数据：100+ partner 集成包，覆盖几乎所有主流 LLM/向量库/工具。

Runnable 协议深度解析

Runnable 是 LangChain 最核心的抽象，定义了 5 个标准方法：

class Runnable(Generic[Input, Output], ABC):
    def invoke(self, input: Input, config: RunnableConfig) -> Output: ...
    async def ainvoke(self, input: Input, config: RunnableConfig) -> Output: ...
    def batch(self, inputs: List[Input], ...) -> List[Output]: ...
    def stream(self, input: Input, ...) -> Iterator[Output]: ...
    async def astream(self, input: Input, ...) -> AsyncIterator[Output]: ...

每个 Runnable 都自动拥有：同步/异步/批量/流式 四种调用模式。这意味着你写一个组件，免费获得四种执行策略。

LCEL 组合子

可观测性：LangSmith

LangSmith 是 LangChain 的商业可观测性平台，提供：全链路 trace 可视化、每步 token 用量统计、Prompt 版本管理、回归测试集。核心价值是把”链”的执行过程变成可审计的时间线。

设计取舍分析

优势：

• 生态碾压：138K star + 100+ 集成，任何新 LLM/工具几天内就有 partner 包
• LCEL 学习曲线低，Unix 管道直觉
• 文档和教程极其丰富

代价：

• 代码量庞大（15000+ 文件），版本演进有历史包袱
• 抽象层多，调试时 stack trace 很深
• 有状态的复杂 Agent 逻辑被推给了 LangGraph——LangChain 本身是无状态链

2. LangGraph — “Graph is Program”

设计哲学

LangGraph 的核心信念：复杂 Agent 的决策流程本质上是一个有向图。节点（Node）是计算步骤，边（Edge）是条件跳转，状态（State）在节点间通过 Channel 传递。

日常类比：地铁线路图。你从起点出发，遇到换乘站（条件边）按规则选方向，可以做环线（循环图），可以在某站等人（human-in-the-loop），还能在任意站存档（checkpoint）。

Channel 系统——状态管理的核心

LangGraph 定义了两种基本 Channel：

# LastValue Channel：只保留最后一次写入的值
class LastValue(Channel[Value]):
    """覆盖式更新——新值覆盖旧值"""
    value: Value

# BinaryOperatorAggregate Channel：用 reducer 函数聚合所有写入
class BinaryOperatorAggregate(Channel[Value]):
    """累积式更新——用 operator 合并新旧值"""
    value: Value
    operator: Callable[[Value, Value], Value]  # 如 operator.add

为什么这个设计很精彩？

想象多个并行节点同时向同一个 state key 写入结果。LastValue 只保留最后一个（适合”覆盖型”状态如 current_answer），BinaryOperatorAggregate 可以用 operator.add 把所有写入合并（适合”累积型”状态如 messages）。

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 累积消息列表
    current_plan: str                         # 覆盖式更新
    tool_results: Annotated[list, operator.add]  # 累积工具结果

Pregel 执行引擎

LangGraph 的执行引擎叫 Pregel（致敬 Google 的图计算框架）。核心文件 pregel/__init__.py 长达 4380 行，是整个项目最复杂的部分。

执行流程：

1. 初始化 State → 写入所有 Channel
2. 确定可执行节点（入度为 0 或条件满足）
3. 并行执行节点 → 每个节点读取 Channel、计算、写入新值
4. Channel 更新 → 触发 reducer
5. 评估边条件 → 确定下一批可执行节点
6. 循环 2-5 直到：终止节点 / GraphInterrupt / 无可执行节点

关键设计决策：Pregel 是同步超步（superstep）模型——同一超步内的节点并行执行，超步之间严格有序。这保证了状态一致性，代价是不能跨超步流水线化。

Checkpoint 持久化

LangGraph 的 Checkpoint 机制让有向图的执行可以：中断后恢复、回滚到历史状态、fork 分支探索。

# Checkpoint 存储接口
class BaseCheckpointSaver(ABC):
    def put(self, config, checkpoint, metadata): ...
    def get(self, config): ...
    def list(self, config, filter): ...

内置支持：SQLite、PostgreSQL、Redis。Checkpoint 保存的是完整 Channel 快照——不是增量 diff，所以恢复是 O(1) 的。

GraphInterrupt——Human-in-the-Loop 的精髓

# 在审批节点中触发中断
def approval_node(state):
    if state["needs_human_review"]:
        raise GraphInterrupt(
            value={"question": "确认执行此操作?", "context": state["plan"]}
        )
    return state

# 人类审批后恢复
graph.invoke(Command(resume={"approved": True}), config=thread_config)

GraphInterrupt 不是简单的暂停——它把当前状态完整存入 Checkpoint，人类可以在任意时间后恢复，甚至可以修改 state 再恢复。这比传统的回调/webhook 模式强大得多。

设计取舍分析

优势：

• 图模型表达力极强：循环、分支、并行、中断都是一等公民
• 状态管理是最严谨的（Channel + Checkpoint 保证一致性）
• ~400 文件核心精简，可以通读

代价：

• 学习曲线陡峭：Channel/Pregel/Checkpoint 三层抽象需要时间理解
• 简单场景过度工程化——一个线性 Agent 用不到图
• 强绑定 LangChain 生态（虽然理论上可独立使用）

3. CrewAI — “Role Play is All You Need”

设计哲学

CrewAI 的核心洞察：人类协作的最成功模式就是角色分工。与其让开发者思考”图节点”“状态通道”这些抽象概念，不如让他们想”我需要一个研究员、一个写手、一个审核员”。

日常类比：剧组拍电影。导演（Process）安排工作顺序，演员（Agent）各有角色定位，每个演员按剧本（Task）表演。不需要理解”有向图”，只需要知道”谁做什么”。

Agent 定义：Role + Goal + Backstory

researcher = Agent(
    role="高级研究分析师",
    goal="发现 AI Agent 领域的最新技术趋势",
    backstory="""你是一位拥有 10 年经验的技术研究员，
    擅长从海量信息中提炼关键洞察。你特别关注
    技术的实际落地价值而非纯概念创新。""",
    tools=[search_tool, arxiv_tool],
    llm=ChatOpenAI(model="gpt-4"),
    verbose=True,
    allow_delegation=True
)

为什么 Backstory 很重要？ 它不是装饰——它给 LLM 提供了决策的”性格基准”。同一个 Goal，不同 Backstory 的 Agent 会采取完全不同的策略。这是 CrewAI 对 prompt engineering 的结构化封装。

Crew 编排

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 顺序执行
    # process=Process.hierarchical,  # 层级：manager 分配任务
    verbose=True
)
result = crew.kickoff()

两种 Process 模式：

Flows v2——超越单次 Crew 的编排

CrewAI 2025 年引入了 Flows：在多个 Crew 之间建立编排逻辑。

class ContentPipeline(Flow):
    @start()
    def research_phase(self):
        return research_crew.kickoff()

    @listen(research_phase)
    def writing_phase(self, research_output):
        return writing_crew.kickoff(inputs={"research": research_output})

    @router(writing_phase)
    def quality_gate(self, draft):
        if draft.quality_score > 0.8:
            return "publish"
        return "revise"

Flows 弥补了早期 CrewAI 只能做单次线性执行的局限——现在可以做条件路由、循环重试、并行分支。

设计取舍分析

优势：

• 认知负担最低：Role/Goal/Backstory 直觉强，非技术人员也能理解
• 代码量少（~800 文件），上手极快
• 角色设计的 prompt engineering 效果出奇好

代价：

• 状态管理弱——Crew 执行期间没有 LangGraph 那种精细的 Channel/Checkpoint
• Sequential/Hierarchical 二选一太粗粒度，复杂工作流表达力不足
• 对底层 LLM 调用的控制力有限（不如 LangChain 的 Runnable 组合子灵活）

4. Agno — “Minimal Abstraction, Max Power”

设计哲学

Agno（前身 PhiData）的核心信条：Agent 框架不应该比你自己写的代码更复杂。整个 Agent 定义就是一个 Python dataclass，不需要学习任何新的 DSL 或编排概念。

日常类比：Python 脚本。能跑就行，需要什么 import 什么。不搞 Spring Boot 那种 20 个配置文件才能跑一个 Hello World 的事。

1741 行 Agent Dataclass

Agno 的 Agent 类是一个巨大的 dataclass（1741 行），把所有 Agent 能力平铺为字段：

@dataclass
class Agent:
    # 身份
    name: str = "Agent"
    role: Optional[str] = None
    description: Optional[str] = None

    # 模型
    model: Optional[Model] = None
    
    # 工具（190+ 内建）
    tools: Optional[List[Union[Tool, Toolkit, Callable]]] = None
    
    # 记忆
    memory: Optional[AgentMemory] = None
    add_history_to_messages: bool = False
    num_history_responses: int = 3
    
    # 知识库
    knowledge: Optional[AgentKnowledge] = None
    
    # 推理
    reasoning: bool = False
    reasoning_model: Optional[Model] = None
    
    # 结构化输出
    response_model: Optional[Type[BaseModel]] = None
    structured_outputs: bool = False
    
    # ... 更多字段

设计哲学的体现：不用继承、不用装饰器模式、不用 Builder 模式。所有配置都是构造函数参数。你看到 Agent() 的签名就知道它能做什么。

Team 协作：4 种模式

team = Team(
    agents=[researcher, writer, reviewer],
    mode="coordinate",  # 4 种模式之一
)

AgentOS 运行时

Agno 2025 年推出了 AgentOS：一个统一的 Agent 部署运行时，功能包括：

• Agent 注册与发现
• 会话管理（多租户）
• 向量知识库管理
• 监控与日志

from agno.playground import Playground

app = Playground(agents=[researcher, writer]).get_app()
# 一行代码部署 Agent 为 Web 服务

190+ 内建工具

Agno 的工具覆盖极广：

搜索类：DuckDuckGo / Google / Bing / Arxiv / Wikipedia
代码类：Python / Shell / Docker
数据类：SQL / CSV / JSON / Pandas
多媒体：Image / Audio / Video / PDF
网络类：HTTP / WebSocket / Email
文件类：File / S3 / GCS

对比：LangChain 的工具通过 partner 包分散分布，Agno 全部内建——不需要额外安装。

LearningMachine

Agno 独有的”自我学习”机制：Agent 在执行过程中可以把成功策略存入长期记忆，下次遇到类似任务直接复用。

agent = Agent(
    learning=True,  # 开启自动学习
    learning_db=PgVector(table="agent_learnings"),
)

设计取舍分析

优势：

• 代码量最少的全功能框架（~600 文件）
• 零学习曲线：会 Python 就会用
• 190+ 内建工具，不需要到处找集成包
• AgentOS 运行时让部署很简单

代价：

• 1741 行 dataclass 违反了单一职责原则——Agent 类做了太多事
• 状态管理不如 LangGraph 精细（没有 Channel 系统）
• Team 的 4 种模式相对静态，不如图模型灵活
• 社区生态不如 LangChain（39K vs 138K）

5. tRPC-Agent-Go — “Agent = Microservice”

设计哲学

tRPC-Agent-Go 的核心信念：Agent 本质上就是一个微服务。既然如此，为什么不直接用微服务框架来构建 Agent？所有微服务治理能力（服务注册发现、负载均衡、熔断限流、链路追踪）Agent 天然就需要。

日常类比：公司组织架构。每个部门（Agent）是独立运作的团队，通过公司 OA 系统（tRPC 框架）互相发工单协作。不需要额外造一套”Agent 通信协议”——企业已有的组织管理方式就够用了。

4-Method Agent Interface

整个 Agent 定义只需要实现 4 个方法：

type Agent interface {
    // Run：Agent 的核心执行逻辑
    Run(ctx context.Context, input *Input) (*Output, error)
    
    // Tools：Agent 可以使用的工具列表
    Tools() []Tool
    
    // Info：Agent 的元信息（名称、描述、能力）
    Info() *AgentInfo
    
    // SubAgents：Agent 可以调度的子 Agent 列表
    SubAgents() []Agent
}

极简但完备：4 个方法覆盖了 Agent 的全部能力——执行（Run）、工具使用（Tools）、自我描述（Info）、层级协作（SubAgents）。对比 LangChain 的 Runnable 需要 5 个方法 + 一堆 Mixin，tRPC-Agent 的接口更干净。

2339 行 Invocation Context

Invocation 是 Agent 执行时的上下文对象，包含：

type Invocation struct {
    // 身份
    Agent      Agent
    AgentInfo  *AgentInfo
    
    // 状态
    State      *State
    Memory     *Memory
    
    // 工具
    Tools      []Tool
    ToolResults map[string]*ToolResult
    
    // 通信
    Messages   []*Message
    Events     chan *Event
    
    // 控制
    Plan       *Plan
    MaxSteps   int
    CurrentStep int
    
    // 链路
    TraceID    string
    SpanID     string
    ParentSpanID string
}

Invocation 是 tRPC-Agent 最重的对象（2339 行），集中了执行期间的所有上下文。设计取舍：集中式 context 方便访问但耦合度高。

3971 行 Runner

Runner 是 Agent 的执行引擎：

type Runner struct {
    agent       Agent
    model       Model
    memory      Memory
    tools       []Tool
    subAgents   []Agent
    planner     Planner
    maxSteps    int
    // ...
}

func (r *Runner) Run(ctx context.Context, input string) (*Output, error) {
    // 1. 初始化 Invocation
    // 2. 加载 Memory
    // 3. 执行 Planning（如果有 Planner）
    // 4. ReAct 循环：Think → Act → Observe
    // 5. 保存 Memory
    // 6. 返回结果
}

Runner 的 ReAct 循环是核心执行模式：Agent 思考（Think）要做什么 → 调用工具（Act）→ 观察结果（Observe）→ 决定是否继续。这和 LangChain 的 AgentExecutor 类似，但 Runner 额外管理了 Memory 和 Planning。

Team 双模式：Coordinator + Swarm

// Coordinator 模式：有中心调度者
team := NewTeam(
    WithMode(ModeCoordinator),
    WithAgents(researcher, writer, reviewer),
    WithCoordinatorAgent(manager),  // 显式指定协调者
)

// Swarm 模式：去中心化，Agent 间直接握手转交
team := NewTeam(
    WithMode(ModeSwarm),
    WithAgents(researcher, writer, reviewer),
    // 无协调者——Agent 根据 HandoffConfig 自行决定转交给谁
)

StateGraph——LangGraph 的 Go 等价物

tRPC-Agent-Go 内建了一个 6192 行 的 StateGraph 实现，功能等价于 LangGraph：

graph := NewStateGraph(AgentState{})

graph.AddNode("research", researchNode)
graph.AddNode("write", writeNode)
graph.AddNode("review", reviewNode)

graph.AddEdge("research", "write")
graph.AddConditionalEdge("write", routeByQuality, map[string]string{
    "pass": "end",
    "fail": "review",
})
graph.AddEdge("review", "write")  // 循环边

graph.SetEntryPoint("research")
app := graph.Compile()

双引擎架构：开发者可以选择 Team（高层协作抽象）或 StateGraph（底层图控制），甚至可以把 Team 作为 StateGraph 的一个节点嵌入。

Memory 双类型：Fact + Episode

type Memory interface {
    // Fact Memory：提炼后的知识（如"用户偏好中文回复"）
    StoreFact(ctx context.Context, fact *Fact) error
    QueryFacts(ctx context.Context, query string) ([]*Fact, error)
    
    // Episode Memory：原始对话历史
    StoreEpisode(ctx context.Context, episode *Episode) error
    QueryEpisodes(ctx context.Context, query string) ([]*Episode, error)
}

9 种 Session 后端：Redis / PostgreSQL / MySQL / MongoDB / SQLite / DynamoDB / Firestore / Memory / Custom。生产环境的选择极其灵活。

ReAct Planner

type Planner interface {
    Plan(ctx context.Context, inv *Invocation) (*Plan, error)
    Replan(ctx context.Context, inv *Invocation, feedback string) (*Plan, error)
}

Planner 在 Runner 循环开始前生成执行计划，执行过程中可以根据反馈 Replan。这给了 Agent 一个”先想后做”的能力——不是每一步都问 LLM”接下来干嘛”，而是先规划全局再逐步执行。

协议生态：A2A + AG-UI + MCP + OpenAI

四协议全覆盖意味着 tRPC-Agent 构建的 Agent 可以：被其他 Agent 调用（A2A）、接入任意前端（AG-UI）、使用任意工具（MCP）、兼容 OpenAI 生态（OpenAI API）。

可观测性：OpenTelemetry + Langfuse

// 自动注入 trace 到每次 LLM 调用和工具调用
runner := NewRunner(
    WithAgent(myAgent),
    WithTracing(OpenTelemetryTracer{}),  // OTel 标准
    WithLangfuse(langfuseConfig),         // Langfuse 可视化
)

继承 tRPC 框架的全链路追踪能力，同时适配 AI 专用的 Langfuse 平台。Agent 的每一步（LLM 调用、工具执行、Agent 间通信）都自动产生 span。

设计取舍分析

优势：

• Go 语言在 Agent 框架赛道几乎无竞品——微服务场景有天然优势
• 微服务治理能力（负载均衡/熔断/限流/链路追踪）零成本继承
• 腾讯内部大规模生产验证（元宝、腾讯体育）
• 代码精简（~200 文件），通读成本最低
• 四协议覆盖，互操作性最强

代价：

• Star 数 <1K，社区生态尚未成熟
• Go 语言的 AI/ML 生态弱于 Python（模型推理、数据处理库少）
• 文档和教程相对匮乏（相比 LangChain/CrewAI）
• StateGraph 6192 行说明复杂度集中在少数文件中——维护门槛高

6. tRPC-Agent-Python — “LangGraph Wrapper + tRPC Ecosystem”

设计定位

tRPC-Agent-Python 不是 Go 版的 Python 翻译——它是用 Python 生态（Pydantic + LangGraph）封装一层，接入 tRPC 后端服务治理。

核心架构

from pydantic import BaseModel
from langgraph.graph import StateGraph

class MyAgentState(BaseModel):
    messages: list = []
    tool_results: list = []
    current_step: str = "init"

class MyAgent(BaseAgent):
    state_schema = MyAgentState
    
    def build_graph(self) -> StateGraph:
        graph = StateGraph(self.state_schema)
        graph.add_node("think", self.think)
        graph.add_node("act", self.act)
        graph.add_conditional_edges("think", self.should_act)
        return graph

Filter Pipeline

tRPC-Agent-Python 独有的 Filter 机制：在 Agent 执行前后插入处理逻辑。

class LoggingFilter(Filter):
    def before(self, ctx, input):
        logger.info(f"Agent {ctx.agent_name} 开始执行")
        return input
    
    def after(self, ctx, output):
        logger.info(f"Agent {ctx.agent_name} 完成，耗时 {ctx.elapsed}ms")
        return output

agent = MyAgent(filters=[LoggingFilter(), AuthFilter(), RateLimitFilter()])

Filter 链的设计类似 Java 的 Servlet Filter 或 Go 的 Middleware——在不修改 Agent 逻辑的前提下注入横切关注点（日志、鉴权、限流）。

Parent/Sub Agent 树

class OrchestratorAgent(BaseAgent):
    sub_agents = [ResearchAgent(), WritingAgent(), ReviewAgent()]
    
    def route(self, state):
        # 根据状态决定交给哪个子 Agent
        if state.needs_research:
            return "research"
        elif state.needs_writing:
            return "writing"
        return "review"

设计取舍分析

优势：

• Pydantic 类型安全 + IDE 自动补全
• 复用 LangGraph 成熟的图引擎（不重复造轮子）
• Filter 机制优雅解决横切关注点
• Python 生态丰富，AI/ML 库随手可得

代价：

• 本质是 LangGraph 的 wrapper——受限于 LangGraph 的设计边界
• 相比 Go 版缺少 StateGraph 级别的自研深度
• 社区更小，文档更少

横向对比：6 大维度

维度 1：Agent 定义方式

维度 2：状态管理

维度 3：多 Agent 协作

维度 4：生产就绪度

维度 5：框架代码规模

LangChain:          ~15,000+ files  — 生态庞大，通读不现实
CrewAI:              ~800 files     — 中等规模
Agno:                ~600 files     — 精简
LangGraph:           ~400 files     — 核心精简
tRPC-Agent-Go:       ~100 files     — 最精简（Go 版核心）
tRPC-Agent-Python:   ~100 files     — 最精简

维度 6：协议支持

框架演进趋势

趋势 1：从链到图到微服务

2022: LangChain — 链式调用（线性）
2023: LangGraph — 有向图（非线性 + 状态）
2024: tRPC-Agent — 微服务原生（分布式 + 治理）

每一代解决上一代的核心痛点：链无法表达循环→图解决；图无法分布式扩展→微服务解决。

趋势 2：从重抽象到极简

LangChain 全家桶 → CrewAI 角色简化 → Agno 极简 dataclass

市场在”功能完备”和”使用简单”之间反复摇摆。当一个框架变得太复杂，就会有新框架以”更简单”切入。

趋势 3：协议标准化

2025-2026 年涌现的四大协议（MCP/A2A/AG-UI/OpenAI API）正在定义 Agent 世界的”互操作性标准”。未来框架的竞争力不只看功能——还看”能不能和别人对话”。

趋势 4：从 Python 独占到多语言

LangChain/LangGraph 已有 JS/TS 版本，tRPC-Agent 以 Go 为主。未来 Agent 框架必然多语言化——因为生产环境不可能只有 Python。

趋势 5：可观测性从附加项变为核心

早期框架把可观测性当商业增值（LangSmith），新一代框架（tRPC-Agent/Agno）把它当内建能力。在 Agent 行为不可预测的背景下，”看不到就管不了”。

贡献机会分析

高价值贡献方向（按 ROI 排序）

犀牛鸟参与者的建议路径

1. 第 1-2 周：通读 tRPC-Agent-Go 核心代码（~100 文件），理解 4-Method Interface + Runner + Team + StateGraph
2. 第 3 周：对比 LangGraph 的 Channel/Pregel 设计，理解两种状态管理哲学的差异
3. 第 4 周：选择一个贡献方向，提交第一个 PR
4. 后续：根据 Mentor 反馈迭代，逐步深入 StateGraph 或 Memory 子系统

总结

六大框架代表了 Agent 框架设计的四个流派：

1. 管道流派（LangChain）：万物皆 Runnable，组合子拼接
2. 图流派（LangGraph）：程序即有向图，Channel 传状态
3. 角色流派（CrewAI/Agno）：Agent 就是人，协作就是分工
4. 微服务流派（tRPC-Agent）：Agent 就是服务，RPC 就是协作

没有”最好”的框架——只有最适合场景的设计。但从”学习 Agent 框架设计”的角度，tRPC-Agent-Go 的代码精简度 + 微服务原生设计 + 四协议覆盖使其成为最高效的学习标的。