第十章：跨框架对比——生产就绪度与协议生态

本章目标：理解”从 Demo 到生产”的鸿沟，掌握可观测性、水平扩展、容错机制和四大 Agent 协议的战略意义，建立”能跑不等于能上线”的认知。

10.1 什么是”生产就绪”？

日常类比

你在家做了一道菜，味道不错。但如果你要开一家餐厅，”味道不错”远远不够——你还需要：

1. 可观测：厨房里有温度计、计时器、监控摄像头（知道每个环节的状态）
2. 可扩展：中午高峰期来了200个客人，你得有足够的灶台和厨师（应对流量增长）
3. 可恢复：某个菜烧糊了，你能快速重做（不影响其他桌）
4. 可限制：客人不能冲进后厨自己动手（安全边界和权限控制）

Agent 系统也一样。Demo 跑通了只是”在家做菜”；生产就绪要求可观测 + 可扩展 + 可恢复 + 可限制。

技术定义

生产就绪（Production Readiness） 是一组能力的集合：

Demo 与生产的鸿沟

Demo 阶段关心的：                生产阶段关心的：
- 能不能跑通？                   - 跑 10000 次能稳定吗？
- 输出对不对？                   - 每次耗费多少 token/钱？
- 好不好玩？                     - 出错了怎么发现、怎么恢复？
                                  - 100个用户同时用会怎样？
                                  - Agent 做了危险操作怎么拦？
                                  - 能接入公司现有的监控系统吗？
                                  - 法规合规（数据不出境、操作可审计）？

10.2 可观测性对比

为什么 Agent 的可观测性特别难？

传统软件的可观测性相对简单——你知道代码会按确定路径执行。但 Agent 系统的挑战在于：

1. 路径不确定：LLM 可能调用任意工具、任意次数，执行路径由运行时决定
2. 成本不可预测：一次对话可能花 $0.01 也可能花 $5，取决于 LLM 决定循环多少次
3. 质量难量化：Agent 的输出是自然语言，怎么自动判断”质量好不好”？
4. 链路长：一次用户请求可能经过 5 个 Agent、20 次工具调用、50 次 LLM 调用

tRPC-Agent：OTel + Langfuse + tRPC 全套

tRPC-Agent 继承了 tRPC 框架的企业级监控能力，并在此基础上增加了 Agent 特有的观测：

// tRPC-Agent 的可观测性层次

// 第1层：tRPC 基础设施监控（自动）
// - 请求/响应延迟（P50/P95/P99）
// - 错误率和错误类型分布
// - 连接数、并发数
// - 服务间调用链路（分布式 trace）

// 第2层：Agent 特有指标
agent := NewAgent(
    WithLLM(gpt4),
    WithMetrics(MetricsConfig{
        // Token 消耗监控
        TrackTokenUsage: true,    // 每次 LLM 调用的输入/输出 token
        TrackCost:       true,    // 按模型定价计算实际成本
        
        // 执行轨迹
        TrackToolCalls:  true,    // 每次工具调用的入参/出参/耗时
        TrackLoopCount:  true,    // ReAct 循环了多少轮
        
        // 质量指标
        TrackLatency:    true,    // 端到端响应时间
        TrackRetries:    true,    // 重试次数
    }),
    
    // 第3层：OpenTelemetry 集成
    WithOTelTracer(tracer),      // 标准 OTel trace
    WithOTelMeter(meter),        // 标准 OTel metrics
    
    // 第4层：Langfuse 集成（AI 专属可观测平台）
    WithLangfuse(LangfuseConfig{
        PublicKey: "pk-xxx",
        SecretKey: "sk-xxx",
        // Langfuse 提供：
        // - 对话回放（逐步查看 Agent 的决策过程）
        // - 成本分析（按用户/按功能/按时间的成本breakdown）
        // - 质量评估（自动/人工打分）
        // - A/B 测试（不同 prompt 的效果对比）
    }),
)

// 实际监控面板能看到：
// ┌──────────────────────────────────────┐
// │ Agent "research_bot" - 最近 1 小时    │
// ├──────────────────────────────────────┤
// │ 请求数: 1,234                         │
// │ 平均延迟: 3.2s (P95: 8.1s)           │
// │ 平均 Token: 2,800/请求               │
// │ 平均成本: $0.04/请求                  │
// │ 工具调用分布: search(45%), read(30%), │
// │              write(15%), other(10%)   │
// │ 循环次数分布: 1-3次(70%), 4-7次(25%),│
// │              8+次(5%)                 │
// │ 错误率: 2.1% (主要是 API timeout)    │
// └──────────────────────────────────────┘

核心优势：

• 零侵入：大部分指标自动采集，不需要在业务代码中手动埋点
• 标准化：基于 OpenTelemetry 标准，可接入任何兼容的监控后端（Prometheus, Grafana, Datadog…）
• 多层次：从基础设施到 Agent 业务逻辑，全链路覆盖
• 成本可控：实时看到”这个功能每天花多少钱”，及时发现成本异常

