# 记忆系统：技术参考

本文档提供了记忆系统组件的实现细节。

## 向量存储实现

### 基础向量存储

```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,
)
```