多Agent记忆与协作系统完全教程

从零构建具备记忆能力的多智能体协作系统：架构设计、记忆管理、工作流编排与实战

前言

单个AI Agent的能力终究有限。当面对复杂任务时，我们需要多个专业化Agent协同工作——一个负责搜索，一个负责分析，一个负责代码生成，一个负责质量检查。而要让这些Agent真正高效协作，记忆系统是关键。

没有记忆的Agent就像一个每次对话都失忆的人——它无法从过去的经验中学习，无法在多轮协作中保持上下文，也无法与其他Agent共享知识。

本教程将系统讲解多Agent记忆与协作系统的完整架构与开发技术，帮助你构建真正具备记忆能力的多智能体协作系统。

第一章：Agent记忆架构

1.1 记忆的认知科学基础

人类记忆系统为Agent记忆设计提供了重要启发：

1.2 记忆系统架构设计

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from datetime import datetime
from enum import Enum
import json
import hashlib

class MemoryType(Enum):
    """记忆类型枚举"""
    WORKING = "working"         # 工作记忆（当前上下文）
    SHORT_TERM = "short_term"   # 短期记忆（会话摘要）
    LONG_TERM = "long_term"     # 长期记忆（持久知识）
    EPISODIC = "episodic"       # 情景记忆（事件记录）
    SEMANTIC = "semantic"       # 语义记忆（概念知识）

@dataclass
class MemoryItem:
    """记忆单元"""
    id: str
    content: str
    memory_type: MemoryType
    agent_id: str               # 所属Agent
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    importance: float = 0.5     # 重要性评分 0-1
    emotion: float = 0.0        # 情感标记 -1到1
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None
    related_memories: List[str] = field(default_factory=list)  # 关联记忆ID
    
    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "content": self.content,
            "memory_type": self.memory_type.value,
            "agent_id": self.agent_id,
            "created_at": self.created_at.isoformat(),
            "last_accessed": self.last_accessed.isoformat(),
            "access_count": self.access_count,
            "importance": self.importance,
            "tags": self.tags,
            "metadata": self.metadata
        }

class MemoryManager:
    """Agent记忆管理器 - 核心类"""
    
    def __init__(self, agent_id: str, embedding_model=None, vector_store=None):
        self.agent_id = agent_id
        self.embedding_model = embedding_model
        self.vector_store = vector_store
        
        # 各类型记忆存储
        self.working_memory: List[MemoryItem] = []     # 工作记忆（有限容量）
        self.short_term_store: List[MemoryItem] = []    # 短期记忆
        self.long_term_store: List[MemoryItem] = []     # 长期记忆
        self.episodic_store: List[MemoryItem] = []      # 情景记忆
        
        self.working_memory_limit = 20  # 工作记忆容量限制
    
    def add_memory(self, content: str, memory_type: MemoryType,
                   importance: float = 0.5, tags: List[str] = None,
                   metadata: dict = None) -> MemoryItem:
        """添加新记忆"""
        memory_id = hashlib.md5(
            f"{self.agent_id}:{content}:{datetime.now().isoformat()}".encode()
        ).hexdigest()[:12]
        
        # 生成embedding
        embedding = None
        if self.embedding_model:
            embedding = self.embedding_model.encode(content).tolist()
        
        item = MemoryItem(
            id=memory_id,
            content=content,
            memory_type=memory_type,
            agent_id=self.agent_id,
            importance=importance,
            tags=tags or [],
            metadata=metadata or {},
            embedding=embedding
        )
        
        # 存入对应存储
        if memory_type == MemoryType.WORKING:
            self.working_memory.append(item)
            self._enforce_working_memory_limit()
        elif memory_type == MemoryType.SHORT_TERM:
            self.short_term_store.append(item)
        elif memory_type == MemoryType.LONG_TERM:
            self.long_term_store.append(item)
        elif memory_type == MemoryType.EPISODIC:
            self.episodic_store.append(item)
        
        # 如果有向量存储，同步索引
        if self.vector_store and embedding:
            self.vector_store.add(memory_id, embedding, item.to_dict())
        
        return item
    
    def _enforce_working_memory_limit(self):
        """强制工作记忆容量限制（遗忘机制）"""
        if len(self.working_memory) > self.working_memory_limit:
            # 按重要性和访问时间排序，移除最不重要的
            self.working_memory.sort(
                key=lambda x: (x.importance, x.last_accessed),
                reverse=True
            )
            # 将被移除的记忆降级为短期记忆
            evicted = self.working_memory[self.working_memory_limit:]
            self.working_memory = self.working_memory[:self.working_memory_limit]
            
            for item in evicted:
                item.memory_type = MemoryType.SHORT_TERM
                self.short_term_store.append(item)
    
    def retrieve(self, query: str, memory_types: List[MemoryType] = None,
                 top_k: int = 5, min_importance: float = 0.0) -> List[MemoryItem]:
        """检索相关记忆"""
        if memory_types is None:
            memory_types = [MemoryType.WORKING, MemoryType.SHORT_TERM, 
                           MemoryType.LONG_TERM, MemoryType.EPISODIC]
        
        candidates = []
        for mtype in memory_types:
            store = self._get_store(mtype)
            candidates.extend(store)
        
        # 过滤低重要性
        candidates = [c for c in candidates if c.importance >= min_importance]
        
        if self.embedding_model and self.vector_store:
            # 向量语义检索
            query_embedding = self.embedding_model.encode(query).tolist()
            results = self.vector_store.search(query_embedding, top_k=top_k)
            memory_ids = {r["id"] for r in results}
            retrieved = [c for c in candidates if c.id in memory_ids]
        else:
            # 关键词匹配回退
            retrieved = self._keyword_search(query, candidates, top_k)
        
        # 更新访问记录
        for item in retrieved:
            item.last_accessed = datetime.now()
            item.access_count += 1
        
        return retrieved
    
    def _get_store(self, memory_type: MemoryType) -> List[MemoryItem]:
        """获取对应类型的存储"""
        mapping = {
            MemoryType.WORKING: self.working_memory,
            MemoryType.SHORT_TERM: self.short_term_store,
            MemoryType.LONG_TERM: self.long_term_store,
            MemoryType.EPISODIC: self.episodic_store,
        }
        return mapping.get(memory_type, [])
    
    def _keyword_search(self, query: str, candidates: List[MemoryItem],
                        top_k: int) -> List[MemoryItem]:
        """关键词回退检索"""
        query_terms = set(query.lower().split())
        scored = []
        
        for item in candidates:
            content_terms = set(item.content.lower().split())
            overlap = len(query_terms & content_terms)
            score = overlap / max(len(query_terms), 1) * item.importance
            if score > 0:
                scored.append((item, score))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        return [item for item, _ in scored[:top_k]]
    
    def consolidate_memories(self):
        """
        记忆整合：将短期记忆中的重要内容提升为长期记忆
        类似人类睡眠时的记忆巩固过程
        """
        promoted = []
        kept = []
        
        for item in self.short_term_store:
            # 提升条件：高重要性或高访问频率
            if item.importance > 0.7 or item.access_count > 3:
                item.memory_type = MemoryType.LONG_TERM
                self.long_term_store.append(item)
                promoted.append(item)
            else:
                kept.append(item)
        
        self.short_term_store = kept
        return promoted
    
    def forget(self, memory_id: str):
        """主动遗忘特定记忆"""
        for store in [self.working_memory, self.short_term_store, 
                      self.long_term_store, self.episodic_store]:
            store[:] = [item for item in store if item.id != memory_id]
        
        if self.vector_store:
            self.vector_store.delete(memory_id)
    
    def get_context_summary(self, max_tokens: int = 2000) -> str:
        """生成当前记忆上下文摘要，用于注入到Agent的系统提示"""
        parts = []
        
        # 工作记忆
        if self.working_memory:
            parts.append("## 当前上下文")
            for item in self.working_memory[-5:]:
                parts.append(f"- {item.content[:200]}")
        
        # 相关长期记忆
        if self.long_term_store:
            important = sorted(self.long_term_store, 
                             key=lambda x: x.importance, reverse=True)[:3]
            parts.append("\n## 相关知识")
            for item in important:
                parts.append(f"- [{', '.join(item.tags)}] {item.content[:200]}")
        
        return "\n".join(parts)