LangGraph：LangSmith（商业平台）

LangGraph 的可观测性依赖 LangSmith——LangChain 公司的商业 SaaS 平台：

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-xxx"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

# 配置后，所有 LangGraph/LangChain 的调用自动上报到 LangSmith

# LangSmith 提供：
# 1. Trace 可视化：完整的执行图，每个节点的输入/输出/耗时
# 2. 对话回放：逐步回放 Agent 的决策过程
# 3. 数据集管理：收集真实案例，用于评估和改进
# 4. 在线评估：自动评估 Agent 输出质量
# 5. 提示版本管理：A/B 测试不同的 prompt
# 6. 成本追踪：token 使用量和成本统计

优势：

• 与 LangGraph 深度集成，零配置即可获得丰富的 trace 信息
• 可视化界面优秀——图的执行路径、节点颜色标注
• 提供 evaluation 框架——可以系统性地评估 Agent 质量

劣势：

• 商业付费：免费 tier 有限制（每月 5000 次 trace）
• 数据外发：trace 数据发送到 LangSmith 的服务器（有 self-hosted 选项但成本高）
• 锁定效应：深度绑定 LangChain/LangGraph 生态
• 不支持 OTel 标准：不能直接接入企业已有的 Prometheus/Grafana 监控

CrewAI：CrewAI+（商业平台）

import os
os.environ["CREWAI_API_KEY"] = "ck-xxx"

# CrewAI+ 提供：
# 1. 执行监控：Crew 的执行状态和进度
# 2. Agent 性能：每个 Agent 的耗时和 token 使用
# 3. Task 追踪：哪个 Task 成功/失败/重试了
# 4. 基础成本统计

劣势：

• 功能比 LangSmith 少（没有 evaluation 框架、没有数据集管理）
• 同样是商业付费 + 数据外发
• 不支持 OTel 标准

Agno：内建 + AgentOS

from agno.monitoring import Monitor

# Agno 内建监控
agent = Agent(
    model=OpenAIChat(id="gpt-4"),
    monitoring=True,  # 开启后自动上报到 agno.com
)

# AgentOS 提供：
# - Agent 执行追踪
# - 会话管理
# - 基础指标

可观测性对比总结

关键洞察：tRPC-Agent 走的是”标准化 + 自主可控”路线，适合企业内部部署；LangGraph/CrewAI/Agno 走的是”商业平台 + 开箱即用”路线，适合快速起步。选择取决于你对数据主权和已有监控体系的要求。

10.3 水平扩展对比

为什么 Agent 系统需要水平扩展？

当你的 Agent 系统从”自己用”变成”给 1000 个用户用”时：

• 1000 个并发用户，每人一个长对话 → 1000 个活跃 session
• 每个 session 平均 5 轮 LLM 调用 → 5000 个并发 LLM 请求
• 每轮可能有工具调用 → 额外的计算和 IO 压力

单机扛不住，需要多实例水平扩展。

tRPC-Agent：tRPC 负载均衡

tRPC-Agent 天然继承了 tRPC 的分布式能力：

// tRPC-Agent 的部署架构
//
//        ┌─── Agent Instance 1 ───┐
// LB ──→ ├─── Agent Instance 2 ───┤ ──→ LLM API
//        ├─── Agent Instance 3 ───┤ ──→ Tool Services
//        └─── Agent Instance N ───┘
//                   ↕
//            Session Storage (Redis)
//
// 关键点：
// 1. Agent 本身无状态——状态在外部 Session Storage
// 2. 任何实例都能处理任何请求（加载对应 session 即可）
// 3. 通过 tRPC 的负载均衡自然分流

// 配置示例
agentService := NewAgentService(
    WithLLM(gpt4),
    WithSessionStorage(NewRedisStorage(redisClient)),  // 共享存储
    WithLoadBalancing(LoadBalanceConfig{
        Strategy: "round_robin",  // 或 "least_connections"
    }),
)

// 扩容 = 启动更多实例，注册到服务发现
// 缩容 = 下线实例（请求自动路由到其他实例）

优势：

• 天然无状态：Agent 实例不持有状态（状态在 Redis/DB），可任意水平扩展
• 成熟的 tRPC 生态：服务发现、负载均衡、健康检查、优雅关闭——全部现成
• 弹性伸缩：可以结合 K8s HPA 根据负载自动扩缩容
• 零感知迁移：用户的请求被路由到不同实例，用户完全无感

LangGraph：LangGraph Cloud

LangGraph 提供了托管部署方案 LangGraph Cloud：

# LangGraph Cloud 的部署方式
# 1. 定义 graph
# 2. 部署到 LangGraph Cloud（或 self-hosted LangGraph Platform）
# 3. 通过 REST API 调用

# langgraph.json 配置文件
{
    "graphs": {
        "my_agent": "./agent.py:graph"
    },
    "dependencies": ["langchain", "langgraph"],
    "env": {
        "OPENAI_API_KEY": "..."
    }
}

