Multi-Agent 与 A2A 协作：从角色分工到协议化运行

1
query_order()      是工具/服务
2
search_policy()    是检索模块
3
verify_amount()    是规则函数
4
compose_reply()    可以是一次模型调用

1
Context Isolation
2
  不同专家只看必要数据
3

4
Capability Isolation
5
  不同身份只持有最小工具权限
6

7
Parallelism
8
  独立子任务并发执行
9

10
Independent Review
11
  生成与验证分离
12

13
Organizational Boundary
14
  跨团队/供应商/系统协作
15

16
Failure Isolation
17
  某个专家失败不污染整个 Runtime

1
只是想让代码更模块化
2
任务步骤固定且简单
3
所有角色看到相同上下文、使用相同模型/工具
4
没有可测的专业分工
5
延迟和成本很敏感
6
单 Agent 已稳定完成
7
多角色只是在重复投票

1
Baseline：当前单 Agent 指标
2
Bottleneck：它具体失败在哪里
3
Hypothesis：哪个边界拆分会改善什么指标
4
Cost：新增消息、token、延迟、故障面
5
Experiment：怎样通过消融证明贡献
6
Rollback：失败时如何回到单 Agent

1
Supervisor
2
  -> Worker A
3
  -> Worker B
4
  -> Worker C
5
  -> aggregate/verify

1
Planner Agent -> Plan
2
Executor Agent/Runtime -> Observations
3
Planner/Repairer -> Updated Plan

1
Researcher -> Evidence Bundle
2
Writer -> Draft with Claims
3
Reviewer -> Claim/Evidence Review

1
Triage Agent --handoff--> Refund Specialist

1
Manager Agent --call--> Specialist Agent --result--> Manager

1
Producer -> Artifact
2
Verifier -> pass/fail/unknown
3
Repairer -> revised artifact

1
Agents <-> Shared Blackboard

1
from typing import Literal
2
from pydantic import BaseModel
3

4

5
class AgentContract(BaseModel):
6
    agent_id: str
7
    version: str
8
    role: str
9
    accepted_task_types: list[str]
10
    input_schema_ref: str
11
    output_schema_ref: str
12
    allowed_capabilities: list[str]
13
    allowed_data_classes: list[str]
14
    max_side_effect: Literal["none", "read", "reversible_write", "irreversible"]
15
    default_budget: dict
16
    terminal_conditions: list[str]
17
    escalation_routes: list[str]
18
    owner: str

1
你是一个聪明的业务专家，请努力解决问题。

1
Policy Agent
2
Input: verified facts + policy query requirements
3
Output: evidence bundle + missing/conflicting requirements
4
Tools: published policy search/read only
5
Cannot: decide user identity, execute refund, write memory
6
Success: all required claims have versioned citations or explicit insufficiency

1
Context Agent：抽取/整理，不查询敏感业务
2
Business Service：返回权威事实，不做语言决策
3
Policy Agent：检索证据，不执行动作
4
Decision Agent：提出建议，不授权
5
Policy Engine：确定性裁决
6
Execution Service：执行已授权命令
7
Verification Service：查询最终状态

1
class AgentTask(BaseModel):
2
    task_id: str
3
    context_id: str
4
    parent_task_id: str | None
5
    requester_agent_id: str
6
    assignee_agent_id: str
7
    task_type: str
8
    goal: str
9
    input_refs: list[str]
10
    constraints: list[str]
11
    expected_artifact_schema: str
12
    success_conditions: list[str]
13
    deadline_at: str
14
    budget: dict
15
    authz_context_ref: str
16
    idempotency_key: str
17
    trace_id: str

1
created
2
-> assigned
3
-> accepted
4
-> working
5
-> input_required
6
-> completed
7
-> failed
8
-> cancelled
9
-> expired

1
Task：有生命周期、目标和最终产物
2
Message：Agent/User 之间的一次交流
3
Artifact：Task 生成的结果对象
4
Event：状态或增量变化通知

1
class AgentMessage(BaseModel):
2
    message_id: str
3
    task_id: str
4
    context_id: str
5
    sender: str
6
    recipient: str
