Hotdry.
归档
ai-systems

Memori 中 episodic 记忆模块的设计：高效检索交互历史支持 LLM 代理多轮决策

探讨 Memori 框架下 episodic 记忆模块的设计，聚焦交互历史的检索优化，以提升 LLM 代理在多轮对话中的决策能力。提供具体参数配置和实现清单。

在大型语言模型（LLM）驱动的代理系统中，记忆机制是实现长期上下文理解和决策连续性的关键。其中，episodic memory（情节记忆）作为一种存储特定交互事件和历史序列的机制，能够帮助代理在多轮对话中进行选择性回忆，避免无关信息的干扰，从而提升决策效率。本文聚焦于开源记忆引擎 Memori 框架下 episodic 记忆模块的设计，探讨如何高效检索交互历史，支持 LLM 代理在复杂多转场景下的智能行为。Memori 以 SQL 原生方式存储记忆，天然适合构建结构化的 episodic 模块，我们将从设计原理、检索优化到落地参数进行剖析。

episodic 记忆在 LLM 代理中的核心作用

episodic memory 不同于语义记忆（semantic memory），它强调对具体事件的时间序列和上下文细节的保留，例如用户在多轮交互中提到的偏好、事件触发或决策路径。在 LLM 代理应用中，如客服机器人或游戏 NPC，多轮场景往往涉及数百次交互，如果缺乏高效的 episodic 检索，代理容易遗忘关键历史，导致决策偏差或重复查询。根据 Memori 的架构，记忆存储在标准 SQL 数据库中，支持全文搜索索引，这为 episodic 模块提供了坚实基础。[1] Memori 通过拦截 LLM 调用，在预调用阶段注入相关上下文，并在后调用阶段提取实体并分类存储，从而实现记忆的动态更新。

设计 episodic 模块时，需要考虑代理的决策需求：例如，在一个多代理协作系统中，代理需回忆特定交互（如 “用户上轮拒绝了方案 A”），以避免重复提案。这要求模块支持时间戳排序、事件关联和选择性过滤，避免全量加载历史导致的 token 消耗过高。Memori 的 Retrieval Agent 在 auto 模式下动态搜索相关记忆，正是 episodic 检索的理想起点。通过扩展此机制，我们可以构建专属的 episodic 层，确保检索仅限于交互历史子集。

Memori 中 episodic 记忆模块的架构设计

Memori 的核心是其拦截器系统：当代理发起 LLM 调用时，模块首先从 SQL 数据库中检索相关记忆，然后注入消息提示中。针对 episodic memory，我们设计一个分层模块：底层存储层使用 SQL 表 schema 记录交互事件，包括字段如 event_id（唯一 ID）、timestamp（时间戳）、user_id（用户标识）、content（交互文本）、entities（提取实体，如人名、意图）、relations（事件关联，如 “导致” 或 “跟随”）。例如，一个交互历史可表示为：{"event_id": 123, "timestamp": "2025-11-20T10:00:00", "content": "用户查询产品价格", "entities": ["产品 A", "价格"], "relations": "query -> productA"}。

在上层，Retrieval Agent 被定制为 Episodic Retriever：它使用 SQL 查询结合嵌入向量（可选使用轻量嵌入模型如 sentence-transformers）进行混合检索。首先，基于时间窗口过滤（如最近 7 天交互），然后通过语义相似度阈值（e.g., cosine similarity > 0.7）筛选相关事件。这避免了纯关键字搜索的噪声问题。在 conscious 模式下，Conscious Agent 可背景运行，每 6 小时分析交互模式，将高频或关键 episodic 事件提升至短期工作记忆（short-term store），如使用 Redis 缓存最近 50 条高优先级历史。

证据显示，这种设计在多轮场景中显著提升性能。Memori 的例子中，多用户隔离机制确保每个代理的 episodic 记忆独立存储，避免跨用户污染。[2] 在一个 FastAPI 多用户应用示例中，代理能回忆特定用户的交互历史，实现个性化决策。相比传统向量数据库，SQL 原生的 episodic 存储减少了 80% 的成本，同时保持可审计性。

高效检索优化的关键策略

高效检索是 episodic 模块的核心挑战，尤其在交互历史积累到数万条时。优化从索引入手：为 timestamp 和 entities 字段创建复合索引（e.g., CREATE INDEX idx_episodic_time_entity ON episodic_events (timestamp, entities)），支持快速范围查询。同时，引入分片策略：根据 user_id 或 session_id 分表存储，减少单表扫描开销。在 Memori 的 auto_ingest 模式下，Episodic Retriever 可配置查询限额，如 LIMIT 20 以控制注入 token 数。