# 部署后通过 REST 调用
# POST /threads/{thread_id}/runs
# {
#   "assistant_id": "my_agent",
#   "input": {"messages": [{"role": "user", "content": "..."}]}
# }

# LangGraph Cloud 提供：
# - 自动扩缩容
# - Checkpoint 持久化（自动）
# - 长时间运行支持（后台任务）
# - Cron 定时任务
# - Webhook 回调

优势：

• 全托管——不需要自己运维
• Checkpoint 自动处理（不需要配置存储后端）
• 支持长时间后台运行（agent 可以跑数小时）

劣势：

• 商业付费（按执行时间计费）
• 锁定：只能在 LangGraph Platform 上运行
• Self-hosted 复杂：自己部署需要 Docker + PostgreSQL + Redis
• 网络延迟：如果 Agent 需要访问内部系统，Cloud 部署有网络边界问题

CrewAI：应用层自理

CrewAI 不提供内建的水平扩展方案——它定位为”库”而非”平台”：

# CrewAI 的部署通常是"包在 Web 服务里"

from fastapi import FastAPI
from crewai import Crew

app = FastAPI()

@app.post("/research")
async def research(request: ResearchRequest):
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task],
    )
    result = crew.kickoff(inputs={"topic": request.topic})
    return {"result": result.raw}

# 水平扩展 = 多实例部署 FastAPI
# 但注意：CrewAI 执行是同步阻塞的
# 一个请求可能跑几分钟，占用一个 worker
# 需要额外的任务队列（Celery/RQ）来处理

实际做法：

# 生产部署通常这样：
# FastAPI → Celery → CrewAI Crew
# 
# FastAPI 接收请求，丢到 Celery 队列
# Celery Worker 执行 Crew（可能跑几分钟）
# 完成后通过 Webhook/WebSocket 通知客户端

from celery import Celery

celery_app = Celery("tasks", broker="redis://localhost")

@celery_app.task
def run_crew(topic):
    crew = Crew(agents=[...], tasks=[...])
    result = crew.kickoff(inputs={"topic": topic})
    # 存储结果到 DB
    save_result(result)
    return result.raw

水平扩展对比

10.4 容错机制深度对比

生产环境的故障类型

1. LLM API 故障
   - 超时（网络抖动）
   - 限流（rate limit exceeded）
   - 服务降级（模型不可用）

2. 工具调用故障
   - 外部 API 返回错误
   - 数据库连接断开
   - 文件系统写入失败

3. Agent 逻辑故障
   - 无限循环（LLM 反复调用同一工具）
   - 格式错误（LLM 输出不符合预期格式）
   - 幻觉导致的错误操作

4. 基础设施故障
   - 内存溢出（状态过大）
   - 进程崩溃
   - 网络分区

tRPC-Agent：框架级熔断限流

tRPC-Agent 的容错能力来自两个层面——tRPC 基础设施层和 Agent 应用层：

// ===== 第1层：tRPC 基础设施容错 =====

// 熔断器（Circuit Breaker）
// 连续 N 次失败后"断路"，一段时间内不再尝试
agent := NewAgent(
    WithCircuitBreaker(CircuitBreakerConfig{
        FailureThreshold:  5,              // 5次失败触发断路
        SuccessThreshold:  2,              // 2次成功恢复
        Timeout:           30 * time.Second, // 断路30秒后尝试恢复
        // 状态机：Closed → Open → Half-Open → Closed
    }),
)

// 限流（Rate Limiting）
// 防止对 LLM API 发送过多请求
agent := NewAgent(
    WithRateLimit(RateLimitConfig{
        RequestsPerSecond: 10,    // 每秒最多10个LLM请求
        BurstSize:         20,    // 允许突发20个
        WaitStrategy:      "queue", // 超限时排队等待（而非直接拒绝）
    }),
)

// 超时控制
agent := NewAgent(
    WithTimeout(TimeoutConfig{
        LLMCallTimeout:    30 * time.Second,  // 单次LLM调用超时
        ToolCallTimeout:   10 * time.Second,  // 单次工具调用超时
        TotalTimeout:      5 * time.Minute,   // 整体执行超时
    }),
)

// ===== 第2层：Agent 应用层容错 =====

// Plugin 机制处理工具调用错误
type ToolRetryPlugin struct{}

func (p *ToolRetryPlugin) OnToolError(ctx context.Context, err error, call ToolCall) Action {
    if isTransient(err) && getAttempt(ctx) < 3 {
        return RetryWithBackoff(time.Second * time.Duration(getAttempt(ctx)))
    }
    if isRateLimit(err) {
        return RetryAfter(getRetryAfter(err))
    }
    // 不可恢复的错误：让 LLM 知道工具失败了，自己决定下一步
    return InjectErrorMessage(fmt.Sprintf("工具 %s 执行失败: %v", call.Name, err))
}

