tiphys.memory

20@dataclass
21class MemoryEntry:
22    """A single memory entry."""
23
24    id: str
25    content: str
26    metadata: dict[str, Any] = field(default_factory=dict)
27    created_at: datetime = field(default_factory=datetime.utcnow)
28    updated_at: datetime = field(default_factory=datetime.utcnow)
29    embedding: list[float] | None = None
30    content_hash: str = ""
31
32    def __post_init__(self) -> None:
33        """Calculate content hash if not provided."""
34        if not self.content_hash:
35            self.content_hash = hashlib.sha256(self.content.encode()).hexdigest()[:16]

 47class MemoryStore:
 48    """
 49    In-memory storage for agent memories.
 50
 51    Provides basic CRUD operations and keyword search.
 52    Can be extended with vector search when embeddings are available.
 53    """
 54
 55    def __init__(self) -> None:
 56        """Initialize the memory store."""
 57        self._entries: dict[str, MemoryEntry] = {}
 58        self._by_hash: dict[str, str] = {}  # content_hash -> id
 59
 60    def add(
 61        self,
 62        content: str,
 63        metadata: dict[str, Any] | None = None,
 64        entry_id: str | None = None,
 65        embedding: list[float] | None = None,
 66    ) -> MemoryEntry:
 67        """
 68        Add a memory entry.
 69
 70        Args:
 71            content: The memory content.
 72            metadata: Optional metadata dict.
 73            entry_id: Optional custom ID (generated if not provided).
 74            embedding: Optional pre-computed embedding vector.
 75
 76        Returns:
 77            The created memory entry.
 78        """
 79        entry_id = entry_id or str(uuid4())
 80
 81        entry = MemoryEntry(
 82            id=entry_id,
 83            content=content,
 84            metadata=metadata or {},
 85            embedding=embedding,
 86        )
 87
 88        # Check for duplicate content
 89        if entry.content_hash in self._by_hash:
 90            existing_id = self._by_hash[entry.content_hash]
 91            existing = self._entries.get(existing_id)
 92            if existing:
 93                # Update existing entry
 94                existing.updated_at = datetime.utcnow()
 95                if metadata:
 96                    existing.metadata.update(metadata)
 97                if embedding:
 98                    existing.embedding = embedding
 99                return existing
100
101        self._entries[entry_id] = entry
102        self._by_hash[entry.content_hash] = entry_id
103
104        return entry
105
106    def get(self, entry_id: str) -> MemoryEntry | None:
107        """Get a memory entry by ID."""
108        return self._entries.get(entry_id)
109
110    def update(
111        self,
112        entry_id: str,
113        content: str | None = None,
114        metadata: dict[str, Any] | None = None,
115        embedding: list[float] | None = None,
116    ) -> MemoryEntry | None:
117        """
118        Update a memory entry.
119
120        Args:
121            entry_id: ID of the entry to update.
122            content: New content (optional).
123            metadata: Metadata to merge (optional).
124            embedding: New embedding (optional).
125
126        Returns:
127            Updated entry or None if not found.
128        """
129        entry = self._entries.get(entry_id)
130        if not entry:
131            return None
132
133        if content is not None:
134            # Remove old hash mapping
135            self._by_hash.pop(entry.content_hash, None)
136
137            entry.content = content
138            entry.content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
139
140            # Add new hash mapping
141            self._by_hash[entry.content_hash] = entry_id
142
143        if metadata is not None:
144            entry.metadata.update(metadata)
145
146        if embedding is not None:
147            entry.embedding = embedding
148
149        entry.updated_at = datetime.utcnow()
150
151        return entry
152
153    def delete(self, entry_id: str) -> bool:
154        """
155        Delete a memory entry.
156
157        Args:
158            entry_id: ID of the entry to delete.
159
160        Returns:
161            True if deleted, False if not found.
162        """
163        entry = self._entries.pop(entry_id, None)
164        if entry:
165            self._by_hash.pop(entry.content_hash, None)
166            return True
167        return False
168
169    def search(
170        self,
171        query: str,
172        limit: int = 10,
173        metadata_filter: dict[str, Any] | None = None,
174    ) -> list[SearchResult]:
175        """
176        Search memories using keyword matching.
177
178        Args:
179            query: Search query string.
180            limit: Maximum results to return.
181            metadata_filter: Filter by metadata fields.
182
183        Returns:
184            List of search results sorted by relevance.
185        """
186        query_lower = query.lower()
187        query_words = set(query_lower.split())
188        results: list[SearchResult] = []
189
190        for entry in self._entries.values():
191            # Apply metadata filter
192            if metadata_filter and not all(
193                entry.metadata.get(k) == v for k, v in metadata_filter.items()
194            ):
195                continue
196
197            # Score based on keyword matching
198            content_lower = entry.content.lower()
199
200            # Exact phrase match
201            if query_lower in content_lower:
202                results.append(SearchResult(entry=entry, score=1.0, match_type="exact"))
203                continue
204
205            # Word overlap score
206            content_words = set(content_lower.split())
207            overlap = len(query_words & content_words)
208            if overlap > 0:
209                score = overlap / len(query_words)
210                results.append(SearchResult(entry=entry, score=score, match_type="keyword"))
211
212        # Sort by score descending
213        results.sort(key=lambda r: r.score, reverse=True)
214
215        return results[:limit]
216
217    def search_by_metadata(
218        self,
219        metadata_filter: dict[str, Any],
220        limit: int = 100,
221    ) -> list[MemoryEntry]:
222        """
223        Search memories by metadata fields.
224
225        Args:
226            metadata_filter: Metadata fields to match.
227            limit: Maximum results.
228
229        Returns:
230            List of matching entries.
231        """
232        results: list[MemoryEntry] = []
233
234        for entry in self._entries.values():
235            if all(entry.metadata.get(k) == v for k, v in metadata_filter.items()):
236                results.append(entry)
237                if len(results) >= limit:
238                    break
239
240        return results
241
242    def list_all(self, limit: int = 100, offset: int = 0) -> list[MemoryEntry]:
243        """
244        List all memory entries.
245
246        Args:
247            limit: Maximum entries to return.
248            offset: Number of entries to skip.
249
250        Returns:
251            List of memory entries.
252        """
253        entries = list(self._entries.values())
254        # Sort by created_at descending (newest first)
255        entries.sort(key=lambda e: e.created_at, reverse=True)
256        return entries[offset : offset + limit]
257
258    def count(self) -> int:
259        """Get total number of entries."""
260        return len(self._entries)
261
262    def clear(self) -> int:
263        """
264        Clear all entries.
265
266        Returns:
267            Number of entries cleared.
268        """
269        count = len(self._entries)
270        self._entries.clear()
271        self._by_hash.clear()
272        return count

