A2A Protocol

Agent Gateway Protocol (AGP):实践教程与规范

MILO
Share
Agent Gateway Protocol (AGP): Practical Tutorial and Specification

Agent Gateway Protocol (AGP) 实践教程

本指南将带你了解 Agent Gateway Protocol (AGP):它是什么、解决了哪些问题、核心规范,以及如何运行一个可工作的仿真。为便于检索,全文多处出现关键词 Agent Gateway Protocol。

什么是 Agent Gateway Protocol (AGP)

Agent Gateway Protocol (AGP) 是面向多智能体系统的分层路由层,通过引入面向域的网关与基于策略的路由增强了扁平 A2A 网状结构。受 BGP 启发,Agent Gateway Protocol 基于已声明的能力(Capabilities)、成本与策略,把意图(Intent)路由到合适的小队(Squads),从而实现跨自治小队的可扩展、安全且合规的编排。

  • Agent Gateway Protocol 强调:
    • 来自小队的能力公告(Capability announcements)
    • 基于策略约束的意图路由(Intent routing)
    • 以成本为导向的路径选择
    • 企业规模的分层域组织(如 Finance、Engineering、HR)

Agent Gateway Protocol 解决了什么问题

  • 可扩展性:扁平网随规模扩大变得嘈杂低效。Agent Gateway Protocol 引入分层网关与路由聚合。
  • 策略合规:敏感负载需要策略执行(如 PII 处理、安全等级、地域)。Agent Gateway Protocol 将意图与已公告的策略匹配。
  • 成本优化:多个提供方可满足相同能力;Agent Gateway Protocol 会在满足约束的前提下选择最低成本路径。
  • 鲁棒性:Agent Gateway Protocol 为缺失路由、策略违规、表陈旧等定义了明确的错误码,确保行为可预期。

Agent Gateway Protocol 规范(概览)

能力公告(Capability Announcement)

小队公告其提供的能力,包括版本、成本与策略。

{
  "capability": "financial_analysis:quarterly",
  "version": "1.5",
  "cost": 0.05,
  "policy": {
    "requires_auth": "level_3"
  }
}

字段说明:

  • capability (string):功能/技能键(如 "financial_analysis:quarterly")。
  • version (string):接口/模式版本。
  • cost (number,可选):成本度量(USD、tokens 等)。
  • policy (object):策略属性(如 PII、安全等级、地域等)。

意图负载(Intent Payload)

客户端表达目标与约束;网关寻找满足策略的路由。

{
  "target_capability": "billing:invoice:generate",
  "payload": {
    "customer_id": 123,
    "amount": 99.99
  },
  "policy_constraints": {
    "requires_pii": true
  }
}

字段说明:

  • target_capability (string):要满足的能力。
  • payload (object):输入参数。
  • policy_constraints (object,可选):客户侧要求的策略约束,用于与公告策略匹配。

路由结构(Routing Structures)

  • AGPTable:映射 capabilityRouteEntry 列表。
  • RouteEntry 字段:path(目的地)、cost(用于选择)、policy(用于匹配约束)。

Agent Card 中的代理声明

网关必须通过 A2A 扩展机制声明其角色与支持版本:

{
  "uri": "https://github.com/a2aproject/a2a-samples/tree/main/extensions/agp",
  "params": {
    "agent_role": "gateway",
    "supported_agp_versions": ["1.0"]
  }
}

错误码(JSON-RPC)

  • -32200 AGP_ROUTE_NOT_FOUND — 所请求能力没有已公告路由。
  • -32201 AGP_POLICY_VIOLATION — 找到路由,但无一满足策略约束。
  • -32202 AGP_TABLE_STALE — 路由表已陈旧;需要刷新。

详细规范(在 spec.md 基础上扩展)

CapabilityAnnouncement — 字段与语义

Field Type Required Description
capability string Yes 功能/技能键(如 financial_analysis:quarterly)。
version string Yes 能力的模式/接口版本(如 1.5)。
cost number No 估算成本度量(如 0.05 美元、tokens)。
policy object Yes 策略属性(如 requires_pii:truesecurity_level:5)。

示例:

{
  "capability": "financial_analysis:quarterly",
  "version": "1.5",
  "cost": 0.05,
  "policy": {
    "requires_auth": "level_3"
  }
}

Intent — 字段与语义

Field Type Required Description
target_capability string Yes 意图希望满足的能力。
payload object Yes 任务的输入参数。
policy_constraints object No 客户端侧的策略约束,用于与公告策略匹配。

示例:

{
  "target_capability": "billing:invoice:generate",
  "payload": {
    "customer_id": 123,
    "amount": 99.99
  },
  "policy_constraints": {
    "requires_pii": true
  }
}