// 循环检测：防止 Agent 陷入无限循环
agent := NewAgent(
    WithMaxIterations(20),       // 最多20轮循环
    WithLoopDetection(LoopDetectionConfig{
        WindowSize:      5,       // 检查最近5次工具调用
        SimilarityThreshold: 0.9, // 如果90%相似，判定为循环
        OnLoop:          "inject_warning", // 注入提示让LLM换个方法
    }),
)

LangGraph：Checkpoint + retry_policy + 错误节点

from langgraph.graph import StateGraph
from langgraph.pregel import RetryPolicy

# 节点级重试
graph.add_node(
    "call_api", 
    api_node,
    retry=RetryPolicy(
        max_attempts=3,
        initial_interval=1.0,
        backoff_factor=2.0,
        max_interval=10.0,
        retry_on=(TimeoutError, ConnectionError),
    )
)

# 错误处理节点 + 条件路由
def error_handler_node(state):
    """统一错误处理逻辑"""
    error = state.get("last_error")
    error_count = state.get("error_count", 0)
    
    if error_count > 3:
        return {"action": "abort", "reason": "过多错误，终止执行"}
    
    if "rate_limit" in str(error):
        return {"action": "wait", "wait_seconds": 60}
    elif "timeout" in str(error):
        return {"action": "retry"}
    else:
        return {"action": "fallback", "reason": str(error)}

graph.add_node("error_handler", error_handler_node)

# 条件路由：正常→下一步，错误→错误处理
def check_error(state):
    if state.get("last_error"):
        return "error_handler"
    return "next_step"

graph.add_conditional_edges("risky_node", check_error, {
    "error_handler": "error_handler",
    "next_step": "next_step",
})

# Checkpoint 恢复（最终兜底）
# 即使所有重试都失败了，Checkpoint 保证了：
# 1. 最后一个成功状态被保存
# 2. 修复问题后可以从那里恢复
# 3. 不需要从头跑

CrewAI：基础重试

from crewai import Agent, Task, Crew

# Agent 级别配置
agent = Agent(
    role="Researcher",
    max_retry_limit=3,        # 最多重试3次
    max_execution_time=600,   # 10分钟超时
    verbose=True,
)

# Task 级别配置
task = Task(
    description="...",
    agent=agent,
    max_retries=2,
)

# Crew 级别
crew = Crew(
    agents=[agent],
    tasks=[task],
    max_rpm=10,  # 每分钟最多10次请求（简单限流）
)

# 没有：
# - 熔断器
# - 指数退避
# - 自定义重试条件
# - 循环检测
# - 错误路由
# - Checkpoint 恢复

容错能力对比总结

10.5 四大 Agent 协议详解

为什么协议很重要？

日常类比：想象世界上只有一种手机品牌可以互相打电话。如果你用的手机打不了其他品牌的电话，你就被困在自己的”生态圈”里。协议就是让不同系统能互相”打电话”的标准。

Agent 系统面临同样的问题：不同团队用不同框架构建的 Agent 怎么互相协作？Agent 怎么调用不同供应商的工具？Agent 怎么把结果展示在不同的前端上？

2024-2025年涌现了四大 Agent 协议，每个解决不同层面的”互通”问题。

协议1：MCP (Model Context Protocol) — 工具标准化接入

解决的问题：Agent 怎么调用外部工具/数据源？

日常类比：USB 接口标准。以前每个设备都有自己独特的接口（苹果30pin、诺基亚小头、三星宽口），换个设备就要换线。USB 统一后，一根线接所有设备。MCP 就是 Agent 世界的 USB——让任何 Agent 都能接入任何工具。

// MCP Server 定义（工具提供方）
{
  "name": "weather-service",
  "version": "1.0.0",
  "tools": [
    {
      "name": "get_weather",
      "description": "获取指定城市的天气信息",
      "inputSchema": {
        "type": "object",
        "properties": {
          "city": {"type": "string", "description": "城市名"},
          "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
      }
    }
  ],
  "resources": [
    {
      "name": "weather_history",
      "description": "历史天气数据",
      "mimeType": "application/json"
    }
  ]
}

// MCP Client 调用（Agent 框架侧）
// Agent 框架只需要实现 MCP Client 协议
// 就能自动发现和调用任何 MCP Server 提供的工具

核心架构：

Agent 框架 (MCP Client)
    ↕ MCP 协议（JSON-RPC over stdio/HTTP）
MCP Server（工具提供方）
    ↕
实际服务/数据库/API

MCP 的价值：

• 工具生态共享：一个人写的 MCP Server（比如 GitHub 集成），所有 Agent 框架都能用
• 解耦开发：工具开发者不需要了解 Agent 框架的内部实现
• 标准发现：Agent 可以自动发现 MCP Server 提供哪些工具，无需硬编码
• 安全边界：MCP Server 可以控制 Agent 的访问权限

协议2：A2A (Agent-to-Agent) — Agent 间标准通信

解决的问题：不同框架/不同组织构建的 Agent 怎么互相协作？

日常类比：外交协议。中国外交官和法国外交官说不同的语言、有不同的行事风格，但通过外交协议（递交国书、照会、备忘录的标准格式），他们可以有效沟通合作。A2A 就是 Agent 之间的”外交协议”。

// A2A Agent Card（自我介绍卡）
{
  "name": "research-agent",
  "description": "专门负责学术研究和信息检索的Agent",
  "url": "https://research-agent.example.com",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "academic-search",
      "name": "学术搜索",
      "description": "搜索和分析学术论文",
      "inputModes": ["text"],
      "outputModes": ["text", "file"]
    }
  ],
  "authentication": {
    "schemes": ["oauth2", "api-key"]
  }
}