7
    role: Literal["requester", "assignee"]
8
    parts: list[dict]
9
    in_reply_to: str | None
10
    created_at: str
11
    data_classification: str

1
Text Part：简短解释/问题
2
Data Part：结构化 JSON
3
File Part：文件或受控 URI
4
Reference Part：事实、证据、Trace、Artifact 引用

1
class AgentArtifact(BaseModel):
2
    artifact_id: str
3
    task_id: str
4
    name: str
5
    schema_ref: str
6
    parts: list[dict]
7
    source_refs: list[str]
8
    producer_agent: str
9
    producer_version: str
10
    content_hash: str
11
    created_at: str
12
    status: Literal["draft", "verified", "rejected"]

1
不要传完整对话，如果只需 order_id + intent
2
不要传原始身份证明，如果只需 verified_subject_ref
3
不要传工具 Secret
4
不要把 Reviewer 的内部安全策略暴露给外部 Agent

1
Task/Message/Event 记录协作事实
2
Shared State 保存规范化当前视图
3
Artifact Store 保存不可变产物
4
Business System 保存最终业务真相

1
class BlackboardEntry(BaseModel):
2
    entry_id: str
3
    task_id: str
4
    kind: Literal["fact", "hypothesis", "decision", "question", "artifact_ref"]
5
    value: dict
6
    source_refs: list[str]
7
    author_agent: str
8
    confidence: float | None
9
    status: Literal["proposed", "verified", "invalidated"]
10
    version: int

1
Context Task
2
    ↓
3
Business Facts ─┐
4
                ├-> Decision -> Execute -> Verify
5
Policy Evidence ┘

1
class TaskLease(BaseModel):
2
    task_id: str
3
    worker_agent_id: str
4
    lease_token: str
5
    acquired_at: str
6
    expires_at: str
7
    attempt: int

1
Policy Query 可能依赖 product_type/order_time

1
max_active_subtasks
2
max_depth
3
max_total_agents
4
max_messages
5
max_tokens/cost
6
deadline

1
等待必需分支
2
对可选分支设 deadline
3
取消重复/低价值分支
4
返回 partial artifact + missing parts

1
Triage -> Specialist

1
handoff reason
2
context package
3
allowed capabilities
4
return/escalation condition
5
user-visible disclosure

1
Coordinator -> Policy Agent: produce Evidence Bundle
2
Policy Agent -> Coordinator: Artifact

1
class DelegationResult(BaseModel):
2
    task_id: str
3
    status: Literal["completed", "failed", "input_required", "partial"]
4
    artifact_refs: list[str]
5
    missing_requirements: list[str]
6
    warnings: list[str]
7
    usage: dict
8
    trace_ref: str

1
目标
2
已验证事实
3
相关消息摘要/引用
4
策略与权限
5
预期输出
6
剩余预算

1
A2A Client Agent
2
  -> discover Agent Card
3
  -> authenticate
4
  -> send message/task
5
  -> stream/poll/push status
6
  -> receive artifact

1
{
2
  "name": "Order Policy Agent",
3
  "description": "Returns versioned policy evidence for order fulfillment tasks",
4
  "url": "${POLICY_AGENT_A2A_URL}",
5
  "version": "3.4.0",
6
  "capabilities": {
7
    "streaming": true,
8
    "pushNotifications": false
9
  },
10
  "defaultInputModes": ["application/json"],
11
  "defaultOutputModes": ["application/json"],
12
  "skills": [
13
    {
14
      "id": "refund-policy-evidence",
15
      "name": "Refund policy evidence",
16
      "description": "Produces evidence bundles; does not authorize refunds"
17
    }
18
  ]
19
}

1
验证发布者/endpoint
2
认证 Agent
3
检查本地 allowlist
4
最小授权 scope
5
每 Task 资源级 Policy

1
来源 DNS/TLS
2
组织目录/签名
3
Card version/hash
4
Owner/review/expiry
5
安全方案
6
能力变化 diff

1
message/stream
2
-> task status updates
3
-> artifact updates
4
-> terminal state

1
验证 callback URL
2
防 SSRF
3
回调认证/签名
4
重放保护
5
事件幂等
6
不在 URL/query 泄露 Secret

