| """ |
| STEW — Vector Memory Engine (ChromaDB) |
| Long-term semantic memory: stores facts, documents, and project context |
| that can be retrieved by semantic similarity. |
| """ |
| import hashlib |
| from typing import Optional |
| from config import Config |
|
|
|
|
| class VectorMemory: |
| """ |
| Persistent semantic memory using ChromaDB. |
| Falls back to in-memory store if ChromaDB is not available. |
| """ |
|
|
| def __init__(self, persist_dir: str = "/tmp/stew_vector_db"): |
| self._client = None |
| self._collection = None |
| self._persist_dir = persist_dir |
| self._fallback: list[dict] = [] |
|
|
| def _init_chroma(self): |
| if self._collection is not None: |
| return True |
| try: |
| import chromadb |
| self._client = chromadb.PersistentClient(path=self._persist_dir) |
| self._collection = self._client.get_or_create_collection( |
| name="stew_memory", |
| metadata={"hnsw:space": "cosine"}, |
| ) |
| return True |
| except Exception: |
| return False |
|
|
| |
|
|
| def store(self, text: str, metadata: dict | None = None, doc_id: str | None = None) -> str: |
| """Store a text chunk in vector memory.""" |
| if not doc_id: |
| doc_id = hashlib.sha256(text.encode()).hexdigest()[:16] |
| meta = metadata or {} |
|
|
| if self._init_chroma(): |
| try: |
| self._collection.upsert( |
| ids=[doc_id], |
| documents=[text], |
| metadatas=[meta], |
| ) |
| return f"✅ Stored in vector memory (id: {doc_id})" |
| except Exception as e: |
| return f"Vector store error: {e}" |
| else: |
| |
| self._fallback.append({"id": doc_id, "text": text, "meta": meta}) |
| return f"✅ Stored in fallback memory (id: {doc_id})" |
|
|
| def search(self, query: str, n_results: int = 5) -> str: |
| """Semantic search over stored memory.""" |
| if self._init_chroma(): |
| try: |
| results = self._collection.query( |
| query_texts=[query], |
| n_results=min(n_results, self._collection.count() or 1), |
| ) |
| docs = results.get("documents", [[]])[0] |
| if not docs: |
| return "No relevant memories found." |
| lines = [f"[{i+1}] {d[:300]}" for i, d in enumerate(docs)] |
| return "🧠 Relevant memories:\n" + "\n\n".join(lines) |
| except Exception as e: |
| return f"Vector search error: {e}" |
| else: |
| |
| matches = [f["text"] for f in self._fallback if query.lower() in f["text"].lower()] |
| return ("🧠 Matches:\n" + "\n".join(m[:300] for m in matches[:5])) if matches else "No matches." |
|
|
| def store_project(self, project_name: str, description: str) -> str: |
| """Store a project description.""" |
| return self.store( |
| f"Project: {project_name}\n{description}", |
| metadata={"type": "project", "name": project_name}, |
| doc_id=f"project_{project_name[:20]}", |
| ) |
|
|
| def count(self) -> int: |
| if self._init_chroma(): |
| try: |
| return self._collection.count() |
| except Exception: |
| pass |
| return len(self._fallback) |
|
|
| def clear(self) -> str: |
| if self._init_chroma(): |
| try: |
| self._client.delete_collection("stew_memory") |
| self._collection = None |
| return "🗑️ Vector memory cleared." |
| except Exception as e: |
| return f"Clear error: {e}" |
| self._fallback.clear() |
| return "🗑️ Fallback memory cleared." |
|
|
|
|
| |
| _vector_memory: Optional[VectorMemory] = None |
|
|
| def get_vector_memory() -> VectorMemory: |
| global _vector_memory |
| if _vector_memory is None: |
| _vector_memory = VectorMemory() |
| return _vector_memory |
| |