// A2A Task 请求
// POST /tasks/send
{
  "id": "task-001",
  "message": {
    "role": "user",
    "parts": [
      {"type": "text", "text": "查找2024年关于LLM Agent的综述论文"}
    ]
  }
}

// A2A Task 响应（支持流式）
{
  "id": "task-001",
  "status": {"state": "completed"},
  "artifacts": [
    {
      "name": "research-results",
      "parts": [
        {"type": "text", "text": "找到以下综述论文..."},
        {"type": "file", "file": {"mimeType": "application/pdf", "url": "..."}}
      ]
    }
  ]
}

A2A 的生命周期：

Agent A 发现 Agent B（读取 Agent Card）
  → Agent A 发送 Task 给 Agent B
  → Agent B 执行（可流式返回中间状态）
  → Agent B 返回最终 Artifact
  → Agent A 使用结果

A2A vs 直接 API 调用的区别：

• A2A 是对话式的（Task 可以有多轮交互）
• A2A 支持能力发现（Agent Card 声明自己能做什么）
• A2A 是异步友好的（长时间任务通过状态更新通知）
• A2A 有标准的身份验证机制

协议3：AG-UI — Agent 与前端 UI 标准交互

解决的问题：Agent 的输出怎么实时展示给用户？用户怎么与正在运行的 Agent 交互？

日常类比：电视直播协议。摄像机拍摄的画面，通过标准的直播协议（RTMP/HLS），可以在任何电视/手机/电脑上实时观看。AG-UI 就是让 Agent 的执行过程能在任何前端上”直播”的协议。

// AG-UI 事件流（Agent → Frontend）
// Agent 执行过程中产生一系列事件，前端实时渲染

// 事件类型：
interface AgentEvent {
  type: "text_delta"           // 文本增量（流式输出）
       | "tool_call_start"     // 开始调用工具
       | "tool_call_end"       // 工具调用完成
       | "state_update"        // 状态变化
       | "interrupt"           // 需要人类介入
       | "step_start"          // 新步骤开始
       | "step_end"            // 步骤完成
       | "error";              // 错误发生
  data: any;
}

// 前端实时渲染示例：
// ┌─────────────────────────────────────┐
// │ 🤖 Agent 正在工作...                 │
// │                                      │
// │ [Step 1] 搜索相关文献  ✅ 完成       │
// │   └─ 🔧 调用 search_papers("LLM")   │
// │   └─ 📄 找到 15 篇相关论文           │
// │                                      │
// │ [Step 2] 分析关键发现  ⏳ 进行中...  │
// │   └─ 正在阅读 paper_3.pdf...        │
// │                                      │
// │ ⚠️ [需要确认] Agent 想执行：          │
// │    删除文件 old_data.csv             │
// │    [确认] [取消] [修改]              │
// └─────────────────────────────────────┘

AG-UI 的核心价值：

• 标准化前端渲染：不同框架的 Agent 可以在同一个 UI 组件中展示
• 实时流式：不是等 Agent 全部做完再展示，而是过程中实时更新
• 交互能力：人类可以在 Agent 执行中途介入（确认/取消/修改）
• 前后端解耦：前端开发者不需要了解 Agent 框架的内部实现

协议4：OpenAI API 兼容 — 事实标准的兼容层

解决的问题：怎么让 Agent 框架能使用任意 LLM 供应商？

日常类比：电源插座标准。虽然有国标/美标/欧标的区别，但只要用个转接头就能通用。OpenAI API 格式已经成了事实标准——其他 LLM 提供商（Anthropic、Google、开源模型）都提供了兼容接口。

# OpenAI API 兼容的价值：
# 同一套代码，切换不同的 LLM 供应商

# 使用 OpenAI
from openai import OpenAI
client = OpenAI(api_key="sk-...")
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)

# 使用 Anthropic（通过兼容层）
client = OpenAI(
    api_key="sk-ant-...",
    base_url="https://api.anthropic.com/v1/openai/"  # 兼容端点
)

# 使用本地模型（通过 Ollama/vLLM）
client = OpenAI(
    api_key="not-needed",
    base_url="http://localhost:11434/v1"  # Ollama 兼容端点
)

