9.1 多智能体系统的四大上下文挑战

在单Agent场景下，上下文工程关注的问题相对简单：如何把正确的信息以正确的密度放进同一个上下文窗口。多Agent场景则一下子引入了四个全新的、棘手的挑战——状态共享、污染防范、可追溯性、一致性。这四个问题要是解决不好，多Agent系统就很容易陷入“一群聪明人共同做出愚蠢决策”的尴尬境地。

9.1.1 挑战一：状态共享

核心问题在于：当多个Agent协同工作时，哪些状态应该共享，哪些又必须被隔离？如果什么都不共享，每个Agent都是信息孤岛，根本无法协同；可如果什么都共享，上下文窗口很快就会塞满无关信息，同时还伴随着状态污染的风险。

更微妙的问题是上下文爆炸。在一个多Agent系统中，每个Agent在运行循环中都会持续生成大量数据（工具调用、观测结果、推理步骤）。如果所有这些数据都被所有Agent共享，整个系统的上下文总大小会随时间呈指数级增长。别忘了，LLM的注意力预算是有限的，本来就存在“lost-in-the-middle”现象，上下文爆炸会迅速把系统性能拖到谷底。

从量化角度看：假设一个Agent平均每轮生成2000 token，10个Agent协作50轮，就会产生100万token。即便模型支持百万token窗口，ChromaDB 2025年的研究已经证明，准确率会因此下降30%以上（参见第7章7.1节Context Rot）。

9.1.2 挑战二：污染防范

核心问题同样严峻：一个Agent产生的错误记忆或错误推理，会不会传播到其他Agent的上下文，进而污染整个系统？在单Agent场景中，上下文污染只影响自己；但在多Agent场景中，错误可以像传染病一样扩散——Agent A错了，Agent B基于A的错误继续推理，结果就是错上加错。

典型表现包括：

• 会话内污染：临时上下文被永久化存储到共享记忆空间，导致数据库中堆积大量无用记录
• 跨会话污染：用户的偏好随时间变化，旧版本的偏好污染了新版本的认知
• 群体污染：一个Agent的幻觉被其他Agent反复引用，最终整个系统都相信了这个幻觉

9.1.3 挑战三：可追溯性

核心问题很直接：当多Agent系统产生了一个错误结果，我们能否回答：“哪一步出了问题？哪个Agent做出了错误决策？决策依赖于哪些前面的上下文？”

在传统的紧耦合架构中，Session、Harness、Sandbox都耦合在同一个容器里，失败后根本无法定位是哪个组件出了错——你只有一个完整的WebSocket事件流，得自己从中找问题。可追溯性要求每个事件都有迹可循，每个决策都有源可溯。

9.1.4 挑战四：一致性

核心问题也很现实：当多个Agent对同一个状态做出不同的修改时，如何保持状态的一致性？

典型场景有：

• Summary Drift（摘要漂移）：连续多次总结同一个上下文，信息会逐步偏离原始内容，就像“传话游戏”——每次总结都损失一点信息，最后总结出来的东西和原始内容完全不一样了。
• 记忆冲突：同一个用户，不同Agent从不同时间点读取了偏好，Agent A认为用户喜欢Python，Agent B认为用户喜欢Go——哪个才是对的？
• 并发修改：两个Agent同时修改同一个文件，哪个版本应该保留？

9.1.5 已成熟的解决方案矩阵

9.2 2026年主流多智能体上下文架构

到2026年，工业界已经形成了两个被广泛验证的主流架构方案：Google的ADK三层上下文栈和Anthropic的Managed Agents三层隔离架构。两者在设计哲学上有所不同，但解决的是同一个核心问题——通过架构分层来分离不同生命周期的上下文，从根源上降低污染和爆炸的风险。

9.2.1 Google ADK三层上下文栈：存储 → 处理器管道 → 编译后工作上下文

Google ADK（Agent Development Kit）是Google在2025年开源的全生命周期Agent开发框架，它在架构层面强制分离了会话内临时上下文和跨会话长期知识——这是它最核心的设计贡献。

ADK三层架构：