第二章：共享记忆空间

2.1 为什么需要共享记忆

在多Agent系统中，每个Agent都有自己的私有记忆。但协作需要共享知识：

• 团队知识：项目背景、约定俗成的规则
• 任务状态：当前进展、已完成的子任务
• 共享发现：某个Agent发现的信息，其他Agent可能需要

2.2 共享记忆空间实现

from typing import Set
import threading

class SharedMemorySpace:
    """
    共享记忆空间 - 多Agent之间的公共知识库
    实现了读写锁、命名空间隔离、权限控制
    """
    
    def __init__(self, space_id: str):
        self.space_id = space_id
        self.memories: Dict[str, MemoryItem] = {}
        self.namespaces: Dict[str, Set[str]] = {}  # namespace -> memory_ids
        self.permissions: Dict[str, Dict[str, Set[str]]] = {}  # memory_id -> {readers, writers}
        self._lock = threading.RLock()
        
        # 事件通知
        self.subscribers: Dict[str, List[callable]] = {}  # event -> callbacks
    
    def publish(self, agent_id: str, content: str, namespace: str = "general",
                importance: float = 0.5, tags: List[str] = None,
                readable_by: Set[str] = None, writable_by: Set[str] = None) -> MemoryItem:
        """
        发布记忆到共享空间
        
        Args:
            agent_id: 发布者ID
            content: 记忆内容
            namespace: 命名空间（如 "project", "task", "knowledge"）
            importance: 重要性
            tags: 标签
            readable_by: 可读的Agent集合，None表示所有人可读
            writable_by: 可写的Agent集合，None表示仅创建者可写
        """
        with self._lock:
            memory_id = hashlib.md5(
                f"{self.space_id}:{namespace}:{content}".encode()
            ).hexdigest()[:12]
            
            item = MemoryItem(
                id=memory_id,
                content=content,
                memory_type=MemoryType.SEMANTIC,
                agent_id=agent_id,
                importance=importance,
                tags=tags or [],
                metadata={"namespace": namespace, "shared": True}
            )
            
            self.memories[memory_id] = item
            
            # 命名空间索引
            if namespace not in self.namespaces:
                self.namespaces[namespace] = set()
            self.namespaces[namespace].add(memory_id)
            
            # 权限设置
            self.permissions[memory_id] = {
                "readers": readable_by,  # None = 所有人
                "writers": writable_by or {agent_id}
            }
            
            # 通知订阅者
            self._notify("memory_published", {
                "memory_id": memory_id,
                "agent_id": agent_id,
                "namespace": namespace
            })
            
            return item
    
    def query(self, agent_id: str, query: str = None, 
              namespace: str = None, tags: List[str] = None,
              top_k: int = 10) -> List[MemoryItem]:
        """查询共享记忆"""
        with self._lock:
            candidates = []
            
            # 按命名空间过滤
            if namespace and namespace in self.namespaces:
                memory_ids = self.namespaces[namespace]
                candidates = [self.memories[mid] for mid in memory_ids 
                             if mid in self.memories]
            else:
                candidates = list(self.memories.values())
            
            # 权限过滤
            candidates = [
                item for item in candidates
                if self._can_read(item.id, agent_id)
            ]
            
            # 标签过滤
            if tags:
                candidates = [
                    item for item in candidates
                    if any(tag in item.tags for tag in tags)
                ]
            
            # 按查询相关性排序（简化版）
            if query:
                query_terms = set(query.lower().split())
                scored = []
                for item in candidates:
                    content_terms = set(item.content.lower().split())
                    relevance = len(query_terms & content_terms) / max(len(query_terms), 1)
                    score = relevance * 0.5 + item.importance * 0.5
                    scored.append((item, score))
                scored.sort(key=lambda x: x[1], reverse=True)
                candidates = [item for item, _ in scored[:top_k]]
            else:
                candidates.sort(key=lambda x: x.importance, reverse=True)
                candidates = candidates[:top_k]
            
            return candidates
    
    def update(self, agent_id: str, memory_id: str, 
               content: str = None, importance: float = None) -> bool:
        """更新共享记忆"""
        with self._lock:
            if memory_id not in self.memories:
                return False
            if not self._can_write(memory_id, agent_id):
                return False
            
            item = self.memories[memory_id]
            if content:
                item.content = content
            if importance is not None:
                item.importance = importance
            item.last_accessed = datetime.now()
            
            self._notify("memory_updated", {"memory_id": memory_id, "agent_id": agent_id})
            return True
    
    def subscribe(self, event: str, callback: callable):
        """订阅事件"""
        if event not in self.subscribers:
            self.subscribers[event] = []
        self.subscribers[event].append(callback)
    
    def _can_read(self, memory_id: str, agent_id: str) -> bool:
        perm = self.permissions.get(memory_id, {})
        readers = perm.get("readers")
        return readers is None or agent_id in readers
    
    def _can_write(self, memory_id: str, agent_id: str) -> bool:
        perm = self.permissions.get(memory_id, {})
        writers = perm.get("writers", set())
        return agent_id in writers
    
    def _notify(self, event: str, data: dict):
        for callback in self.subscribers.get(event, []):
            try:
                callback(data)
            except Exception:
                pass