38@dataclass
39class SearchResult:
40    """Result from a memory search."""
41
42    entry: MemoryEntry
43    score: float = 1.0
44    match_type: str = "exact"  # "exact", "keyword", "semantic"

457def get_memory_store() -> ScopedMemoryStore:
458    """Get the global memory store instance."""
459    global _memory_store
460    if _memory_store is None:
461        from pathlib import Path
462
463        path = str(Path.home() / ".tiphys" / "memory")
464        _memory_store = ScopedMemoryStore(base_path=path)
465    return _memory_store

17class EmbeddingProvider(ABC):
18    """Abstract base class for embedding providers."""
19
20    @property
21    @abstractmethod
22    def dimension(self) -> int:
23        """Get the embedding dimension."""
24        ...
25
26    @property
27    @abstractmethod
28    def model_name(self) -> str:
29        """Get the model name."""
30        ...
31
32    @abstractmethod
33    async def embed(self, text: str) -> list[float]:
34        """
35        Generate embedding for a single text.
36
37        Args:
38            text: Text to embed.
39
40        Returns:
41            Embedding vector.
42        """
43        ...
44
45    @abstractmethod
46    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
47        """
48        Generate embeddings for multiple texts.
49
50        Args:
51            texts: List of texts to embed.
52
53        Returns:
54            List of embedding vectors.
55        """
56        ...

59class DummyEmbeddingProvider(EmbeddingProvider):
60    """
61    Dummy embedding provider for testing.
62
63    Generates deterministic hash-based embeddings.
64    """
65
66    def __init__(self, dimension: int = 384) -> None:
67        """Initialize with specified dimension."""
68        self._dimension = dimension
69
70    @property
71    def dimension(self) -> int:
72        """Get the embedding dimension."""
73        return self._dimension
74
75    @property
76    def model_name(self) -> str:
77        """Get the model name."""
78        return "dummy"
79
80    async def embed(self, text: str) -> list[float]:
81        """Generate a deterministic embedding based on text hash."""
82        # Use text hash to generate reproducible embedding
83        text_hash = hashlib.sha256(text.encode()).digest()
84
85        # Convert bytes to floats in range [-1, 1]
86        embedding: list[float] = []
87        for i in range(self._dimension):
88            byte_idx = i % len(text_hash)
89            # Normalize to [-1, 1]
90            value = (text_hash[byte_idx] / 127.5) - 1.0
91            embedding.append(value)
92
93        return embedding
94
95    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
96        """Generate embeddings for multiple texts."""
97        return [await self.embed(text) for text in texts]

