Agent间通信协议设计：为什么你的Multi-Agent系统需要一套对话规范

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List
from enum import Enum
from datetime import datetime

class MessageType(str, Enum):
    DIRECT = "direct"           # 点对点消息
    BROADCAST = "broadcast"     # 广播
    REQUEST = "request"         # 请求（需响应）
    RESPONSE = "response"       # 响应
    ERROR = "error"             # 错误
    HEARTBEAT = "heartbeat"     # 心跳
    EVENT = "event"             # 事件通知

class AgentMessage(BaseModel):
    # === 身份标识 ===
    message_id: str = Field(..., description="全局唯一消息ID")
    correlation_id: str = Field(..., description="关联ID，用于追踪请求链")
    trace_id: str = Field(..., description="分布式追踪ID")
    
    # === 发送方/接收方 ===
    sender: str = Field(..., description="发送者Agent ID")
    recipient: Optional[str] = Field(None, description="接收者Agent ID（广播时为空）")
    
    # === 内容载荷 ===
    message_type: MessageType
    payload: Dict[str, Any] = Field(default_factory=dict, description="业务数据")
    metadata: Dict[str, Any] = Field(default_factory=dict, description="扩展元数据")
    
    # === 协议控制 ===
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    ttl: int = Field(default=300, description="消息生存时间（秒）")
    priority: int = Field(default=5, description="优先级 1-10，1最高")
    
    # === 版本控制 ===
    protocol_version: str = Field(default="1.0")
    schema_version: str = Field(default="1.0", description="payload schema版本")

# 同步调用示例：Agent A 需要 Agent B 的分析结果才能继续
class SyncClient:
    def request(self, target_agent: str, message: AgentMessage, timeout: float = 30.0) -> AgentMessage:
        # 发送请求并阻塞等待响应
        response = self.transport.send_and_wait(message, timeout)
        if response.message_type == MessageType.ERROR:
            raise AgentCommunicationError(response.payload)
        return response

# 异步发布示例
class AsyncPublisher:
    def publish(self, message: AgentMessage):
        # 发送到消息队列，不等待响应
        self.message_broker.publish(
            exchange="agent.events",
            routing_key=f"agent.{message.recipient}",
            body=message.json()
        )

class HybridMessage(BaseModel):
    base: AgentMessage
    reply_to: Optional[str] = None        # 响应队列/回调地址
    callback_type: str = "queue"          # queue / webhook / polling

# 直接路由到特定Agent
message = AgentMessage(
    sender="planner_agent",
    recipient="coder_agent",  # 明确指定接收者
    message_type=MessageType.REQUEST,
    payload={"task": "实现用户认证模块"}
)

# 发布到主题，感兴趣的Agent订阅
message = AgentMessage(
    sender="monitor_agent",
    recipient=None,  # 广播
    message_type=MessageType.EVENT,
    payload={"event": "system_load_high", "cpu_percent": 95}
)

class CapabilityRouter:
    def route(self, message: AgentMessage) -> List[str]:
        required_capability = message.payload.get("required_capability")
        # 查询注册中心：哪些Agent具备此能力且当前健康？
        candidates = self.registry.find_agents(
            capability=required_capability,
            healthy=True,
            load_threshold=0.8
        )
        # 负载均衡选择
        return self.load_balancer.select(candidates, strategy="least_connections")

class ErrorCode(str, Enum):
    TIMEOUT = "AGENT_TIMEOUT"
    CAPACITY_EXCEEDED = "CAPACITY_EXCEEDED"
    INVALID_PAYLOAD = "INVALID_PAYLOAD"
    SCHEMA_MISMATCH = "SCHEMA_MISMATCH"
    DEPENDENCY_FAILED = "DEPENDENCY_FAILED"
    PERMISSION_DENIED = "PERMISSION_DENIED"
    UNKNOWN = "UNKNOWN"

class ErrorPayload(BaseModel):
    code: ErrorCode
    message: str
    details: Dict[str, Any] = Field(default_factory=dict)
    retryable: bool = Field(default=False)
    suggested_retry_after: Optional[int] = None  # 建议多久后重试（秒）
    failed_agent: str  # 发生错误的Agent ID
    stack_trace: Optional[str] = None  # 调试信息（仅在开发环境）

class ErrorHandler:
    def handle(self, error_msg: AgentMessage, context: ExecutionContext):
        error = ErrorPayload(**error_msg.payload)
        
        if not error.retryable:
            # 不可重试：立即向上传播或进入死信队列
            self.circuit_breaker.record_failure(error.failed_agent)
            raise NonRetryableError(error)
        
        if context.retry_count < context.max_retries:
            # 指数退避重试
            backoff = min(2 ** context.retry_count, 60)
            self.scheduler.schedule_retry(error_msg, delay=backoff)
        else:
            # 超过重试次数：进入死信队列，人工介入
            self.dead_letter_queue.enqueue(error_msg)
            self.alert_manager.notify(error)

class ExternalCallRequest(BaseModel):
    # 继承自AgentMessage的上下文
    trace_id: str
    correlation_id: str
    
    # 外部调用元数据
    target_system: str          # 如 "openai", "slack", "internal_db"
    endpoint: str               # API端点或函数名
    method: str                 # GET/POST/或函数调用
    parameters: Dict[str, Any]  # 请求参数
    
    # 安全与治理
    api_key_ref: str            # 密钥引用（非明文），如 "vault://openai-key"
    rate_limit_group: str       # 用于限流分组
    timeout: float = 30.0
    max_retries: int = 3

class ExternalCallResponse(BaseModel):
    success: bool
    status_code: Optional[int] = None
    data: Optional[Dict[str, Any]] = None
    error: Optional[ErrorPayload] = None
    latency_ms: float
    tokens_used: Optional[int] = None  # LLM调用特有

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Agent A    │     │  Agent B    │     │  Agent C    │
└──────┬──────┘     └──────┬──────┘     └──────┬──────┘
       │                   │                   │
       └───────────────────┼───────────────────┘
                           ▼
                  ┌─────────────────┐
                  │  Egress Gateway │  ← 统一鉴权、限流、日志、重试
                  │  (Sidecar模式)  │
                  └────────┬────────┘
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
           OpenAI      Database      Slack API

┌─────────────────────────────────────────┐
│           Application Layer             │  ← 业务Agent逻辑
│    (Planner / Coder / Reviewer ...)     │
├─────────────────────────────────────────┤
│         Agent SDK / Runtime             │  ← 生命周期管理、状态机
├─────────────────────────────────────────┤
│      Inter-Agent Communication          │  ← 本文讨论的协议层
│   (Message Bus / RPC / Event Stream)    │
├─────────────────────────────────────────┤
│         Transport Layer                 │  ← Redis/RabbitMQ/gRPC/HTTP
└─────────────────────────────────────────┘

# 适配器模式：将框架原生消息转换为标准协议
class AutoGenAdapter:
    def to_standard(self, autogen_message) -> AgentMessage:
        return AgentMessage(
            message_id=generate_uuid(),
            correlation_id=autogen_message.get("id"),
            trace_id=get_current_trace_id(),
            sender=autogen_message["name"],
            recipient=autogen_message.get("recipient"),
            message_type=MessageType.DIRECT,
            payload={"content": autogen_message["content"]},
            metadata={"source_framework": "autogen"}
        )
    
    def from_standard(self, msg: AgentMessage):
        # 转换为AutoGen期望的格式
        ...