另一个策略是选择性回忆机制：使用规则 - based 过滤器结合 LLM 辅助。例如，预定义回忆触发器（如关键词 “回顾上次”），触发时查询相关事件链（通过 relations 字段递归关联）。为应对多转决策，引入图结构表示：将交互历史建模为事件图，节点为事件，边为因果关系，使用 Neo4j 或纯 SQL 视图实现子图检索。这在复杂代理如 CrewAI 集成中特别有用，代理可回忆 “事件 A 导致决策 B” 的路径，支持 counterfactual 推理。

潜在风险包括检索延迟和一致性问题。在高并发多代理系统中，CRDT（Conflict-free Replicated Data Types）可确保分布式 episodic 存储的一致性，但 Memori 当前依赖 SQL 事务，故建议添加乐观锁（e.g., version 字段）。限制造约：数据库规模超过 1GB 时，迁移至 PostgreSQL 以利用其全文搜索扩展（如 pg_trgm）。

可落地参数配置与实现清单

为快速部署 episodic 记忆模块，以下提供具体参数和清单，基于 Memori 的 ConfigManager。

  1. 初始化配置：

    • database_connect: "postgresql://user:pass@localhost/memori"（优先 PostgreSQL 以支持高级索引）。
    • conscious_ingest: True（启用短期 episodic 缓存）。
    • auto_ingest: True（动态检索交互历史）。
    • episodic_threshold: 0.75（语义相似度阈值，低于此不注入）。
    • time_window: "7d"（检索最近 7 天历史，单位：d/h/m）。

  2. SQL Schema 创建（运行一次）：

    CREATE TABLE episodic_events (
    event_id SERIAL PRIMARY KEY,
    user_id VARCHAR(50) NOT NULL,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    content TEXT,
    entities JSONB,
    relations JSONB,
    priority INT DEFAULT 0  -- 0: low, 1: medium, 2: high
);
CREATE INDEX idx_episodic_user_time ON episodic_events (user_id, timestamp DESC);
CREATE INDEX idx_episodic_entities ON episodic_events USING GIN (entities);

  3. 检索函数扩展（Python 代码片段，集成 Memori）：

    from memori import Memori
import psycopg2  # 或 sqlite3

memori = Memori(database_connect="postgresql://...", conscious_ingest=True, auto_ingest=True)
memori.enable()

def retrieve_episodic_history(user_id, query_embedding, time_window='7d'):
    conn = psycopg2.connect(memori.database_connect)
    cur = conn.cursor()
    # 时间过滤 + 相似度
    cur.execute("""
        SELECT content, entities FROM episodic_events 
        WHERE user_id = %s AND timestamp > NOW() - INTERVAL %s 
        ORDER BY timestamp DESC LIMIT 20
    """, (user_id, time_window))
    events = cur.fetchall()
    # 进一步语义过滤（使用 cosine sim > 0.75）
    relevant = [e for e in events if semantic_sim(query_embedding, e[1]) > 0.75]
    cur.close()
    conn.close()
    return relevant

  4. 监控与回滚策略：

    • 监控点：检索延迟（<200ms）、命中率（> 80% 相关事件注入）、token 消耗（< 2000 / 调用）。
    • 阈值警报：如果延迟 > 500ms，回滚至 conscious 模式仅用缓存。
    • 回滚清单：1. 禁用 auto_ingest；2. 清空短期缓存；3. 验证 SQL 索引完整性。

  5. 测试清单：

    • 单元测试：模拟 100 轮交互，验证检索准确率 > 90%。
    • 集成测试：与 OpenAI client 结合，检查多轮决策一致性。
    • 负载测试：100 用户并发，监控数据库 CPU < 70%。

通过这些参数，开发者可在 Memori 上快速构建 episodic 模块，实现代理的记忆增强。实际部署中，结合 Memori 的多框架支持（如 LangChain），可扩展至生产级应用。

结语与资料来源

episodic 记忆模块的设计不仅提升了 LLM 代理的智能性，还降低了运维复杂度。在 Memori 的 SQL 生态中，这种优化特别高效，适用于从个人助手到企业代理的各种场景。未来，可进一步集成多模态 episodic 数据（如图像事件），扩展其边界。

资料来源： [1] Memori GitHub 仓库：https://github.com/GibsonAI/Memori （描述了记忆拦截和存储机制）。 [2] Memori 文档：https://memorilabs.ai/docs （提供了架构细节和例子）。

（本文约 1250 字）

查看归档