智能体核心技术七大模块之（四）

记忆与上下文管理——构建智能体的长期与短期记忆

本文是七大模块的第四部分，深入探讨智能体的记忆与上下文管理模块。记忆是智能体持续学习和个性化交互的基础，它使智能体能够记住用户偏好、历史对话、中间结果，并在需要时检索相关信息。我们将通过UML建模、详细设计和完整代码实现，构建一个分层、可扩展的记忆系统，支持短期对话记忆、长期知识存储以及语义检索。

---

1. 核心概念与设计目标

1.1 什么是记忆模块

记忆模块负责存储和管理智能体与用户交互过程中的所有信息，包括：

• 短期记忆（工作记忆）：当前对话轮次的消息、临时中间结果。
• 长期记忆（情景记忆/语义记忆）：用户偏好、个人信息、历史重要事件、学习到的知识。
• 记忆检索：根据当前上下文，从长期记忆中检索相关信息，辅助决策。

1.2 设计目标

• 分层存储：区分短期和长期记忆，不同存储策略。
• 可扩展性：支持不同类型的记忆存储（如内存、文件、向量数据库）。
• 高效检索：支持基于关键词、时间、语义的检索。
• 上下文构建：能智能地选择最相关的记忆片段，构建供LLM使用的上下文。
• 持久化：长期记忆可持久化保存，重启后恢复。
• 与规划/推理模块集成：记忆模块为其他模块提供信息查询和更新接口。

---

2. 系统架构与UML建模

2.1 核心类设计

2.2 记忆存储与检索时序图

---

3. 详细设计

3.1 记忆项模型 (MemoryItem)

每个记忆项包含：

• id：唯一标识符（如UUID）。
• content：记忆内容（字符串、字典等）。
• timestamp：创建时间。
• metadata：附加信息（如角色、工具名、重要性评分）。
• embedding：可选，用于向量检索的嵌入向量。

3.2 短期记忆 (ShortTermMemory)

短期记忆存储最近N条消息，容量有限（例如20条），采用FIFO策略。主要提供：

• add(item)：添加新项，如果超出容量则移除最旧的。
• get_context(max_items)：获取最近的若干项，用于构建当前上下文。
• clear()：清空。

3.3 长期记忆 (LongTermMemory)

长期记忆存储用户偏好、重要事实等，可持久化。实现方式可以是内存字典+文件持久化，或数据库。主要功能：

• add(item)：添加或更新。
• get(id)：按ID获取。
• query_by_keyword(keyword)：关键词匹配（简单文本搜索）。
• save_to_disk(path) / load_from_disk(path)：持久化。

3.4 向量记忆 (VectorMemory)

向量记忆利用嵌入模型将记忆内容转为向量，支持语义搜索。需要依赖向量数据库（如FAISS、Chroma）或自己实现简单的向量存储。这里我们提供一个抽象接口，并实现一个基于内存和余弦相似度的简单版本，以便演示。主要功能：

• add(item)：生成嵌入并存储。
• query_similar(query_embedding, top_k)：返回最相似的top_k个记忆项。

3.5 记忆管理器 (MemoryManager)

记忆管理器整合各类记忆，提供统一的接口：

• add_message(role, content, metadata)：将消息添加到短期记忆，并可选择性地添加到长期记忆（如标记为重要）。
• build_context(current_input, max_tokens=2000)：构建供LLM使用的上下文文本，策略包括：
  • 最近短期记忆（最后N条）。
  • 从长期记忆中检索与当前输入关键词匹配的项。
  • 如果支持向量记忆，检索语义相似项。
  • 组合、排序、截断以满足token限制。
• search_memory(query)：跨记忆类型搜索。

3.6 与智能体其他模块的集成

• State：之前我们已经有State类包含Memory，现在可以改为包含MemoryManager。State.get_context()将委托给MemoryManager.build_context()。
• Agent：在run方法中，通过memory_manager添加用户消息和助手消息，并获取上下文。
• Reflector：反思模块可以将总结存入长期记忆。

---

4. 项目文件结构

在原有项目基础上，新增/修改以下文件：

