殷龙飞的博客

A2A（Agent2Agent）系列专题 (九) A2A 的通信机制：HTTP vs WebSocket

 协议/大模型/A2A  3476  7 分钟

A2A 的通信机制：HTTP vs WebSocket

摘要：通信机制是 A2A（Agent2Agent）协议的核心，决定了代理间如何高效交换 AgentCard、任务和状态更新。A2A 支持 HTTP 和 WebSocket，分别满足简单请求和实时交互的需求。本文深入剖析 A2A 的通信机制，聚焦 HTTP 和 WebSocket 的设计、实现细节、性能对比及其在代理协作中的应用。结合 GitHub 仓库的实现、Mermaid 图表和代码示例，我们将揭示 A2A 如何通过硬核的通信设计支持动态协作，为开发者提供深入的技术洞察。

1. 引言：通信机制的基石

在企业 AI 系统中，代理（Agent）通过通信交换任务、元数据和状态更新，驱动从费用报销到实时客服的复杂协作。Google 的 A2A（Agent2Agent） 协议采用 HTTP 和 WebSocket 作为通信机制，分别满足同步请求和实时交互的需求。这种双协议设计借鉴了分布式系统的通信模式（如 REST 和 WebSocket），但针对 AI 代理的动态性进行了优化。

HTTP 提供简单、可靠的请求-响应模型，适合获取 AgentCard 或提交任务；WebSocket 提供持久连接，支持实时状态更新和多模态交互。本文将深入解析 A2A 的通信机制，结合 Google A2A GitHub 仓库 的实现，揭示其硬核内核。

2. A2A 通信机制概览

A2A 的通信机制围绕代理间的核心交互设计，包括：

  • AgentCard 交换：Host Agent 获取 Remote Agent 的元数据。
  • 任务提交与处理：Host Agent 提交任务，Remote Agent 返回结果。
  • 状态更新：实时通知任务状态变化（如 in_progress 到 completed）。
  • 动态交互：支持文本、表单、音视频等模式的协商和传输。

以下是通信机制的架构示意图：

graph TD
    A[Host Agent] -->|HTTP| B[Remote Agent]
    A -->|WebSocket| C[Remote Agent]
    B --> D[AgentCard Response]
    B --> E[Task Response]
    C --> F[Status Updates]
    C --> G[Dynamic Interaction]
    style B fill:#bbf,stroke:#333
    style C fill:#bfb,stroke:#333

2.1 通信需求

A2A 的通信机制需满足以下需求：

  • 可靠性：确保消息不丢失，状态一致。
  • 动态性：支持运行时协商和多模态交互。
  • 性能：低延迟，适配高并发场景。
  • 兼容性：跨平台、跨供应商的代理协作。

2.2 HTTP vs WebSocket

  • HTTP：基于请求-响应模型，适合简单、同步的任务。
  • WebSocket：基于持久连接，适合实时、双向的交互。

3. HTTP：简单可靠的通信

3.1 设计与实现

HTTP 是 A2A 的基础通信协议，基于 REST 风格，适用于以下场景：

  • 获取 AgentCard：GET /agentcard 返回代理元数据。
  • 提交任务：POST /task 发送任务数据。
  • 查询状态：GET /task/{taskId} 获取任务状态。

HTTP 请求示例（提交任务）：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

POST /a2a/task HTTP/1.1
Host: example.com
Content-Type: application/json
Authorization: Bearer token123

{
  "taskId": "task-001",
  "type": "expense",
  "data": {
    "amount": 100,
    "currency": "USD"
  },
  "status": "created"
}

响应：

1
2
3
4

{
  "taskId": "task-001",
  "status": "accepted"
}

3.2 优势

  • 简单性：基于成熟的 HTTP 协议，易于实现和调试。
  • 兼容性：支持广泛的客户端和服务器框架（如 Flask、Express）。
  • 幂等性：GET 和某些 POST 请求可重复执行，适合任务重试。

3.3 局限

  • 实时性不足：需要轮询（GET /task/{taskId}）获取状态更新，增加延迟。
  • 单向性：请求-响应模型不适合双向或流式交互。
  • 开销：HTTP 头可能增加数据量，影响高频通信。

3.4 使用场景

  • 获取 AgentCard 或任务初始提交。
  • 低频、同步的任务（如验证费用报销）。
  • 与传统 REST API 集成的场景。

4. WebSocket：实时动态的通信

4.1 设计与实现

WebSocket 是一个全双工协议，基于持久连接，适用于以下场景：

  • 状态更新：推送任务状态变化（如 in_progress 到 completed）。
  • 动态交互：支持流式传输（音视频）或表单请求。
  • 推送通知：实时通知代理或用户（如任务失败警告）。