┌─────────────────────────────────────────────────────────┐
│                        Agent 层                         │
│                  定义智能体类型与行为逻辑                 │
├─────────────────────────────────────────────────────────┤
│                       Runtime 层                        │
│             管理 Session 生命周期和 Event 流             │
│核心：Event 驱动模型，每次操作封装为 Event               │
│追加到事件流，可任意时刻回放决策过程                     │
├─────────────────────────────────────────────────────────┤
│                       Service 层                        │
│           SessionService          MemoryService         │
│       ┌──────────────┐       ┌──────────────┐          │
│       │ 会话内上下文  │       │ 跨会话长期记忆 │          │
│       │内存/数据库/云托管│    │ 向量检索+混合查询│         │
│       └──────────────┘       └──────────────┘          │
└─────────────────────────────────────────────────────────┘

State Prefix机制：细化作用域

ADK通过前缀命名空间来明确状态的作用域，这是一个非常简单但极其有效的设计：

# ADK state prefix 机制示例
# user: 前缀 — 用户所有 Session 间共享
state["user:language_preference"] = "Python"
# app: 前缀 — 应用级别共享
state["app:api_endpoint"] = "https://api.example.com"
# temp: 前缀 — 仅在当前 Session 有效
state["temp:current_query"] = "Milvus 连接超时怎么办?"

这种前缀机制天然实现了“哪些共享、哪些隔离”的决策：

• 共享：user:和app:前缀的状态会被所有会话看到
• 隔离：temp:前缀的状态只在当前会话有效，不会污染其他会话

ADK与Milvus集成的长期记忆示例：

# Google ADK + Milvus 长期记忆实现
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from pymilvus import connections, Collection, DataType
import google.generativeai as genai

# 1. SessionService — 管理会话内临时上下文
#每个会话独立，天然隔离
session_service = InMemorySessionService()

# 2. MemoryService — 通过 Milvus 实现跨会话长期记忆
#Schema: user_id | session_id | question | solution | embedding | timestamp
connections.connect(host="localhost", port="19530")

def recall_memory(query: str, user_id: str, top_k: int = 3) -> str:
    """从记忆库检索相关历史案例"""
    query_embedding = genai.embed_content(model="models/text-embedding-004",
                                          content=query)["embedding"]
    # 向量检索 + 用户隔离的标量过滤
    results = memory_collection.search(data=[query_embedding],
                                       anns_field="embedding",
                                       param={"metric_type": "COSINE", "params": {"nprobe": 10}},
                                       limit=top_k,
                                       expr=f'user_id == "{user_id}"',  # 用户隔离的关键
                                       output_fields=["question", "solution", "timestamp"])
    formatted = []
    for i, hit in enumerate(results[0]):
        q = hit.entity.get("question")
        s = hit.entity.get("solution")
        formatted.append(f"{i+1}. {q}\n  解决方案: {s[:200]}...")
    return "\n\n".join(formatted)

recall_memory_tool = FunctionTool(name="recall_memory",
                                  description="从长期记忆中检索相关历史案例。当处理重复性问题时先调用此工具。",
                                  function=recall_memory)

# 3. Agent 定义
support_agent = Agent(model="gemini-2.5-flash-lite",
                      name="support_agent",
                      description="技术支持专家 Agent",
                      instruction="""你是一个技术支持专家。严格按照以下流程处理:
1. 立即调用 recall_memory 工具查找历史案例
2. 根据检索结果回答
3. 回答完毕后询问用户问题是否解决
4. 用户确认解决后调用 store_memory 存储新案例""",
                      tools=[recall_memory_tool, store_memory_tool])

# 4. Runner 运行
runner = Runner(agent=support_agent,
                app_name="tech_support",
                session_service=session_service)

ADK核心设计原则：

1. Event驱动模型：每次操作都封装成Event追加到事件流，可以在任意时刻回放决策过程
2. 接口抽象：SessionService和MemoryService都是接口，具体存储后端由开发者选择（内存、数据库、云托管）
3. 架构强制分离：框架从设计上就不允许开发者将会话临时数据和跨会话长期数据混在一起，从根源上降低了污染风险