# Agent 框架只需要实现一次 OpenAI API 调用逻辑
# 就能对接所有兼容的 LLM 供应商

四大协议的层次关系

┌─────────────────────────────────────────────┐
│            用户 (User Interface)             │
└─────────────────┬───────────────────────────┘
                  │ AG-UI (展示层协议)
┌─────────────────▼───────────────────────────┐
│           Agent 框架 (Orchestration)         │
├─────────────────────────────────────────────┤
│ ┌─────────┐  ┌──────────┐  ┌────────────┐  │
│ │  Agent  │──│  Agent   │──│   Agent    │  │
│ │   A     │  │    B     │  │     C      │  │
│ └────┬────┘  └────┬─────┘  └─────┬──────┘  │
│      │ A2A协议     │              │         │
│      │ (Agent间)   │              │         │
└──────┼─────────────┼──────────────┼─────────┘
       │             │              │
       │ MCP (工具层协议)           │
┌──────▼─────┐ ┌─────▼────┐ ┌──────▼──────┐
│ MCP Server │ │MCP Server│ │ MCP Server  │
│ (GitHub)   │ │ (DB)     │ │ (Search)    │
└────────────┘ └──────────┘ └─────────────┘

底层 LLM 调用全部走 OpenAI API 兼容格式

每层解决的问题：

10.6 各框架的协议支持度对比

协议支持矩阵

详细分析

tRPC-Agent：协议支持最全面

// MCP 支持
agent := NewAgent(
    WithMCPServers(
        MCPServer{Name: "github", Command: "npx", Args: []string{"@github/mcp-server"}},
        MCPServer{Name: "db", URL: "http://localhost:3001/mcp"},
    ),
)

// A2A 支持
// tRPC-Agent 可以作为 A2A Server 暴露能力
// 也可以作为 A2A Client 调用其他 A2A Agent
a2aClient := NewA2AClient("https://research-agent.example.com")
task := a2aClient.SendTask(TaskRequest{
    Message: "搜索最新论文",
})

// AG-UI 支持
// Agent 执行过程产生标准 AG-UI 事件流
// 前端可以用任何 AG-UI 兼容的组件渲染
handler := NewAGUIHandler(agent)
http.Handle("/agent/stream", handler)

为什么 tRPC-Agent 协议支持最全？

这与其定位有关——tRPC-Agent 源自企业内部的 Agent 平台需求，企业场景下”与其他系统互通”是硬需求。你不能只和自己的 Agent 玩，必须能调用其他团队的 Agent（A2A）、接入公司已有的工具服务（MCP）、在公司统一的前端展示（AG-UI）。

LangGraph：专注自身生态

LangGraph 的策略是构建自有的完整生态（LangSmith + LangGraph Cloud + LangChain），而不是积极拥抱外部协议。这带来了深度集成的优势，但也意味着与非 LangChain 生态的互通需要额外适配。

CrewAI / Agno：选择性支持

CrewAI 和 Agno 优先支持了 MCP（因为工具接入是最急迫的需求），但对 A2A 和 AG-UI 的支持暂时缺失。

协议覆盖的战略意义

"能和谁对话"决定了生态位

                    高
                    │
    生态            │         tRPC-Agent
    开放度          │         (MCP + A2A + AG-UI)
                    │
                    │    Agno / CrewAI
                    │    (MCP + OpenAI兼容)
                    │
                    │              LangGraph
                    │              (自有生态为主)
                    │
                    低
                    └────────────────────────────
                    低           自有生态深度          高

战略分析：

1. tRPC-Agent 的策略：成为”连接器”——我可以和任何人对话。不试图构建封闭生态，而是成为现有生态的接入点。适合企业内部”连接一切”的需求。

2. LangGraph 的策略：构建”围墙花园”——在我的生态内一切体验最好。LangSmith + LangGraph Cloud + LangChain 形成完整闭环。适合”一站式”选择 LangChain 生态的团队。

3. CrewAI/Agno 的策略：务实选择——先接入最急迫的标准（MCP + OpenAI兼容），其他随后。适合快速迭代的创业团队。

10.7 部署模式对比

三种部署模式

Agent 框架的部署形态直接影响运维复杂度、成本和控制力：

库模式（CrewAI, Agno）

# 库模式：Agent 框架是你应用的一个依赖
# pip install crewai

from crewai import Crew
from fastapi import FastAPI

app = FastAPI()

@app.post("/run")
def run_agent(request):
    crew = Crew(...)  # 在你的进程内执行
    return crew.kickoff()

# 部署 = 部署你的应用
# 扩展 = 扩展你的应用
# 监控 = 你自己的监控体系

优势：

• 完全控制——代码在你的进程里，想怎么改怎么改
• 无额外成本——不需要付费给 Agent 框架供应商
• 无网络依赖——Agent 逻辑和业务逻辑在同一进程，零网络延迟

劣势：