第三章：记忆检索与遗忘机制

3.1 基于重要性的遗忘算法

import math
from datetime import datetime, timedelta

class ForgettingCurve:
    """
    基于艾宾浩斯遗忘曲线的记忆衰减模型
    结合重要性、访问频率、情感强度等因素
    """
    
    def __init__(self, decay_rate: float = 0.5):
        self.decay_rate = decay_rate
    
    def compute_strength(self, memory: MemoryItem, 
                         current_time: datetime = None) -> float:
        """
        计算记忆的当前强度（0-1）
        强度越高，越不容易被遗忘
        """
        if current_time is None:
            current_time = datetime.now()
        
        # 时间衰减（艾宾浩斯遗忘曲线）
        time_since_last = (current_time - memory.last_accessed).total_seconds() / 3600
        time_decay = math.exp(-self.decay_rate * time_since_last / 24)
        
        # 访问频率加成（间隔重复效应）
        frequency_boost = min(1.0, memory.access_count / 5) * 0.3
        
        # 重要性加成
        importance_boost = memory.importance * 0.3
        
        # 情感加成（情感强烈的记忆更持久）
        emotion_boost = abs(memory.emotion) * 0.1
        
        # 综合强度
        strength = time_decay * (0.6 + frequency_boost + importance_boost + emotion_boost)
        return max(0, min(1, strength))
    
    def should_forget(self, memory: MemoryItem, 
                      threshold: float = 0.1) -> bool:
        """判断是否应该遗忘"""
        return self.compute_strength(memory) < threshold


class MemoryConsolidator:
    """
    记忆整合器 - 定期运行，管理记忆生命周期
    """
    
    def __init__(self, memory_manager: MemoryManager):
        self.memory_manager = memory_manager
        self.forgetting_curve = ForgettingCurve()
    
    def run_consolidation(self) -> dict:
        """执行一次记忆整合"""
        stats = {
            "promoted": 0,
            "forgotten": 0,
            "compressed": 0
        }
        
        # 1. 短期记忆 -> 长期记忆（提升）
        promoted = self.memory_manager.consolidate_memories()
        stats["promoted"] = len(promoted)
        
        # 2. 衰减过期的记忆（遗忘）
        all_stores = [
            self.memory_manager.short_term_store,
            self.memory_manager.episodic_store
        ]
        
        for store in all_stores:
            to_forget = []
            for item in store:
                if self.forgetting_curve.should_forget(item):
                    to_forget.append(item)
            
            for item in to_forget:
                self.memory_manager.forget(item.id)
            stats["forgotten"] += len(to_forget)
        
        # 3. 压缩相似记忆
        stats["compressed"] = self._compress_similar_memories()
        
        return stats
    
    def _compress_similar_memories(self) -> int:
        """合并相似的短期记忆"""
        store = self.memory_manager.short_term_store
        if len(store) < 2:
            return 0
        
        compressed = 0
        merged_ids = set()
        
        for i in range(len(store)):
            if store[i].id in merged_ids:
                continue
            for j in range(i + 1, len(store)):
                if store[j].id in merged_ids:
                    continue
                
                # 简单相似度检测
                similarity = self._text_similarity(
                    store[i].content, store[j].content
                )
                if similarity > 0.8:
                    # 合并：保留更重要的，更新内容
                    primary, secondary = (store[i], store[j]) \
                        if store[i].importance >= store[j].importance \
                        else (store[j], store[i])
                    
                    primary.content = f"{primary.content}\n补充：{secondary.content}"
                    primary.importance = max(primary.importance, secondary.importance)
                    merged_ids.add(secondary.id)
                    compressed += 1
        
        # 移除被合并的记忆
        self.memory_manager.short_term_store = [
            item for item in store if item.id not in merged_ids
        ]
        
        return compressed
    
    def _text_similarity(self, text1: str, text2: str) -> float:
        """简单文本相似度"""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        if not words1 or not words2:
            return 0
        intersection = len(words1 & words2)
        union = len(words1 | words2)
        return intersection / union

