8.0
热点
AI SCORE
技术实践2026-08-13 22:14
A2A 协议详解：多 Agent 通信架构与实战实现
dev.to · AI#Agent#多Agent#架构
Editor brief · 编辑速览
从零讲解 Agent 间通信问题，对比函数调用/REST/队列/专用协议等方案，并手把手实现一套完整的 A2A 系统，适合构建多 Agent 协同流水线。
AI 智能体正在变得越来越不像孤立的聊天机器人，而更像软件组件。
一个智能体可以做研究。
另一个可以写代码。
还有一个可以测试代码。
还有一个可以部署代码。
这立刻产生了一个新的工程问题：
一个智能体应该如何与另一个智能体通信？
你可以用简单的函数调用来解决这个问题。
可以用 REST API 来解决。
可以使用队列或事件流。
或者，你可以定义一个 proper 的 Agent-to-Agent 通信协议。
本文从基础开始解释这个问题，比较各种替代方案，并逐步构建一个可工作的 A2A 系统。
最简单的形式下，Agent-to-Agent 通信意味着：
Agent A
|
| request
v
Agent B
|
| result
v
Agent A
Agent A 请求 Agent B 执行某些工作。
Research Agent
"Find the best authentication approach for our API."
|
v
Coding Agent
"Implement JWT authentication."
|
v
Testing Agent
"Run the test suite."
|
v
Review Agent
"Review the implementation."
每个智能体专精于某一领域。
我们不做一个大而全的智能体，而是划分职责。
假设一切都在一个 Python 进程内。
coding_agent.generate_code(request)
但真实系统通常是这样的：
research-agent.company.internal
coding-agent.company.internal
testing-agent.company.internal
review-agent.company.internal
现在智能体是独立的服务。
谁在发送请求？
谁应该接收它？
正在请求什么任务？
需要什么能力？
提供了什么输入？
任务状态是什么？
结果在哪里？
请求失败怎么办？
请求重复了怎么办？
接收方智能体如何验证调用者身份？
这就是协议变得有用的真正原因。
A2A 最容易理解为一个契约。
它定义了智能体之间如何交换工作。
{
"task_id": "task-123",
"sender": "research-agent",
"receiver": "coding-agent",
"capability": "generate_code",
"input": {
"language": "python",
"requirements": [
"FastAPI",
"JWT authentication"
]
}
}
重要的不是 JSON 本身。
重要的是两个智能体对相同的含义有共同的理解。
我在请求 capability = generate_code
我收到了一个 generate_code 任务
这种共同的理解就是协议。
这个区别很重要。
A2A 是通信契约。
你可以把它想象成：
┌───────────────────────────┐
│ A2A Protocol │
│ │
│ Task │
│ Message │
│ Capability │
│ State │
│ Result │
│ Error │
└─────────────┬─────────────┘
|
v
┌───────────────────────────┐
│ Transport │
│ │
│ HTTP / WebSocket / Queue │
└───────────────────────────┘
今天你可能使用 HTTP。
明天你可能使用消息代理。
协议概念可以保持不变。
没有单一的架构。
有多个层级。
Level 1 - 直接函数调用
最简单的方法。
Agent A
|
| function call
v
Agent B
result = coding_agent.generate_code(
"Build a FastAPI API"
)
智能体之间紧密耦合。
它们必须存在于同一个应用或运行时中。
你无法轻松地独立部署它们。
现在把智能体变成独立的服务。
┌──────────────────┐
│ Research Agent │
└────────┬─────────┘
|
| HTTP
v
┌──────────────────┐
│ Coding Agent │
└──────────────────┘
Research Agent 可能这样调用：
POST /generate-code
{
"language": "python",
"requirement": "Build a FastAPI API"
}
这已经很有用了。
但现在每个智能体都需要自己的 API 契约。
你可能最终得到：
POST /generate-code
POST /review-code
POST /run-tests
POST /deploy
POST /summarize
随着智能体数量的增长，集成变得越来越困难。
不要为每个能力创建不同的 API 契约：
/generate-code
/review-code
/run-tests
我们定义一个通用的通信端点：
POST /a2a
操作在消息内部。
{
"type": "task.create",
"task_id": "task-123",
"capability": "generate_code",
"input": {
"language": "python"
}
}
现在端点保持稳定。
能力在变化。
这给了我们一个更清晰的抽象。
有些智能体任务需要时间。
LLM reasoning
+
database lookup
+
code generation
+
test execution
+
security scan
可能需要 30 秒或几分钟。
Agent A
|
| request
v
Agent B
|
| wait...
|
| wait...
|
v
result
我们可以使通信异步化。
Agent A
|
| task.create
v
Agent B
|
| task.accept
v
Agent A
...work happens...
Agent B
|
| task.complete
v
Agent A
这更适合生产系统。
Agent A
|
| publish
v
┌─────────────────┐
│ Message Broker │
└───────┬─────────┘
|
+─────────────+
| |
v v
Agent B Agent C
可能的技术包括：
Kafka
NATS
RabbitMQ
Redis Streams
Cloud queues
现在智能体不需要直接互相调用。
它们发布和消费事件。
这提供了更好的解耦和扩展性，但也引入了更多的基础设施。
没有 universal 的答案。
一个有用的演进路径是：
Same process
↓
Function call
Separate services
↓
HTTP
Long-running tasks
↓
HTTP + Task Store + Worker
Large distributed system
↓
Broker / Event-driven architecture
错误是从最复杂的架构开始。
从解决实际问题所需的最小架构开始。
让我们在本文中用一个具体的例子来贯穿始终。
我们正在构建一个软件开发系统。
有三个智能体：
┌──────────────────┐
│ Research Agent │
│ │
│ Finds solution │
└────────┬─────────┘
|
v
┌──────────────────┐
│ Coding Agent │
│ │
│ Writes code │
└────────┬─────────┘
|
v
┌──────────────────┐
│ Testing Agent │
│ │
│ Runs tests │
└──────────────────┘
"Build a Python REST API for user authentication."
User
|
v
Research Agent
|
| create task
v
Coding Agent
|
| code
v
Testing Agent
|
| test result
v
Research Agent
|
v
Final response
现在让我们来实现它。
首先，不用 A2A 来解决这个问题。
一切都在一个应用内运行。
class ResearchAgent:
def create_plan(self, requirement):
return {
"language": "python",
"requirement": requirement
}
class CodingAgent:
def generate_code(self, plan):
return {
"files": [
{
"name": "main.py",
"content": "print('FastAPI app')"
}
]
}
class TestingAgent:
def run_tests(self, code):
return {
"passed": 10,
"failed": 0
}
编排器调用每个智能体：
research = ResearchAgent()
coding = CodingAgent()
testing = TestingAgent()
plan = research.create_plan(
"Build authentication API"
)
code = coding.generate_code(plan)
tests = testing.run_tests(code)
print(tests)
{
"passed": 10,
"failed": 0
}
这完全 valid。
这个架构没有问题。
看看耦合。
Orchestrator
|
+--> ResearchAgent
|
+--> CodingAgent
|
+--> TestingAgent
编排器知道：
class names
method names
argument formats
return formats
implementation details
现在假设 Coding Agent 变成了一个独立的服务。
我们可以把 Coding Agent 移动到另一个服务。
Research Agent
|
| HTTP
v
Coding Service
import httpx
response = httpx.post(
"http://coding-agent/generate-code",
json={
"language": "python",
"requirement": "Build authentication API"
}
)
code = response.json()
但现在 Research Agent 依赖于：
URL
endpoint
request schema
response schema
authentication
error handling
Coding Agent 有一个 API。
Research Agent 必须理解那个 API。
现在定义一个共享的消息。
from pydantic import BaseModel
class A2AMessage(BaseModel):
message_id: str
task_id: str
type: str
sender: str
receiver: str
capability: str
input: dict
{
"message_id": "msg-100",
"task_id": "task-100",
"type": "task.create",
"sender": "research-agent",
"receiver": "coding-agent",
"capability": "generate_code",
"input": {
"language": "python",
"requirement": "Build authentication API"
}
}
现在每个智能体都理解了同一个基础契约。
class Agent:
def __init__(self, agent_id):
self.agent_id = agent_id
self.capabilities = {}
def register(self, name, handler):
self.capabilities[name] = handler
async def execute(self, capability, input):
handler = self.capabilities.get(capability)
if not handler:
raise ValueError(
f"Capability '{capability}' not found"
)
return await handler(input)
现在注册一个 coding 能力。
coding_agent = Agent("coding-agent")
async def generate_code(input):
return {
"language": input["language"],
"artifacts": [
{
"type": "file",
"name": "main.py",
"content": "print('FastAPI')"
}
]
}
coding_agent.register(
"generate_code",
generate_code
)
from fastapi import FastAPI
app = FastAPI()
@app.post("/a2a")
async def receive_message(
message: A2AMessage
):
result = await coding_agent.execute(
message.capability,
message.input
)
return {
"message_id": "msg-101",
"task_id": message.task_id,
"type": "task.complete",
"sender": "coding-agent",
"receiver": message.sender,
"output": result
}
uvicorn server:app --port 8001
现在 Coding Agent 是一个独立的服务。
Research Agent 发送：
import httpx
message = {
"message_id": "msg-100",
"task_id": "task-100",
"type": "task.create",
"sender": "research-agent",
"receiver": "coding-agent",
"capability": "generate_code",
"input": {
"language": "python",
"requirement": "Build authentication API"
}
}
response = httpx.post(
"http://localhost:8001/a2a",
json=message
)
print(response.json())
Coding Agent 返回：
{
"message_id": "msg-101",
"task_id": "task-100",
"type": "task.complete",
"sender": "coding-agent",
"receiver": "research-agent",
"output": {
"language": "python",
"artifacts": [
{
"type": "file",
"name": "main.py",
"content": "print('FastAPI')"
}
]
}
}
现在通信是显式的：
Research Agent
|
| task.create
|
| task_id = task-100
v
Coding Agent
|
| task.complete
|
| task_id = task-100
v
Research Agent
重要的是不要声称 A2A 能让智能体自动变得更聪明。
模型还是那个模型。
改进在于系统架构。
Agent A
|
| custom API
v
Agent B
Agent A 需要了解 Agent B 的 API。
Agent A
|
| common protocol
v
Agent B
双方理解相同的通信契约。
智能体不需要知道彼此的内部实现。
不同智能体可以使用相同契约进行通信。
只要支持相同协议，你就可以用另一个实现替换 Coding Agent。
任务可以无需保持 HTTP 请求打开而继续执行。
每个任务都可以有唯一的 ID。
新能力可以无需创建全新通信架构而引入。
生产级协议通常需要的不只是：
sender
receiver
input
output
有用的能力包括：
Task identity
Message identity
Capability discovery
Task states
Structured results
Error handling
Retries
Timeouts
Authentication
Authorization
Idempotency
Streaming
Observability
Versioning
Artifact references
来理解一下为什么。
{
"task_id": "task-123"
}
这允许整个工作流引用同一操作。
task-123
|
+-- task.create
+-- task.accept
+-- task.progress
+-- task.complete
{
"message_id": "msg-456"
}
这标识了一次通信事件。
deduplication
logging
retries
debugging
tracing
"What endpoint does this service have?"
"What can this agent do?"
{
"agent_id": "coding-agent",
"capabilities": [
"generate_code",
"review_code",
"run_tests"
]
}
这在智能体动态部署时特别有用。
{
"output": "Everything looks good."
}
{
"output": {
"status": "approved",
"issues": [],
"artifacts": []
}
}
结构化输出允许另一个智能体无需解析自然语言即可消费结果。
协议应该定义错误。
{
"type": "task.failed",
"task_id": "task-123",
"error": {
"code": "CAPABILITY_NOT_SUPPORTED",
"message": "Agent cannot generate code",
"retryable": false
}
}
现在调用方知道该怎么做。
Agent A
|
| task.create
v
Agent B
|
| executes
|
X network failure
但如果 Agent B 已经执行了任务，我们不希望：
execute
execute
message_id = msg-100
msg-100 -> completed
重复消息可以返回现有结果。
一个有用的生产流程是：
task.create
Agent A ----------------------> Agent B
|
v
persist task
|
v
task.accept
|
v
execute task
|
v
task.complete
|
v
Agent A <-------------------------
现在任务可以运行数分钟而无需保持 HTTP 连接打开。
有些智能体会产生增量输出。
request
|
| ............
|
v
complete response
task.progress
task.progress
task.progress
task.complete
{
"type": "task.progress",
"task_id": "task-123",
"progress": 50,
"message": "Running integration tests"
}
这对于长时间运行的工作流很有用。
一个严肃的 A2A 系统必须回答：
这个智能体是谁？
它被允许做什么？
它可以访问哪些数据？
它可以调用哪些能力？
认证可能使用：
mTLS
JWT
OAuth
API keys
workload identity
然后授权控制能力。
research-agent
|
+--> generate_code allowed
|
+--> run_tests allowed
|
+--> production_deploy denied
一个有用的架构是：
User
|
v
Orchestrator
|
v
┌─────────────────┐
│ A2A Protocol │
└───────┬─────────┘
|
┌──────────┼──────────┐
| | |
v v v
Research Coding Testing
Agent Agent Agent
| | |
└──────────┼──────────┘
|
v
Tools
A2A 是智能体组件之间的通信层。
A2A 并不总是最佳答案。
你甚至可以结合使用它们。
A2A
|
+-- HTTP for control
|
+-- queue for long-running work
|
+-- object storage for artifacts
这些也是不同的概念。
编排器决定：
接下来应该发生什么？
智能体如何通信工作？
Orchestrator
1. Ask Research Agent
2. Ask Coding Agent
3. Ask Testing Agent
4. Ask Review Agent
A2A 处理通信：
task.create
task.accept
task.complete
task.failed
所以它们是互补的。
在以下情况使用正式的 A2A 风格协议：
智能体独立部署
+
智能体可能来自不同团队
+
智能体有不同能力
+
任务是异步的
+
你需要可追溯性
+
你需要标准化
在以下情况可能不需要：
一切都在一个 Python 进程中
+
只有两个智能体存在
+
通信很简单
+
你控制一切
agent_b.run(task)
就是很好的工程实践。
不要直接跳入分布式架构。
使用这个渐进路径：
Stage 1
Function calls
Stage 2
HTTP + JSON
Stage 3
Common A2A message format
Stage 4
Task persistence + async workers
Stage 5
Discovery + security + observability
Stage 6
Message broker + streaming + horizontal scaling
这保持复杂度与你的实际需求成正比。
A2A 不能让智能体更聪明。
相反，它使智能体系统更具可组合性。
没有通用协议：
Agent A
↓ custom integration
Agent B
↓ another integration
Agent C
↓ another integration
Agent D
有了通用协议：
A2A
|
┌────┼────┐
| | |
A B C
|
D
每个智能体只需理解该协议。
这就是架构上的胜利。
可以把 A2A 理解为三层结构。
这个智能体能做什么？
我们要求它执行什么任务？
如何传达这项任务？
完整的流程如下：
智能体 A
|
| 创建
v
任务（Task）
|
| 编码为
v
A2A 消息（A2A Message）
|
| 通过以下方式传输
v
HTTP / 队列 / WebSocket
|
v
智能体 B
|
| 执行
v
结果（Result）
这就是核心概念。
智能体之间的通信并没有任何魔法。
最初，它可以简单到：
coding_agent.generate_code(task)
当智能体成为独立服务时，我们可以迁移到：
HTTP + JSON
当多个独立智能体需要共享的通信模型时，我们引入 A2A 协议：
task（任务）
message（消息）
capability（能力）
state（状态）
result（结果）
error（错误）
然后，当系统规模扩大时，我们再添加：
异步执行
队列
worker
服务发现
认证
授权
重试
流式传输
可观测性
重要的架构教训是：
不要因为协议听起来很高级就从一个协议开始。先从通信问题入手。当独立智能体之间的通用契约能带来真正价值时，再引入 A2A。
最简单的实现几行代码就能构建。
同一个契约以后可以演变成一个分布式多智能体平台。
这就是 A2A 的价值所在：它不是让智能体更聪明，而是让智能体更容易连接、更容易替换、更容易扩展、更容易观测、更容易组合。