agent_core/
├── agent/
│   ├── core/
│   │   ├── __init__.py
│   │   ├── memory/
│   │   │   ├── __init__.py
│   │   │   ├── base.py            # Memory基类, MemoryItem
│   │   │   ├── short_term.py      # ShortTermMemory
│   │   │   ├── long_term.py       # LongTermMemory
│   │   │   ├── vector_memory.py   # VectorMemory（简单实现）
│   │   │   └── manager.py         # MemoryManager
│   │   ├── state.py                # 修改，使用MemoryManager
│   │   ├── agent.py                # 修改，初始化MemoryManager
│   │   └── ... (其他不变)
│   ├── llm/... (不变)
│   ├── tools/... (不变)
│   ├── nlu/... (不变)
│   ├── planners/... (不变)
│   └── utils/... (不变)
├── examples/
│   └── memory_demo.py              # 新示例，展示记忆功能
└── requirements.txt                 # 添加依赖（如需向量嵌入，可选numpy、scikit-learn等）

---

5. 源代码完整实现

5.1 依赖安装

为了向量记忆，我们使用简单的numpy计算余弦相似度，不需要外部向量数据库。但如果需要更高级的检索，可安装faiss等。此处我们只依赖numpy（可选），但为了演示，我们将实现一个纯Python的简单向量检索。

pip install numpy  # 可选，用于向量运算

5.2 记忆项模型 (agent/core/memory/base.py)

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional

@dataclass
class MemoryItem:
    """记忆项"""
    content: Any
    id: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: Optional[List[float]] = None

    def __post_init__(self):
        if self.id is None:
            import uuid
            self.id = str(uuid.uuid4())

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "content": self.content,
            "timestamp": self.timestamp.isoformat(),
            "metadata": self.metadata,
            "embedding": self.embedding
        }

    @classmethod
    def from_dict(cls, data: dict) -> "MemoryItem":
        data["timestamp"] = datetime.fromisoformat(data["timestamp"])
        return cls(**data)

5.3 短期记忆 (agent/core/memory/short_term.py)

from typing import List, Optional
from collections import deque
from .base import MemoryItem

class ShortTermMemory:
    """短期记忆，固定容量FIFO"""
    def __init__(self, capacity: int = 20):
        self.capacity = capacity
        self.items = deque(maxlen=capacity)

    def add(self, item: MemoryItem) -> None:
        self.items.append(item)

    def get_recent(self, count: Optional[int] = None) -> List[MemoryItem]:
        """获取最近count条，默认全部"""
        if count is None:
            return list(self.items)
        return list(self.items)[-count:]

    def get_all(self) -> List[MemoryItem]:
        return list(self.items)

    def clear(self) -> None:
        self.items.clear()

5.4 长期记忆 (agent/core/memory/long_term.py)

import json
import os
from typing import Dict, List, Optional
from .base import MemoryItem

class LongTermMemory:
    """长期记忆，基于字典，支持持久化"""
    def __init__(self):
        self.items: Dict[str, MemoryItem] = {}

    def add(self, item: MemoryItem) -> None:
        self.items[item.id] = item

    def get(self, item_id: str) -> Optional[MemoryItem]:
        return self.items.get(item_id)

    def update(self, item_id: str, **kwargs) -> bool:
        """更新记忆项的属性（如metadata）"""
        if item_id in self.items:
            for key, value in kwargs.items():
                setattr(self.items[item_id], key, value)
            return True
        return False

    def query_by_keyword(self, keyword: str) -> List[MemoryItem]:
        """简单关键词搜索（在content和metadata中匹配）"""
        results = []
        for item in self.items.values():
            if isinstance(item.content, str) and keyword.lower() in item.content.lower():
                results.append(item)
            elif isinstance(item.content, dict):
                # 简单处理，将字典转为字符串搜索
                if keyword.lower() in str(item.content).lower():
                    results.append(item)
            # 也可搜索metadata中的值
            for k, v in item.metadata.items():
                if isinstance(v, str) and keyword.lower() in v.lower():
                    results.append(item)
                    break
        return results

    def get_all(self) -> List[MemoryItem]:
        return list(self.items.values())

    def save_to_file(self, filepath: str) -> None:
        """保存到JSON文件"""
        data = [item.to_dict() for item in self.items.values()]
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def load_from_file(self, filepath: str) -> None:
        """从JSON文件加载"""
        if not os.path.exists(filepath):
            return
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        for item_data in data:
            item = MemoryItem.from_dict(item_data)
            self.items[item.id] = item