第四章：多Agent通信协议

4.1 消息类型与通信模式

from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Awaitable
import asyncio
from datetime import datetime

class MessageType(Enum):
    """消息类型"""
    REQUEST = "request"         # 请求（需要回复）
    RESPONSE = "response"       # 响应
    BROADCAST = "broadcast"     # 广播（不需要回复）
    DELEGATE = "delegate"       # 任务委托
    REPORT = "report"           # 进度报告
    QUERY = "query"             # 知识查询
    SHARE = "share"             # 记忆共享
    HEARTBEAT = "heartbeat"     # 心跳

@dataclass
class AgentMessage:
    """Agent间通信消息"""
    id: str
    sender: str                 # 发送者Agent ID
    receiver: str               # 接收者Agent ID（"*"表示广播）
    msg_type: MessageType
    content: Any
    reply_to: Optional[str] = None  # 回复的目标消息ID
    timestamp: datetime = field(default_factory=datetime.now)
    priority: int = 5           # 优先级 1-10
    ttl: int = 300              # 消息存活时间（秒）
    metadata: Dict[str, Any] = field(default_factory=dict)

class MessageBus:
    """
    Agent通信总线 - 实现发布/订阅和点对点通信
    """
    
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}  # agent_id -> handlers
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.message_history: List[AgentMessage] = []
        self.pending_replies: Dict[str, asyncio.Future] = {}
        self._running = False
    
    def subscribe(self, agent_id: str, handler: Callable[[AgentMessage], Awaitable[None]]):
        """Agent订阅消息"""
        if agent_id not in self.subscribers:
            self.subscribers[agent_id] = []
        self.subscribers[agent_id].append(handler)
    
    async def send(self, message: AgentMessage):
        """发送消息"""
        self.message_history.append(message)
        
        if message.receiver == "*":
            # 广播
            for agent_id, handlers in self.subscribers.items():
                if agent_id != message.sender:
                    for handler in handlers:
                        asyncio.create_task(handler(message))
        else:
            # 点对点
            handlers = self.subscribers.get(message.receiver, [])
            for handler in handlers:
                asyncio.create_task(handler(message))
        
        # 如果是回复消息，唤醒等待的Future
        if message.reply_to and message.reply_to in self.pending_replies:
            self.pending_replies[message.reply_to].set_result(message)
    
    async def request(self, sender: str, receiver: str, content: Any,
                       timeout: float = 30) -> AgentMessage:
        """发送请求并等待回复（RPC模式）"""
        msg_id = hashlib.md5(f"{sender}:{receiver}:{datetime.now()}".encode()).hexdigest()[:8]
        
        message = AgentMessage(
            id=msg_id,
            sender=sender,
            receiver=receiver,
            msg_type=MessageType.REQUEST,
            content=content
        )
        
        # 创建Future等待回复
        future = asyncio.get_event_loop().create_future()
        self.pending_replies[msg_id] = future
        
        await self.send(message)
        
        try:
            response = await asyncio.wait_for(future, timeout=timeout)
            return response
        except asyncio.TimeoutError:
            del self.pending_replies[msg_id]
            raise TimeoutError(f"Agent {receiver} 未在 {timeout}s 内回复")
    
    def get_history(self, agent_id: str = None, 
                    msg_type: MessageType = None,
                    limit: int = 50) -> List[AgentMessage]:
        """获取消息历史"""
        filtered = self.message_history
        
        if agent_id:
            filtered = [m for m in filtered 
                       if m.sender == agent_id or m.receiver == agent_id]
        if msg_type:
            filtered = [m for m in filtered if m.msg_type == msg_type]
        
        return filtered[-limit:]

第五章：Agent角色定义与能力边界

5.1 Agent基类设计

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional

class AgentCapability(Enum):
    """Agent能力类型"""
    SEARCH = "search"           # 信息检索
    ANALYSIS = "analysis"       # 数据分析
    CODE = "code"               # 代码生成
    WRITING = "writing"         # 文本创作
    REVIEW = "review"           # 审核校对
    PLANNING = "planning"       # 规划决策
    TOOL_USE = "tool_use"       # 工具调用

@dataclass
class AgentProfile:
    """Agent角色配置"""
    agent_id: str
    name: str
    description: str
    capabilities: List[AgentCapability]
    system_prompt: str
    model: str = "gpt-4o"
    max_iterations: int = 10
    memory_config: dict = field(default_factory=dict)