9.2.2 Anthropic Managed Agents：Session / Harness / Sandbox 三层隔离

Anthropic在2026年4月正式公测的Managed Agents提出了另一种架构哲学——三层解耦，让每个组件只做一件事。核心灵感来自操作系统的设计：“为尚未被设想的程序设计系统”——接口保持稳定，实现可以随时间演进。

三层解耦架构：

┌─────────────────────────────────────────────────────────┐
│              Managed Agents Architecture                │
│                                                         │
│  ┌──────────────┐  ┌──────────────┐  ┌────────────┐    │
│  │   Session    │  │   Harness    │  │  Sandbox   │    │
│  │ (记忆/账本)  │  │ (编排/控制面) │  │ (执行/手)  │    │
│  │              │  │              │  │            │    │
│  │ • 持久事件日志 │  │ • 工具调用决策 │  │ • 代码执行 │    │
│  │ • 上下文外存储 │  │ • 上下文管理  │  │ • 文件编辑 │    │
│  │ • 可恢复/可查询│  │ • 错误恢复   │  │ • 进程隔离 │    │
│  │ • getEvents() │  │ • 无状态设计 │  │ • 检查点   │    │
│  └──────┬───────┘  └──────┬───────┘  └─────┬──────┘    │
│         │                 │                 │          │
│         │    getSession() │  execute(name,  │          │
│         │ ◄───────────────│  input) → string│          │
│         │    emitEvent()  │────────────────►│          │
│         │                 │                 │          │
│  ┌──────┴───────────────────┴───────────────────┴──────┐ │
│  │            Credential Proxy / Vault                  │ │
│  │ 凭证永远不会进入 Sandbox，防止 Prompt Injection      │ │
│  └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘

关键接口定义：

# Session 接口 — 持久化append-only事件日志
def getEvents(sessionId: str, range=None) -> Event[]
def emitEvent(sessionId: str, event: Event) -> None
def wake(sessionId: str) -> HarnessState

# Harness 接口 — 编排和控制面
def execute(name: str, input: str) -> str
# 统一工具调用接口
def provision(resources: ResourceSpec) -> Sandbox  # 按需初始化执行环境

# Sandbox 接口 — 执行环境
def execute(name: str, input: str) -> str
def checkpoint() -> Snapshot              # 保存执行状态
def restore(snapshot: Snapshot) -> None   # 恢复执行状态

Brain/Hands解耦：这是最关键的设计决策

Anthropic将整个系统拆解为两个独立部分：

• Brain = Claude + Harness（控制面，无状态）
• Hands = Sandbox + MCP Server + 外部工具（执行资源，可重建）

这种解耦带来了三个重大工程收益：

1. Sandbox崩溃不代表任务死亡。Harness会捕获失败作为工具调用错误，传给Claude决策重试。用户不需要重启整个任务，只需要重建Sandbox从检查点恢复。
2. TTFT（首Token时间）大幅优化。Brain可以先工作，Sandbox后加载。p50 TTFT降低约60%，p95 TTFT降低超过90%。
3. 一个Brain可以调度多个Hands。支持多执行环境并行，一个Agent可以同时在多个Sandbox中执行不同任务。

Session ≠ Context Window——这是根本性的认知升级

Managed Agents最根本的洞见是：Session不应该等于Claude的上下文窗口。它们是两个完全不同的东西：

Context Builder（从Session构建当前工作上下文）：

class ContextBuilder:
    def build(self, session: Session, current_task: Task) -> Context:
        # 从 Session Event Log 中选择当前需要的部分
        recent_events = session.getEvents(range=(-20, -1))  # 最近20轮
        relevant_memories = self.memory_retriever.query(current_task)  # 相关记忆
        active_skills = self.skill_manager.get_active()  # 当前激活的技能
        return Context(system_prompt=self.prompt_loader.load(),
                       events=recent_events,
                       memories=relevant_memories,
                       skills=active_skills,
                       tools=self.tool_registry.get_relevant(current_task))

凭证安全边界：

Managed Agents从架构层面强制解决了凭证泄露问题：