159class EmbeddingCache:
160    """
161    Cache for embeddings indexed by content hash.
162
163    Avoids redundant embedding generation for repeated content.
164    """
165
166    def __init__(self) -> None:
167        """Initialize the cache."""
168        self._cache: dict[str, list[float]] = {}
169        self._hits = 0
170        self._misses = 0
171
172    def _hash_text(self, text: str) -> str:
173        """Generate hash for text."""
174        return hashlib.sha256(text.encode()).hexdigest()[:32]
175
176    def get(self, text: str) -> list[float] | None:
177        """
178        Get cached embedding for text.
179
180        Args:
181            text: The text to look up.
182
183        Returns:
184            Cached embedding or None if not found.
185        """
186        key = self._hash_text(text)
187        embedding = self._cache.get(key)
188
189        if embedding is not None:
190            self._hits += 1
191        else:
192            self._misses += 1
193
194        return embedding
195
196    def set(self, text: str, embedding: list[float]) -> None:
197        """
198        Cache an embedding.
199
200        Args:
201            text: The text that was embedded.
202            embedding: The embedding vector.
203        """
204        key = self._hash_text(text)
205        self._cache[key] = embedding
206
207    def clear(self) -> int:
208        """
209        Clear the cache.
210
211        Returns:
212            Number of entries cleared.
213        """
214        count = len(self._cache)
215        self._cache.clear()
216        return count
217
218    @property
219    def size(self) -> int:
220        """Get number of cached embeddings."""
221        return len(self._cache)
222
223    @property
224    def hit_rate(self) -> float:
225        """Get cache hit rate."""
226        total = self._hits + self._misses
227        if total == 0:
228            return 0.0
229        return self._hits / total

232class EmbeddingService:
233    """
234    High-level service for generating embeddings with caching.
235
236    Combines a provider with a cache for efficient embedding generation.
237    """
238
239    def __init__(
240        self,
241        provider: EmbeddingProvider | None = None,
242        cache: EmbeddingCache | None = None,
243    ) -> None:
244        """
245        Initialize the embedding service.
246
247        Args:
248            provider: Embedding provider. Uses dummy provider if not provided.
249            cache: Embedding cache. Creates new cache if not provided.
250        """
251        self._provider = provider or DummyEmbeddingProvider()
252        self._cache = cache or EmbeddingCache()
253
254    @property
255    def provider(self) -> EmbeddingProvider:
256        """Get the embedding provider."""
257        return self._provider
258
259    @property
260    def cache(self) -> EmbeddingCache:
261        """Get the embedding cache."""
262        return self._cache
263
264    @property
265    def dimension(self) -> int:
266        """Get the embedding dimension."""
267        return self._provider.dimension
268
269    async def embed(self, text: str, use_cache: bool = True) -> list[float]:
270        """
271        Generate embedding for text.
272
273        Args:
274            text: Text to embed.
275            use_cache: Whether to use caching.
276
277        Returns:
278            Embedding vector.
279        """
280        if use_cache:
281            cached = self._cache.get(text)
282            if cached is not None:
283                return cached
284
285        embedding = await self._provider.embed(text)
286
287        if use_cache:
288            self._cache.set(text, embedding)
289
290        return embedding
291
292    async def embed_batch(self, texts: list[str], use_cache: bool = True) -> list[list[float]]:
293        """
294        Generate embeddings for multiple texts.
295
296        Uses cache for individual texts that have been seen before.
297
298        Args:
299            texts: List of texts to embed.
300            use_cache: Whether to use caching.
301
302        Returns:
303            List of embedding vectors.
304        """
305        if not use_cache:
306            return await self._provider.embed_batch(texts)
307
308        embeddings: list[list[float] | None] = []
309        texts_to_embed: list[tuple[int, str]] = []
310
311        # Check cache for each text
312        for i, text in enumerate(texts):
313            cached = self._cache.get(text)
314            if cached is not None:
315                embeddings.append(cached)
316            else:
317                embeddings.append(None)
318                texts_to_embed.append((i, text))
319
320        # Generate missing embeddings
321        if texts_to_embed:
322            missing_texts = [t[1] for t in texts_to_embed]
323            new_embeddings = await self._provider.embed_batch(missing_texts)
324
325            # Fill in results and cache
326            for (idx, text), embedding in zip(texts_to_embed, new_embeddings):
327                embeddings[idx] = embedding
328                self._cache.set(text, embedding)
329
330        # Type narrowing - all None values should be filled
331        return [e for e in embeddings if e is not None]

334def cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
335    """
336    Calculate cosine similarity between two vectors.
337
338    Args:
339        vec1: First vector.
340        vec2: Second vector.
341
342    Returns:
343        Cosine similarity in range [-1, 1].
344    """
345    if len(vec1) != len(vec2):
346        raise ValueError("Vectors must have the same dimension")
347
348    dot_product: float = sum(a * b for a, b in zip(vec1, vec2))
349    norm1: float = sum(a * a for a in vec1) ** 0.5
350    norm2: float = sum(b * b for b in vec2) ** 0.5
351
352    if norm1 == 0 or norm2 == 0:
353        return 0.0
354
355    result: float = dot_product / (norm1 * norm2)
356    return result

363def get_embedding_service() -> EmbeddingService:
364    """Get the global embedding service instance."""
365    global _embedding_service
366    if _embedding_service is None:
367        _embedding_service = EmbeddingService()
368    return _embedding_service