class BaseAgent(ABC):
    """
    Agent基类 - 所有Agent的抽象父类
    """
    
    def __init__(self, profile: AgentProfile, 
                 memory_manager: MemoryManager,
                 message_bus: MessageBus,
                 shared_memory: SharedMemorySpace = None):
        self.profile = profile
        self.memory = memory_manager
        self.message_bus = message_bus
        self.shared_memory = shared_memory
        
        self.state = "idle"  # idle, working, waiting, error
        self.current_task: Optional[str] = None
        self.task_history: List[dict] = []
        
        # 注册消息处理器
        self.message_bus.subscribe(profile.agent_id, self._handle_message)
    
    @abstractmethod
    async def execute(self, task: str, context: dict = None) -> dict:
        """执行任务的核心方法（子类实现）"""
        pass
    
    async def _handle_message(self, message: AgentMessage):
        """处理接收到的消息"""
        if message.msg_type == MessageType.REQUEST:
            result = await self.execute(message.content)
            reply = AgentMessage(
                id=f"reply-{message.id}",
                sender=self.profile.agent_id,
                receiver=message.sender,
                msg_type=MessageType.RESPONSE,
                content=result,
                reply_to=message.id
            )
            await self.message_bus.send(reply)
        
        elif message.msg_type == MessageType.SHARE:
            # 存入共享记忆
            if self.shared_memory:
                self.shared_memory.publish(
                    agent_id=message.sender,
                    content=message.content,
                    namespace="shared"
                )
        
        elif message.msg_type == MessageType.DELEGATE:
            # 接受任务委托
            asyncio.create_task(self._accept_delegation(message))
    
    async def _accept_delegation(self, message: AgentMessage):
        """接受并执行委托的任务"""
        task = message.content
        self.state = "working"
        self.current_task = task.get("description", "")
        
        try:
            result = await self.execute(
                task["description"], 
                context=task.get("context")
            )
            
            # 上报结果
            report = AgentMessage(
                id=f"report-{message.id}",
                sender=self.profile.agent_id,
                receiver=message.sender,
                msg_type=MessageType.REPORT,
                content={"status": "completed", "result": result},
                reply_to=message.id
            )
            await self.message_bus.send(report)
        except Exception as e:
            report = AgentMessage(
                id=f"error-{message.id}",
                sender=self.profile.agent_id,
                receiver=message.sender,
                msg_type=MessageType.REPORT,
                content={"status": "error", "error": str(e)},
                reply_to=message.id
            )
            await self.message_bus.send(report)
        finally:
            self.state = "idle"
            self.current_task = None
    
    def remember(self, content: str, memory_type: MemoryType = MemoryType.EPISODIC,
                 importance: float = 0.5, tags: List[str] = None):
        """记录记忆"""
        self.memory.add_memory(
            content=content,
            memory_type=memory_type,
            importance=importance,
            tags=tags or []
        )
    
    def recall(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        """回忆相关记忆"""
        return self.memory.retrieve(query, top_k=top_k)
    
    def get_context(self) -> str:
        """获取当前上下文（用于注入到prompt）"""
        memory_context = self.memory.get_context_summary()
        shared_context = ""
        
        if self.shared_memory:
            shared_items = self.shared_memory.query(
                self.profile.agent_id, top_k=3
            )
            if shared_items:
                shared_context = "\n## 团队共享知识\n" + "\n".join(
                    f"- {item.content[:150]}" for item in shared_items
                )
        
        return f"{memory_context}\n{shared_context}"

5.2 具体Agent实现示例

class ResearchAgent(BaseAgent):
    """研究Agent - 专门负责信息检索和调研"""
    
    def __init__(self, profile, memory, message_bus, shared_memory, 
                 search_engine=None):
        super().__init__(profile, memory, message_bus, shared_memory)
        self.search_engine = search_engine
    
    async def execute(self, task: str, context: dict = None) -> dict:
        """执行研究任务"""
        self.remember(f"开始研究任务: {task}", MemoryType.EPISODIC, importance=0.6)
        
        # 回忆相关知识
        relevant_memories = self.recall(task)
        existing_knowledge = "\n".join([m.content for m in relevant_memories])
        
        # 执行搜索
        search_results = []
        if self.search_engine:
            search_results = await self.search_engine.search(task)
        
        # 综合分析
        summary = await self._synthesize(task, search_results, existing_knowledge)
        
        # 存储研究成果
        self.remember(
            f"研究结论 [{task}]: {summary[:500]}",
            MemoryType.LONG_TERM,
            importance=0.8,
            tags=["research", "conclusion"]
        )
        
        # 共享给团队
        if self.shared_memory:
            self.shared_memory.publish(
                agent_id=self.profile.agent_id,
                content=f"研究发现 - {task}: {summary[:300]}",
                namespace="research",
                importance=0.7,
                tags=["research"]
            )
        
        return {"task": task, "summary": summary, "sources": len(search_results)}
    
    async def _synthesize(self, task, results, existing_knowledge) -> str:
        """综合分析（实际实现会调用LLM）"""
        # 简化示例
        return f"关于'{task}'的研究综合报告..."

第六章：任务分解与依赖管理

6.1 任务图（Task DAG）

from enum import Enum
from typing import Set
import asyncio

class TaskStatus(Enum):
    PENDING = "pending"
    READY = "ready"        # 依赖已满足
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class Task:
    task_id: str
    description: str
    assigned_agent: str    # 分配的Agent ID
    dependencies: Set[str] = field(default_factory=set)  # 依赖的任务ID
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: Optional[str] = None
    priority: int = 5
    created_at: datetime = field(default_factory=datetime.now)
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None

class TaskGraph:
    """
    有向无环图（DAG）任务管理器
    支持任务依赖、并行执行、故障恢复
    """
    
    def __init__(self, message_bus: MessageBus):
        self.tasks: Dict[str, Task] = {}
        self.message_bus = message_bus
        self.completion_events: Dict[str, asyncio.Event] = {}
    
    def add_task(self, task_id: str, description: str, 
                 assigned_agent: str, dependencies: Set[str] = None,
                 priority: int = 5) -> Task:
        """添加任务到图中"""
        task = Task(
            task_id=task_id,
            description=description,
            assigned_agent=assigned_agent,
            dependencies=dependencies or set(),
            priority=priority
        )
        self.tasks[task_id] = task
        self.completion_events[task_id] = asyncio.Event()
        return task
    
    def get_ready_tasks(self) -> List[Task]:
        """获取所有依赖已满足的待执行任务"""
        ready = []
        for task in self.tasks.values():
            if task.status != TaskStatus.PENDING:
                continue
            
            # 检查所有依赖是否完成
            deps_met = all(
                self.tasks[dep].status == TaskStatus.COMPLETED
                for dep in task.dependencies
                if dep in self.tasks
            )
            
            if deps_met:
                task.status = TaskStatus.READY
                ready.append(task)
        
        # 按优先级排序
        ready.sort(key=lambda t: t.priority, reverse=True)
        return ready
    
    async def execute_graph(self, max_parallel: int = 5) -> Dict[str, Any]:
        """
        执行整个任务图
        支持并行执行和依赖等待
        """
        semaphore = asyncio.Semaphore(max_parallel)
        results = {}
        
        async def run_task(task: Task):
            async with semaphore:
                task.status = TaskStatus.RUNNING
                task.started_at = datetime.now()
                
                try:
                    # 收集依赖结果
                    dep_results = {}
                    for dep_id in task.dependencies:
                        if dep_id in self.tasks:
                            dep_task = self.tasks[dep_id]
                            if dep_task.status == TaskStatus.COMPLETED:
                                dep_results[dep_id] = dep_task.result
                    
                    # 发送任务委托消息
                    response = await self.message_bus.request(
                        sender="orchestrator",
                        receiver=task.assigned_agent,
                        content={
                            "description": task.description,
                            "dependencies": dep_results
                        },
                        timeout=300
                    )
                    
                    task.result = response.content
                    task.status = TaskStatus.COMPLETED
                    task.completed_at = datetime.now()
                    
                    # 触发完成事件
                    self.completion_events[task.task_id].set()
                    
                except Exception as e:
                    task.status = TaskStatus.FAILED
                    task.error = str(e)
                    self.completion_events[task.task_id].set()
        
        # 循环执行直到所有任务完成
        while True:
            ready = self.get_ready_tasks()
            if not ready:
                # 检查是否全部完成
                all_done = all(
                    t.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED)
                    for t in self.tasks.values()
                )
                if all_done:
                    break
                
                # 等待正在运行的任务完成
                running = [t for t in self.tasks.values() if t.status == TaskStatus.RUNNING]
                if running:
                    await asyncio.gather(*[
                        self.completion_events[t.task_id].wait() 
                        for t in running
                    ])
                else:
                    break  # 死锁检测
            else:
                # 并行启动就绪任务
                await asyncio.gather(*[run_task(task) for task in ready])
        
        # 收集结果
        for task_id, task in self.tasks.items():
            results[task_id] = {
                "status": task.status.value,
                "result": task.result,
                "error": task.error,
                "duration": str(task.completed_at - task.started_at) 
                    if task.completed_at and task.started_at else None
            }
        
        return results
    
    def visualize(self) -> str:
        """生成任务图的文本可视化"""
        lines = ["任务依赖图:"]
        for task in self.tasks.values():
            status_icon = {
                TaskStatus.PENDING: "⏳",
                TaskStatus.READY: "🟢",
                TaskStatus.RUNNING: "🔄",
                TaskStatus.COMPLETED: "✅",
                TaskStatus.FAILED: "❌",
                TaskStatus.CANCELLED: "⛔"
            }.get(task.status, "❓")
            
            deps = ", ".join(task.dependencies) if task.dependencies else "无"
            lines.append(f"  {status_icon} [{task.task_id}] {task.description}")
            lines.append(f"     分配给: {task.assigned_agent} | 依赖: {deps}")
        
        return "\n".join(lines)