模式一：Git凭证绑定到资源
Sandbox初始化时用scoped token clone仓库
→ Token永远不会以明文形式进入Sandbox文件系统
→ 只有git命令可以使用它
模式二：Vault + 袋里
Claude → MCP Proxy（会话级token） → Vault（凭证存储） → 外部服务
→ Harness永远不会知道任何明文凭证
→ 完全防止Prompt Injection窃取凭证

9.2.3 Claude Code的“同步共享、异步隔离、转录留痕”设计哲学

Claude Code作为最广泛使用的交互式Agent开发环境，其实践了一套非常务实的多Agent上下文设计哲学，可以总结为三句话：同步共享上下文、异步隔离执行环境、基于转录的完整审计追踪。

同步共享（Synchronous Sharing）：

Claude Code和开发者共享同一个工作环境——同一个文件系统、同一个终端、同一个编辑器。开发者实时看到Agent做了什么，Agent也实时看到开发者改了什么。这是交互式编程最自然的协作模式：

• 文件修改实时同步，开发者即刻可见
• Agent通过文件系统读取开发者的修改，继续推理
• 结构化提问工具AskUserQuestion实现人机同步协作

异步隔离（Asynchronous Isolation）：

每个Subagent或Agent Team中的成员都拥有独立的上下文窗口和独立的Sandbox执行环境：

• OS级别的文件系统隔离（bubblewrap/seatbelt）
• 网络隔离，袋里验证域名白名单
• 凭证永远不进入Sandbox，通过袋里访问

即使Prompt Injection成功，攻击者也无法窃取SSH密钥或外联到攻击者服务器。内部数据显示沙盒化将权限提示减少了84%。

转录留痕（Transcript-based Audit Trail）：

每一轮对话、每一次工具调用、每一次文件修改都完整记录在transcript中。这与Managed Agents的Session Event Log哲学完全一致——transcript是账本，prompt是工作区。任何故障都可以通过转录完整复现，大大缩短调试时间。

三层知识渐进式披露架构：

Claude Code的知识管理本身就是渐进披露原则的完美实践：

第0层 · 索引 (200-500 tokens)
→ 能力导航，标注可用工具、调用时机、功能范围
→ 启动时只加载这一层
第1层 · 模式卡片 (500-1500 tokens)
→ 标准化操作手册，附示例 + 反面案例
→ 模型判断相关时才加载
第2层 · 完整手册 (2000 tokens 以上)
→ 细节文档，仅模型需要深度信息时动态加载
→ 控制递归读取深度，质量无提升时立刻终止

9.3 上下文隔离与共享机制：哪些状态该共享，哪些必须隔离

经过Google ADK和Anthropic Managed Architects的实践，到2026年已经形成了一套清晰的决策框架——不是全共享也不是全隔离，而是根据状态的生命周期和安全边界分类处理。

9.3.1 状态分类矩阵

9.3.2 决策框架

判断一个状态应该共享还是隔离，回答四个问题：

1. 生命周期：此状态的生命周期是什么？
   - 单次推理 → Context Window（隔离）
   - 单次Session → Session State（隔离）
   - 跨Session → Memory Service（共享，但按user_id隔离）
   - 永久 → Profile/Knowledge Base（共享）
2. 安全边界：此状态包含敏感信息吗？
   - 凭证 → Vault（绝对隔离，连Sandbox都不能访问）
   - PII个人信息 → user级别 + 加密存储
   - 公开信息 → 可共享
3. 污染风险：此状态被错误Agent读取会造成什么后果？
   - 高风险 → 严格隔离 + 访问控制审计日志
   - 低风险 → 可共享，但记录审计日志
4. 协作需求：多个Agent是否需要同时访问此状态？
   - 需要 → 共享，但需要冲突解决机制（版本号、优先级）
   - 不需要 → 隔离

9.3.3 五层上下文管理模式（分层组合）

在实践中，成熟的多Agent系统会组合使用五层上下文管理：