路由数据结构(Routing Data Structures)

  • RouteEntry
    • path (string):目标小队/API 路径(如 Squad_Finance/gateway)。
    • cost (number):用于路由选择的成本度量。
    • policy (object):用于约束匹配的目标策略。
  • AGPTable
    • 由网关根据能力公告维护,从 capability 映射到 [RouteEntry]。

代理声明(Agent Card 扩展)

网关必须通过 A2A 扩展机制声明其角色与支持的 Agent Gateway Protocol 版本:

{
  "uri": "https://github.com/a2aproject/a2a-samples/tree/main/extensions/agp",
  "params": {
    "agent_role": "gateway",
    "supported_agp_versions": ["1.0"]
  }
}

错误码(运行语义)

  • -32200 AGP_ROUTE_NOT_FOUNDtarget_capability 无路由。调用方可稍后重试,或通过带外流程触发发现/公告。
  • -32201 AGP_POLICY_VIOLATION:存在路由,但无一满足 policy_constraints。调用方可放宽约束或改用其他能力。
  • -32202 AGP_TABLE_STALE:表已陈旧。网关应先与对等方刷新后再尝试;若失败则返回错误。

路由算法(先策略,后成本)

Agent Gateway Protocol 推荐确定性的选择策略:

  1. 在 AGPTable 中按 target_capability 过滤候选。如果没有,返回 -32200。
  2. 对每个候选路线进行策略充分性检查:
    • 数值阈值:candidate.policy.value >= constraint.value(如 security_level)。
    • 布尔标志:当约束要求时 candidate.policy.flag == true(如 requires_PII)。
    • 任意属性:默认精确匹配,除非定义了领域特定匹配器(如 geo)。
    • 若无候选满足全部约束,返回 -32201。
  3. 在满足策略的候选中,选择最小 cost
  4. 若 AGPTable 已知陈旧,则先刷新;刷新失败返回 -32202。

并列裁决(若存在多个最低成本路由):

  • 优先选择最近的公告,然后按 path 字典序(稳定排序),除非领域策略另有定义。

表维护与新鲜度(Freshness)

  • 入站公告:网关从小队/对等网关摄取能力公告。
  • 过期/刷新:公告应包含 TTL/版本(带外);过期时移除路由,-32202 风险升高。
  • 局部陈旧:网关可仅标记某些能力为陈旧,而非整表,以降低影响面。

安全与合规注意事项

  • 最小特权路由:在意图中仅表达必要的最少 policy_constraints;交由 Agent Gateway Protocol 执行。
  • PII/数据驻留:使用 requires_PIIgeo 等显式字段,限制路由域。
  • 可审计性:记录意图路由决策、被拒候选与返回的错误码。

版本管理与向后兼容

  • 能力携带 version 字段以稳定小队与客户端间的契约。
  • 网关应同时支持多个版本,并在未被策略覆盖时偏好最新匹配版本。

扩展流程(时序图)

sequenceDiagram
  autonumber
  participant Client as 客户端
  participant Gateway as 网关
  participant SquadA as Squad A (Eng)
  participant SquadB as Squad B (Vendor)

  Client->>Gateway: Intent(target_capability, payload, policy_constraints)
  Gateway->>Gateway: 查找 AGPTable[target_capability]
  alt 无路由
    Gateway-->>Client: 错误 -32200 (AGP_ROUTE_NOT_FOUND)
  else 存在候选路由
    Gateway->>Gateway: 过滤策略充分性
    alt 无候选满足
      Gateway-->>Client: 错误 -32201 (AGP_POLICY_VIOLATION)
    else 有候选满足
      Gateway->>Gateway: 选择最低成本(含并列裁决)
      Gateway->>SquadB: 转发请求(若被选中)
      SquadB-->>Gateway: 响应 / 结果
      Gateway-->>Client: 成功
    end
  end

运维指南(Operational Guidance)

  • 监控:跟踪 -32200/-32201/-32202 的比率,用于识别拓扑缺口、策略漂移与陈旧。
  • SLO:定义路由成功延迟与新鲜度预算;为表刷新路径埋点。
  • 事件响应:当系统性出现 -32202 时,优先刷新/扩散;尽量优雅降级至最安全的内部路由。

常见问题(FAQ)

  • 问:Agent Gateway Protocol 是服务发现的替代品吗?
    • 答:不是。Agent Gateway Protocol 是一个基于策略的意图路由器;它可以消费发现数据,但关注点在能力/策略语义。
  • 问:如何新增一个域小队?
    • 答:部署小队网关,发布能力公告,确保上游网关将其摄入 AGPTable。
  • 问:能否优先考虑安全而非成本?
    • 答:可以。在意图中编码更高的 security_level 约束;Agent Gateway Protocol 只会考虑满足策略的路由,即便更昂贵。

