chore: import zh skill memory-systems

2026-07-13 21:36:25 +08:00
commit 89f0a1771f
4 changed files with 1409 additions and 0 deletions
@@ -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 清单整理
+- 原作者、版权和许可证信息以上游仓库为准
@@ -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
@@ -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,
+)
+```
@@ -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")