┌─────────────────────────────────────────────┐
│ 1. Tool Management          ← 定义什么可以进入context
│   · MCP token audit、schema校验             │
├─────────────────────────────────────────────┤
│ 2. Progressive Disclosure    ← 按需分层加载 │
│   · 发现层 → 激活层 → 执行层               │
├─────────────────────────────────────────────┤
│ 3. Context Routing          ← 管理执行期间保留什么
│   · 相关性分类，只保留相关信息              │
├─────────────────────────────────────────────┤
│ 4. Context Compression      ← 压缩历史上下文
│   · 滑动窗口 + 摘要混合策略                │
├─────────────────────────────────────────────┤
│ 5. Evolved Retrieval        ← 按需从外部获取
│   · Agentic / Graph / Self-RAG             │
└─────────────────────────────────────────────┘

成熟度：渐进披露（高）> 上下文压缩（高）> 工具管理（中）> 上下文路由（中）> 演化检索（中）

9.4 智能体间通信协议双雄：MCP vs A2A

到2026年，多Agent生态系统已经形成了“一个协议栈，两个层次”的清晰分工：MCP负责Agent与工具/数据的连接，A2A负责Agent与Agent的通信。两者互补而非竞争，共同构成了完整的多Agent通信标准。

9.4.1 MCP（模型上下文协议）：Anthropic主导，标准化Agent-工具连接

发展时间线：

协议架构：

MCP采用经典的Client-Server模式：

┌──────────────┐                    ┌──────────────┐
│  MCP Client  │◄────REST────►       │  MCP Server  │
│ (AI应用内)   │   Streamable HTTP   │ (外部工具/数据) │
│              │                     │              │
│ • 发送请求   │                     │ • 托管工具   │
│ • 解析响应   │                     │ • 数据源     │
│ • 注入上下文 │                     │ • API资源    │
└──────────────┘                     └──────────────┘

三种传输模式演进：

Streamable HTTP的核心优势：单一端点替代双端点（HTTP + SSE），支持双向流式通信，防火墙穿透更简单。

核心设计理念：

MCP被比喻为AI世界的USB-C接口——通过标准化协议将AI模型与外部世界连接。开发者只需要实现一次MCP服务器，所有支持MCP的AI应用（Claude Code、ChatGPT桌面、Cursor、Google ADK）都能直接使用。

传统对接方式：                        MCP方式：
Agent A ──→ API A（单独对接）         Agent ──→ MCP Client ──→ MCP Server A ──→ Tool A
Agent B ──→ API B（单独对接）                          └─→ MCP Server B ──→ Tool B
Agent C ──→ API C（单独对接）                          └─→ MCP Server C ──→ Data Source C

代码示例：Python MCP Server

# MCP Server 基本结构（Python SDK）
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("my-documentation-mcp-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="Search the documentation for a given query",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query keywords"
                    }
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = search_index(arguments["query"])
        return [TextContent(type="text", text=str(results))]
    raise ValueError(f"Unknown tool: {name}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(app))

四个未解决的工程问题（2026年Q1现状）：

1. 描述质量问题：大多数MCP Server作者为人类而非模型编写tool description，要么太模糊（模型选错工具）要么太冗长（单个schema浪费大量context）
2. 工具重叠问题：两个不同MCP Server可能提供相似能力，缺乏去重和偏好逻辑
3. 版本控制缺失：MCP Server更新schema时，Agent无法感知；缓存中的 stale descriptions 导致静默失败
4. 安全攻击面：每个连接的MCP Server都是一个攻击面，tool outputs 可能包含 prompt injection 尝试

9.4.2 A2A（Agent-to-Agent Protocol）：Google主导，标准化Agent-Agent通信

基本信息：

• 发布时间：2025年4月9日，由Google主导，超过150家组织支持
• 开源地址：github.com/google/A2A
• 定位：让不同框架、不同厂商的AI Agent能够直接互通与协作
• 2026年4月：达到v1.0 production-ready

协议架构：

A2A同样基于Client-Server模式，建立在HTTP + JSON-RPC 2.0之上：

┌──────────────────────────────────────────────────────────┐
│                       A2A Protocol                       │
│                                                          │
│  ┌──────────────┐                         ┌──────────────┐│
│  │ Client Agent │◄──────────────────────►│ Server Agent ││
│  │ (任务发起者) │  HTTP + SSE/JSON-RPC   │ (任务执行者) ││
│  │              │                         │              ││
│  │ • 发现能力   │                         │ • 公开能力   ││
│  │ • 发起 Task  │                         │ • 执行 Task  ││
│  │ • 接收结果   │                         │ • 返回结果   ││
│  └──────┬───────┘                         └──────┬───────┘│
│         │                                         │      │
│         │               AgentCard发现              │      │
│         │             /.well-known/agent.json      │      │
│         │ ◄───────────────────────────────────────────      │
│         │                                         │      │
│         │           Task / Message / Artifact      │      │
│         │ ────────────────────────────────────────►│      │
└──────────────────────────────────────────────────────────┘

四大核心概念：

1. AgentCard：Server Agent的“名片”，公开能力和认证方式。标准发现路径：https://${host}/.well-known/agent-card.json

{
  "name": "Google Maps Agent",
  "description": "Plan routes, remember places, and generate directions",
  "url": "https://maps-agent.google.com",
  "version": "1.0.0",
  "authentication": { "schemes": ["OAuth2"] },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/html"],
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "route-planner",
      "name": "Route planning",
      "description": "Helps plan routing between two locations",
      "tags": ["maps", "routing", "na vigation"],
      "examples": ["plan my route from Sunnyvale to Mountain View"]
    }
  ]
}