• 运维全靠自己——扩展、容错、监控都要自建
• 依赖管理复杂——Agent 框架的依赖可能和你的应用冲突
• 升级耦合——升级 Agent 框架版本可能影响你的应用

服务模式（tRPC-Agent）

// 服务模式：Agent 是独立的微服务
// 你的应用通过 tRPC 调用 Agent 服务

// Agent 服务（独立部署）
func main() {
    server := trpc.NewServer()
    agentService := NewAgentService(config)
    server.Register(agentService)
    server.Serve()
}

// 业务应用（调用 Agent 服务）
func handleUserRequest(req Request) Response {
    client := NewAgentClient("agent-service")
    result, _ := client.Chat(context.Background(), &ChatRequest{
        SessionID: req.UserID,
        Message:   req.Content,
    })
    return Response{Content: result.Reply}
}

优势：

• 独立扩展——Agent 服务和业务服务独立扩缩容
• 故障隔离——Agent 崩溃不影响主业务
• 团队解耦——Agent 团队和业务团队独立开发部署
• 语言无关——业务是 Java/Python 也能调 Go 写的 Agent 服务

劣势：

• 网络开销——每次调用都有网络延迟
• 运维复杂度——多了一个需要维护的服务
• 分布式问题——网络超时、序列化、版本兼容

云模式（LangGraph Cloud）

# 云模式：上传代码，平台帮你运行
# langgraph.json
{
  "graphs": {"agent": "./agent.py:graph"},
  "env": {"OPENAI_API_KEY": "..."}
}

# 部署命令
# langgraph deploy --name my-agent

# 使用：通过 REST API 调用
# POST https://api.langraph.cloud/threads/xxx/runs

优势：

• 零运维——扩展、容错、监控全托管
• 快速上线——写好代码就能部署
• 内建能力——Checkpoint、长时间运行、Cron 全部现成

劣势：

• 成本高——按执行时间计费，规模大时很贵
• 控制力低——不能深度定制运行时行为
• 锁定——迁移成本高
• 数据安全——代码和数据在供应商的环境中

部署模式选型指南

初创团队 / 快速原型
  → 库模式（CrewAI）或 云模式（LangGraph Cloud）
  → 理由：最小运维负担

成长期团队 / 内部工具
  → 服务模式（tRPC-Agent）
  → 理由：可控成本，独立扩展，团队解耦

大型企业 / 严格合规
  → 服务模式（tRPC-Agent）自建部署
  → 理由：数据主权，接入已有基础设施，安全审计

10.8 安全与合规

Agent 系统特有的安全挑战

传统 Web 服务的安全问题（SQL注入、XSS等）在 Agent 系统中依然存在，但 Agent 还有独特的安全挑战：

1. Prompt Injection：恶意用户通过输入干扰 Agent 行为
2. 工具滥用：Agent 被误导执行危险操作（删库、发邮件）
3. 数据泄露：Agent 把不该暴露的信息包含在回复中
4. 成本攻击：恶意用户让 Agent 无限循环，消耗大量 token
5. 权限逃逸：Agent 通过工具链获得超出预期的系统权限

各框架的安全措施

tRPC-Agent：企业级安全

agent := NewAgent(
    // 工具权限白名单
    WithToolPermissions(PermissionConfig{
        AllowList: []string{"search", "read_file"},  // 只允许这些
        DenyList:  []string{"delete_*", "admin_*"},  // 禁止这些
    }),
    
    // Token 消耗限制
    WithTokenBudget(TokenBudget{
        MaxPerRequest:  10000,    // 单次请求最多用10K token
        MaxPerSession:  100000,   // 单个session最多100K
        MaxPerDay:      1000000,  // 每天最多1M（按用户）
    }),
    
    // 输出过滤
    WithOutputFilter(OutputFilterConfig{
        PII:     true,  // 过滤个人隐私信息
        Secrets: true,  // 过滤密钥/密码
        Custom:  myFilter,
    }),
    
    // 审计日志
    WithAuditLog(AuditConfig{
        LogToolCalls:  true,  // 记录所有工具调用
        LogLLMInputs:  true,  // 记录所有LLM输入
        RetentionDays: 90,    // 保留90天
    }),
)

LangGraph：通过 Human-in-the-Loop 实现安全

# LangGraph 的安全主要通过"关键节点前中断"实现
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_sql", "send_email", "delete_resource"],
    # 这些危险操作必须人类确认
)

CrewAI：基础限制

agent = Agent(
    max_execution_time=300,  # 超时限制
    max_iter=15,             # 循环限制
    # 安全相关配置较少，主要依赖应用层自己实现
)

10.9 综合评估矩阵

生产就绪度总评

选型总决策

【选 tRPC-Agent 如果】：
- 企业内部部署，有数据主权要求
- 需要接入已有的监控/服务治理体系
- 需要与其他系统（Agent/工具/UI）标准化互通
- 需要生产级的熔断限流
- 团队熟悉 Go / tRPC 生态

