多Agent共享记忆六栈：缓存一致性到知识同步实测

graph TB
    subgraph "单Agent记忆模型"
        A1[用户输入] --> A2[上下文窗口]
        A2 --> A3[长期记忆]
        A3 --> A2
    end

    subgraph "多Agent共享记忆模型"
        B1[Agent A] --> B4[共享工作记忆<br/>Working Memory]
        B2[Agent B] --> B4
        B3[Agent C] --> B4
        B4 --> B5[一致性协议<br/>MESI类比]
        B5 --> B6[长期知识库<br/>Long-term Store]
        B6 --> B4
        B1 -.->|环境标记| B7[Stigmergy空间]
        B2 -.->|环境标记| B7
        B3 -.->|环境标记| B7
        B7 -.-> B1
        B7 -.-> B2
        B7 -.-> B3
    end

    style B4 fill:#4a9eff,color:#fff
    style B5 fill:#ff6b6b,color:#fff
    style B7 fill:#51cf66,color:#fff

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph

class SharedState(TypedDict):
    research_findings: list[str]      # Agent A写入，Agent B读取
    analysis_results: dict            # Agent B写入，Agent C读取
    final_report: str                 # Agent C写入
    messages: Annotated[list, "add"]  # 累积式记忆

def researcher(state: SharedState) -> dict:
    findings = search_and_analyze(state["messages"][-1])
    return {"research_findings": findings}

def analyst(state: SharedState) -> dict:
    # 直接读取researcher的输出，无需对话
    results = process(state["research_findings"])
    return {"analysis_results": results}

graph = StateGraph(SharedState)
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_edge("researcher", "analyst")

from crewai import Agent, Task, Crew, Process
from crewai.memory import LongTermMemory

# Agent各自带独立记忆
researcher = Agent(
    role="Researcher",
    goal="Find relevant information",
    backstory="Expert researcher",
    memory=True,  # 启用独立记忆
    allow_delegation=False
)

writer = Agent(
    role="Writer",
    goal="Write engaging content",
    backstory="Professional writer",
    memory=True,
)

# 任务链传递上下文，而非共享状态
research_task = Task(
    description="Research {topic}",
    agent=researcher,
    expected_output="Research findings"
)

write_task = Task(
    description="Based on findings, write article",
    agent=writer,
    expected_output="Final article",
    context=[research_task]  # 通过任务引用传递上下文
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential
)

import mem0

# 共享记忆实例，多Agent共用
shared_memory = mem0.Memory.from_config({
    "version": "v1.1",
    "embedder": {
        "provider": "openai",
        "model": "text-embedding-3-small"
    },
    "llm": {"provider": "openai", "model": "gpt-4o"},
    "graph_store": {
        "provider": "neo4j",
        "url": "neo4j://localhost:7687",
        "username": "neo4j",
        "password": "password"
    }
})

# Agent A写入发现
shared_memory.add(
    "User prefers Python over JavaScript for data analysis",
    user_id="team_alpha",
    agent_id="researcher_01",
    metadata={"confidence": 0.92, "source": "conversation"}
)

# Agent B读取，无需对话
results = shared_memory.search(
    "What language does the user prefer?",
    user_id="team_alpha",
    agent_id="writer_02"
)

import json
from datetime import datetime

class StigmergySpace:
    """共享环境空间——Agent通过读写环境标记间接通信"""
    def __init__(self):
        self.signals = {}  # 信号空间

    def leave_signal(self, agent_id: str, signal_type: str, 
                     content: dict, intensity: float = 1.0):
        self.signals[f"{agent_id}_{signal_type}"] = {
            "content": content,
            "intensity": intensity,
            "timestamp": datetime.now().isoformat(),
            "decay_rate": 0.1  # 信息素衰减
        }

    def read_signals(self, signal_type: str = None):
        results = []
        for key, sig in self.signals.items():
            if signal_type and signal_type not in key:
                continue
            # 模拟信息素衰减
            elapsed = (datetime.now() - datetime.fromisoformat(sig["timestamp"])).seconds
            sig["current_intensity"] = sig["intensity"] * (1 - sig["decay_rate"] * elapsed / 3600)
            results.append(sig)
        return sorted(results, key=lambda x: x["current_intensity"], reverse=True)

# Agent A：研究者留下发现标记
space = StigmergySpace()
space.leave_signal("researcher", "finding", 
    {"topic": "quantum computing", "summary": "..."},
    intensity=0.9)

# Agent B：写作者读取环境中的发现，无需对话
findings = space.read_signals("finding")

# Google ADK 分层上下文架构（概念示意）
from google.adk import Agent, Context