2. Task：具有明确状态的任务实体，由Client Agent创建，Server Agent维护状态。状态流转：

submitted → working → input-required → completed / canceled / failed

每个Task拥有唯一id和sessionId，多个Task可以共享同一个sessionId。

3. Message：Agent间交互的基本单元，由多个Part组成，支持TextPart / FilePart / DataPart。

4. Artifact：任务执行后的输出结果，不可变、可命名、支持渐进式构建（append + lastChunk）。

典型交互流程：

1. Client发现Server能力 → GET /.well-known/agent-card.json → AgentCard
2. Client发起Task → JSON-RPC: tasks/send → Task创建
3. Client监听状态更新 → SSE or 轮询调用 tasks/get
4. Server完成 → Task状态变为 completed，返回Artifact

9.4.3 MCP vs A2A互补全景

MCP和A2A并非竞争关系，而是在协议栈的不同层次分工合作，完美互补：

┌─────────────────────────────────────────────────────────┐
│                      Agent 协作生态                      │
│                                                         │
│  ┌───────────────────────────────────────────────────┐  │
│  │         A2A: Agent-to-Agent 通信层                │  │
│  │            "AI 如何与 AI 协作"                     │  │
│  │  • Agent 发现 (AgentCard)                         │  │
│  │  • 任务委托与分发 (Task lifecycle)                │  │
│  │  • 跨厂商互通                                     │  │
│  └───────────────────────────────────────────────────┘  │
│                 ▲                                       │
│                 │ 每个 Agent 内部使用 MCP               │
│                 ▼                                       │
│  ┌───────────────────────────────────────────────────┐  │
│  │         MCP: 工具/数据连接层                      │  │
│  │          "AI 如何与外部世界交互"                   │  │
│  │  • 标准化工具接口 (Tool)                          │  │
│  │  • 数据源访问 (Resources)                         │  │
│  │  • 提示模板 (Prompts)                             │  │
│  │  • 安全传输 (Streamable HTTP/stdio)               │  │
│  └───────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

详细对比：

组合使用案例：企业活动策划系统

主智能体（A2A Client）
├── A2A → 场地智能体（A2A Server）
│       └── MCP → 地图API、日历API
├── A2A → 餐饮智能体（A2A Server）
│       └── MCP → 菜单数据库、供应商API
└── A2A → 嘉宾智能体（A2A Server）
        └── MCP → 邮件API、CRM系统

9.5 多智能体上下文冲突解决

多个Agent协作，就会有冲突。冲突主要分为三类：记忆冲突、资源竞争、知识冲突。本节讨论如何系统性地解决这些冲突。

9.5.1 冲突分类

9.5.2 版本管理：时间戳 + 置信度 + 优先级

最实用的解决方案是为每个记忆条目增加版本信息，基于时间戳和置信度解决冲突：