5.5 向量记忆 (agent/core/memory/vector_memory.py)

使用简单的numpy实现余弦相似度检索。如果没有numpy，可以自己实现向量点积。这里我们假设有numpy。

import numpy as np
from typing import List, Optional
from .base import MemoryItem

class VectorMemory:
    """基于向量的语义记忆，使用余弦相似度"""
    def __init__(self):
        self.items: List[MemoryItem] = []
        self.embeddings: Optional[np.ndarray] = None  # 矩阵，每行一个向量

    def add(self, item: MemoryItem) -> None:
        if item.embedding is None:
            raise ValueError("MemoryItem must have embedding to add to VectorMemory")
        self.items.append(item)
        self._rebuild_embeddings_matrix()

    def add_batch(self, items: List[MemoryItem]) -> None:
        for item in items:
            if item.embedding is None:
                raise ValueError("All MemoryItems must have embeddings")
        self.items.extend(items)
        self._rebuild_embeddings_matrix()

    def _rebuild_embeddings_matrix(self):
        if self.items:
            self.embeddings = np.array([item.embedding for item in self.items])
        else:
            self.embeddings = None

    def query_similar(self, query_embedding: List[float], top_k: int = 5) -> List[MemoryItem]:
        """返回最相似的top_k个记忆项"""
        if not self.items or self.embeddings is None:
            return []
        query_vec = np.array(query_embedding).reshape(1, -1)
        # 计算余弦相似度
        norms = np.linalg.norm(self.embeddings, axis=1, keepdims=True)
        query_norm = np.linalg.norm(query_vec)
        if query_norm == 0:
            return []
        similarities = np.dot(self.embeddings, query_vec.T).flatten() / (norms.flatten() * query_norm)
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.items[i] for i in top_indices]

注意：实际应用中，嵌入向量通常由专门的嵌入模型生成（如OpenAI的text-embedding-ada-002）。此处我们仅提供存储和检索机制，嵌入生成需在添加时由外部提供。

5.6 记忆管理器 (agent/core/memory/manager.py)

from typing import List, Optional, Dict, Any
from .base import MemoryItem
from .short_term import ShortTermMemory
from .long_term import LongTermMemory
from .vector_memory import VectorMemory

class MemoryManager:
    """整合多种记忆的统一管理器"""
    def __init__(self, short_term_capacity: int = 20):
        self.short_term = ShortTermMemory(capacity=short_term_capacity)
        self.long_term = LongTermMemory()
        self.vector_memory: Optional[VectorMemory] = None

    def enable_vector_memory(self, vector_memory: VectorMemory):
        self.vector_memory = vector_memory

    def add_message(self, role: str, content: str, metadata: Optional[Dict] = None,
                    store_long_term: bool = False, embedding: Optional[List[float]] = None) -> MemoryItem:
        """添加一条消息到短期记忆，可选长期存储和向量存储"""
        item = MemoryItem(
            content=content,
            metadata={"role": role, **(metadata or {})},
            embedding=embedding
        )
        self.short_term.add(item)
        if store_long_term:
            self.long_term.add(item)
        if embedding is not None and self.vector_memory:
            self.vector_memory.add(item)
        return item

    def add_to_long_term(self, content: Any, metadata: Optional[Dict] = None,
                         embedding: Optional[List[float]] = None) -> MemoryItem:
        """直接添加到长期记忆"""
        item = MemoryItem(content=content, metadata=metadata, embedding=embedding)
        self.long_term.add(item)
        if embedding and self.vector_memory:
            self.vector_memory.add(item)
        return item

    def build_context(self, current_input: str, max_tokens: int = 2000,
                      include_short_term: int = 10,
                      include_long_term_keywords: List[str] = None,
                      include_vector_similar: bool = True) -> str:
        """
        构建供LLM使用的上下文文本
        :param current_input: 当前用户输入，用于检索
        :param max_tokens: 最大token数（近似）
        :param include_short_term: 包含最近多少条短期记忆
        :param include_long_term_keywords: 用哪些关键词检索长期记忆
        :param include_vector_similar: 是否进行向量相似检索
        :return: 组合后的上下文文本
        """
        context_parts = []

        # 1. 短期记忆
        if include_short_term > 0:
            recent = self.short_term.get_recent(include_short_term)
            if recent:
                context_parts.append("【近期对话】")
                for item in recent:
                    role = item.metadata.get("role", "unknown")
                    context_parts.append(f"{role}: {item.content}")

        # 2. 长期记忆关键词检索
        if include_long_term_keywords:
            for kw in include_long_term_keywords:
                items = self.long_term.query_by_keyword(kw)
                if items:
                    context_parts.append(f"\n【关于“{kw}”的记忆】")
                    for item in items:
                        context_parts.append(f"- {item.content}")

        # 3. 向量相似检索（如果有）
        if include_vector_similar and self.vector_memory:
            # 这里需要当前输入的嵌入，但管理器不知道如何生成，需要外部传入
            # 简单起见，我们假设外部已计算好嵌入并存储，这里无法动态获取。
            # 实际使用时，可在调用前计算当前输入的嵌入，然后传入。
            pass

        # 合并并截断（这里简单按字符数估算，实际应根据tokenizer）
        full_context = "\n".join(context_parts)
        # 简单截断，保留最后max_tokens*4个字符（假设平均每个token4字符）
        if len(full_context) > max_tokens * 4:
            full_context = full_context[-(max_tokens * 4):]
        return full_context

    def search_memory(self, query: str, search_long_term: bool = True,
                      search_vector: bool = True, top_k: int = 5) -> List[MemoryItem]:
        """跨记忆搜索"""
        results = []
        if search_long_term:
            results.extend(self.long_term.query_by_keyword(query))
        # 如果需要向量搜索，需要当前query的嵌入，但此处没有，故略。
        # 可返回去重结果
        return results

