From 89f0a1771f10841c6faef46afe0afea9a8cc818c Mon Sep 17 00:00:00 2001 From: wehub-skill-sync Date: Mon, 13 Jul 2026 21:36:25 +0800 Subject: [PATCH] chore: import zh skill memory-systems --- README.wehub.md | 9 + SKILL.md | 234 +++++++++++++ references/implementation.md | 550 +++++++++++++++++++++++++++++++ scripts/memory_store.py | 616 +++++++++++++++++++++++++++++++++++ 4 files changed, 1409 insertions(+) create mode 100644 README.wehub.md create mode 100644 SKILL.md create mode 100644 references/implementation.md create mode 100644 scripts/memory_store.py diff --git a/README.wehub.md b/README.wehub.md new file mode 100644 index 0000000..c5942ff --- /dev/null +++ b/README.wehub.md @@ -0,0 +1,9 @@ +# WeHub 来源说明 + +- Skill 名称：`memory-systems` +- 中文类目：Agent 长期/语义记忆架构 +- 上游仓库：`muratcankoylan__Agent-Skills-for-Context-Engineering` +- 上游路径：`skills/memory-systems/SKILL.md` +- 上游链接：https://github.com/muratcankoylan/Agent-Skills-for-Context-Engineering/blob/HEAD/skills/memory-systems/SKILL.md +- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理 +- 原作者、版权和许可证信息以上游仓库为准 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..8371cf1 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,234 @@ +--- +name: memory-systems +description: > + 该技能应用于智能体系统中的持久化语义记忆： + 跨会话知识保留、实体追踪、时间有效性、 + 图或向量检索、记忆整合、以及记忆基准测试的选择。 + 文件型临时存储请交给 filesystem-context，交接摘要请交给 context-compression， + 令牌效率策略请交给 context-optimization。 +--- + +# 记忆系统设计 + +记忆提供了持久化层，使智能体能够跨会话保持连续性，并基于累积的知识进行推理。简单的智能体完全依赖上下文作为记忆，会话结束时所有状态都会丢失。复杂的智能体则实现分层记忆架构，在即时上下文需求与长期知识保留之间取得平衡。从向量存储到知识图谱再到时序知识图谱的演进，代表着在结构化记忆方面投入不断增加，以改善检索和推理能力。 + +## 何时激活 + +在以下情况激活该技能： +- 构建必须跨会话持久化知识的智能体时 +- 在记忆框架（Mem0、Zep/Graphiti、Letta、LangMem、Cognee）之间做选择时 +- 需要在跨对话中保持实体一致性时 +- 实现基于累积知识的推理时 +- 设计可在生产环境中扩展的记忆架构时 +- 根据基准测试（LoCoMo、LongMemEval、DMR）评估记忆系统时 +- 构建具有自动实体/关系提取和自我改进记忆的动态记忆系统（Cognee）时 + +对于归属于其他技能的相邻工作，请勿激活该技能： +- 文件型临时存储、运行日志和工具输出卸载：`filesystem-context` +- 对话压缩或人类可读的交接摘要：`context-compression` +- 掩码、前缀缓存、令牌预算或单次轨迹内的检索范围控制：`context-optimization` +- 基于 RDF 状态的正式信念/愿望/意图模型：`bdi-mental-states` + +## 核心概念 + +将记忆视为从易失的上下文窗口到持久存储的一个光谱。默认选择能满足检索需求的最简单层级，因为基准测试证据表明，对于某些记忆工作负载，工具复杂度的重要性低于可靠的检索（claim-memory-locomo-filesystem-baseline）。只有在检索质量下降，或者智能体需要多跳推理、关系遍历或时间旅行查询时，才增加结构（图、时间有效性）。 + +## 详细主题 + +### 生产框架全景 + +根据智能体所需的主要检索模式来选择框架。使用下表缩小候选范围，然后通过下面的基准测试数据进行验证。 + +| 框架 | 架构 | 最适合 | 权衡 | +|-----------|-------------|----------|-----------| +| **Mem0** | 向量存储 + 图记忆，可插拔后端 | 多租户系统，广泛的集成 | 在多智能体场景下不够专业 | +| **Zep/Graphiti** | 时序知识图谱，双时态模型 | 需要关系建模 + 时序推理的企业级场景 | 高级功能锁定在云端 | +| **Letta** | 自编辑记忆，分层存储（上下文内/核心/归档） | 完整的智能体内省，有状态服务 | 对简单用例过于复杂 | +| **Cognee** | 通过可定制的 ECL 管道与可定制任务构建的多层语义图 | 不断演化的智能体记忆，能够自适应和学习；多跳推理 | 摄入时处理更重 | +| **LangMem** | 用于 LangGraph 工作流的记忆工具 | 已在 LangGraph 上的团队 | 与 LangGraph 紧密耦合 | +| **文件系统** | 带命名约定的纯文本文件 | 简单智能体、原型开发 | 无语义搜索，无关系 | + +当智能体需要双时态建模（同时追踪事件发生时间与数据摄入时间）时，选择 Zep/Graphiti，因为其三层级知识图谱（事件、语义实体、社区子图）在时序查询方面表现出色。当优先考虑通过托管基础设施快速上线时，选择 Mem0。当智能体需要通过其智能体开发环境进行深度自我内省时，选择 Letta。当智能体必须构建密集的多层语义图时，选择 Cognee——它将文本块和实体类型作为节点，并带有详细的关系边，每个核心组件（摄入、实体提取、后处理、检索）都是可定制的。 + +**基准测试性能对比** + +参考以下基准测试来设定预期，但应将其视为检索维度上的特定来源信号，而非绝对排名。没有单一基准测试是决定性的。 + +| 系统 | DMR 准确率 | LoCoMo | HotPotQA（多跳） | 延迟 | +|--------|-------------|--------|---------------------|---------| +| Cognee | — | — | 公布的分数较高 | 视情况而定 | +| Zep（时序知识图谱） | 公布的分数较高 | — | 各项指标中等 | 报告为低延迟 | +| Letta（文件系统） | — | 公布的文件系统基线 | — | — | +| Mem0 | — | 公布的专业工具基线 | 在某项对比中较低 | — | +| MemGPT | 公布的分数较高 | — | — | 视情况而定 | +| GraphRAG | 公布的中/高范围 | — | — | 视情况而定 | +| 向量 RAG 基线 | 公布的较低范围 | — | — | 快速 | + +关键结论：根据检索形态来比较记忆系统，而不是看品牌。将基准测试数据视为过时的证据，在用于产品宣传之前必须重新核实；稳定的设计原则是：从浅层开始，衡量检索质量，然后仅在更简单的层级无法满足需求时才添加语义或图结构。 + +### 记忆层级（决策要点） + +选择能满足持久化需求的最浅记忆层级。每增加一个层级都会带来基础设施成本和运维复杂度，因此只有在浅层无法满足检索或持久化需求时才进行升级。 + +| 层级 | 持久性 | 实现方式 | 何时使用 | +|-------|------------|----------------|-------------| +| **工作记忆** | 仅上下文窗口 | 系统提示中的临时记录 | 始终使用——通过注意力优先位置进行优化 | +| **短期记忆** | 会话范围 | 文件系统、内存缓存 | 中间工具结果、对话状态 | +| **长期记忆** | 跨会话 | 键值存储 → 图数据库 | 用户偏好、领域知识、实体注册表 | +| **实体记忆** | 跨会话 | 实体注册表 + 属性 | 维护身份一致性（"张三"在跨对话中为同一个人） | +| **时序知识图谱** | 跨会话 + 历史记录 | 带有效时间区间的图 | 随时间变化的事实、时间旅行查询、防止上下文冲突 | + +### 检索策略 + +根据查询形态匹配检索策略。语义搜索擅长处理直接事实查询，但在多跳推理上表现不佳；基于实体的遍历擅长处理"关于 X 的一切"查询，但需要图结构；时序过滤擅长处理变化中的事实，但需要有效性元数据。当准确率至关重要且基础设施预算允许时，将多种策略结合为混合检索。 + +| 策略 | 何时使用 | 局限性 | +|----------|----------|------------| +| **语义**（嵌入相似度） | 直接事实查询 | 多跳推理时表现不佳 | +| **基于实体**（图遍历） | "告诉我关于 X 的一切" | 需要图结构 | +| **时序**（有效性过滤） | 事实随时间变化 | 需要有效性元数据 | +| **混合**（语义 + 关键词 + 图） | 总体准确率最佳 | 基础设施需求最大 | + +混合方法通过仅检索相关的子图或记忆来减少活跃上下文。Cognee 通过在图、向量和关系存储之间提供多种搜索模式来实现混合检索，使智能体能够根据查询类型选择合适的检索策略，而不是采用一刀切的方法。 + +### 记忆整合 + +定期执行记忆整合以防止无限制增长，因为不受控制的记忆累积会随着时间推移降低检索质量。**标记为无效而非丢弃**——保留历史记录对于需要重建过去状态的时序查询至关重要。在记忆数量达到阈值、检索质量下降或按预设时间间隔触发整合。参见[实现参考](./references/implementation.md)获取可工作的整合代码。 + +## 实践指导 + +### 选择记忆架构 + +**从最简单的可行层级开始，仅当检索质量下降时才增加复杂度。** 大多数智能体在第一天并不需要时序知识图谱。遵循以下升级路径： + +1. **原型阶段**：使用文件系统记忆。将事实存储为带时间戳的结构化 JSON。这能在投入基础设施之前验证智能体行为。 +2. **扩展阶段**：当智能体需要语义搜索和多租户隔离时，迁移到 Mem0 或向量存储（带元数据），因为基于文件的查找无法处理相似度查询。 +3. **复杂推理**：当智能体需要关系遍历、时间有效性或跨会话综合时，添加 Zep/Graphiti。Graphiti 使用带有通用关系的结构化连接，保持图简单且易于推理；Cognee 构建更密集的多层语义图，带有详细的关系边——根据智能体是需要时序双模型（Graphiti）还是更丰富的互联知识结构（Cognee）来选择。 +4. **完全控制**：当智能体必须通过深度内省来自我管理其记忆时，使用 Letta 或 Cognee，因为这些框架将记忆操作暴露为智能体的一等动作。 + +### 与上下文的集成 + +采用即时加载记忆的方式，而非预先加载全部内容，因为大的上下文负载成本高昂且会降低注意力质量。将检索到的记忆放置在注意力优先的位置（上下文开头或结尾），以最大化其对生成结果的影响力。 + +### 错误恢复 + +优雅地处理检索失败，因为记忆系统天生具有噪声。按顺序应用以下恢复策略： + +- **检索为空**：回退到更广泛的搜索（移除实体过滤条件，扩大时间范围）。如果仍然为空，提示用户澄清。 +- **结果过时**：检查 `valid_until` 时间戳。如果大多数结果已过期，在重试之前触发整合。 +- **事实冲突**：优先选择 `valid_from` 最近的事实。如果置信度较低，向用户展示冲突。 +- **存储失败**：将写入操作排队等待重试。绝不能让记忆写入阻塞智能体的响应。 + +## 示例 + +**示例 1：Mem0 集成** +```python +from mem0 import Memory + +m = Memory() +m.add("用户偏好深色模式和 Python 3.12", user_id="alice") +m.add("用户切换到浅色模式", user_id="alice") + +# 检索当前偏好（浅色模式），而非过时的 +results = m.search("用户偏好什么主题？", user_id="alice") +``` + +**示例 2：时序查询** +```python +# 使用有效时间区间追踪实体 +graph.create_temporal_relationship( + source_id=user_node, + rel_type="LIVES_AT", + target_id=address_node, + valid_from=datetime(2024, 1, 15), + valid_until=datetime(2024, 9, 1), # 已搬走 +) + +# 查询：用户在 2024 年 3 月 1 日住在哪里？ +results = graph.query_at_time( + {"type": "LIVES_AT", "source_label": "User"}, + query_time=datetime(2024, 3, 1) +) +``` + +**示例 3：Cognee 记忆摄入与搜索** +```python +import cognee +from cognee.modules.search.types import SearchType + +# 摄入并构建知识图谱 +await cognee.add("./docs/") +await cognee.add("任何数据") +await cognee.cognify() + +# 丰富记忆 +await cognee.memify() + +# 智能体检索具有关系感知能力的上下文 +results = await cognee.search( + query_text="对你的记忆进行任意查询", + query_type=SearchType.GRAPH_COMPLETION, +) +``` + +## 指导原则 + +1. 从文件系统记忆开始；只有在检索质量要求时才增加复杂度 +2. 对任何可能随时间变化的事实追踪时间有效性 +3. 使用混合检索（语义 + 关键词 + 图）以获得最佳准确率 +4. 定期整合记忆——标记为无效但不丢弃 +5. 为检索失败做好设计：当记忆查找未返回结果时，始终准备一个回退方案 +6. 考虑持久化记忆的隐私影响（保留策略、删除权限） +7. 在每次变更前后，使用 LoCoMo 或 LongMemEval 对你的记忆系统进行基准测试 +8. 在生产环境中监控记忆增长和检索延迟 + +## 注意事项 + +1. **将所有内容塞入上下文**：将可用的记忆全部加载到提示中成本高昂且会降低注意力质量。使用即时检索并配合相关性过滤。 +2. **忽略时间有效性**：事实会过时。如果没有时间有效性追踪，过时的信息会污染上下文，智能体会基于错误的假设行事。 +3. **过早过度设计**：在部分基准测试中，简单的文件系统记忆可能优于更专业的工具（claim-memory-locomo-filesystem-baseline）。只有在简单方法明显失败时才增加复杂度。 +4. **没有整合策略**：不受控制的记忆增长会随时间推移降低检索质量。设置记忆数量阈值或定时计划来触发整合。 +5. **嵌入模型不匹配**：用一个嵌入模型写入记忆，用另一个模型读取记忆，会导致检索效果不佳，因为向量空间不可互换。为每个记忆存储固定一个嵌入模型，如果模型变更，则重新嵌入所有条目。 +6. **图谱模式僵化**：过度结构化的图谱模式（固定的节点类型、固定的关系标签）在领域演化时会失效。优先使用通用关系类型和灵活的属性包，以便新实体类型无需模式迁移。 +7. **过时记忆中毒**：与当前状态相矛盾的旧记忆会静默地破坏智能体行为。实施过期策略或置信度衰减机制，使智能体降低对过时事实的优先级排序，并在检测到矛盾时显式呈现。 +8. **记忆-上下文不匹配**：检索到的记忆在主题上相关但在上下文上错误（例如，当智能体讨论 Python 编程语言时，检索到了关于"蟒蛇"的记忆）。通过在记忆条目中包含会话或领域元数据，并在检索时进行过滤来缓解此问题。 + +## 集成 + +该技能负责持久化语义记忆。相邻技能负责临时存储、压缩和上下文策略： + +- `filesystem-context`：在需要语义检索之前的文件型临时记录、日志和简单运行状态。 +- `context-compression`：以散文形式保存会话状态的摘要和交接记录。 +- `context-optimization`：在活跃上下文预算内进行即时记忆加载和检索范围控制。 +- `context-degradation`：将过时或冲突的记忆视为上下文中毒或冲突。 +- `bdi-mental-states`：当信念、愿望、意图和来源链很重要时的正式心理状态建模。 +- `multi-agent-patterns`：跨智能体的共享记忆。 +- `evaluation`：记忆质量、检索正确性和基准测试选择。 + +## 参考资料 + +内部参考： +- [实现参考](./references/implementation.md) - 阅读时机：从头实现向量存储、属性图、时序查询或记忆整合逻辑时 + +本集合中的相关技能： +- context-fundamentals - 阅读时机：设计记忆所馈入的上下文层时 +- multi-agent-patterns - 阅读时机：多个智能体需要共享或协调记忆状态时 + +外部资源： +- Zep 时序知识图谱论文（arXiv:2501.13956） - 阅读时机：评估双时态建模或 Graphiti 架构时 +- Mem0 生产架构论文（arXiv:2504.19413） - 阅读时机：评估托管记忆基础设施的权衡时 +- Cognee 优化知识图谱 + 大语言模型推理论文（arXiv:2505.24478） - 阅读时机：比较多层语义图方法时 +- LoCoMo 基准测试（Snap Research） - 阅读时机：评估长对话记忆保持能力时 +- MemBench 评估框架（ACL 2025） - 阅读时机：设计记忆评估套件时 +- Graphiti 开源时序知识图谱引擎（github.com/getzep/graphiti） - 阅读时机：实现时序知识图谱时 +- Cognee 开源知识图谱记忆（github.com/topoteretes/cognee） - 阅读时机：为记忆构建可定制的 ECL 管道时 +- [Cognee 对比：形态与功能](https://www.cognee.ai/blog/deep-dives/competition-comparison-form-vs-function) - 阅读时机：比较 Mem0、Graphiti、LightRAG、Cognee 之间的图谱结构时 + +--- + +## 技能元数据 + +**创建日期**：2025-12-20 +**最后更新**：2026-05-15 +**作者**：面向上下文工程的智能体技能贡献者 +**版本**：4.1.0 diff --git a/references/implementation.md b/references/implementation.md new file mode 100644 index 0000000..47f2b17 --- /dev/null +++ b/references/implementation.md @@ -0,0 +1,550 @@ +# 记忆系统：技术参考 + +本文档提供了记忆系统组件的实现细节。 + +## 向量存储实现 + +### 基础向量存储 + +```python +import numpy as np +from typing import List, Dict, Any +import json + + +def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: + """Compute cosine similarity between two vectors.""" + norm_a = np.linalg.norm(a) + norm_b = np.linalg.norm(b) + if norm_a == 0 or norm_b == 0: + return 0.0 + return float(np.dot(a, b) / (norm_a * norm_b)) + + +class VectorStore: + def __init__(self, dimension=768): + self.dimension = dimension + self.vectors = [] + self.metadata = [] + self.texts = [] + + def add(self, text: str, metadata: Dict[str, Any] = None): + """Add document to store.""" + embedding = self._embed(text) + self.vectors.append(embedding) + self.metadata.append(metadata or {}) + self.texts.append(text) + return len(self.vectors) - 1 + + def search(self, query: str, limit: int = 5, + filters: Dict[str, Any] = None) -> List[Dict]: + """Search for similar documents.""" + query_embedding = self._embed(query) + + scores = [] + for i, vec in enumerate(self.vectors): + score = cosine_similarity(query_embedding, vec) + + # Apply filters + if filters and not self._matches_filters(self.metadata[i], filters): + score = -1 # Exclude + + scores.append((i, score)) + + # Sort by score + scores.sort(key=lambda x: x[1], reverse=True) + + # Return top k + results = [] + for idx, score in scores[:limit]: + if score > 0: # Only include positive matches + results.append({ + "index": idx, + "score": score, + "text": self._get_text(idx), + "metadata": self.metadata[idx] + }) + + return results + + def _embed(self, text: str) -> np.ndarray: + """Generate deterministic pseudo-embedding for demonstration. + In production, replace with actual embedding model.""" + np.random.seed(hash(text) % (2**32)) + vec = np.random.randn(self.dimension) + return vec / (np.linalg.norm(vec) + 1e-8) + + def _matches_filters(self, metadata: Dict, filters: Dict) -> bool: + """Check if metadata matches filters.""" + for key, value in filters.items(): + if key not in metadata: + return False + if isinstance(value, list): + if metadata[key] not in value: + return False + elif metadata[key] != value: + return False + return True + + def _get_text(self, index: int) -> str: + """Retrieve original text for index.""" + return self.texts[index] if index < len(self.texts) else "" +``` + +### 元数据增强向量存储 + +```python +class MetadataVectorStore(VectorStore): + def __init__(self, dimension=768): + super().__init__(dimension) + self.entity_index = {} # entity -> [indices] + self.time_index = {} # time_range -> [indices] + + def add(self, text: str, metadata: Dict[str, Any] = None): + """Add with enhanced indexing.""" + metadata = metadata or {} + index = super().add(text, metadata) + + # Index by entity + if "entity" in metadata: + entity = metadata["entity"] + if entity not in self.entity_index: + self.entity_index[entity] = [] + self.entity_index[entity].append(index) + + # Index by time + if "valid_from" in metadata: + time_key = self._time_range_key( + metadata.get("valid_from"), + metadata.get("valid_until") + ) + if time_key not in self.time_index: + self.time_index[time_key] = [] + self.time_index[time_key].append(index) + + return index + + def search_by_entity(self, query: str, entity: str, limit: int = 5) -> List[Dict]: + """Search within specific entity.""" + indices = self.entity_index.get(entity, []) + filtered = [self.metadata[i] for i in indices] + + # Score and rank + query_embedding = self._embed(query) + scored = [] + for i, meta in zip(indices, filtered): + vec = self.vectors[i] + score = cosine_similarity(query_embedding, vec) + scored.append((i, score, meta)) + + scored.sort(key=lambda x: x[1], reverse=True) + + return [{ + "index": idx, + "score": score, + "metadata": meta + } for idx, score, meta in scored[:limit]] +``` + +## 知识图谱实现 + +### 属性图存储 + +```python +from typing import Dict, List, Optional +import uuid + +class PropertyGraph: + def __init__(self): + self.nodes = {} # id -> properties + self.edges = [] # list of edge dicts + self.entity_registry = {} # name -> node_id (maintains identity) + self.indexes = { + "node_label": {}, # label -> [node_ids] + "edge_type": {} # type -> [edge_ids] + } + + def get_or_create_node(self, name: str, label: str, properties: Dict = None) -> str: + """Get existing node by name, or create a new one. + Uses entity_registry to ensure identity across interactions.""" + if name in self.entity_registry: + return self.entity_registry[name] + node_id = self.create_node(label, {**(properties or {}), "name": name}) + self.entity_registry[name] = node_id + return node_id + + def create_node(self, label: str, properties: Dict = None) -> str: + """Create node with label and properties.""" + node_id = str(uuid.uuid4()) + self.nodes[node_id] = { + "label": label, + "properties": properties or {} + } + + # Index by label + if label not in self.indexes["node_label"]: + self.indexes["node_label"][label] = [] + self.indexes["node_label"][label].append(node_id) + + return node_id + + def create_relationship(self, source_id: str, rel_type: str, + target_id: str, properties: Dict = None) -> str: + """Create directed relationship between nodes.""" + edge_id = str(uuid.uuid4()) + self.edges.append({ + "id": edge_id, + "source": source_id, + "target": target_id, + "type": rel_type, + "properties": properties or {} + }) + + # Index by type + if rel_type not in self.indexes["edge_type"]: + self.indexes["edge_type"][rel_type] = [] + self.indexes["edge_type"][rel_type].append(edge_id) + + return edge_id + + def query(self, cypher_like: str, params: Dict = None) -> List[Dict]: + """ + Simple query matching. + + Supports patterns like: + MATCH (e)-[r]->(o) WHERE e.id = $id RETURN r + """ + # In production, use actual graph database + # This is a simplified pattern matcher + results = [] + + if cypher_like.startswith("MATCH"): + # Parse basic pattern + pattern = self._parse_pattern(cypher_like) + results = self._match_pattern(pattern, params or {}) + + return results + + def _parse_pattern(self, query: str) -> Dict: + """Parse simplified MATCH pattern.""" + # Simplified parser for demonstration + return { + "source_label": self._extract_label(query, "source"), + "rel_type": self._extract_type(query), + "target_label": self._extract_label(query, "target"), + "where": self._extract_where(query) + } + + def _match_pattern(self, pattern: Dict, params: Dict) -> List[Dict]: + """Match pattern against graph.""" + results = [] + + for edge in self.edges: + # Match relationship type + if pattern["rel_type"] and edge["type"] != pattern["rel_type"]: + continue + + source = self.nodes.get(edge["source"], {}) + target = self.nodes.get(edge["target"], {}) + + # Match labels + if pattern["source_label"] and source.get("label") != pattern["source_label"]: + continue + if pattern["target_label"] and target.get("label") != pattern["target_label"]: + continue + + # Match where clause + if pattern["where"] and not self._match_where(edge, source, target, params): + continue + + results.append({ + "source": source, + "relationship": edge, + "target": target + }) + + return results +``` + +## 时序知识图谱 + +```python +from datetime import datetime +from typing import Optional + +class TemporalKnowledgeGraph(PropertyGraph): + def __init__(self): + super().__init__() + self.temporal_index = {} # time_range -> [edge_ids] + + def create_temporal_relationship( + self, + source_id: str, + rel_type: str, + target_id: str, + valid_from: datetime, + valid_until: Optional[datetime] = None, + properties: Dict = None + ) -> str: + """Create relationship with temporal validity.""" + edge_id = super().create_relationship( + source_id, rel_type, target_id, properties + ) + + # Index temporally + time_key = self._time_range_key(valid_from, valid_until) + if time_key not in self.temporal_index: + self.temporal_index[time_key] = [] + self.temporal_index[time_key].append(edge_id) + + # Store validity on edge + edge = self._get_edge(edge_id) + edge["valid_from"] = valid_from.isoformat() + edge["valid_until"] = valid_until.isoformat() if valid_until else None + + return edge_id + + def query_at_time(self, query: str, query_time: datetime) -> List[Dict]: + """Query graph state at specific time.""" + # Find edges valid at query time + valid_edges = [] + for edge in self.edges: + valid_from = datetime.fromisoformat(edge.get("valid_from", "1970-01-01")) + valid_until = edge.get("valid_until") + + if valid_from <= query_time: + if valid_until is None or datetime.fromisoformat(valid_until) > query_time: + valid_edges.append(edge) + + # Match against pattern + pattern = self._parse_pattern(query) + results = [] + + for edge in valid_edges: + if pattern["rel_type"] and edge["type"] != pattern["rel_type"]: + continue + + source = self.nodes.get(edge["source"], {}) + target = self.nodes.get(edge["target"], {}) + + results.append({ + "source": source, + "relationship": edge, + "target": target + }) + + return results + + def _time_range_key(self, start: datetime, end: Optional[datetime]) -> str: + """Create time range key for indexing.""" + start_str = start.isoformat() + end_str = end.isoformat() if end else "infinity" + return f"{start_str}::{end_str}" +``` + +## 记忆整合 + +```python +class MemoryConsolidator: + def __init__(self, graph: PropertyGraph, vector_store: VectorStore): + self.graph = graph + self.vector_store = vector_store + self.consolidation_threshold = 1000 # memories before consolidation + + def should_consolidate(self) -> bool: + """Check if consolidation should trigger.""" + total_memories = len(self.graph.nodes) + len(self.graph.edges) + return total_memories > self.consolidation_threshold + + def consolidate(self): + """Run consolidation process.""" + # Step 1: Identify duplicate or merged facts + duplicates = self.find_duplicates() + + # Step 2: Merge related facts + for group in duplicates: + self.merge_fact_group(group) + + # Step 3: Update validity periods + self.update_validity_periods() + + # Step 4: Rebuild indexes + self.rebuild_indexes() + + def find_duplicates(self) -> List[List]: + """Find groups of potentially duplicate facts.""" + # Group by subject and predicate + groups = {} + + for edge in self.graph.edges: + key = (edge["source"], edge["type"]) + if key not in groups: + groups[key] = [] + groups[key].append(edge) + + # Return groups with multiple edges + return [edges for edges in groups.values() if len(edges) > 1] + + def merge_fact_group(self, edges: List[Dict]): + """Merge group of duplicate edges.""" + if len(edges) == 1: + return + + # Keep most recent/relevant + keeper = max(edges, key=lambda e: e.get("properties", {}).get("confidence", 0)) + + # Merge metadata + for edge in edges: + if edge["id"] != keeper["id"]: + self.merge_properties(keeper, edge) + self.graph.edges.remove(edge) + + def merge_properties(self, target: Dict, source: Dict): + """Merge properties from source into target.""" + for key, value in source.get("properties", {}).items(): + if key not in target["properties"]: + target["properties"][key] = value + elif isinstance(value, list): + target["properties"][key].extend(value) +``` + +## 记忆-上下文集成 + +```python +class MemoryContextIntegrator: + def __init__(self, memory_system, context_limit=100000): + self.memory_system = memory_system + self.context_limit = context_limit + + def build_context(self, task: str, current_context: str = "") -> str: + """Build context including relevant memories.""" + # Extract entities from task + entities = self._extract_entities(task) + + # Retrieve memories for each entity + memories = [] + for entity in entities: + entity_memories = self.memory_system.retrieve_entity(entity) + memories.extend(entity_memories) + + # Format memories for context + memory_section = self._format_memories(memories) + + # Combine with current context + combined = current_context + "\n\n" + memory_section + + # Check limit and truncate if needed + if self._token_count(combined) > self.context_limit: + combined = self._truncate_context(combined, self.context_limit) + + return combined + + def _extract_entities(self, task: str) -> List[str]: + """Extract entity mentions from task.""" + # In production, use NER or entity extraction + import re + pattern = r"\[([^\]]+)\]" # [[entity_name]] convention + return re.findall(pattern, task) + + def _format_memories(self, memories: List[Dict]) -> str: + """Format memories for context injection.""" + sections = ["## Relevant Memories"] + + for memory in memories: + formatted = f"- {memory.get('content', '')}" + if "source" in memory: + formatted += f" (Source: {memory['source']})" + if "timestamp" in memory: + formatted += f" [Time: {memory['timestamp']}]" + sections.append(formatted) + + return "\n".join(sections) + + def _token_count(self, text: str) -> int: + """Estimate token count.""" + return len(text) // 4 # Rough approximation + + def _truncate_context(self, context: str, limit: int) -> str: + """Truncate context to fit limit.""" + tokens = context.split() + truncated = [] + count = 0 + + for token in tokens: + if count + 1 > limit: + break + truncated.append(token) + count += 1 + + return " ".join(truncated) +``` + +## 框架集成示例 + +### Mem0 快速入门 + +```python +from mem0 import Memory + +# Initialize with default config (uses local storage) +m = Memory() + +# Store memories with user scoping +m.add("Prefers Python 3.12 with type hints", user_id="dev-alice") +m.add("Working on microservices migration", user_id="dev-alice") + +# Search with natural language +results = m.search("What language does the user prefer?", user_id="dev-alice") + +# Batch operations +m.add([ + "Sprint goal: complete auth service", + "Blocked on database schema review" +], user_id="dev-alice") +``` + +### Graphiti（Zep 开源时序知识图谱引擎） + +```python +from graphiti_core import Graphiti +from graphiti_core.nodes import EpisodeType + +# Initialize with Neo4j backend +graphiti = Graphiti("bolt://localhost:7687", "neo4j", "password") + +# Add episodes (conversations, events) +await graphiti.add_episode( + name="user_conversation_42", + episode_body="Alice mentioned she moved to Berlin in January.", + source=EpisodeType.message, + source_description="Chat with Alice" +) + +# Search combines semantic, keyword, and graph traversal +results = await graphiti.search("Where does Alice live?") +``` + +### Cognee（AI 记忆开源知识引擎） + +```python +import cognee +from cognee.modules.search.types import SearchType + +# ECL pipeline: add → cognify → memify → search +await cognee.add("./docs/") +await cognee.add("any-data") +await cognee.cognify() +await cognee.memify() + +# Graph-aware retrieval (default: GRAPH_COMPLETION) +results = await cognee.search( + query_text="any query to search in memory", + query_type=SearchType.GRAPH_COMPLETION, +) + +# Raw chunks when agent reasons over text itself +chunks = await cognee.search( + query_text="any query to search in memory", + query_type=SearchType.CHUNKS, +) +``` diff --git a/scripts/memory_store.py b/scripts/memory_store.py new file mode 100644 index 0000000..6bb63fa --- /dev/null +++ b/scripts/memory_store.py @@ -0,0 +1,616 @@ +"""Memory System Implementation. + +Provides composable building blocks for agent memory: vector stores with +metadata indexing, property graphs for entity relationships, and temporal +knowledge graphs for facts that change over time. + +Use when: + - Building a memory persistence layer for an agent that must retain + knowledge across sessions. + - Prototyping memory architectures before committing to a production + framework (Mem0, Zep/Graphiti, Letta, Cognee). + - Combining semantic search with graph-based entity retrieval in a + single integrated system. + +Typical usage:: + + from memory_store import IntegratedMemorySystem + mem = IntegratedMemorySystem() + mem.start_session("session-001") + mem.store_fact("Alice prefers dark mode", entity="Alice") + results = mem.retrieve_memories("theme preference") +""" + +import hashlib +import json +from datetime import datetime +from typing import Any, Dict, List, Optional + +import numpy as np + +__all__ = [ + "VectorStore", + "PropertyGraph", + "TemporalKnowledgeGraph", + "IntegratedMemorySystem", +] + + +class VectorStore: + """Simple vector store with metadata indexing. + + Use when: the agent needs semantic similarity search over stored facts + with optional entity and temporal filtering. + """ + + def __init__(self, dimension: int = 768) -> None: + self.dimension: int = dimension + self.vectors: List[np.ndarray] = [] + self.metadata: List[Dict[str, Any]] = [] + self.entity_index: Dict[str, List[int]] = {} + self.time_index: Dict[str, List[int]] = {} + + def add(self, text: str, metadata: Optional[Dict[str, Any]] = None) -> int: + """Add document to store. + + Use when: persisting a new fact or observation that the agent should + be able to retrieve later via semantic search. + """ + metadata = metadata or {} + embedding: np.ndarray = self._embed(text) + index: int = len(self.vectors) + + self.vectors.append(embedding) + self.metadata.append(metadata) + + # Index by entity + if "entity" in metadata: + entity: str = metadata["entity"] + if entity not in self.entity_index: + self.entity_index[entity] = [] + self.entity_index[entity].append(index) + + # Index by time + if "valid_from" in metadata: + time_key: str = self._time_key(metadata["valid_from"]) + if time_key not in self.time_index: + self.time_index[time_key] = [] + self.time_index[time_key].append(index) + + return index + + def search( + self, + query: str, + limit: int = 5, + filters: Optional[Dict[str, Any]] = None, + ) -> List[Dict[str, Any]]: + """Search for similar documents. + + Use when: retrieving memories relevant to a query, optionally + narrowed by metadata filters (entity, session, time range). + """ + query_embedding: np.ndarray = self._embed(query) + + scores: List[tuple[int, float]] = [] + for i, vec in enumerate(self.vectors): + score: float = float( + np.dot(query_embedding, vec) + / (np.linalg.norm(query_embedding) * np.linalg.norm(vec) + 1e-8) + ) + + # Apply filters + if filters and not self._matches_filters(self.metadata[i], filters): + score = -1.0 + + scores.append((i, score)) + + scores.sort(key=lambda x: x[1], reverse=True) + + results: List[Dict[str, Any]] = [] + for idx, score in scores[:limit]: + if score > 0: + results.append( + { + "index": idx, + "score": score, + "text": self.metadata[idx].get("text", ""), + "metadata": self.metadata[idx], + } + ) + + return results + + def search_by_entity( + self, entity: str, query: str = "", limit: int = 5 + ) -> List[Dict[str, Any]]: + """Search within specific entity. + + Use when: the agent needs all memories associated with a known + entity, optionally ranked by relevance to a query. + """ + indices: List[int] = self.entity_index.get(entity, []) + + if not indices: + return [] + + if query: + query_embedding: np.ndarray = self._embed(query) + scored: List[tuple[int, float, Dict[str, Any]]] = [] + for i in indices: + vec: np.ndarray = self.vectors[i] + score: float = float( + np.dot(query_embedding, vec) + / (np.linalg.norm(query_embedding) * np.linalg.norm(vec) + 1e-8) + ) + scored.append((i, score, self.metadata[i])) + + scored.sort(key=lambda x: x[1], reverse=True) + return [ + {"index": i, "score": s, "metadata": m} + for i, s, m in scored[:limit] + ] + else: + return [ + {"index": i, "score": 1.0, "metadata": self.metadata[i]} + for i in indices[:limit] + ] + + def _embed(self, text: str) -> np.ndarray: + """Generate embedding for text. + + In production, replace with an actual embedding model. This + deterministic stub uses the text hash as a random seed so that + identical texts always produce identical vectors. Uses a local + RNG to avoid corrupting global numpy random state. + """ + rng = np.random.default_rng(hash(text) % (2**32)) + return rng.standard_normal(self.dimension) + + def _time_key(self, timestamp: Any) -> str: + """Create time key for indexing.""" + if isinstance(timestamp, datetime): + return timestamp.strftime("%Y-%m") + return str(timestamp) + + def _matches_filters(self, metadata: Dict[str, Any], filters: Dict[str, Any]) -> bool: + """Check if metadata matches filters.""" + for key, value in filters.items(): + if key not in metadata: + return False + if isinstance(value, list): + if metadata[key] not in value: + return False + elif metadata[key] != value: + return False + return True + + +class PropertyGraph: + """Simple property graph storage. + + Use when: the agent needs to maintain entity relationships and + traverse connections between nodes (e.g., "find all projects + associated with this user"). + """ + + def __init__(self) -> None: + self.nodes: Dict[str, Dict[str, Any]] = {} + self.edges: Dict[str, Dict[str, Any]] = {} + self.entity_registry: Dict[str, str] = {} # name -> node_id + self.node_index: Dict[str, List[str]] = {} # label -> node_ids + self.edge_index: Dict[str, List[str]] = {} # type -> edge_ids + + def get_or_create_node( + self, name: str, label: str = "Entity", properties: Optional[Dict[str, Any]] = None + ) -> str: + """Get existing node by name, or create a new one. + + Use when: storing an entity that may already exist. The entity + registry ensures identity is maintained across interactions + ("John Doe" always maps to the same node). + """ + if name in self.entity_registry: + node_id: str = self.entity_registry[name] + if properties: + self.nodes[node_id]["properties"].update(properties) + return node_id + node_id = self.create_node(label, {**(properties or {}), "name": name}) + self.entity_registry[name] = node_id + return node_id + + def create_node(self, label: str, properties: Optional[Dict[str, Any]] = None) -> str: + """Create node with label and properties. + + Use when: adding a new entity to the graph that does not need + identity deduplication (prefer get_or_create_node otherwise). + """ + node_id: str = hashlib.md5(f"{label}{datetime.now().isoformat()}".encode()).hexdigest()[:16] + + self.nodes[node_id] = { + "id": node_id, + "label": label, + "properties": properties or {}, + "created_at": datetime.now().isoformat(), + } + + if label not in self.node_index: + self.node_index[label] = [] + self.node_index[label].append(node_id) + + return node_id + + def create_relationship( + self, + source_id: str, + rel_type: str, + target_id: str, + properties: Optional[Dict[str, Any]] = None, + ) -> str: + """Create directed relationship between nodes. + + Use when: recording a connection between two entities (e.g., + WORKS_AT, LIVES_IN, DEPENDS_ON). + """ + if source_id not in self.nodes: + raise ValueError(f"Unknown source node: {source_id}") + if target_id not in self.nodes: + raise ValueError(f"Unknown target node: {target_id}") + + edge_id: str = hashlib.md5( + f"{source_id}{rel_type}{target_id}{datetime.now().isoformat()}".encode() + ).hexdigest()[:16] + + self.edges[edge_id] = { + "id": edge_id, + "source": source_id, + "target": target_id, + "type": rel_type, + "properties": properties or {}, + "created_at": datetime.now().isoformat(), + } + + if rel_type not in self.edge_index: + self.edge_index[rel_type] = [] + self.edge_index[rel_type].append(edge_id) + + return edge_id + + def query(self, pattern: Dict[str, Any]) -> List[Dict[str, Any]]: + """Query graph with simple pattern matching. + + Use when: finding relationships that match a structural pattern + (e.g., all WORKS_AT edges from Person nodes). + """ + results: List[Dict[str, Any]] = [] + + # Match by edge type + if "type" in pattern: + edge_ids: List[str] = self.edge_index.get(pattern["type"], []) + for eid in edge_ids: + edge: Dict[str, Any] = self.edges[eid] + source: Dict[str, Any] = self.nodes.get(edge["source"], {}) + target: Dict[str, Any] = self.nodes.get(edge["target"], {}) + + # Match source label + if "source_label" in pattern: + if source.get("label") != pattern["source_label"]: + continue + + # Match target label + if "target_label" in pattern: + if target.get("label") != pattern["target_label"]: + continue + + results.append({"source": source, "edge": edge, "target": target}) + + return results + + def get_node(self, node_id: str) -> Optional[Dict[str, Any]]: + """Get node by ID.""" + return self.nodes.get(node_id) + + def get_relationships( + self, node_id: str, direction: str = "both" + ) -> List[Dict[str, Any]]: + """Get relationships for a node. + + Use when: retrieving all connections for a given entity to build + a complete entity context. + """ + relationships: List[Dict[str, Any]] = [] + + for edge in self.edges.values(): + if direction in ["outgoing", "both"] and edge["source"] == node_id: + relationships.append( + { + "edge": edge, + "target": self.nodes.get(edge["target"]), + "direction": "outgoing", + } + ) + if direction in ["incoming", "both"] and edge["target"] == node_id: + relationships.append( + { + "edge": edge, + "source": self.nodes.get(edge["source"]), + "direction": "incoming", + } + ) + + return relationships + + +class TemporalKnowledgeGraph(PropertyGraph): + """Property graph with temporal validity for facts. + + Use when: the agent must track facts that change over time and + answer time-scoped queries (e.g., "where did the user live in + March 2024?"). + """ + + def create_temporal_relationship( + self, + source_id: str, + rel_type: str, + target_id: str, + valid_from: datetime, + valid_until: Optional[datetime] = None, + properties: Optional[Dict[str, Any]] = None, + ) -> str: + """Create relationship with temporal validity. + + Use when: recording a fact that has a known start time and + may expire (e.g., employment, address, subscription status). + """ + edge_id: str = super().create_relationship( + source_id, rel_type, target_id, properties + ) + + # Add temporal properties + self.edges[edge_id]["valid_from"] = valid_from.isoformat() + self.edges[edge_id]["valid_until"] = ( + valid_until.isoformat() if valid_until else None + ) + + return edge_id + + def query_at_time( + self, query: Dict[str, Any], query_time: datetime + ) -> List[Dict[str, Any]]: + """Query graph state at specific time. + + Use when: answering point-in-time questions about entities + (e.g., "what was true on date X?"). + """ + results: List[Dict[str, Any]] = [] + + # Get base query results + base_results: List[Dict[str, Any]] = self.query(query) + + for result in base_results: + edge: Dict[str, Any] = result["edge"] + valid_from: datetime = datetime.fromisoformat( + edge.get("valid_from", "1970-01-01") + ) + valid_until: Optional[str] = edge.get("valid_until") + + # Check temporal validity + if valid_from <= query_time: + if valid_until is None or datetime.fromisoformat(valid_until) > query_time: + results.append( + { + **result, + "valid_from": valid_from, + "valid_until": valid_until, + } + ) + + return results + + def query_time_range( + self, + query: Dict[str, Any], + start_time: datetime, + end_time: datetime, + ) -> List[Dict[str, Any]]: + """Query facts valid during time range. + + Use when: retrieving all facts that overlap with a given time + window (e.g., "what changed between January and June?"). + """ + results: List[Dict[str, Any]] = [] + + base_results: List[Dict[str, Any]] = self.query(query) + + for result in base_results: + edge: Dict[str, Any] = result["edge"] + valid_from: datetime = datetime.fromisoformat( + edge.get("valid_from", "1970-01-01") + ) + valid_until: Optional[str] = edge.get("valid_until") + + # Check if overlaps with query range + until_dt: datetime = ( + datetime.fromisoformat(valid_until) if valid_until else datetime.max + ) + + if until_dt >= start_time and valid_from <= end_time: + results.append( + { + **result, + "valid_from": valid_from, + "valid_until": valid_until, + } + ) + + return results + + +# --------------------------------------------------------------------------- +# Memory System Integration +# --------------------------------------------------------------------------- + + +class IntegratedMemorySystem: + """Integrated memory system combining vector store and graph. + + Use when: the agent needs both semantic search over facts and + graph-based entity relationship traversal in a single unified + interface. This class composes VectorStore and TemporalKnowledgeGraph, + enriching vector search results with graph context. + """ + + def __init__(self) -> None: + self.vector_store: VectorStore = VectorStore() + self.graph: TemporalKnowledgeGraph = TemporalKnowledgeGraph() + self.session_id: str = "" + + def start_session(self, session_id: str) -> None: + """Start a new memory session. + + Use when: beginning a new conversation or task that should + scope its memories to a distinct session identifier. + """ + self.session_id = session_id + + def store_fact( + self, + fact: str, + entity: str, + timestamp: Optional[datetime] = None, + relationships: Optional[List[Dict[str, Any]]] = None, + ) -> None: + """Store a fact with entity and relationships. + + Use when: the agent observes a new piece of information that + should be persisted for future retrieval. Stores in both the + vector store (for semantic search) and the graph (for entity + traversal). + """ + # Store in vector store + self.vector_store.add( + fact, + { + "text": fact, + "entity": entity, + "valid_from": (timestamp or datetime.now()).isoformat(), + "session_id": self.session_id, + }, + ) + + # Get or create entity node (uses registry for identity) + entity_node_id: str = self.graph.get_or_create_node(entity) + + # Create relationships + if relationships: + for rel in relationships: + target_node_id: str = self.graph.get_or_create_node(rel["target"]) + self.graph.create_relationship( + entity_node_id, + rel["type"], + target_node_id, + properties=rel.get("properties", {}), + ) + + def retrieve_memories( + self, + query: str, + entity_filter: Optional[str] = None, + time_filter: Optional[Dict[str, Any]] = None, + limit: int = 5, + ) -> List[Dict[str, Any]]: + """Retrieve memories matching query. + + Use when: the agent needs to recall previously stored facts, + optionally filtered by entity or time. Results are enriched + with graph relationships for each matched entity. + """ + # Vector search + filters: Dict[str, Any] = {"session_id": self.session_id} + if entity_filter: + filters["entity"] = entity_filter + + results: List[Dict[str, Any]] = self.vector_store.search( + query, limit=limit, filters=filters + ) + + # Enrich with graph relationships + for result in results: + entity: Optional[str] = result["metadata"].get("entity") + if entity: + node_id: Optional[str] = self.graph.entity_registry.get(entity) + if node_id: + result["relationships"] = self.graph.get_relationships(node_id) + + return results + + def retrieve_entity_context(self, entity: str) -> Dict[str, Any]: + """Retrieve complete context for an entity. + + Use when: the agent needs a full picture of a single entity + including its properties, all relationships, and associated + vector memories. + """ + node_id: Optional[str] = self.graph.entity_registry.get(entity) + + # Get entity node + entity_node: Optional[Dict[str, Any]] = ( + self.graph.get_node(node_id) if node_id else None + ) + + # Get relationships + relationships: List[Dict[str, Any]] = ( + self.graph.get_relationships(node_id) if node_id else [] + ) + + # Get vector memories + memories: List[Dict[str, Any]] = self.vector_store.search_by_entity( + entity, limit=10 + ) + + return { + "entity": entity_node, + "relationships": relationships, + "memories": memories, + } + + def consolidate(self) -> None: + """Consolidate memories and remove outdated information. + + Use when: memory count exceeds a threshold, retrieval quality + degrades, or on a scheduled interval. In production, implement: + - Merge related facts into summaries + - Update validity periods on stale entries + - Archive obsolete facts (invalidate, do not discard) + """ + pass + + +if __name__ == "__main__": + # Quick smoke test demonstrating the integrated memory system. + mem = IntegratedMemorySystem() + mem.start_session("demo-session") + + # Store facts with entity relationships + mem.store_fact( + "Alice prefers dark mode", + entity="Alice", + relationships=[{"target": "dark mode", "type": "PREFERS"}], + ) + mem.store_fact( + "Alice works at Acme Corp", + entity="Alice", + relationships=[{"target": "Acme Corp", "type": "WORKS_AT"}], + ) + + # Semantic retrieval + results = mem.retrieve_memories("theme preference") + print(f"Search results: {len(results)} memories found") + for r in results: + print(f" score={r['score']:.3f} text={r['text']}") + + # Entity context + context = mem.retrieve_entity_context("Alice") + print(f"\nAlice context: {len(context['relationships'])} relationships, " + f"{len(context['memories'])} memories")