WebSocket 消息示例（状态更新）：

1
2
3
4
5
6

{
  "event": "task_update",
  "taskId": "task-001",
  "status": "in_progress",
  "progress": 50
}

完成通知：

1
2
3
4
5
6
7
8
9

{
  "event": "task_update",
  "taskId": "task-001",
  "status": "completed",
  "result": {
    "status": "approved",
    "message": "Processed 100 USD"
  }
}

A2A 的 WebSocket 实现基于事件驱动模型，定义了以下事件类型：

  • task_update：任务状态或进度更新。
  • interaction_request：请求动态交互（如表单）。
  • error：通信或任务错误。

4.2 优势

  • 实时性：低延迟，适合动态场景（如音视频流）。
  • 双向性：支持服务器主动推送，减少轮询开销。
  • 高效性：持久连接降低协议开销，适配高频通信。

4.3 局限

  • 复杂性：连接管理（断线重连、心跳检测）增加开发成本。
  • 资源占用：持久连接可能消耗更多服务器资源。
  • 兼容性：部分代理环境可能不支持 WebSocket。

4.4 使用场景

  • 实时任务监控（状态更新、进度反馈）。
  • 多模态交互（音视频、表单）。
  • 高并发或动态协作场景。

5. HTTP vs WebSocket：性能与适用性对比

5.1 性能对比

以下是 HTTP 和 WebSocket 的性能对比：

特性HTTPWebSocket
延迟较高（需轮询）较低（实时推送）
吞吐量适中（头开销较大）较高（连接复用）
连接管理无需（短连接）复杂（持久连接）
资源占用较低（按需连接）较高（保持连接）
实时性较差（轮询或长轮询）优秀（双向通信）

5.2 适用性对比

  • HTTP：适合初始化交互（AgentCard 获取、任务提交）或低频任务，优先考虑简单性和兼容性。
  • WebSocket：适合实时更新（任务状态、推送通知）或动态交互（音视频流），优先考虑低延迟和双向性。

5.3 混合模式

A2A 支持混合模式，例如：

  1. 使用 HTTP 获取 AgentCard 和提交任务。
  2. 切换到 WebSocket 监控状态更新或处理动态交互。

以下是混合模式的流程图：

flowchart TD
    A[Host Agent] -->|HTTP: GET /agentcard| B[Remote Agent]
    B -->|AgentCard| A
    A -->|HTTP: POST /task| B
    B -->|Task Accepted| A
    A -->|WebSocket: Connect| B
    B -->|task_update| A
    B -->|interaction_request| A
    A -->|WebSocket: Response| B

GitHub Issues 提到，社区正在优化 WebSocket 的重连机制和 HTTP 的缓存策略，以提升混合模式的性能。

6. 通信安全与可靠性

6.1 安全机制

  • 认证：AgentAuthentication 支持 Bearer 令牌，HTTP 使用 Authorization 头，WebSocket 使用连接参数。
  • 加密：HTTP 使用 HTTPS，WebSocket 使用 WSS，确保数据安全。
  • 访问控制：未来计划支持 OAuth 2.0 和 RBAC（参考 GitHub Issues）。

6.2 可靠性机制

  • HTTP：
    • 幂等性：GET 和 POST 请求通过 taskId 确保重复执行安全。
    • 重试：客户端可实现指数退避重试策略。
  • WebSocket：
    • 心跳检测：定期发送 ping/pong 消息，检测连接状态。
    • 重连机制：断线后自动重连，恢复未完成的任务。
    • 消息确认：支持 ACK 机制，确保消息送达。

7. 代码示例：实现 HTTP 和 WebSocket 通信

以下是一个基于 samples/python/agents/google_adk 的费用报销代理，展示 HTTP 和 WebSocket 的混合使用。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

# Remote Agent：支持 HTTP 和 WebSocket
from a2a import A2AServer, AgentCard
import asyncio
from websockets.server import serve

