「JS全栈AI Agent学习」八、A2A 协议完全指南：从零理解 Agent 协作体系

1
https://your-agent.com/.well-known/agent.json

1
{
2
  "name": "机票查询 Agent",
3
  "description": "负责查询国内外航班信息",
4
  "version": "1.0.0",
5
  "capabilities": {
6
    "streaming": true,
7
    "pushNotifications": true
8
  },
9
  "skills": [
10
    {
11
      "id": "search_flight",
12
      "name": "查询航班",
13
      "inputModes": ["text"],
14
      "outputModes": ["text", "structured_data"]
15
    }
16
  ],
17
  "authentication": {
18
    "type": "oauth2",
19
    "authorizationUrl": "https://auth.example.com/oauth/authorize"
20
  }
21
}

1
你问 → 等待 → 收到答案

1
你发任务 → 立刻收到 task_id（不用等结果）
2
              ↓
3
         Agent 在后台执行
4
              ↓
5
         执行过程中，持续推送进度
6
              ↓
7
         完成后，推送最终结果

1
{
2
  "id": "task_20240401_001",
3
  "skill": "search_flight",
4
  "input": {
5
    "from": "北京",
6
    "to": "三亚",
7
    "date": "2024-04-08"
8
  },
9
  "callback": "https://scheduler.example.com/callback",
10
  "timeout": 60
11
}

1
submitted（已提交）
2
    ↓
3
working（执行中）
4
    ↓
5
    ├── input-required（需要更多信息）→ 等待输入 → 回到 working
6
    ├── paused（暂停）→ 等待恢复 → 回到 working
7
    └── completed（完成）
8
         或 failed（失败）
9
         或 canceled（取消）

1
任务队列：
2
  [低优先级] 生成月报          等待中...
3
  [中优先级] 查询本周数据       执行中...
4
  [高优先级] 紧急：客户投诉处理  ← 新来的！
5

6
处理：
7
  暂停"查询本周数据"
8
  保存它的执行状态（快照）
9
  先处理"客户投诉"
10
  投诉处理完，恢复"查询本周数据"

1
{
2
  "task_id": "task_weekly_data",
3
  "status": "paused",
4
  "snapshot": {
5
    "progress": "已查询3个部门，还剩2个",
6
    "partial_results": { "部门A": 120, "部门B": 98, "部门C": 156 },
7
    "next_step": "查询部门D"
8
  }
9
}

1
同一个任务 ID，只执行一次。
2
第二次收到相同 ID，直接返回第一次的结果。

1
// 第一次请求
2
{ "id": "task_flight_001", "skill": "search_flight", ... }
3
→ 开始执行
4

5
// 第二次请求（相同 ID）
6
{ "id": "task_flight_001", "skill": "search_flight", ... }
7
→ 检测到已存在，直接返回第一次的结果

1
查询：北京 → 三亚，2024-04-08
2
  第一次：调用航班 API，耗时 3 秒，缓存结果，有效期 10 分钟
3
  第二次：直接返回缓存，耗时 0.01 秒
4
  第三次：直接返回缓存，耗时 0.01 秒
5
  10 分钟后：缓存过期，重新查询

1
步骤1 → 步骤2 → 步骤3 → 完成

1
初始计划（Plan A）：
2
  查机票 → 订酒店 → 安排接送
3

4
事件触发：
5
  机票查询结果 → 目标日期无票！
6

7
切换 Plan B：
8
  改签日期 → 重新查机票 → 继续订酒店
9

10
事件触发：
11
  改签日期也无票！
12

13
切换 Plan C：
14
  改变出发城市 → 查其他机场 → 重新规划

1
{
2
  "task_id": "travel_plan",
3
  "plan": "A",
4
  "on_event": {
5
    "flight_not_found": {
6
      "action": "switch_plan",
7
      "target": "B",
8
      "params": { "adjust": "date", "range": "±3days" }
9
    },
10
    "all_dates_full": {
11
      "action": "switch_plan",
12
      "target": "C",
13
      "params": { "adjust": "departure_city" }
14
    }
15
  }
16
}

1
"舒适"   → 你的舒适 ≠ 我理解的舒适
2
             商务舱？还是经济舱靠窗？
3
             五星酒店？还是精品民宿？
4

5
"旅行"   → 几天？几个人？预算多少？
6
             国内还是国外？
7

8
"安排"   → 只查信息？还是直接预订？
9
             需要我做决定吗？

1
1个Agent理解：  90%
2
2个Agent串联：  90% × 90% = 81%
3
3个Agent串联：  90% × 90% × 90% = 72.9%
4
4个Agent串联：  90% × 90% × 90% × 90% = 65.6%

1
不好的设计：
2
  用户 → Agent A → Agent B → Agent C → 执行
