A2A消息类型与格式规范

基于消息特征的自动编码选择算法：

class A2AMessageEncoder:
    """A2A Protocol消息编码器"""
    
    def select_encoding(self, message: Dict, context: Dict) -> str:
        """智能选择最优编码格式"""
        
        # 消息大小评估
        estimated_size = self._estimate_message_size(message)
        
        # 性能要求评估
        performance_requirement = context.get("performance", "normal")
        
        # 兼容性要求
        compatibility_requirement = context.get("compatibility", "standard")
        
        # 决策逻辑
        if estimated_size > 1024 * 1024:  # > 1MB
            return "protobuf" if performance_requirement == "high" else "msgpack"
        
        elif performance_requirement == "high":
            return "msgpack"
        
        elif compatibility_requirement == "maximum":
            return "json"
        
        elif message.get("header", {}).get("messageType") in ["HEARTBEAT", "HANDSHAKE"]:
            return "json"  # 控制消息优先可读性
        
        else:
            return "msgpack"  # 默认高效格式
    
    def encode_message(self, message: Dict, encoding: str = None) -> bytes:
        """编码A2A消息"""
        
        if encoding is None:
            encoding = self.select_encoding(message, {})
        
        # 添加编码元信息
        message["_meta"] = {
            "encoding": encoding,
            "version": "1.0",
            "timestamp": datetime.utcnow().isoformat()
        }
        
        if encoding == "json":
            return json.dumps(message, ensure_ascii=False).encode('utf-8')
        
        elif encoding == "msgpack":
            import msgpack
            return msgpack.packb(message, use_bin_type=True)
        
        elif encoding == "protobuf":
            # 转换为protobuf并序列化
            pb_message = self._dict_to_protobuf(message)
            return pb_message.SerializeToString()
        
        else:
            raise ValueError(f"不支持的编码格式: {encoding}")
    
    def decode_message(self, data: bytes, encoding: str = None) -> Dict:
        """解码A2A消息"""
        
        if encoding is None:
            # 尝试自动检测编码格式
            encoding = self._detect_encoding(data)
        
        if encoding == "json":
            return json.loads(data.decode('utf-8'))
        
        elif encoding == "msgpack":
            import msgpack
            return msgpack.unpackb(data, raw=False, strict_map_key=False)
        
        elif encoding == "protobuf":
            pb_message = self._bytes_to_protobuf(data)
            return self._protobuf_to_dict(pb_message)
        
        else:
            raise ValueError(f"不支持的编码格式: {encoding}")

# 性能基准测试结果
ENCODING_BENCHMARKS = {
    "json": {
        "encode_speed": "1.0x",
        "decode_speed": "1.0x", 
        "size_ratio": "1.0x",
        "cpu_usage": "low"
    },
    "msgpack": {
        "encode_speed": "3.2x",
        "decode_speed": "2.8x",
        "size_ratio": "0.7x",
        "cpu_usage": "medium"
    },
    "protobuf": {
        "encode_speed": "4.1x",
        "decode_speed": "3.9x",
        "size_ratio": "0.5x",
        "cpu_usage": "high"
    }
}