代理通信协议 (ACP)简明教程

ACP 是一个标准化、可执行的协议,为多代理通信带来秩序。它提供了一个完整的框架,说明代理应该如何相互通信、跟踪可靠性并智能地路由任务。

代理通信协议 (ACP)简明教程
微信 ezpoda免费咨询:AI编程 | AI模型微调| AI私有化部署 | Tripo 3D | Meshy AI

随着 AI 系统从单模型应用演变为复杂的多代理架构,一个关键问题已经出现:代理不讲同一种语言

1、问题概述

今天的多代理系统受到五个基本问题的困扰:

  • 非结构化通信

代理传递自由形式的文本或不一致的 JSON 载荷,没有语义保证。一个代理可能发送 {"task": "summarize"},而另一个代理期望 {"action": "summarize", "type": "text"}。这种不一致导致静默失败、误解和脆弱的集成。

  • 没有置信度建模

当代理返回结果时,它有多自信?当前的系统没有标准化的方法来表达不确定性。这使得加权共识变得不可能,并迫使下游代理将所有响应视为同等可靠——这是一个危险的假设。

  • 约束违规

没有正式的机制来指定或执行诸如截止时间、token 限制或质量阈值等约束。被要求在 500ms 内响应的代理可能需要 5 秒,而请求代理无法事先知道这种情况。

  • 信任失败

多代理系统缺乏声誉跟踪。一个持续不可靠的代理受到与高度可靠的代理相同的对待。没有能力验证,失败后没有信任降级,也没有办法阻止有问题的代理。

  • 调试不可能

当 10 个代理的工作流程中出现问题时,事后分析需要进行考古学。如果没有结构化的消息格式和明确的意图语义,追踪失败的根源几乎是不可能的。

2、解决方案:代理通信协议 (ACP)

ACP 是一个标准化、可执行的协议,为多代理通信带来秩序。它提供了一个完整的框架,说明代理应该如何相互通信、跟踪可靠性并智能地路由任务。

2.1 核心架构

ACP 建立在四个支柱之上:

┌─────────────────────────────────────────────────────────────┐
│                    ACP 消息层                              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │   意图     │  │   约束     │  │     置信度         │  │
│  │   分类法   │  │   系统     │  │   + 不确定性       │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
├─────────────────────────────────────────────────────────────┤
│                    信任和路由层                           │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │    信任    │  │   校准     │  │  基于能力的        │  │
│  │   管理     │  │   系统     │  │      路由          │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

2.2 具有语义意图的结构化消息

每条 ACP 消息都遵循严格的模式,并经过 Pydantic 验证:

from acp import ACPMessage, Intent, Constraints, Context

message = ACPMessage(
    sender_id="coordinator",
    recipient_id="summarizer",
    intent=Intent.DELEGATE,
    content={
        "task": "summarize_document",
        "document_id": "doc_123",
    },
    confidence=0.87,
    constraints=Constraints(
        max_tokens=800,
        deadline_ms=5000,
        quality_threshold=0.85,
    ),
    context=Context(
        domain="legal",
        urgency="high",
    ),
)

ACP 定义了 10 个语义意图,每个都有特定的必填字段:

意图 目的 必填内容
REQUEST 询问信息 content.query
DELEGATE 分配任务 content.task
VERIFY 检查正确性 content.target_message_id
CHALLENGE 争议主张 content.reason
PROPOSE 建议操作 content.proposal
ACCEPT 同意提案 parent_message_id
REJECT 拒绝提案 content.reason
ESCALATE 上报给主管 content.reason
REPORT 状态更新 content.status
TERMINATE 结束交互 content.reason

这个分类法消除了歧义。当代理收到 DELEGATE 消息时,它确切知道期望什么。当它发送 REPORT 时,接收者知道这是状态更新,而不是新请求。

2.3 带校准的一等置信度

ACP 将置信度视为一等公民。每条消息都包含 0.01 到 0.99 之间的置信度分数(不允许绝对确定),以及可选的不确定性界限:

message = ACPMessage(
    # ...
    confidence=0.85,
    uncertainty_bounds=(0.78, 0.92),  # 95% 置信区间
)

但是,如果没有验证,声称的置信度就没有任何意义。ACP 的校准系统跟踪代理是否真正像他们声称的那样可靠:

from acp import Calibrator

calibrator = Calibrator()

# 随着时间的推移记录结果
calibrator.record_outcome("agent_01", claimed_confidence=0.90, success=True)
calibrator.record_outcome("agent_01", claimed_confidence=0.90, success=False)
calibrator.record_outcome("agent_01", claimed_confidence=0.85, success=True)

# 检查代理是否校准良好
metrics = calibrator.get_metrics("agent_01")
print(f"Reliability: {metrics.reliability_score:.2%}")
print(f"ECE: {metrics.expected_calibration_error:.4f}")

# 根据历史记录调整未来的声明
adjusted = calibrator.adjust_confidence("agent_01", claimed_confidence=0.90)
# 如果代理历史上过度自信,返回 0.81