from typing import Any, List
from dataclasses import dataclass
import time

@dataclass
class MemoryVersion:
    value: Any
    timestamp: float
    confidence: float
    source_agent: str
    version: int

class VersionedMemory:
    """带版本管理的多Agent记忆系统"""
    def __init__(self):
        self._memories: dict[str, List[MemoryVersion]] = {}

    def write(self, key: str, value: Any, confidence: float = 1.0, source_agent: str = None) -> None:
        """写入一个新版本"""
        if key not in self._memories:
            self._memories[key] = []
        version = MemoryVersion(
            value=value,
            timestamp=time.time(),
            confidence=confidence,
            source_agent=source_agent,
            version=len(self._memories[key]) + 1
        )
        self._memories[key].append(version)

    def read(self, key: str, strategy: str = "latest_high_confidence") -> MemoryVersion:
        """读取最新版本，支持多种策略"""
        versions = self._memories.get(key, [])
        if not versions:
            return None
        if strategy == "latest":
            return versions[-1]
        elif strategy == "latest_high_confidence":
            # 返回最近的高置信度版本
            high_confidence = [v for v in versions if v.confidence > 0.8]
            if not high_confidence:
                return versions[-1]
            return max(high_confidence, key=lambda v: v.timestamp)
        elif strategy == "consensus":
            # 多Agent共识：最近N个版本多数同意的值
            from collections import Counter
            recent = versions[-5:]  # 只考虑最近5个版本
            values = [v.value for v in recent]
            return Counter(values).most_common(1)[0][0]
        else:
            raise ValueError(f"Unknown strategy: {strategy}")

9.5.3 优先级规则：领域权威 + 安全优先 + 时间衰减

在版本管理之上，我们可以定义一套优先级规则帮助自动解决冲突：

from dataclasses import dataclass
from typing import Optional, List

@dataclass
class Conflict:
    key: str
    versions: List[MemoryVersion]

@dataclass
class Resolution:
    choice: Optional[MemoryVersion]
    reason: str
    needs_human_review: bool = False
    candidates: List[MemoryVersion] = None

class ConflictResolver:
    """多Agent上下文冲突解决器"""
    RULES = {
        # 规则一：安全策略永远优先于功能策略
        "security_over_functional": True,
        # 规则二：近期时间衰减（越近越优先）
        "recency_decay_days": 30,
        # 规则三：高置信度覆盖低置信度
        "confidence_threshold": 0.7,
        # 规则四：领域权威有更高权重
        "domain_authority": {
            "security_agent": ["credentials", "permissions", "auth"],
            "database_agent": ["schema", "migration", "query"],
            "frontend_agent": ["css", "layout", "component"],
        }
    }

    def resolve(self, conflict: Conflict) -> Resolution:
        versions = conflict.versions

        # 1. 检查安全规则（最高优先级）
        if self._involves_security(conflict):
            # 安全Agent的版本有最高优先级
            security_version = self._get_security_version(versions)
            if security_version:
                return Resolution(
                    choice=security_version,
                    reason="security policy has highest priority",
                    needs_human_review=False
                )

        # 2. 检查领域权威
        domain = self._detect_domain(conflict)
        if domain in self.RULES["domain_authority"]:
            authority_agent = self.RULES["domain_authority"][domain][0]
            authority_version = self._get_version_by_agent(versions, authority_agent)
            if authority_version:
                return Resolution(
                    choice=authority_version,
                    reason=f"{authority_agent} is domain authority for {domain}",
                    needs_human_review=False
                )

        # 3. 时间衰减 + 置信度加权打分
        scores = {}
        for v in versions:
            recency_score = self._recency_decay(v.timestamp)
            scores[v] = recency_score * v.confidence
        winner = max(scores, key=scores.get)
        top_candidates = sorted(scores, key=scores.get, reverse=True)[:3]

        # 4. 如果最高分和次高分差距太小，触发人工审查
        if len(top_candidates) >= 2:
            score_diff = scores[top_candidates[0]] - scores[top_candidates[1]]
            if score_diff < 0.15:  # 差距小于15%触发人工
                return Resolution(
                    choice=None,
                    reason="scores too close, need human review",
                    needs_human_review=True,
                    candidates=top_candidates
                )

        return Resolution(
            choice=winner,
            reason="weighted score (recency × confidence)",
            needs_human_review=False
        )

