A2A协议实战：Agent协作最佳实践指南

2026-06-20阅读 0热度 0

其他

上次我们讨论了 MCP：Agent 通过标准化协议连接工具服务——工具本身处于被调用位置，等待触发、执行并返回结果。但在实际开发中，有不少场景你需要的不是工具，而是一个具备独立决策能力的 Agent： - 一个调研型 Agent，专攻技术资料的搜集与深度分析； - 一个写作型 Agent，能将分析结论组织成结构清晰的文档； - 一个代码审查 Agent，专注于安全漏洞的自动排查。那么问题来了：三个 Agent 要协同工作，它们之间如何通信？谁来发现彼此的存在？成果如何流转？这正是 **A2A（Agent-to-Agent）协议** 要解决的核心难题。 ``` MCP： Agent ←→ 工具/数据源（垂直集成，Agent 主动调用工具） A2A： Agent ←→ Agent （水平协作，Agent 委托给另一个 Agent） ``` ## A2A 的三个核心概念 ### AgentCard：Agent 的“名片” 每个 Agent 发布一张 `AgentCard`，用来描述自身的服务能力。你可以把它视作 OpenAPI Spec 的 Agent 版本——一份机器可读的能力声明。 ```python from a2a.types import AgentCard, AgentSkill def make_skill(skill_id, name, description, tags): s = AgentSkill() s.id = skill_id s.name = name s.description = description s.tags.extend(tags) return s research_card = AgentCard() research_card.name = "research-agent" research_card.description = "Gathers factual background on technical topics" research_card.skills.append( make_skill("research", "Research", "Collect key facts on a topic", ["research", "facts"]) ) ``` 关键字段包括：`name`、`description`、`skills`（每个 skill 自带 `tags`，用于发现机制）。 ### Task：工作单元 Agent 之间传递的不是函数调用，而是 `Task`。它拥有完整的生命周期：`SUBMITTED → WORKING → COMPLETED / FAILED`。完成时，Agent 将结果追加为 `ROLE_AGENT` 类别的 Message。 ```python from a2a.types import Task, TaskState, TaskStatus, Message, Part, Role task = Task() task.id = str(uuid.uuid4()) task.status.state = TaskState.TASK_STATE_SUBMITTED # 输入：User 角色的 Message msg = Message() msg.role = Role.ROLE_USER part = Part(); part.text = "Should I use Python or Go?" msg.parts.append(part) task.history.append(msg) ``` ### Registry：Agent 的“黄页” `AgentRegistry` 存储所有已注册的 AgentCard，支持按标签 (tag) 进行发现。这样一来，orchestrator 无需硬编码任何 Agent 的名字，只需根据能力标签去查找。 ```python class AgentRegistry: def register(self, card: AgentCard, handler: Callable[[Task], Task]) -> None: self._agents[card.name] = AgentEntry(card=card, handler=handler) def discover(self, tag: str) -> list[AgentCard]: """返回所有 skill 包含指定 tag 的 Agent""" ... def delegate(self, agent_name: str, input_text: str) -> Task: """创建 Task 并通过注册的 handler 执行""" ... ``` ## Demo 1：直接调用——简单但耦合没有协议时，orchestrator 直接调用三个 Python 函数： ```python def direct_orchestrator(question: str) -> str: research = research_agent_fn(question) # 硬依赖 analysis = analysis_agent_fn(research) # 硬依赖 answer = writing_agent_fn(analysis) # 硬依赖 return answer ``` 实测结果： ``` → calling research_agent (direct) → calling analysis_agent (direct) → calling writing_agent (direct) Answer: Choose Python if you need rapid development with broad libraries. Select Go if performance and concurrency are critical... ``` 功能没问题。但问题是：orchestrator 里写死了三个函数名——任何一个 Agent 被替换，都需要修改 orchestrator 代码。如果 orchestrator 部署在另一个服务中，这就是跨服务改代码的场景。 ## Demo 2：A2A AgentCard 注册表注册三个 Agent，各自携带不同的 skill tag： ``` [registry] registered: research-agent [registry] registered: analysis-agent [registry] registered: writing-agent ``` 发现测试： ```python researchers = registry.discover("research") # → Found: research-agent — Gathers factual background on technical topics writers = registry.discover("writing") # → Found: writing-agent — Composes clear technical prose from analysis output ``` orchestrator 完全不写 Agent 名字，只按 tag 发现： ```python def a2a_orchestrator(question: str) -> str: researchers = registry.discover("research") t1 = registry.delegate(researchers[0].name, question) analysts = registry.discover("analysis") t2 = registry.delegate(analysts[0].name, task_output(t1)) writers = registry.discover("writing") t3 = registry.delegate(writers[0].name, task_output(t2)) return task_output(t3) ``` 实测执行： ``` → delegating to research-agent (discovered via tag) → delegating to analysis-agent (discovered via tag) → delegating to writing-agent (discovered via tag) Answer: Choose Python for rapid development; Go for high-throughput performance... ``` **关键差别：orchestrator 代码里没有出现 `research-agent`、`writing-agent` 这些名字**。新注册一个 `writing-agent-v2`（带上相同的 `writing` 标签），orchestrator 立刻就能发现并使用——零代码改动。 ## Demo 3：LLM 驱动的 Agent 路由这是 A2A 最强大的用法：**LLM 读取 AgentCard 目录，自行决定调用哪些 Agent、按什么顺序**。向 LLM 展示 Agent 目录： ```yaml Available agents: research-agent: Gathers factual background [skills: Research(research, facts)] analysis-agent: Analyzes research notes [skills: Analysis(analysis, tradeoffs)] writing-agent: Composes technical prose [skills: Writing(writing, prose)] ``` LLM 输出执行计划： ```json ["research-agent", "analysis-agent", "writing-agent"] ``` 按计划执行： ``` Executing 3 agents: → delegating to research-agent → delegating to analysis-agent → delegating to writing-agent Final answer: Choose Python for rapid development; Go for high-throughput... ``` 这才是终态——无需预先配置 orchestrator，LLM 根据任务需求和 AgentCard 描述，**在运行时自主规划协作链路**。 ## MCP vs A2A vs ANP：协议选型矩阵 | 维度 | MCP | A2A | |---|---|---| | 解决什么问题 | Agent ↔ 工具/数据源 | Agent ↔ Agent | | 发现机制 | `list_tools()`（工具目录） | `discover()`（Agent 注册表） | | 工作单元 | Tool call（同步） | Task（异步就绪） | | 耦合方式 | Agent 直接使用工具 | Orchestrator 委托给 Agent | | 另一端的性质 | 被动的工具服务 | 有自主逻辑的 Agent | | 跨服务 | 工具是独立进程 | Agent 是独立服务 | 四种协作方式的完整选型： | 场景 | 推荐方案 | |---|---| | 同一代码库，调用确定 | 直接函数调用 | | Agent 需要调用外部工具 | MCP 协议（tools as service） | | Agent 委托给专业 Agent | A2A 协议（agents as service） | | 跨组织大规模 Agent 网络 | ANP（去中心化发现，Web3 风格） | ## 设计 Checklist **AgentCard 设计** - [ ] `description` 用一句话概括 Agent 的专长，LLM 会据此做决策 - [ ] `skill.tags` 使用语义清晰的标签（如 `research`、`analysis`、`writing`），避免版本号或 ID - [ ] `AgentCard` 应同时满足机器可读和人类可理解（参考 OpenAPI 的风格） **Task 设计** - [ ] `Task.id` 使用 UUID，便于追踪和幂等重试 - [ ] 利用 `history`（Message 链）传递上下文，而非在 part.text 里拼接所有历史 - [ ] 明确区分 `ROLE_USER`（输入）与 `ROLE_AGENT`（输出）的 Message **Registry 与发现** - [ ] 一个 tag 可对应多个 Agent（实现负载均衡、A/B 测试） - [ ] 支持按 tag 筛选，也支持按 name 精确查找 - [ ] 生产环境使用持久化 Registry（数据库或服务注册中心），而非内存字典 **LLM 驱动路由** - [ ] Agent catalog 要简洁：name + description + skill tags，避免将完整 AgentCard JSON 塞给 LLM - [ ] 解析 LLM 输出时加入校验（`agent_name not in registry`），并准备兜底路径 - [ ] 记录 LLM 生成的执行计划，便于事后分析与调试 ## 总结五个核心结论： 1. **A2A 和 MCP 不竞争，互补**：MCP 是 Agent 调用工具，A2A 是 Agent 委托 Agent；一个系统可同时使用两种协议 2. **AgentCard 是 A2A 的核心**：它让 Agent 转化为可发现、可组合的服务单元，而非硬编码的依赖 3. **Task 比函数调用更丰富**：拥有生命周期状态，支持异步，能携带结构化历史 4. **直接调用 vs A2A 的分界线**：同一团队、同一代码库 → 直接调用；跨服务、需要解耦 → A2A 5. **LLM 驱动路由是 A2A 的最高形态**：Agent 根据任务描述自主规划协作链路，无需预先配置 orchestrator 下一篇：**Agent 评估框架** —— 如何系统性地测试 Agent？用哪些指标衡量优劣？DeepEval 怎么用？ --- ## 参考资料 - A2A Protocol 官方规范 - A2A Python SDK - ANP（Agent Network Protocol） - 本系列完整 Demo 代码：agent-10-a2a

A2A协议实战：Agent协作最佳实践指南

相关阅读

最新教程

最新资讯