期望校准误差 (ECE) 衡量代理的置信度与现实匹配的程度。声称 90% 置信度的代理应该在大约 90% 的时间内成功。校准器检测过度自信和不足自信,从而实现自动调整。

2.4 带自动进度的信任管理

ACP 实现了一个五级信任系统:

级别 分数 描述
VERIFIED 1.0 经过加密验证
TRUSTED 0.8 一致的良好性能
NEUTRAL 0.5 未知或新代理
SUSPICIOUS 0.2 存在警告标志
BLOCKED 0.0 不进行交互

信任级别根据性能自动调整:

from acp import TrustManager, TrustLevel

trust_manager = TrustManager(
    upgrade_threshold=10,      # 升级所需的成功次数
    downgrade_threshold=5,     # 降级所需的失败次数
)

# 新代理从 NEUTRAL 开始
level = trust_manager.get_trust_level("new_agent")  # NEUTRAL

# 记录成功
for _ in range(12):
    trust_manager.record_success("new_agent")

# 代理自动升级
level = trust_manager.get_trust_level("new_agent")  # TRUSTED -> VERIFIED

# 记录另一个代理的失败
for _ in range(6):
    trust_manager.record_failure("bad_agent", "Timeout")

# 代理自动降级
level = trust_manager.get_trust_level("bad_agent")  # SUSPICIOUS -> BLOCKED

被阻止的代理必须显式解除阻止——它们无法通过成功自动恢复,从而防止对系统的游戏化。

2.5 智能基于能力的路由

ACP 的路由器根据多个加权因素为每个任务选择最佳代理:

from acp import Router, AgentCapability

router = Router(weights={
    "reliability": 0.4,   # 基于校准的可靠性
    "trust": 0.3,         # 信任级别分数
    "calibration": 0.2,   # 代理校准得如何
    "capability": 0.1,    # 能力匹配分数
})

# 注册代理及其能力
router.register_agent(AgentCapability(
    agent_id="summarizer_fast",
    capabilities={"summarization", "text_processing"},
    min_deadline_ms=500,
    max_tokens=1000,
))

router.register_agent(AgentCapability(
    agent_id="summarizer_quality",
    capabilities={"summarization", "analysis"},
    min_deadline_ms=2000,
    max_tokens=4000,
))

# 基于消息约束进行路由
best_agent = router.route(
    message,
    calibrator=calibrator,
    trust_manager=trust_manager,
    required_capabilities={"summarization"},
)

路由器自动排除无法满足约束的代理(截止时间太短、token 限制太低),并根据其历史性能对其余代理进行评分。

2.6 多层验证

ACP 在多个级别验证消息:

from acp import ACPValidator

validator = ACPValidator(
    strict_mode=True,
    min_deadline_ms=100,
    max_clock_skew_seconds=60,
)

result = validator.validate(message)

if not result.valid:
    for error in result.errors:
        print(f"Validation failed: {error}")

验证包括:

  • 模式验证:必填字段、正确类型
  • 意图验证:内容匹配意图要求
  • 约束验证:截止日期和阈值是合理的
  • 时间验证:时间戳在可接受的偏移范围内
  • 信任验证:发送者未被阻止

2.7 现实世界集成

ACP 包括流行框架的适配器。这是一个 LangChain 集成示例:

from acp.adapters import ACPLangChainAdapter
from acp import Intent

adapter = ACPLangChainAdapter(
    agent_id="langchain_coordinator",
    default_domain="general",
)

# 将 LangChain 输出转换为 ACP 消息
acp_response = adapter.wrap_langchain_response(
    langchain_output=chain.invoke(input),
    recipient_id="downstream_agent",
    intent=Intent.REPORT,
    confidence=0.88,
)

# 将传入的 ACP 消息转换为 LangChain
chain_input = adapter.create_chain_input(incoming_acp_message)
result = chain.invoke(chain_input)

3、开始使用

使用 pip 安装 ACP:

pip install acp-protocol

# 支持 LangChain
pip install acp-protocol[langchain]

在 10 行以内进行基本使用:

from acp import ACPMessage, Intent, ACPValidator

# 创建消息
msg = ACPMessage(
    sender_id="agent_a",
    recipient_id="agent_b",
    intent=Intent.REQUEST,
    content={"query": "What is capital of France?"},
    confidence=0.95,
)

# 发送前验证
validator = ACPValidator()
validator.validate_or_raise(msg)

# 序列化以便传输
data = msg.model_dump()

4、结束语

多代理 AI 系统是未来,但它们需要可靠的通信基础。ACP 提供了这个基础,具有:

  • 结构化消息,消除歧义
  • 置信度校准,建立问责制
  • 信任管理,奖励可靠性
  • 智能路由,优化任务分配
  • 多层验证,及早发现错误

该协议是开源的、可扩展的,并准备好用于生产环境。停止构建多代理意大利面条——开始使用 ACP 构建。


原文链接: Solving Multi-Agent Communication Crisis: Introducing the Agent Communication Protocol (ACP)

汇智网翻译整理,转载请标明出处