5.7 修改状态模块 (agent/core/state.py)

将原有的Memory替换为MemoryManager。

from typing import List, Optional
from .memory.manager import MemoryManager
from .models import Message
from .plan import Plan, StepStatus

class State:
    def __init__(self):
        self.memory_manager = MemoryManager()
        self.current_plan: Optional[Plan] = None
        self.intermediate_results: dict = {}

    def add_user_message(self, content: str):
        self.memory_manager.add_message(role="user", content=content)

    def add_assistant_message(self, content: str):
        self.memory_manager.add_message(role="assistant", content=content)

    def add_system_message(self, content: str):
        self.memory_manager.add_message(role="system", content=content)

    def add_tool_message(self, tool_name: str, result: str):
        self.memory_manager.add_message(role="tool", content=f"{tool_name}: {result}")

    def get_context(self, current_input: str = "") -> str:
        """构建上下文，用于推理"""
        return self.memory_manager.build_context(
            current_input=current_input,
            include_short_term=10,
            include_long_term_keywords=self._extract_keywords(current_input)
        )

    def _extract_keywords(self, text: str) -> List[str]:
        """简单提取关键词，用于长期记忆检索"""
        # 实际可用更复杂的方法，这里简单返回非停用词
        if not text:
            return []
        words = text.split()
        # 简单过滤长度大于1的词
        return [w for w in words if len(w) > 1][:3]

    # 其他方法（set_plan, update_step等）保持不变
    # ...

5.8 修改Agent主类 (agent/core/agent.py)

需要调整初始化，移除原有的Memory，使用State自带的memory_manager。

class Agent:
    def __init__(self, nlu_engine, planner, llm):
        self.perception = Perception(nlu_engine)
        self.planner = planner
        self.reasoning = Reasoning(llm)
        self.tool_registry = ToolRegistry()
        self.tool_invoker = ToolInvoker(self.tool_registry)
        self.executor = Executor(self.tool_invoker)
        self.state = State()  # 现在包含memory_manager
        self.reflector = Reflector()
        self.max_iterations = 20

    def register_tool(self, tool):
        self.tool_registry.register(tool)

    def run(self, user_input: str) -> str:
        # 1. 感知
        parsed = self.perception.parse(user_input)
        self.state.add_user_message(user_input)

        # 2. 规划
        plan = self.planner.create_plan(parsed, self.tool_registry.list_tools(), self.state)
        self.state.set_plan(plan)

        # 3. ReAct循环
        iteration = 0
        final_answer = None
        while iteration < self.max_iterations:
            iteration += 1
            # 获取上下文（传入当前输入用于关键词提取）
            context = self.state.get_context(current_input=user_input)
            step_info = self._get_step_info()
            full_context = context + "\n" + step_info

            action = self.reasoning.decide(full_context, self.state.memory_manager.short_term.get_recent(5), self.tool_registry.list_tools())

            if action.type == 'final':
                final_answer = action.content
                self.state.add_assistant_message(final_answer)
                break
            elif action.type == 'tool':
                obs = self.executor.execute(action)
                self.state.add_tool_message(action.tool, obs.result if not obs.error else obs.error)
                # 更新计划步骤状态
                self._update_plan_step(action.tool, obs)
            else:
                if action.content:
                    self.state.add_assistant_message(f"思考: {action.content}")

            if self.state.current_plan and self.state.current_plan.is_complete():
                # 计划完成，可生成总结
                pass

        if not final_answer:
            final_answer = "抱歉，无法完成您的请求。"
        return final_answer

    def _get_step_info(self):
        # 获取当前步骤信息
        step = self.state.get_next_step()
        return f"当前步骤: {step.to_dict() if step else '无'}"