1
Human/User Principal
2
Calling Agent Identity
3
Remote Agent Identity
4
Delegated Scope
5
Target Resource

1
当前用户能否调用该 Skill
2
能否发送这些数据
3
能否让远端执行该动作
4
返回 Artifact 能否进入当前 Context/Memory

1
audience=remote-agent
2
subject/calling-agent
3
delegated scopes
4
tenant/resource constraints
5
short expiry
6
task/context binding

1
Schema 校验
2
size/type 限制
3
malware/content scan
4
Prompt Injection 隔离
5
source/provenance 验证
6
不得直接触发本地高风险工具

1
远端区域
2
数据分类
3
保留/训练政策
4
日志和子处理者
5
删除能力
6
合同与合规

1
事实冲突
2
建议冲突
3
资源竞争
4
计划冲突
5
版本冲突

1
A -> B -> A 委派环
2
同一 task fingerprint 重复
3
相同 Message 内容反复
4
Artifact 无新增信息

1
{
2
  "delegation_path": ["coordinator", "policy-agent", "legal-agent"],
3
  "max_depth": 3,
4
  "visited_task_fingerprints": ["sha256:..."]
5
}

1
必需 Agent 失败 -> task fail/input_required/human
2
可选 Agent 失败 -> partial artifact + warning
3
Verifier 失败 -> unknown，不自动 pass

1
停止新委派
2
取消尚未执行子 Task
3
向远端发取消请求
4
记录不可取消的外部动作
5
继续对账必要结果
6
汇总部分完成状态

1
Coordinator
2
  固定 StateGraph + 局部 LLM 路由
3

4
Context Agent
5
  意图、槽位、最小上下文
6

7
Business Facts Service（非 LLM）
8
  订单、物流、工单权威查询
9

10
Policy Agent
11
  RuleRAG + Evidence Bundle
12

13
Decision Agent
14
  基于事实/证据提出结构化建议
15

16
Policy Engine（非 LLM）
17
  权限、金额、条件与确认裁决
18

19
Execution Service（非 LLM）
20
  幂等退款/补偿/工单
21

22
Verification Service（规则 + 查询）
23
  最终业务状态与安全校验

1
class OrderFlowSharedState(BaseModel):
2
    run_id: str
3
    task_id: str
4
    intent_artifact_ref: str | None
5
    fact_bundle_ref: str | None
6
    evidence_bundle_ref: str | None
7
    decision_proposal_ref: str | None
8
    policy_decision_ref: str | None
9
    execution_ref: str | None
10
    verification_ref: str | None
11
    state_version: int

1
Coordinator
2
-> Context Agent: normalized task artifact
3
-> Business Facts + Policy Agent (parallel after minimum facts)
4
-> merge fact/evidence artifacts
5
-> Decision Agent: proposal
6
-> Policy Engine: allow/deny/confirm/human
7
-> Execution Service
8
-> Verification Service
9
-> Coordinator: verified response

1
OrderFlow Coordinator (A2A Client)
2
-> discover approved Policy Agent Card
3
-> send Evidence Task with verified facts refs
4
-> stream Task status
5
-> receive signed Evidence Bundle Artifact
6
-> local schema/source/policy validation

1
class Coordinator:
2
    async def advance(self, run_id: str) -> None:
3
        state = await self.state_store.load_for_update(run_id)
4
        enforce_team_budget(state)
5

6
        ready = self.task_graph.ready_tasks(state)
7
        for task in ready[: state.remaining_parallel_slots]:
8
            contract = self.agent_registry.resolve(task.assignee, task.task_type)
9
            enforce_delegation_policy(state, task, contract)
10
            lease = await self.leases.reserve(task.id, contract.agent_id)
11
            await self.dispatcher.dispatch(
12
                task=build_minimal_task_envelope(task, state),
13
                lease=lease,
14
            )
15

16
        events = await self.inbox.read_new(run_id, after=state.inbox_watermark)
17
        for event in deduplicate_and_order(events):
18
            state = apply_validated_agent_event(state, event)
19

20
        conflicts = detect_conflicts(state.blackboard)
