如何构建一个多Agent系统，并为其安全交互设置协议？

引言：多Agent系统中的信任挑战

随着大模型（LLMs）驱动的Agent技术快速发展，构建复杂的多Agent系统（MAS）已成为AI应用部署的新趋势。然而，当多个自治Agent在分布式环境中协作时，最核心的挑战是如何确保它们之间通信的安全、完整性和可信赖性。恶意Agent注入虚假信息或中间人攻击都可能导致系统崩溃或决策错误。

本篇文章将聚焦于AI基础设施层面，展示如何通过实现基于非对称加密（数字签名）的通信协议，为Agent之间的消息交互提供强大的安全保障。我们将使用Python和FastAPI来模拟Agent的部署环境，并利用cryptography库实现签名和验证机制。

核心协议：非对称加密与数字签名

数字签名的工作流程如下：
1. 密钥对生成： 每个Agent拥有一个私钥（用于签名）和一个公钥（用于验证）。公钥可以公开。
2. 签名： 发送方Agent（如Agent A）使用其私钥对消息内容进行哈希并加密，生成数字签名。
3. 传输： Agent A将原始消息和数字签名一起发送给接收方Agent（Agent B）。
4. 验证： Agent B使用Agent A公开的公钥对接收到的签名进行解密，并独立计算接收消息的哈希值。如果两者哈希值匹配，则消息未被篡改，且确认为Agent A发送。

实战演练：构建安全Agent通信基础设施

步骤一：环境准备与密钥生成

首先，我们需要安装必要的库。

pip install cryptography fastapi uvicorn pydantic

接下来，每个Agent都需要生成自己的私钥和公钥。在生产环境中，这通常由密钥管理服务（KMS）负责。在示例中，我们使用文件存储。

from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# 生成Agent A的密钥对
private_key_a = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key_a = private_key_a.public_key()

# 序列化公钥（Agent B需要）
public_pem_a = public_key_a.public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo
)

# 实际应用中，Agent B也需要生成自己的密钥对，并交换公钥。
# 为简化示例，我们假设 Agent B 已拥有 Agent A 的公钥。

print("Agent A Public Key (for distribution):\n" + public_pem_a.decode())

步骤二：实现签名与验证工具类

这是我们的核心通信协议逻辑，用于处理消息的加密签名和解密验证。

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.exceptions import InvalidSignature
import base64

class SecureAgentProtocol:
    def __init__(self, private_key=None, peer_public_key_pem=None):
        self.private_key = private_key
        if peer_public_key_pem:
            # 加载对端Agent的公钥
            self.peer_public_key = serialization.load_pem_public_key(
                peer_public_key_pem
            )
        else:
            self.peer_public_key = None

    def sign_message(self, message: str) -> str:
        """使用Agent私钥对消息进行签名。"""
        if not self.private_key:
            raise ValueError("Private key required for signing.")

        # 消息转换为字节
        message_bytes = message.encode('utf-8')

        signature = self.private_key.sign(
            message_bytes,
            padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
            hashes.SHA256()
        )

        # 返回Base64编码的签名，便于HTTP传输
        return base64.urlsafe_b64encode(signature).decode('utf-8')

    def verify_signature(self, message: str, signature: str) -> bool:
        """使用对端Agent公钥验证签名。"""
        if not self.peer_public_key:
            raise ValueError("Peer public key required for verification.")

        message_bytes = message.encode('utf-8')
        signature_bytes = base64.urlsafe_b64decode(signature.encode('utf-8'))

        try:
            self.peer_public_key.verify(
                signature_bytes,
                message_bytes,
                padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
                hashes.SHA256()
            )
            print("Signature VERIFIED: Message is authentic and untampered.")
            return True
        except InvalidSignature:
            print("Signature FAILED: Message integrity compromised or sender invalid.")
            return False

步骤三：Agent A (发送方) 部署与通信

Agent A使用FastAPI暴露一个发送消息的接口，并在发送前对消息进行签名。

# 假设我们在 Agent A 的环境中
# ------------------------------------------------
agent_a_protocol = SecureAgentProtocol(private_key=private_key_a) 

# 模拟Agent A发送给Agent B的数据结构
class SignedMessage(BaseModel):
    content: str
    signature: str

# 模拟Agent A发送消息的函数
def agent_a_sends(payload: str):
    signature = agent_a_protocol.sign_message(payload)
    print(f"Agent A signed message: {payload[:20]}...")
    return SignedMessage(content=payload, signature=signature)

# 实际部署时，Agent A会调用Agent B的HTTP API发送此结构体。
# example_message = agent_a_sends("Hello Agent B, the critical task allocation is complete.")
# print(example_message.json())

步骤四：Agent B (接收方) 部署与验证

Agent B接收到消息后，必须首先使用Agent A的公钥验证签名，然后才能处理内容。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json

# ------------------------------------------------
# 假设我们在 Agent B 的环境中
app = FastAPI()

# Agent B 加载 Agent A 的公钥，以进行验证
agent_b_protocol = SecureAgentProtocol(peer_public_key_pem=public_pem_a)

@app.post("/receive_task_update")
async def receive_task_update(signed_message: SignedMessage):
    print(f"Agent B received payload: {signed_message.content[:20]}...")

    # 1. 执行验证协议
    if not agent_b_protocol.verify_signature(signed_message.content, signed_message.signature):
        # 验证失败，拒绝执行指令
        raise HTTPException(status_code=403, detail="Invalid signature or tampered message.")

    # 2. 验证成功，执行Agent逻辑

    # 假设这是Agent B的核心逻辑:
    task_data = json.loads(signed_message.content) 
    print(f"[AGENT B SUCCESS] Safely processing critical task: {task_data['task_id']}")

    return {"status": "ok", "message": "Task received and verified successfully."}

# 如何测试:
# 1. 启动 Agent B (假设在终端): uvicorn your_script_name:app --reload --port 8001
# 2. 模拟 Agent A 发送请求:
# import requests
# import json

# critical_payload = json.dumps({"task_id": "T-402", "priority": 5, "instruction": "Start optimization phase"})
# msg = agent_a_sends(critical_payload)

# response = requests.post(
#     "http://localhost:8001/receive_task_update", 
#     json=msg.dict()
# )
# print(response.json())

# 3. 模拟篡改攻击 (如中间人修改 content，但 signature 保持不变):
# tampered_msg = msg.copy(deep=True)
# tampered_msg.content = "ATTACKER: Do something malicious!"
# response_tampered = requests.post(
#     "http://localhost:8001/receive_task_update", 
#     json=tampered_msg.dict()
# )
# print(response_tampered.json())
# 此时 Agent B 将输出 "Signature FAILED" 并返回 403 错误。

总结

通过在分布式Agent通信中嵌入数字签名协议，我们有效地在应用层（而非仅仅传输层）解决了信任问题。这种方法确保了消息的发送者身份经过验证，并且内容在传输过程中保持了完整性。这是构建可靠、安全和高性能AI基础设施的关键一环，尤其适用于处理高风险决策或敏感数据的多Agent协作场景。