第七章：协作工作流编排

7.1 基于LangGraph的工作流

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

# 定义工作流状态
class WorkflowState(TypedDict):
    task: str
    research_results: Annotated[list, operator.add]
    analysis: str
    code_output: str
    review_feedback: str
    final_output: str
    iteration: int

def create_research_workflow():
    """
    创建研究分析工作流
    流程：研究 → 分析 → 代码生成 → 审核 → 输出
    """
    
    graph = StateGraph(WorkflowState)
    
    # 定义节点
    async def research_node(state: WorkflowState) -> dict:
        """研究节点：收集信息"""
        task = state["task"]
        # 调用研究Agent
        result = f"关于'{task}'的研究结果..."
        return {"research_results": [result]}
    
    async def analysis_node(state: WorkflowState) -> dict:
        """分析节点：深度分析研究结果"""
        research = state["research_results"]
        analysis = f"基于研究的深度分析..."
        return {"analysis": analysis}
    
    async def code_node(state: WorkflowState) -> dict:
        """代码节点：根据分析生成代码"""
        analysis = state["analysis"]
        code = f"# 基于分析生成的代码\nprint('Hello')"
        return {"code_output": code}
    
    async def review_node(state: WorkflowState) -> dict:
        """审核节点：检查质量和正确性"""
        code = state["code_output"]
        analysis = state["analysis"]
        feedback = "审核通过，质量良好。"
        return {"review_feedback": feedback}
    
    async def final_node(state: WorkflowState) -> dict:
        """最终输出节点"""
        return {"final_output": f"分析报告\n{state['analysis']}\n\n代码\n{state['code_output']}"}
    
    # 条件路由
    def should_continue(state: WorkflowState) -> str:
        if "需要修改" in state.get("review_feedback", ""):
            if state["iteration"] < 3:
                return "code"  # 返回代码节点重新生成
        return "final"
    
    # 构建图
    graph.add_node("research", research_node)
    graph.add_node("analysis", analysis_node)
    graph.add_node("code", code_node)
    graph.add_node("review", review_node)
    graph.add_node("final", final_node)
    
    # 定义边
    graph.set_entry_point("research")
    graph.add_edge("research", "analysis")
    graph.add_edge("analysis", "code")
    graph.add_edge("code", "review")
    graph.add_conditional_edges("review", should_continue, {
        "code": "code",
        "final": "final"
    })
    graph.add_edge("final", END)
    
    return graph.compile()

7.2 基于CrewAI的协作模式