5.9 运行示例 (examples/memory_demo.py)

创建一个演示，展示记忆模块的功能：包括短期记忆、长期记忆存储和检索。

import sys
sys.path.append("..")

from agent.core.agent import Agent
from agent.nlu.rule_based import RuleBasedNLU
from agent.planners.template_planner import TemplatePlanner
from agent.llm.mock import MockLLM
from agent.tools.calculator import CalculatorTool
from agent.tools.weather import WeatherTool

def main():
    nlu = RuleBasedNLU()
    planner = TemplatePlanner()
    llm = MockLLM()
    agent = Agent(nlu, planner, llm)
    agent.register_tool(CalculatorTool())
    agent.register_tool(WeatherTool())

    print("带有记忆管理的智能体已启动")
    # 先让用户输入一些信息，建立长期记忆
    agent.run("我的名字是张三")
    agent.run("我喜欢晴天")
    agent.run("我住在北京")

    # 然后提问，期望能利用长期记忆
    response = agent.run("你知道我住在哪里吗？")
    print(f"助手: {response}")

    # 查看长期记忆内容
    print("\n长期记忆中的内容：")
    for item in agent.state.memory_manager.long_term.get_all():
        print(f"- {item.content} (metadata: {item.metadata})")

if __name__ == "__main__":
    main()

注意：由于模拟LLM无法真正理解长期记忆，实际效果可能不明显。但在真实LLM中，通过构建包含长期记忆的上下文，智能体可以正确回答。此处仅展示记忆模块的功能。

---

6. 总结与扩展

通过本文，我们为智能体构建了完整的记忆与上下文管理系统：

• 设计了分层记忆结构：短期记忆（FIFO）、长期记忆（键值存储）、向量记忆（语义检索）。
• 实现了记忆管理器，统一管理各类记忆，并提供上下文构建策略。
• 与状态和Agent主类集成，使智能体具备记忆能力。
• 支持持久化（长期记忆可保存到文件）。

未来扩展方向：

• 嵌入模型集成：实际使用时，需要将文本转换为嵌入向量，可集成OpenAI Embeddings或本地模型。
• 更智能的上下文构建：根据重要性、相关性、时间衰减选择记忆。
• 记忆压缩与摘要：当短期记忆过长时，自动生成摘要并存入长期记忆。
• 遗忘机制：自动删除或归档旧记忆。
• 数据库后端：使用Redis、MongoDB等支持更大规模存储。

现在，你的智能体已经拥有了记忆，能够记住用户信息和历史对话，为个性化交互和持续学习奠定基础。

---

附录：更新后的项目文件清单

agent_core/
├── agent/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── agent.py                # 修改
│   │   ├── state.py                 # 修改
│   │   ├── memory/
│   │   │   ├── __init__.py
│   │   │   ├── base.py
│   │   │   ├── short_term.py
│   │   │   ├── long_term.py
│   │   │   ├── vector_memory.py
│   │   │   └── manager.py
│   │   ├── ... (其他不变)
│   ├── tools/... (不变)
│   ├── nlu/... (不变)
│   ├── planners/... (不变)
│   ├── llm/... (不变)
│   └── utils/... (不变)
├── examples/
│   └── memory_demo.py
└── requirements.txt                 # 如果使用numpy，需添加

requirements.txt 补充：

numpy>=1.19.0   # 可选，用于向量记忆

现在，你可以基于此框架构建真正具备记忆能力的智能体应用。