多 Agent 协同架构：解决长期记忆问题的共享记忆方案

前言

多 Agent 系统最大的问题是什么？Agent A 知道的信息，Agent B 不知道。

本文们做了一个多 Agent 客服系统，三个 Agent 分别处理订单、物流、售后。结果用户问一句"本文的订单怎么样了"，三个 Agent 各答各的，用户骂死了。

后来搞了一套共享长期记忆架构，才解决了这个问题。

一、底层原理

1.1 多 Agent 记忆共享的难点

每个 Agent 有自己的上下文，怎么共享？

graph TD
    A["Agent A 订单"] --> B["私有记忆"]
    C["Agent B 物流"] --> D["私有记忆"]
    E["Agent C 售后"] --> F["私有记忆"]
    G["共享记忆层"] --> H["统一存储"]
    H --> B
    H --> D
    H --> F
    B --> I["A 知道订单信息"]
    D --> I["B 知道物流信息"]
    F --> I["C 知道售后信息"]
    I --> J["统一回复"]

核心挑战：

• 每个 Agent 有独立上下文
• Agent 间信息不互通
• 缺乏统一记忆管理
• 用户需要重复描述

1.2 记忆方案对比

方案	一致性	性能	复杂度
Agent 私有	差	好	低
共享内存	中	好	中
中心化存储	好	中	中
分布式存储	好	差	高

二、快速上手

2.1 基础共享记忆

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

@dataclass
class MemoryItem:
    key: str
    value: Any
    agent_source: str
    timestamp: float

class SharedMemory:
    def __init__(self):
        self.store: Dict[str, MemoryItem] = {}
    
    def write(self, key: str, value: Any, agent: str):
        import time
        self.store[key] = MemoryItem(
            key=key, value=value,
            agent_source=agent,
            timestamp=time.time()
        )
    
    def read(self, key: str) -> Any:
        item = self.store.get(key)
        return item.value if item else None
    
    def get_all(self) -> Dict:
        return {k: v.value for k, v in self.store.items()}

class Agent:
    def __init__(self, name: str, memory: SharedMemory):
        self.name = name
        self.memory = memory
    
    def remember(self, key: str, value: Any):
        self.memory.write(key, value, self.name)
    
    def recall(self, key: str) -> Any:
        return self.memory.read(key)

三、核心 API / 深水区

3.1 共享记忆组件速查

组件	职责	实现要点
共享存储	存取数据	键值对
版本控制	冲突解决	时间戳
过期策略	自动清理	TTL
权限控制	Agent 隔离	读写权限

3.2 带过期和版本控制的记忆

import time

class VersionedMemory:
    def __init__(self, default_ttl=3600):
        self.store = {}
        self.default_ttl = default_ttl
    
    def set(self, key: str, value: Any, agent: str, ttl=None):
        ttl = ttl or self.default_ttl
        self.store[key] = {
            "value": value,
            "agent": agent,
            "version": self.store.get(key, {}).get("version", 0) + 1,
            "expire_at": time.time() + ttl,
            "created_at": time.time()
        }
    
    def get(self, key: str) -> Any:
        item = self.store.get(key)
        if not item:
            return None
        if time.time() > item["expire_at"]:
            del self.store[key]
            return None
        return item["value"]
    
    def get_latest_by_agent(self, agent: str) -> Dict:
        return {
            k: v for k, v in self.store.items()
            if v["agent"] == agent and time.time() <= v["expire_at"]
        }

3.3 上下文合并

class ContextMerger:
    def merge(self, contexts: List[Dict]) -> str:
        merged = {}
        for ctx in contexts:
            for k, v in ctx.items():
                if k not in merged:
                    merged[k] = v
        return "\n".join(f"{k}: {v}" for k, v in merged.items())

四、实战演练

完整的多 Agent 记忆共享系统：

from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import json
import time

@dataclass
class AgentContext:
    agent_name: str
    messages: List[Dict]
    knowledge: Dict[str, Any]

class MemoryOrchestrator:
    def __init__(self):
        self.global_memory = GlobalMemory()
        self.agents: Dict[str, Agent] = {}
    
    def register_agent(self, name: str, capabilities: List[str]):
        agent = Agent(name, capabilities, self.global_memory)
        self.agents[name] = agent
        return agent
    
    def handle_query(self, query: str) -> str:
        # 1. 查询全局记忆
        context = self.global_memory.get_relevant(query)
        
        # 2. 找到合适的 Agent
        agent = self._route_to_agent(query)
        
        # 3. 带上共享记忆处理
        response = agent.process(query, context)
        
        # 4. 存储这次交互
        self.global_memory.store_interaction(query, response, agent.name)
        
        return response
    
    def _route_to_agent(self, query: str) -> 'Agent':
        if "订单" in query:
            return self.agents.get("order_agent")
        elif "物流" in query:
            return self.agents.get("logistics_agent")
        else:
            return self.agents.get("general_agent")