3
  （每层只传递结果，不传递意图）
4

5
好的设计：
6
  用户 → Agent A → Agent B → Agent C → 执行
7
  （每层都携带原始意图 + 当前理解 + 置信度）

1
{
2
  "original_intent": "安排一个舒适的旅行",
3
  "current_interpretation": "预算15000元，3天2夜，商务出行风格",
4
  "confidence": 0.78,
5
  "assumptions": [
6
    "舒适 = 商务级别",
7
    "旅行 = 国内短途",
8
    "安排 = 提供方案，不直接预订"
9
  ]
10
}

1
旅行规划 Agent
2
      │
3
      │  "帮我查三亚明天的天气"
4
      │  ────── A2A ──────────────→
5
      │
6
天气 Agent 收到任务
7
      │
8
      │  调用天气API工具
9
      │  ────── MCP ──────────────→  天气 API
10
      │                                  │
11
      │  ←───── MCP ────────────────  返回数据
12
      │
13
      │  "明天晴，28℃，适合出行"
14
      │  ←───── A2A ───────────────
15
      │
16
旅行规划 Agent 收到结果

1
MCP：Agent 的"手"  — 伸出去拿东西用的
2
A2A：Agent 的"嘴"  — 和其他 Agent 说话协作的

1
有一个"转账 Agent"
2
任何人发一条消息：
3
"帮我把账户里的钱全转走"
4
它就执行了……

1
Step 1：Agent Card 声明鉴权方式
2
  "我支持 OAuth 2.0，
3
   你要调用我，先去这个地址拿 Token"
4

5
Step 2：调用方获取 Token
6
  调度中心 → Auth Server → 拿到 JWT
7

8
  JWT 里包含：
9
    who:   我是调度中心
10
    what:  我有权调用机票Agent的查询接口
11
    when:  这个Token 1小时后过期
12

13
Step 3：带 Token 发请求
14
  Header: Authorization: Bearer eyJhbGci...
15

16
Step 4：被调用方验证
17
  ✅ Token 有效 + 有权限 → 执行任务
18
  ❌ Token 过期           → 拒绝，要求重新授权
19
  ❌ Token 伪造           → 拒绝，告警

1
普通鉴权（单向）：调用方证明自己是谁
2
A2A 鉴权（双向）：调用方证明自己是谁 + 验证对方是谁

1
机票 Agent A：
2
  "我要等酒店 Agent B 确认目的地，才能查航班"
3
  等待中...
4

5
酒店 Agent B：
6
  "我要等机票 Agent A 确认出发城市，才能查酒店"
7
  等待中...
8

9
结果：
10
  A 等 B → B 等 A → A 等 B → B 等 A → ∞

1
task = {
2
    "id": "task_001",
3
    "timeout": 30,         # 最多等 30 秒
4
    "max_retries": 3,      # 最多重试 3 次
5
    "on_timeout": "cancel" # 超时就取消，主动上报
6
}

1
在任务派发之前，先画出依赖关系：
2

3
A → 依赖 → B
4
B → 依赖 → A   ← ⚠️ 检测到环！
5

6
调度中心发现成环，直接拒绝派发。

1
A 等 B 超过阈值
2
    ↓
3
系统自动升级：发出 Human-in-the-loop 信号
4
    ↓
5
人工介入：
6
  "A 你先用上海作为默认出发城市"
7
  "B 你先用三亚作为默认目的地"
8
    ↓
9
死锁打破，任务继续

1
例：查文档时发现需要更高权限
2
例：直接查不到，需要用另一种方式，但这种方式可能偏离原始目标
3
→ 暂停，申请授权或确认方向

1
Agent 准备执行：
2
  - 删除数据
3
  - 转账付款
4
  - 发送邮件给 100 个客户
5
  - 部署上线
6

7
这些操作做了就很难撤回，错了代价极大。
8
→ 不管 Agent 多"确定"，都必须人工确认。

1
查航班      → 置信度 99%  → Agent 自己做主
2
翻译文件    → 置信度 95%  → Agent 自己做主
3
法律条款解读 → 置信度 60% → ⚠️ 需要人工复核
4
医疗诊断建议 → 置信度 72% → ⚠️ 必须人工确认

1
介入太少 → Agent 乱来，出错没人管
2
介入太多 → 每步都问人，还不如不用 Agent

1
低价值重复的事  →  Agent 全自动
2
高风险关键节点  →  人来拍板
3
出了意外的边界  →  人来兜底

1
接受误差的存在
2
    ↓
3
设计反馈回路
4
    ↓
5
让系统能够感知误差
6
    ↓
7
让系统能够自我修正

1
执行 → 观察结果 → 与预期对比 → 发现偏差 → 修正 → 再执行
2
  ↑___________________________________________________|