class TieredContextAgent:
    def __init__(self):
        # Tier 1: 静态上下文（系统指令+工具定义）— 缓存命中率>95%
        self.system_context = Context.tier1(
            instructions="You are a research analyst...",
            tools=[search_tool, analysis_tool]
        )
        # Tier 2: 会话摘要 — 每5轮对话压缩一次
        self.session_summary = Context.tier2(
            compression_ratio=0.3,  # 保留30%关键信息
            update_frequency=5       # 每5轮更新
        )
        # Tier 3: 动态检索 — 按相关性从向量库取
        self.dynamic_memory = Context.tier3(
            vector_store="vertex_ai_vector_search",
            top_k=5,
            relevance_threshold=0.7
        )

    async def run(self, query: str):
        # 组装上下文：Tier1 + Tier2 + Tier3 + 当前query
        full_context = self.system_context + self.session_summary + self.dynamic_memory.retrieve(query)
        return await self.agent.execute(full_context + query)

# Anthropic 多Agent研究系统架构（概念还原）
import anthropic
import asyncio

client = anthropic.Anthropic()

async def lead_agent(query: str):
    # Step 1: 分解任务
    subtasks = decompose_query(query)  # ["search web", "search docs", "analyze data"]

    # Step 2: 并行启动Sub-Agents
    results = await asyncio.gather(*[
        sub_agent(task) for task in subtasks
    ])

    # Step 3: Lead Agent融合结果（唯一的知识汇聚点）
    synthesis = client.messages.create(
        model="claude-sonnet-4-20250514",
        system="Synthesize research results. Resolve contradictions.",
        messages=[{
            "role": "user",
            "content": f"Query: {query}\n\nResults:\n" + 
                       "\n---\n".join([compress(r) for r in results])
        }]
    )
    return synthesis

async def sub_agent(task: dict):
    # 独立上下文，不共享其他Agent状态
    result = client.messages.create(
        model="claude-sonnet-4-20250514",
        system=f"You are a specialized {task['role']}. Return structured findings.",
        messages=[{"role": "user", "content": task["description"]}]
    )
    return result

graph LR
    subgraph "Agent A"
        A_IO["I/O缓冲<br/>Modified"]
        A_WM["工作记忆<br/>Shared"]
    end
    subgraph "Agent B"
        B_IO["I/O缓冲<br/>Invalid"]
        B_WM["工作记忆<br/>Shared"]
    end
    subgraph "Agent C"
        C_IO["I/O缓冲<br/>Exclusive"]
        C_WM["工作记忆<br/>Invalid"]
    end

    A_IO -->|写入| LT["长期记忆<br/>(Shared Store)"]
    B_IO -->|Invalid→拉取| LT
    C_IO -->|独占读写| LT
    A_WM -.->|MESI协议| B_WM
    B_WM -.->|MESI协议| C_WM

    style A_IO fill:#ff6b6b,color:#fff
    style B_IO fill:#ffd43b,color:#333
    style C_IO fill:#51cf66,color:#fff
    style LT fill:#4a9eff,color:#fff

graph TB
    subgraph "路径1: 全连接广播 O(N²)"
        A1[Agent A] <--> A2[Agent B]
        A1 <--> A3[Agent C]
        A2 <--> A3
    end

    subgraph "路径2: Stigmergy O(N)"
        B1[Agent A] --> B4[共享环境]
        B2[Agent B] --> B4
        B3[Agent C] --> B4
        B4 --> B1
        B4 --> B2
        B4 --> B3
    end

    subgraph "路径3: Lead-Sub O(N)"
        C1[Lead Agent] --> C2[Sub A]
        C1 --> C3[Sub B]
        C1 --> C4[Sub C]
        C2 --> C1
        C3 --> C1
        C4 --> C1
    end

    style B4 fill:#51cf66,color:#fff
    style C1 fill:#4a9eff,color:#fff

graph TD
    START[需要多Agent共享记忆] --> Q1{Agent数量?}
    Q1 -->|> 10个| Q2{需要强一致?}
    Q1 -->|≤ 10个| Q3{预算敏感?}

    Q2 -->|是| LANG[LangGraph状态图<br/>强一致但状态膨胀]
    Q2 -->|否| Q4{有图数据库?}

    Q4 -->|有| MEM0[Mem0图谱层<br/>因果一致+SOTA基准]
    Q4 -->|无| STIG[Stigmergy标记<br/>最轻量但工程成熟度低]

    Q3 -->|是| Q5{GCP用户?}
    Q3 -->|否| CREW[CrewAI隔离记忆<br/>上手最快但Token浪费]

    Q5 -->|是| ADK[Google ADK分层<br/>生产级但GCP锁定]
    Q5 -->|否| ANTH[Anthropic Lead-Sub<br/>架构最简但Lead瓶颈]

    style LANG fill:#4a9eff,color:#fff
    style MEM0 fill:#9775fa,color:#fff
    style STIG fill:#51cf66,color:#fff
    style ADK fill:#fcc419,color:#333
    style ANTH fill:#ff6b6b,color:#fff
    style CREW fill:#ffd43b,color:#333