class CrewOrchestrator:
    """
    多Agent团队编排器
    类似CrewAI的协作模式实现
    """
    
    def __init__(self, agents: Dict[str, BaseAgent], message_bus: MessageBus):
        self.agents = agents
        self.message_bus = message_bus
    
    async def run_sequential(self, tasks: List[dict]) -> List[dict]:
        """顺序执行模式：Agent依次完成任务"""
        results = []
        context = {}
        
        for task_config in tasks:
            agent_id = task_config["agent"]
            description = task_config["description"]
            
            agent = self.agents[agent_id]
            result = await agent.execute(description, context)
            results.append(result)
            
            # 将结果传递给下一个任务作为上下文
            context["previous_result"] = result
        
        return results
    
    async def run_parallel(self, tasks: List[dict]) -> List[dict]:
        """并行执行模式：多个Agent同时工作"""
        async def run_one(task_config):
            agent_id = task_config["agent"]
            agent = self.agents[agent_id]
            return await agent.execute(task_config["description"])
        
        results = await asyncio.gather(*[run_one(t) for t in tasks])
        return results
    
    async def run_debate(self, topic: str, agent_ids: List[str],
                         rounds: int = 3) -> List[dict]:
        """
        辩论模式：多个Agent就一个话题展开多轮辩论
        适合需要多角度分析的复杂问题
        """
        debate_history = []
        
        for round_num in range(rounds):
            round_results = []
            
            for agent_id in agent_ids:
                agent = self.agents[agent_id]
                
                # 构建辩论上下文
                debate_context = f"话题：{topic}\n\n"
                debate_context += f"第{round_num + 1}轮辩论\n"
                
                if debate_history:
                    debate_context += "\n之前的论点：\n"
                    for entry in debate_history:
                        debate_context += f"- {entry['agent']}: {entry['argument'][:200]}\n"
                
                debate_context += f"\n请从你的专业角度发表看法。"
                
                result = await agent.execute(debate_context)
                round_results.append({
                    "agent": agent_id,
                    "round": round_num + 1,
                    "argument": result.get("result", str(result))
                })
            
            debate_history.extend(round_results)
        
        return debate_history
    
    async def run_hierarchy(self, leader_id: str, task: str) -> dict:
        """
        层级模式：一个Leader Agent分解任务并分配给下属Agent
        """
        leader = self.agents[leader_id]
        
        # Leader分解任务
        plan = await leader.execute(
            f"将以下任务分解为子任务，分配给团队成员：\n任务：{task}\n"
            f"可用的团队成员：{list(self.agents.keys())}\n"
            f"返回JSON格式：{{\"subtasks\": [{{\"agent\": \"agent_id\", \"description\": \"子任务描述\"}}]}}"
        )
        
        # 执行子任务
        subtasks = plan.get("subtasks", [])
        results = {}
        
        for subtask in subtasks:
            agent_id = subtask["agent"]
            if agent_id in self.agents and agent_id != leader_id:
                agent = self.agents[agent_id]
                result = await agent.execute(subtask["description"])
                results[agent_id] = result
        
        # Leader汇总
        summary = await leader.execute(
            f"汇总以下子任务结果：\n{json.dumps(results, ensure_ascii=False, indent=2)}"
        )
        
        return {"plan": plan, "subtask_results": results, "summary": summary}

第八章：状态持久化与恢复

8.1 系统状态快照

import json
import pickle
from pathlib import Path