架构与流程(Architecture and Flow)

flowchart TD
  A[小队宣告能力<br/>能力、版本、成本、策略] -->|构建| B[AGP 路由表]
  C[客户端意图<br/>目标能力、负载、<br/>策略约束] --> D[网关]
  B --> D
  D --> E{策略匹配?}
  E -- 否 --> F[错误 -32201<br/>AGP_策略违规]
  E -- 是 --> G{能力已知?}
  G -- 否 --> H[错误 -32200<br/>AGP_路由未找到]
  G -- 是 --> I{选择最低成本<br/>合规路由}
  I --> J[转发到小队路径]
  D --> K{路由表过期?}
  K -- 是 --> L[刷新路由表或返回 -32202]
  L --> D

代码示例与说明

以下仿真展示了 Agent Gateway Protocol 如何执行基于策略、以成本为导向的路由。它会初始化 AGP 表,为三个小队发布能力,然后用不同约束路由四个意图。

来源:extensions/agp/agp_run.py

初始化路由表与网关

from agp_protocol import AGPTable, AgentGatewayProtocol

corporate_agp_table = AGPTable()
corporate_gateway = AgentGatewayProtocol(
    squad_name='Corporate_GW', agp_table=corporate_agp_table
)

发布能力公告

# 工程小队(安全高,成本更高)
eng_announcement = CapabilityAnnouncement(
    capability='infra:provision:vm',
    version='1.0',
    cost=0.10,
    policy={'security_level': 5, 'requires_PII': True},
)
corporate_gateway.announce_capability(
    eng_announcement, path='Squad_Engineering/vm_provisioner'
)

# 外部供应商(最便宜,安全较低)
vendor_announcement = CapabilityAnnouncement(
    capability='infra:provision:vm',
    version='1.1',
    cost=0.05,
    policy={'security_level': 3, 'requires_PII': False},
)
corporate_gateway.announce_capability(
    vendor_announcement, path='External_Vendor/vm_provisioning_api'
)

# 财务小队(标准分析)
finance_announcement = CapabilityAnnouncement(
    capability='financial_analysis:quarterly',
    version='2.0',
    cost=0.15,
    policy={'security_level': 3, 'geo': 'US'},
)
corporate_gateway.announce_capability(
    finance_announcement, path='Squad_Finance/analysis_tool'
)

使用不同约束路由意图

# 意图 A —— 成本驱动,最小策略
intent_a = IntentPayload(
    target_capability='infra:provision:vm',
    payload={'type': 'standard', 'user': 'bob'},
    policy_constraints={'security_level': 3},
)
corporate_gateway.route_intent(intent_a)

# 意图 B —— 敏感场景,需要 PII 与安全等级 5
intent_b = IntentPayload(
    target_capability='infra:provision:vm',
    payload={'type': 'sensitive', 'user': 'alice', 'data': 'ssn_data'},
    policy_constraints={'security_level': 5, 'requires_PII': True},
)
corporate_gateway.route_intent(intent_b)

# 意图 C —— 策略不匹配(安全等级 7)
intent_c = IntentPayload(
    target_capability='infra:provision:vm',
    payload={'type': 'max_security'},
    policy_constraints={'security_level': 7},
)
corporate_gateway.route_intent(intent_c)

# 意图 D —— 未知能力
intent_d = IntentPayload(
    target_capability='hr:onboard:new_hire',
    payload={'employee': 'Charlie'},
    policy_constraints={},
)
corporate_gateway.route_intent(intent_d)

预期结果:

  • 意图 A 路由至外部供应商(最低成本,策略满足)。
  • 意图 B 路由至工程小队(要求 PII,供应商被拒)。
  • 意图 C 因策略不满足失败(-32201)。
  • 意图 D 因无路由失败(-32200)。

如何运行仿真

前置条件:Python 3.9+。

如在 extensions/agp 使用 Poetry 安装依赖:

cd extensions/agp
poetry install

或确保环境中已提供 pydantic(参考 pyproject.toml)。

运行仿真:

python extensions/agp/agp_run.py

Agent Gateway Protocol 最佳实践

  • 将能力公告视为控制平面;保持其新鲜度与版本化。
  • 在公告中编码“最小可行”的策略;更严格的约束由客户端意图表达并由网关执行。
  • 偏好为每个能力版本定义显式意图模式,以稳定契约。
  • 监控错误码比率(-32200/-32201/-32202)以识别拓扑缺口、策略空白或陈旧。

关键要点

  • Agent Gateway Protocol 使多智能体路由成为一个以策略驱动、成本感知、分层的系统。
  • 能力公告与意图约束是一等公民。
  • 明确的错误语义为大规模运维提供可操作性清晰度。

A2A 扩展列表