class GlobalMemory:
    def __init__(self):
        self.memory = {}
        self.interactions = []
    
    def store(self, key: str, value: Any, agent: str):
        self.memory[key] = {
            "value": value,
            "agent": agent,
            "time": time.time()
        }
    
    def get(self, key: str) -> Any:
        item = self.memory.get(key)
        return item["value"] if item else None
    
    def get_relevant(self, query: str) -> str:
        results = []
        for key, item in self.memory.items():
            if any(w in key for w in query.split()):
                results.append(f"{key}: {item['value']}")
        return "\n".join(results[-5:])
    
    def store_interaction(self, query: str, response: str, agent: str):
        self.interactions.append({
            "query": query,
            "response": response,
            "agent": agent,
            "time": time.time()
        })
        self.memory[f"query_{len(self.interactions)}"] = {
            "value": f"Q: {query} A: {response}",
            "agent": agent,
            "time": time.time()
        }

class Agent:
    def __init__(self, name: str, capabilities: List[str], memory: GlobalMemory):
        self.name = name
        self.capabilities = capabilities
        self.memory = memory
    
    def process(self, query: str, context: str) -> str:
        response = f"[{self.name}] 收到查询: {query}"
        self.memory.store(f"last_response_{self.name}", response, self.name)
        return response

orchestrator = MemoryOrchestrator()
order_agent = orchestrator.register_agent("order_agent", ["订单查询"])
logistics_agent = orchestrator.register_agent("logistics_agent", ["物流查询"])

r1 = orchestrator.handle_query("本文的订单 12345 怎么样了")
r2 = orchestrator.handle_query("物流到哪了")
print(r1, r2)

五、避坑指南与最佳实践

💡 **技巧：共享记忆不能太大
只存关键信息，不要把所有对话都存进去。

⚠️ **警告：注意权限隔离
Agent A 存的数据，Agent B 不一定能读。

✅ **推荐：用 TTL 自动过期
长期不用的记忆自动清理，节省空间。

六、综合实战演示

生产级多 Agent 记忆架构：

from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import json
import time

@dataclass
class MemoryConfig:
    ttl_seconds: int = 3600
    max_items_per_agent: int = 100
    enable_compression: bool = True

class ProductionSharedMemory:
    def __init__(self, config: MemoryConfig):
        self.config = config
        self.store: Dict[str, Dict] = {}
        self.agent_index: Dict[str, List[str]] = {}
    
    def store(self, key: str, value: Any, agent: str, ttl: Optional[int] = None):
        ttl = ttl or self.config.ttl_seconds
        self.store[key] = {
            "value": value,
            "agent": agent,
            "ttl": ttl,
            "expires_at": time.time() + ttl,
            "created_at": time.time()
        }
        if agent not in self.agent_index:
            self.agent_index[agent] = []
        self.agent_index[agent].append(key)
        self._cleanup_agent(agent)
    
    def _cleanup_agent(self, agent: str):
        keys = self.agent_index.get(agent, [])
        if len(keys) > self.config.max_items_per_agent:
            to_remove = keys[:-self.config.max_items_per_agent]
            for k in to_remove:
                self.store.pop(k, None)
            self.agent_index[agent] = keys[-self.config.max_items_per_agent:]
    
    def recall(self, key: str) -> Optional[Any]:
        item = self.store.get(key)
        if not item:
            return None
        if time.time() > item["expires_at"]:
            del self.store[key]
            return None
        return item["value"]
    
    def recall_by_agent(self, agent: str) -> Dict:
        keys = self.agent_index.get(agent, [])
        result = {}
        for k in keys:
            v = self.recall(k)
            if v is not None:
                result[k] = v
        return result

class MultiAgentSystem:
    def __init__(self, llm):
        self.llm = llm
        config = MemoryConfig(ttl_seconds=3600)
        self.memory = ProductionSharedMemory(config)
        self.agents: Dict[str, Any] = {}
    
    def register_agent(self, name: str, handler):
        self.agents[name] = handler
    
    def process(self, query: str) -> str:
        self.memory.store("last_query", query, "system", ttl=600)
        
        for agent_name, handler in self.agents.items():
            if agent_name in query:
                context = self.memory.recall_by_agent(agent_name)
                response = handler(query, context)
                self.memory.store("last_response", response, agent_name)
                return response
        
        return "无法处理该查询"

system = MultiAgentSystem(llm)
system.register_agent("订单", lambda q, c: f"订单查询: {q}")
system.register_agent("售后", lambda q, c: f"售后处理: {q}")

print(system.process("本文的订单呢"))
print(system.process("售后怎么联系"))
print(system.memory.recall("last_query"))

七、总结

多 Agent 长期记忆共享要点：

• 全局共享记忆层
• 版本 + TTL 管理
• Agent 间权限控制
• 上下文合并

搞定了共享记忆，多 Agent 系统就不再"各自为政"了。