21
        if conflicts:
22
            state = route_conflicts_to_verifier(state, conflicts)
23

24
        await self.state_store.compare_and_swap(state)

1
调度依据 Task Graph，不依赖自由聊天
2
每次委派前重新做 Policy
3
任务包按最小数据构建
4
结果通过 Inbox 去重/排序/Schema 校验
5
事实冲突进入 Verifier
6
共享状态用 CAS 提交

1
验证 task/attempt/lease token
2
-> 若新 Attempt 已完成，不覆盖当前结果
3
-> 保存为 late artifact 供诊断
4
-> 如内容仍有价值，由 Verifier 显式合并

1
Single strong Agent
2
Single Agent + deterministic tools
3
Workflow + specialist calls
4
Full Multi-Agent

1
Task Allocation Accuracy
2
Handoff Precision/Recall
3
Subtask Success Rate
4
Artifact Schema Validity
5
Message Efficiency
6
Duplicate Work Rate
7
Conflict Resolution Accuracy
8
Loop/Deadlock Rate

1
Verified Task Success
2
Policy Compliance
3
Unsafe Action Rate
4
Latency
5
Cost per Verified Success
6
Human Escalation

1
移除 Agent
2
替换为 deterministic node
3
替换为 oracle artifact
4
打乱/隐藏它的输出
5
限制其工具

1
Δ success
2
Δ cost
3
Δ latency
4
Δ safety

1
完整 history
2
结构化 artifact only
3
minimal facts + references

1
Agent Discovery Success
2
Card Trust/Version Failure
3
Task Completion/Expiry
4
Stream Reconnect Success
5
Push Delivery Duplicate Rate
6
Artifact Assembly Accuracy
7
Cross-agent Trace Completeness
8
Delegated Scope Violation

1
root agent.run span
2
  -> coordinator.allocate
3
     -> a2a.client.send (task_id, remote_agent_id)
4
        [trace propagation across boundary]
5
        -> remote agent.task
6
           -> remote tool/retrieval spans
7
     -> a2a.client.artifact.receive
8
  -> artifact.verify
9
  -> decision.merge

1
traceparent/correlation id（按信任策略）
2
task/message/artifact ids
3
Agent Card version/hash
4
remote agent version
5
latency/usage summary
6
terminal error code
7
artifact content hash

1
Useful Work
2
  直接产生被采用 Artifact 的模型/工具成本
3

4
Coordination
5
  路由、消息、状态同步、聚合
6

7
Redundancy
8
  重复研究/重复调用
9

10
Verification
11
  冲突与产物校验
12

13
Recovery
14
  重试、迟到结果、重新委派

1
max remote agents per run
2
max delegation depth
3
max cross-boundary bytes
4
max coordination token ratio
5
max duplicate task rate
6
kill switch per agent/version
7
fallback to local workflow

1
协议版本协商
2
Agent Card 必需/未知字段
3
输入输出 MIME mode
4
Task state 映射
5
Message Part 类型
6
Artifact 增量组装
7
stream reconnect/resubscribe
8
push event duplicate/out-of-order
9
错误对象与重试语义

1
agent-card-vN.json
2
task-working.json
3
task-input-required.json
4
artifact-update-1.json
5
artifact-update-final.json
6
error-auth-expired.json

1
Task/Message/Artifact Schema
2
版本兼容
3
未知 Part/MIME
4
大型 Artifact
5
重复/乱序事件

1
Supervisor 错误路由
2
两个 Worker 抢同一 lease
3
Worker 超时后旧结果到达
4
并行事实版本冲突
5
Agent A/B 委派循环

1
Agent Card 变更/伪造
2
认证过期
3
Task input_required
4
stream 断线与 resubscribe
5
push 重放/SSRF
6
取消与完成竞态

1
远端 Artifact Prompt Injection
2
过宽 delegation token
3
跨租户 reference
4
Agent Card Skill 描述投毒
5
远端请求额外敏感数据

1
Coordinator 崩溃
2
Task 已完成但 Artifact 未登记
3
Artifact 已登记但 Shared State CAS 失败
4
Verifier 不可用
5
部分分支成功