class StatePersistence:
    """
    系统状态持久化
    支持快照保存与恢复
    """
    
    def __init__(self, storage_dir: str = "./agent_state"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(parents=True, exist_ok=True)
    
    def save_snapshot(self, snapshot_id: str, 
                      agents: Dict[str, BaseAgent],
                      task_graph: TaskGraph = None,
                      shared_memory: SharedMemorySpace = None):
        """保存系统状态快照"""
        snapshot = {
            "snapshot_id": snapshot_id,
            "timestamp": datetime.now().isoformat(),
            "agents": {},
            "task_graph": None,
            "shared_memory": None
        }
        
        # 保存各Agent状态
        for agent_id, agent in agents.items():
            snapshot["agents"][agent_id] = {
                "state": agent.state,
                "current_task": agent.current_task,
                "memory": {
                    "working": [m.to_dict() for m in agent.memory.working_memory],
                    "short_term": [m.to_dict() for m in agent.memory.short_term_store],
                    "long_term": [m.to_dict() for m in agent.memory.long_term_store],
                    "episodic": [m.to_dict() for m in agent.memory.episodic_store],
                }
            }
        
        # 保存任务图
        if task_graph:
            snapshot["task_graph"] = {
                task_id: {
                    "status": task.status.value,
                    "result": task.result,
                    "description": task.description
                }
                for task_id, task in task_graph.tasks.items()
            }
        
        # 保存共享记忆
        if shared_memory:
            snapshot["shared_memory"] = {
                "memories": {
                    mid: item.to_dict() 
                    for mid, item in shared_memory.memories.items()
                },
                "namespaces": {
                    ns: list(mids) 
                    for ns, mids in shared_memory.namespaces.items()
                }
            }
        
        # 写入文件
        filepath = self.storage_dir / f"{snapshot_id}.json"
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(snapshot, f, ensure_ascii=False, indent=2)
        
        return str(filepath)
    
    def load_snapshot(self, snapshot_id: str) -> dict:
        """加载状态快照"""
        filepath = self.storage_dir / f"{snapshot_id}.json"
        if not filepath.exists():
            raise FileNotFoundError(f"快照 {snapshot_id} 不存在")
        
        with open(filepath, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def list_snapshots(self) -> List[dict]:
        """列出所有快照"""
        snapshots = []
        for filepath in self.storage_dir.glob("*.json"):
            with open(filepath, 'r') as f:
                data = json.load(f)
                snapshots.append({
                    "id": data["snapshot_id"],
                    "timestamp": data["timestamp"],
                    "agents": list(data.get("agents", {}).keys())
                })
        return sorted(snapshots, key=lambda x: x["timestamp"], reverse=True)

第九章：人机协作交互

9.1 Human-in-the-Loop机制

class HumanInTheLoop:
    """
    人机协作控制器
    在关键决策点引入人类审核
    """
    
    def __init__(self, message_bus: MessageBus):
        self.message_bus = message_bus
        self.pending_approvals: Dict[str, dict] = {}
        self.approval_callbacks: Dict[str, asyncio.Future] = {}
    
    async def request_approval(self, agent_id: str, action: str,
                                context: dict, options: List[str] = None,
                                timeout: float = 300) -> dict:
        """
        请求人类审批
        
        Args:
            agent_id: 请求审批的Agent
            action: 需要审批的操作描述
            context: 相关上下文
            options: 可选操作列表
            timeout: 超时时间
        """
        approval_id = hashlib.md5(
            f"{agent_id}:{action}:{datetime.now()}".encode()
        ).hexdigest()[:8]
        
        self.pending_approvals[approval_id] = {
            "agent_id": agent_id,
            "action": action,
            "context": context,
            "options": options or ["approve", "reject", "modify"],
            "created_at": datetime.now()
        }
        
        # 创建Future等待人类响应
        future = asyncio.get_event_loop().create_future()
        self.approval_callbacks[approval_id] = future
        
        # 通知人类（通过消息总线或UI）
        await self.message_bus.send(AgentMessage(
            id=f"approval-{approval_id}",
            sender=agent_id,
            receiver="human",
            msg_type=MessageType.REQUEST,
            content={
                "type": "approval_request",
                "approval_id": approval_id,
                "action": action,
                "context": context,
                "options": options
            }
        ))
        
        try:
            response = await asyncio.wait_for(future, timeout=timeout)
            return response
        except asyncio.TimeoutError:
            del self.pending_approvals[approval_id]
            del self.approval_callbacks[approval_id]
            return {"decision": "timeout", "message": "审批超时，操作已取消"}
    
    def submit_approval(self, approval_id: str, decision: str, 
                        feedback: str = None):
        """提交人类审批结果"""
        if approval_id in self.approval_callbacks:
            result = {
                "decision": decision,
                "feedback": feedback,
                "timestamp": datetime.now().isoformat()
            }
            self.approval_callbacks[approval_id].set_result(result)
            
            # 清理
            del self.pending_approvals[approval_id]
            del self.approval_callbacks[approval_id]
    
    def get_pending_approvals(self) -> List[dict]:
        """获取所有待审批项"""
        return [
            {**info, "id": aid}
            for aid, info in self.pending_approvals.items()
        ]

第十章：完整系统集成示例

async def build_multi_agent_system():
    """
    构建完整的多Agent协作系统
    """
    # 1. 创建基础设施
    message_bus = MessageBus()
    shared_memory = SharedMemorySpace("main-project")
    
    # 2. 定义Agent角色
    profiles = {
        "researcher": AgentProfile(
            agent_id="researcher",
            name="研究员",
            description="负责信息检索和调研",
            capabilities=[AgentCapability.SEARCH],
            system_prompt="你是一个专业的研究员，擅长信息检索和综合分析。"
        ),
        "analyst": AgentProfile(
            agent_id="analyst",
            name="分析师",
            description="负责数据分析和洞察",
            capabilities=[AgentCapability.ANALYSIS],
            system_prompt="你是一个数据分析师，擅长从数据中发现模式和洞察。"
        ),
        "coder": AgentProfile(
            agent_id="coder",
            name="程序员",
            description="负责代码生成和实现",
            capabilities=[AgentCapability.CODE],
            system_prompt="你是一个资深程序员，擅长编写高质量代码。"
        ),
        "reviewer": AgentProfile(
            agent_id="reviewer",
            name="审核员",
            description="负责质量审核和反馈",
            capabilities=[AgentCapability.REVIEW],
            system_prompt="你是一个严格的质量审核员，负责确保输出质量。"
        )
    }
    
    # 3. 创建Agent实例
    agents = {}
    for agent_id, profile in profiles.items():
        memory = MemoryManager(agent_id)
        agents[agent_id] = ResearchAgent(
            profile=profile,
            memory=memory,
            message_bus=message_bus,
            shared_memory=shared_memory
        )
    
    # 4. 创建编排器
    orchestrator = CrewOrchestrator(agents, message_bus)
    
    # 5. 创建人机协作控制器
    human_loop = HumanInTheLoop(message_bus)
    
    # 6. 创建任务图
    task_graph = TaskGraph(message_bus)
    task_graph.add_task("research", "调研AI搜索引擎最新进展", "researcher")
    task_graph.add_task("analyze", "分析技术趋势", "analyst", {"research"})
    task_graph.add_task("implement", "实现原型代码", "coder", {"analyze"})
    task_graph.add_task("review", "审核代码质量", "reviewer", {"implement"})
    
    # 7. 执行
    results = await task_graph.execute_graph(max_parallel=3)
    
    # 8. 保存状态
    persistence = StatePersistence()
    persistence.save_snapshot("v1", agents, task_graph, shared_memory)
    
    return results


# 运行
if __name__ == "__main__":
    results = asyncio.run(build_multi_agent_system())
    print(json.dumps(results, ensure_ascii=False, indent=2))

总结

本教程系统讲解了多Agent记忆与协作系统的完整技术栈：

核心设计原则：

1. 记忆是Agent的灵魂——好的记忆系统让Agent能学习、能积累、能协作
2. 通信是协作的基础——清晰的消息协议让Agent间高效协同
3. 编排是复杂任务的关键——DAG任务图和多种协作模式应对不同场景
4. 人机协作不可少——关键决策引入人类审核，确保安全可控

下一步学习建议：

• 尝试用LangGraph实现更复杂的多Agent工作流
• 探索Agent自我反思与自我优化机制
• 研究多Agent系统的可观测性与调试
• 学习分布式Agent系统（跨机器协作）

本教程内容约5000字，涵盖多Agent记忆与协作系统的核心架构与实战代码。希望对你的多Agent项目有所帮助！