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

多 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( keykey, valuevalue, agent_sourceagent, timestamptime.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_ttl3600): self.store {} self.default_ttl default_ttl def set(self, key: str, value: Any, agent: str, ttlNone): 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[fquery_{len(self.interactions)}] { value: fQ: {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(flast_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_seconds3600) 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, ttl600) 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 系统就不再各自为政了。