class ExpenseAgent(A2AServer):
    def __init__(self):
        card = AgentCard(
            name="ExpenseAgent",
            description="Processes expense reimbursements",
            url="http://localhost:8080/a2a",
            capabilities={
                "streaming": False,
                "pushNotifications": True,
                "interactionModes": ["text"]
            },
            schema={
                "input": {
                    "type": "object",
                    "properties": {
                        "amount": {"type": "number"},
                        "currency": {"type": "string"}
                    },
                    "required": ["amount", "currency"]
                }
            }
        )
        super().__init__(card=card)

    async def handle_task(self, task: dict) -> dict:
        await self.notify_status(task["taskId"], "in_progress")
        if task["type"] != "expense":
            await self.notify_status(task["taskId"], "failed")
            return {"status": "failed", "error": "Invalid task type"}
        amount = task["data"]["amount"]
        if amount <= 0:
            await self.notify_status(task["taskId"], "failed")
            return {"status": "failed", "error": "Invalid amount"}
        await asyncio.sleep(1)  # 模拟处理
        result = {"status": "approved", "message": f"Processed {amount} {task['data']['currency']}"}
        await self.notify_status(task["taskId"], "completed")
        return {"status": "completed", "result": result}

    async def websocket_handler(self, websocket, path):
        async for message in websocket:
            data = json.loads(message)
            if data["event"] == "subscribe":
                task_id = data["taskId"]
                # 模拟状态更新
                await websocket.send(json.dumps({
                    "event": "task_update",
                    "taskId": task_id,
                    "status": "in_progress",
                    "progress": 50
                }))

# Host Agent：使用 HTTP 和 WebSocket
from a2a import A2AClient
import websockets
import json

async def expense_client(remote_url: str):
    client = A2AClient(remote_url)

    # HTTP：获取 AgentCard
    agent_card = await client.get_agent_card()
    print(f"Agent: {agent_card['name']}")

    # HTTP：提交任务
    task = {
        "taskId": "task-001",
        "type": "expense",
        "data": {"amount": 100, "currency": "USD"}
    }
    response = await client.submit_task(task)
    print(f"Task submitted: {response}")

    # WebSocket：订阅状态更新
    async with websockets.connect("ws://localhost:8080/a2a/ws") as ws:
        await ws.send(json.dumps({"event": "subscribe", "taskId": "task-001"}))
        async for message in ws:
            update = json.loads(message)
            print(f"Status update: {update}")
            if update["status"] in ["completed", "failed"]:
                break

if __name__ == "__main__":
    server = ExpenseAgent()
    asyncio.run(expense_client("http://localhost:8080/a2a"))
    server.run(port=8080)

代码解析

  1. 服务器：实现 HTTP（任务提交）和 WebSocket（状态更新），支持混合通信。
  2. 客户端：使用 HTTP 获取 AgentCard 和提交任务，通过 WebSocket 订阅状态更新。
  3. 异步处理：基于 asyncio 和 websockets，确保高并发性能。
  4. 事件驱动：WebSocket 使用事件（如 task_update）处理动态交互。

8. 硬核设计：通信机制的权衡

8.1 HTTP 的权衡

  • 优势：简单、兼容性强，适合初始化和低频交互。
  • 挑战：轮询增加延迟，头开销影响高频通信。
  • 优化：支持 HTTP/2 减少头开销，缓存 AgentCard 降低请求频率。

8.2 WebSocket 的权衡

  • 优势：实时性强，适合动态交互和高并发。
  • 挑战：连接管理复杂，可能消耗更多资源。
  • 优化：实现心跳检测和重连机制，压缩消息数据。

8.3 分布式场景

在分布式系统中，通信机制面临以下问题：

  • 负载均衡：如何在多服务器间分配 HTTP 和 WebSocket 请求？
  • 一致性：WebSocket 断连后，如何恢复状态？
  • 扩展性：高并发下，WebSocket 的连接池管理需优化。

GitHub Issues 提到，社区计划引入 WebSocket 集群支持（如通过 Redis 同步连接状态）。

9. 应用场景与展望

A2A 的通信机制适用于以下场景：

  • 企业自动化：HTTP 用于任务初始化，WebSocket 用于状态监控。
  • 实时交互：WebSocket 支持客服场景的音视频流。
  • 跨平台协作：HTTP 的兼容性确保不同供应商的代理集成。

未来，A2A 可能引入以下改进：

  • 协议扩展：支持 gRPC 或 MQTT，适配更多场景。
  • 性能优化：压缩 WebSocket 消息，优化 HTTP 缓存。
  • 安全性：增强认证和加密机制（如零信任模型）。

10. 结语：通信机制的未来

A2A 的 HTTP 和 WebSocket 通信机制为代理间协作提供了可靠性和动态性的平衡。HTTP 的简单性和 WebSocket 的实时性共同支持了从简单任务到复杂交互的场景。未来，A2A 将在性能、安全性和分布式支持上进一步突破，为企业 AI 系统提供更强大的通信框架。

在下一篇文章中，我们将探讨 A2A 的性能优化，深入分析高并发和分布式场景的瓶颈与解决方案。欢迎访问 A2A GitHub 仓库，加入社区，探索 AI 协作的未来！

updatedupdated2025-04-172025-04-17

相关文章：