👁 90 次阅读
# AutoGen 企业级代理治理：加密动作收据（AAR）实现可验证审计追踪
现象描述
企业级多智能体系统（Multi-Agent System）在生产环境中面临一个核心矛盾：Agent 之间的对话记录、工具调用日志以及 LLM 输出虽然能呈现“发生了什么”，却无法向外部审计员或合规系统证明“未被篡改”。例如，一个金融合规场景中，Agent A 执行了“转账 100 万”的操作，Agent B 随后确认了该交易。事后若出现纠纷，仅凭日志无法排除以下风险：
- 日志在传输或存储过程中被修改；
- 某个 Agent 声称它只执行了指令 A，但日志显示它执行了指令 B；
- 不同的 Agent 对同一动作产生了矛盾的记录，且无法确定哪一份是真实的。
原因分析
传统日志系统（如 LLM 追踪、对话历史）依赖中心化存储和信任假设，具有以下根本弱点：
1. 缺乏不可否认性：日志可以被写入者或存储中间人修改，事后无法证明原始动作的真实性。 2. 无链式关联：多个 Agent 在同一个工作流中的动作之间没有密码学联系，无法判断后一个动作是否基于前一个动作的合法输出。 3. 审计成本高：审计员需要手动比对多个日志系统，且无法以密码学方式验证数据完整性。 4. 合规标准缺失：GDPR、SOX、HIPAA 等法规要求“可验证的审计轨迹”，传统日志无法满足“数字签名+时间戳+防篡改”的硬性要求。
加密动作收据（AAR）的核心理念是：将每个 Agent 的动作（包括接收到的指令、执行的操作、消费的数据、产生的输出）打包成一个带签名的结构化收据，并可选地链接到前一个收据，形成 Merkle 链或哈希链。这样，审计员仅需持有可信的公钥，就可以独立验证整个工作流的所有动作是否被篡改。
完整解决方案
架构设计
AAR 在 AutoGen 中通过以下组件实现：
- Agent 签名模块：每个 Agent 加载一个私钥（由企业密钥管理系统管理），对动作内容进行签名。
- 收据生成器：在 Agent 每次执行
send、receive、run_tool等操作后，自动生成收据并广播给下游 Agent 或审计存储。 - 收据验证器：审计员或合规系统通过公钥和收据链验证所有动作的完整性。
代码示例
以下示例演示如何在 AutoGen 中启用 AAR 并生成可验证收据。假设已安装 autogen 和 cryptography 库。
import json
import hashlib
from datetime import datetime, timezone
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
from cryptography.exceptions import InvalidSignature
class ActionReceipt:
"""加密动作收据类"""
def __init__(self, agent_id: str, action: str, input_data: str, output_data: str,
parent_receipt_hash: str = None, private_key: ec.EllipticCurvePrivateKey = None):
self.agent_id = agent_id
self.action = action # 例如 "send", "run_tool", "generate_reply"
self.input_data = input_data
self.output_data = output_data
self.timestamp = datetime.now(timezone.utc).isoformat()
self.parent_receipt_hash = parent_receipt_hash or ""
self.receipt_hash = self._compute_hash()
self.signature = self._sign(private_key) if private_key else None
def _compute_hash(self) -> str:
content = json.dumps({
"agent_id": self.agent_id,
"action": self.action,
"input_data": self.input_data,
"output_data": self.output_data,
"timestamp": self.timestamp,
"parent_receipt_hash": self.parent_receipt_hash
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def _sign(self, private_key: ec.EllipticCurvePrivateKey) -> str:
data = self.receipt_hash.encode()
signature = private_key.sign(data, ec.ECDSA(hashes.SHA256()))
return signature.hex()
def to_dict(self) -> dict:
return {
"agent_id": self.agent_id,
"action": self.action,
"input_data": self.input_data,
"output_data": self.output_data,
"timestamp": self.timestamp,
"parent_receipt_hash": self.parent_receipt_hash,
"receipt_hash": self.receipt_hash,
"signature": self.signature
}
@staticmethod
def verify(receipt_dict: dict, public_key: ec.EllipticCurvePublicKey) -> bool:
"""验证收据签名及哈希完整性"""
signature = bytes.fromhex(receipt_dict.pop("signature"))
expected_hash = receipt_dict.pop("receipt_hash")
# 重新计算哈希
computed_hash = hashlib.sha256(
json.dumps(receipt_dict, sort_keys=True).encode()
).hexdigest()
if computed_hash != expected_hash:
return False
# 验证签名
try:
public_key.verify(signature, expected_hash.encode(), ec.ECDSA(hashes.SHA256()))
return True
except InvalidSignature:
return False
在 AutoGen Agent 中集成收据生成（以 AssistantAgent 为例）：
from autogen import AssistantAgent, UserProxyAgent, ConversableAgent
class AuditableAssistantAgent(AssistantAgent):
def __init__(self, name, private_key, public_key, **kwargs):
super().__init__(name, **kwargs)
self._private_key = private_key
self._public_key = public_key
self._last_receipt_hash = None # 用于链式链接
def _generate_receipt(self, action, input_data, output_data):
receipt = ActionReceipt(
agent_id=self.name,
action=action,
input_data=input_data,
output_data=output_data,
parent_receipt_hash=self._last_receipt_hash,
private_key=self._private_key
)
self._last_receipt_hash = receipt.receipt_hash
# 将收据存入专用存储（例如数据库或Kafka）
# store_receipt(receipt.to_dict())
return receipt
def send(self, message, recipient, request_reply=None, silent=False):
# 调用父类send
result = super().send(message, recipient, request_reply, silent)
# 生成收据
receipt = self._generate_receipt(
action="send",
input_data=str(message),
output_data=result
)
# 可选：将收据哈希附加到消息中，供接收方验证
# message["receipt_hash"] = receipt.receipt_hash
return result
def receive(self, message, sender, request_reply=None, silent=False):
receipt = self._generate_receipt(
action="receive",
input_data=str(message),
output_data=""
)
return super().receive(message, sender, request_reply, silent)
收据验证示例（审计方）：
# 假设审计员持有所有Agent的公钥字典
public_keys = {
"agent_trade": agent_trade_pub_key,
"agent_risk": agent_risk_pub_key,
}
# 从存储中读取收据链
receipts = load_receipts_from_db()
for receipt_dict in receipts:
agent_id = receipt_dict["agent_id"]
pub_key = public_keys[agent_id]
is_valid = ActionReceipt.verify(receipt_dict, pub_key)
if not is_valid:
raise Exception(f"收据验证失败: {receipt_dict['receipt_hash']}")
print("所有收据验证通过")
参数详解表格
| 字段 | 类型 | 必填 | 说明 |
agent_id | string | 是 | 执行动作的 Agent 唯一标识符，建议采用 UUID 或企业级名称 |
action | string | 是 | 动作类型，如 send、receive、run_tool、generate_reply 等 |
input_data | string | 是 | 动作的输入数据，通常为 JSON 序列化的消息或工具参数 |
output_data | string | 是 | 动作的输出数据，缺失时可为空字符串 |
timestamp | string (ISO 8601) | 是 | 动作发生时的 UTC 时间戳，需来自可信时间源（如 NTP） |
parent_receipt_hash | string | 否 | 前一个收据的 SHA-256 哈希，用于构建链式结构。若无则留空 |
receipt_hash | string（自动生成） | 是 | 本收据所有字段（不含签名）的 SHA-256 哈希 |
signature | string (hex) | 是 | 对 receipt_hash 的 ECDSA 签名，使用 Agent 私钥 |
常见错误排查
错误 1：哈希不匹配导致验证失败
现象：verify() 返回 False，且日志显示 computed_hash != expected_hash。
原因：收据在传输或存储过程中被修改，或者 to_dict() 序列化时字段顺序不一致（Python 字典键顺序）。
解决：确保序列化时使用 sort_keys=True；验证时使用与生成时完全相同的字段集合（排除 signature 和 receipt_hash 本身）。建议将收据存储为 JSON 并保留原始字符串。
错误 2：私钥未正确加载
现象：生成收据时抛出 TypeError: 'NoneType' object is not callable 或签名函数返回空。
原因：Agent 未正确配置私钥，或者私钥使用了错误格式（如 PEM 文件损坏）。
解决：确认私钥加载路径，使用 cryptography 的 load_pem_private_key 并传入密码。建议在 Agent 初始化时验证私钥与公钥是否匹配：
public_key = private_key.public_key()
# 再存储公钥到审计员信任列表错误 3：链式收据的 parent_receipt_hash 无法追踪
现象：审计时发现收据链断裂，某个收据的 parent_receipt_hash 无法在数据库中匹配。
原因：Agent 重启后状态丢失，导致 _last_receipt_hash 重置为 None；或者收据存储出现异步写入延迟。
解决：将 _last_receipt_hash 持久化到数据库或分布式缓存（如 Redis），并在 Agent 加载时恢复。对于高并发场景，使用乐观锁保证收据写入顺序。
错误 4：时间戳偏差导致合规性问题
现象：审计员发现收据时间戳早于系统启动时间，或晚于当前时间过多。
原因：Agent 所在服务器的时钟未同步，或使用了系统本地时间而非 UTC。
解决：强制使用 datetime.now(timezone.utc) 生成 UTC 时间戳；生产环境建议部署 NTP 服务并定期检查时钟偏移。对于关键场景，可引入外部时间戳服务（如 RFC 3161）。
错误 5：公钥分发的信任问题
现象：审计员无法确定收到的公钥是否属于真正的 Agent。
原因：公钥通过不安全通道分发，可能被中间人替换。
解决：采用 PKI 体系，每个 Agent 的公钥由企业 CA 签发证书，审计员仅信任 CA 根证书。或者使用区块链等去中心化方式存储公钥哈希。
总结
加密动作收据（AAR）为 AutoGen 企业级部署提供了密码学级的审计保障，解决了传统日志无法满足的不可否认性、链式关联和防篡改问题。通过在每个 Agent 动作后生成带签名的结构化收据，并形成哈希链，审计员可以独立验证整个多智能体工作流的完整性。实际部署时需注意私钥管理、时间同步、链式持久化等关键环节，确保收据的可信度。
该方案已由社区在 issue #7353 中详细讨论，并计划集成到 AutoGen 核心库中。企业用户可基于上述代码示例先行在自定义 Agent 中实现，等待官方发布更完善的 AAR API。
FAQ
Q1：加密动作收据与现有的 LLM 追踪（如 LangSmith、Weights & Biases）有何区别？
A：LLM 追踪关注的是模型调用的可观测性，记录“发生了什么”，但缺乏密码学证明。AAR 提供的是可验证性——审计员可以证明记录未被篡改。两者可以互补：追踪用于调试，AAR 用于合规。
Q2：收据的签名会带来显著性能开销吗？
A：ECDSA P-256 签名生成约需 0.5ms，验证约需 0.2ms（现代 CPU）。对于大部分 Agent 交互（每秒几十到几百次），性能影响可忽略。若需高频签名（如流式推理），建议使用 Ed25519（更快的签名算法）。
Q3：如果需要删除某些敏感数据，收据链如何维护？
A：收据链本身不可删除（否则破坏链式完整性）。合规场景下，建议采用“加密存储+访问控制”策略：收据内容可加密，审计员需要解密密钥才能查看。删除操作应记录为“删除收据”动作，并生成新的收据指向被删除收据的哈希（证明该收据已被销毁）。
Q4：是否支持跨组织或跨云的 Agent 协作？
A：支持。每个 Agent 持有自己的签名密钥，其他组织只需信任该 Agent 的 CA 签发的公钥证书即可。收据可以通过标准消息队列（如 Kafka）跨网络传输，接收方验证签名后即可纳入本地的审计链。
Q5：收据的存储空间需求大吗？
A：每个收据 JSON 大小约 200-500 字节（含签名）。假设一个工作流包含 1000 次 Agent 交互，总存储约 200-500KB。对于企业级审计，通常要求保留 3-7 年，可通过压缩或归档至冷存储来降低成本。