9.5.4 人工复核机制：人机协同的最后一道防线

对于高风险操作或自动决策置信度不足的冲突，应该触发人工复核——这是企业级落地的关键组件：

from datetime import datetime, timedelta

class HumanReviewGateway:
    """人工审查网关"""
    REVIEW_TRIGGERS = {
        "confidence_gap_threshold": 0.15,  # 最高与次高置信度差距小于阈值
        "high_risk_operations": [           # 高风险操作列表
            "delete_production_data",
            "modify_security_policy",
            "commit_to_main_branch",
            "change_infrastructure",
        ],
        "first_time_pattern": True,        # 首次出现的新冲突模式
    }

    def should_review(self, resolution: Resolution, operation: str) -> bool:
        """判断这个冲突是否需要人工审查"""
        # 触发条件一：自动决策置信度不足
        if not resolution.choice:
            return True
        # 触发条件二：操作属于高风险
        if operation in self.REVIEW_TRIGGERS["high_risk_operations"]:
            return True
        # 触发条件三：首次出现的新模式
        if self.REVIEW_TRIGGERS["first_time_pattern"] and self._is_novel_pattern(resolution):
            return True
        return False

    def create_review_task(self, resolution: Resolution) -> dict:
        """创建一个人工审查任务"""
        return {
            "title": f"Agent Context Conflict: {resolution.reason}",
            "description": self._format_description(resolution),
            "options": self._format_candidates(resolution.candidates),
            "deadline": datetime.now() + timedelta(hours=4),
            "auto_default": resolution.candidates[0]  # 超时自动选择
        }

实践经验：在企业级多Agent系统中，你不可能指望100%自动解决所有冲突。保留一个人工复核闸门，在自动决策不确定的时候把问题交给人，这比“强制自动决策导致错误”要好得多。

本章小结

本章从多智能体系统的四大上下文挑战出发，系统性地阐述了2026年主流的架构方案、共享隔离决策框架、通信协议栈和冲突解决机制。

9.1节 定义了多Agent系统独特的四大挑战——上下文爆炸（状态共享）、状态污染（污染防范）、可追溯性（失败定位）、一致性（冲突解决）。这四大挑战是单Agent场景不存在的，也是多Agent上下文工程必须解决的核心问题。

9.2节 深入剖析了两种主流架构方案：Google ADK三层上下文栈（通过前缀机制强制分离会话内上下文和跨会话长期记忆）和Anthropic Managed Agents三层隔离（Session/Harness/Sandbox解耦，Brain/Hands解耦，Session不等于Context Window），以及Claude Code的“同步共享/异步隔离/转录留痕”设计哲学。两种架构哲学不同，但核心都是通过架构分层从根源降低污染风险。

9.3节 给出了清晰的共享隔离决策框架——状态分类矩阵（用户/应用/会话/长期/凭证五级分类），基于生命周期、安全边界、污染风险、协作需求四个问题的决策树，以及五层上下文管理组合模式。

9.4节 讨论了通信协议双雄——MCP（Anthropic主导，Agent-工具连接）和A2A（Google主导，Agent-Agent通信），分析了它们的互补关系：MCP负责连接层，A2A负责通信层，两者组合使用覆盖整个Agent生态。

9.5节 讨论了多Agent上下文冲突解决——版本管理（时间戳+置信度）、优先级规则（领域权威+安全优先+时间衰减）、以及人机协同的人工复核机制——这是企业落地的最后一道安全闸门。

从单Agent到多Agent，上下文工程的复杂度提升了一个量级，但核心原则其实没变——清晰的边界、明确的生命周期、隔离可变和不变、分离不同信任域。这些原则说起来简单，但在多Agent场景下一不小心就会违反。在下一章，我们将讨论一个更进阶的主题：上下文安全与可观测性——如何保护上下文不被攻击，如何诊断上下文问题。