QwenPaw源码解析系列(三):记忆系统设计与实现-夜雨聆风

QwenPaw源码解析系列(三):记忆系统设计与实现

前言

记忆系统是AI助手能够”越用越聪明”的关键。QwenPaw实现了分层记忆架构，支持长期记忆、情境记忆和主动记忆。今天让我们深入解析这套记忆系统的设计与实现。

记忆系统的分层架构

QwenPaw的记忆系统分为三层：


┌─────────────────────────────────────┐
│         长期记忆 (Long-term)         │
│    BaseMemoryManager 实现           │
├─────────────────────────────────────┤
│         情境记忆 (Context)           │
│    BaseContextManager 实现          │
├─────────────────────────────────────┤
│         短期记忆 (Working)           │
│    InMemoryMemory                   │
└─────────────────────────────────────┘

各层职责

层级	特点	实现
短期记忆	当前会话，容量有限	InMemoryMemory
情境记忆	会话级上下文，自动压缩	BaseContextManager
长期记忆	持久化，支持检索和总结	BaseMemoryManager

BaseMemoryManager：抽象基类

核心接口


class BaseMemoryManager(ABC):
    """记忆管理器抽象基类"""

    def __init__(self, working_dir: str, agent_id: str):
        self.working_dir: str = working_dir
        self.agent_id: str = agent_id
        self._task_queue: asyncio.Queue = asyncio.Queue()
        self._worker_task: asyncio.Task | None = None

    @abstractmethod
    async def start(self) -> None:
        """初始化存储后端"""

    @abstractmethod
    async def close(self) -> bool:
        """刷新状态并释放资源"""

    @abstractmethod
    def get_memory_prompt(self, language: str = "zh") -> str:
        """返回记忆引导提示词"""

    @abstractmethod
    def list_memory_tools(self) -> list[Callable[..., ToolResponse]]:
        """返回暴露给智能体的记忆工具"""

    async def summarize(self, messages: list[Msg], **kwargs) -> str:
        """总结对话消息并持久化（可选实现）"""
        return ""

    async def retrieve(
        self,
        messages: list[Msg] | Msg,
        **kwargs,
    ) -> dict | None:
        """基于消息检索相关记忆（可选实现）"""
        return None

    async def dream(self, **kwargs) -> None:
        """通过后台智能体优化记忆文件（可选实现）"""
        return None

后台任务处理

记忆管理器使用异步队列处理后台任务：


async def _summarize_worker(self) -> None:
    """后台工作器，串行处理总结任务"""
    while True:
        task_id, messages, kwargs = await self._task_queue.get()
        info = self._summary_task_info.get(task_id)

        info["status"] = "running"
        try:
            result = await self.summarize(messages=messages, **kwargs)
            info["status"] = "completed"
            info["result"] = result
        except Exception as e:
            info["status"] = "failed"
            info["error"] = str(e)

def add_summarize_task(self, messages: list[Msg], **kwargs):
    """调度后台总结任务（非阻塞）"""
    if self._worker_task is None or self._worker_task.done():
        self._worker_task = asyncio.create_task(self._summarize_worker())

    self._task_counter += 1
    task_id = f"task_{self._task_counter}"

    self._summary_task_info[task_id] = {
        "task_id": task_id,
        "start_time": datetime.now(),
        "status": "pending",
        "result": None,
        "error": None,
    }

    self._task_queue.put_nowait((task_id, messages, kwargs))

RemeLightMemoryManager：轻量级实现

RemeLightMemoryManager是QwenPaw的默认记忆管理器实现：


class RemeLightMemoryManager(BaseMemoryManager):
    """轻量级记忆管理器，使用本地文件存储"""

工作原理


用户交互 → 对话总结 → 存储记忆 → 检索匹配 → 上下文注入

核心流程

对话结束触发总结

•

当对话轮次达到阈值时触发

•

后台异步执行，不阻塞当前交互

记忆存储格式


{
  "timestamp": "2026-04-28T10:00:00",
  "summary": "用户询问了关于Python异步编程的问题",
  "key_points": ["async/await语法", "事件循环", "协程"],
  "context": {...}
}

语义检索


async def retrieve(self, messages, **kwargs) -> dict | None:
    # 构建检索向量
    # 与历史记忆进行相似度匹配
    # 返回相关记忆片段

AgentContext：情境记忆

AgentContext继承自InMemoryMemory，提供了更强大的内存管理能力：


class AgentContext(InMemoryMemory):
    """带有token感知的情境记忆实现"""

关键特性

Token计数