【选 LangGraph 如果】：
- 需要复杂的工作流编排（条件+并行+循环+子图）
- 需要强大的状态管理和 checkpoint
- 接受使用 LangSmith 付费平台
- 团队使用 Python + 熟悉 LangChain 生态
- 需要精细的 Human-in-the-Loop 控制

【选 CrewAI 如果】：
- 快速原型/MVP
- 任务是"流水线"模式（严格顺序）
- 团队非技术背景居多（PM、运营也要能看懂）
- 不需要复杂的状态管理和并发
- 优先级是"快速跑通"而非"长期维护"

【选 Agno 如果】：
- 需要多种预制协作模式快速对比
- 轻量级场景
- 想用最少代码实现 Agent
- 作为学习/实验工具

10.10 协议的未来趋势

协议统一的必然性

当前四大协议各管一层，但存在交叉和重叠：

• A2A 和 MCP 都涉及”工具调用”，边界模糊
• AG-UI 和 A2A 都涉及”消息传递”
• 各框架对协议的支持程度不一

预测：2-3 年内可能出现以下趋势：

1. MCP 成为事实标准：工具接入层已经获得广泛采纳（Anthropic 主导 + 微软支持），最有可能第一个成为”HTTP 级别”的共识
2. A2A 可能被 MCP 扩展吸收：如果 MCP 扩展了 Agent-to-Agent 的能力（已有讨论），A2A 可能不需要独立存在
3. AG-UI 需要更广泛的采纳：目前主要是 CopilotKit 推动，需要更多框架和前端库的支持才能成为标准
4. OpenAI API 兼容将持续：除非出现完全颠覆性的模型架构

对开发者的建议

1. 优先支持 MCP：无论用哪个框架，确保你的工具通过 MCP 暴露——这是复用价值最大的标准
2. 关注 A2A 动态：如果你在构建”Agent 平台”（多个 Agent 协作），A2A 值得投入
3. AG-UI 看场景：如果你的 Agent 有复杂的前端交互需求，AG-UI 值得尝试
4. OpenAI 兼容是底线：任何不兼容 OpenAI API 格式的 LLM 接入方案都要谨慎

10.11 本章总结

核心认知

1. “能跑”≠”能上线”：Demo 到生产的鸿沟主要在可观测性、扩展性、容错性和安全性四个方面。

2. 协议 = 生态位：一个框架支持多少协议，决定了它能和多少其他系统协作。在 Agent 时代，”孤岛”是最大的风险。

3. 没有全能选手：tRPC-Agent 生产能力最强但学习曲线不平，LangGraph 编排能力最强但生态封闭，CrewAI 上手最快但生产就绪度不足。

4. 标准正在形成：2024-2025 是 Agent 协议的”战国时代”，MCP 领先一个身位，其他协议仍在竞争中。选择支持标准协议的框架，是对未来的最佳对冲。

一个类比总结四大协议

想象你在一栋办公楼里：

• OpenAI API = 电话线路标准（能打电话给任何人）
• MCP = 办公用品标准接口（打印机、扫描仪、任何工位都能用）
• A2A = 部门间公文标准（财务部和人事部用统一格式传递申请）
• AG-UI = 前台接待协议（任何访客用统一流程进出大楼）

思考题

1. 分析题：为什么 tRPC-Agent 选择基于 OpenTelemetry 标准做可观测性，而 LangGraph 选择构建自己的 LangSmith 平台？这两种策略各自的商业逻辑是什么？如果你是一个 50 人的创业公司 CTO，你会选哪种？如果是 5000 人的企业呢？

2. 设计题：假设你的 Agent 系统平均每次请求消耗 3000 token（约 $0.03），日均 10 万次请求。月成本约 $90,000。老板说”太贵了，砍一半”。你有哪些策略来降低成本？（提示：从模型选择、Context 管理、缓存、路由分流等多角度思考。）

3. 对比题：MCP 和 A2A 都可以用来”调用另一个 Agent 的能力”。它们的本质区别是什么？在什么场景下你应该把一个 Agent 的能力暴露为 MCP Server？什么场景下应该暴露为 A2A Agent？

4. 综合题：你需要为一个金融公司构建一个”自动化研报生成 Agent 系统”。要求：数据不出公司网络，所有操作可审计，支持100个分析师同时使用，错误时人类必须确认才能重试。你会选哪个框架？怎么部署？需要实现哪些安全措施？

5. 预测题：假设 3 年后 Agent 协议已经统一成一个标准（类似 HTTP 统一了 Web 通信）。这个统一标准更可能长什么样——更像 MCP 的”工具接口”风格，还是更像 A2A 的”对话交互”风格？为什么？统一后，Agent 框架的竞争焦点会转移到哪里？

上一章：第九章 跨框架对比——多 Agent 协作模式

下一章：第十一章 Ralph Loop 与可验证 Agent 执行

返回目录：Agent 框架深度导读系列