def count_tokens(self) -> int:
    """计算当前上下文使用的token数量"""

消息标记


_MemoryMark = {
    HINT = "hint",        # 提示信息
    SYSTEM = "system",   # 系统消息
    COMPACTED = "compacted",  # 已压缩
}

动态修剪


def prune_old_messages(self, max_tokens: int) -> None:
    """当token超限时自动修剪旧消息"""

BaseContextManager：上下文管理器

上下文管理器负责管理会话级上下文，包括压缩和修剪：


class BaseContextManager(ABC):
    """上下文管理器抽象基类"""

生命周期钩子


@abstractmethod
async def pre_reasoning(
    self,
    agent: Any,
    kwargs: dict[str, Any],
) -> dict[str, Any] | None:
    """推理前钩子：检查上下文健康状态"""

@abstractmethod
async def post_acting(
    self,
    agent: Any,
    kwargs: dict[str, Any],
    output: Any,
) -> Msg | None:
    """行动后钩子：修剪过长的工具输出"""

@abstractmethod
async def compact_context(
    self,
    messages: list[Msg],
    previous_summary: str = "",
    extra_instruction: str = "",
) -> dict:
    """压缩消息为摘要"""

上下文压实流程


┌──────────────┐
│ 检测Token上限 │
└──────┬───────┘
       ↓
┌──────────────┐
│  选择待压缩段 │
└──────┬───────┘
       ↓
┌──────────────┐     ┌──────────────┐
│  调用LLM压缩  │ ──▶ │  生成摘要    │
└──────┬───────┘     └──────────────┘
       ↓
┌──────────────┐
│  替换原消息   │
└──────┬───────┘
       ↓
┌──────────────┐
│  更新Token数 │
└──────────────┘

工具结果修剪

工具输出的修剪策略：


class ToolResultPruningConfig:
    pruning_recent_msg_max_bytes: int = 20000  # 最近消息最大字节数
    pruning_tool_result_max_bytes: int = 8000   # 工具结果最大字节数

Proactive：主动交互系统

Proactive模块让智能体能够主动发起交互，而非被动响应：


┌─────────────────────────────────────────┐
│             Proactive 模块              │
├─────────────────────────────────────────┤
│  ProactiveTrigger: 触发条件检测         │
│  ProactiveResponder: 响应策略生成        │
│  ProactivePrompts: 提示词模板           │
└─────────────────────────────────────────┘

触发类型

时间触发：定时任务、周期性检查

事件触发：新消息、外部信号

条件触发：满足特定条件时

响应策略


async def proactive_respond(self, trigger: Trigger) -> Msg | None:
    """根据触发器生成主动响应"""
    # 1. 评估触发优先级
    # 2. 生成响应策略
    # 3. 执行动作
    # 4. 记录结果

记忆进化：Dream功能

“梦境”功能通过后台智能体优化记忆文件：


async def dream(self, **kwargs) -> None:
    """通过轻量级智能体优化记忆"""
    # 1. 加载历史记忆
    # 2. 识别冗余和过时内容
    # 3. 生成优化后的记忆
    # 4. 替换旧文件

优化策略：

•

合并相似记忆

•

删除过时信息

•

提炼关键知识点

•

增强关联性

Registry模式：可插拔架构


memory_registry: Registry[BaseMemoryManager] = Registry()

def get_memory_manager_backend(backend: str) -> type[BaseMemoryManager]:
    """根据后端名称获取记忆管理器类"""
    cls = memory_registry.get(backend)
    if cls is None:
        # 回退到第一个注册的类
        registered = memory_registry.list_registered()
        fallback = registered[0]
        cls = memory_registry.get(fallback)
    return cls

注册示例：


@memory_registry.register("reme_light")
class RemeLightMemoryManager(BaseMemoryManager):
    ...

总结

QwenPaw的记忆系统设计体现了以下核心原则：

分层管理：短期、情境、长期记忆各有分工

异步非阻塞：后台任务不干扰主流程

可插拔：通过Registry模式支持多种后端

主动进化：通过Dream功能持续优化记忆

Token感知：精确控制上下文长度

通过这套记忆系统，QwenPaw能够记住用户偏好、积累知识，并在后续对话中主动提供有价值的信息。

往期回顾：

•

系列一：项目概览与整体架构

•

系列二：QwenPawAgent核心实现

下期预告：

•

Skills系统：技能加载与执行机制

•

MCP协议：多智能体通信

•

安全防护：ToolGuard与命令守卫

如果对你有帮助，欢迎点赞、在看！