feat(llm): add embedding adapter with cache and similarity utils (S1.3)
Add OpenAI-compatible embedding support (works with both OpenAI and OpenRouter), file-based embedding cache with content-digest invalidation, and pure-Python cosine similarity utilities for downstream redundancy detection. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
64
markitect/llm/embedding_cache.py
Normal file
64
markitect/llm/embedding_cache.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""
|
||||
File-based embedding cache.
|
||||
|
||||
Stores embedding vectors in a single JSON file keyed by entity slug.
|
||||
Each entry includes a content digest so stale embeddings are
|
||||
automatically invalidated when entity content changes.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class EmbeddingCache:
|
||||
"""Persistent cache for embedding vectors.
|
||||
|
||||
Structure on disk (``embeddings.json``)::
|
||||
|
||||
{
|
||||
"division-of-labour": {"digest": "abc123", "vector": [0.1, ...]},
|
||||
...
|
||||
}
|
||||
"""
|
||||
|
||||
def __init__(self, cache_dir: Path):
|
||||
self._path = cache_dir / "embeddings.json"
|
||||
self._data: dict[str, dict] = {}
|
||||
self._hits = 0
|
||||
self._misses = 0
|
||||
self._load()
|
||||
|
||||
def get(self, slug: str, content_digest: str) -> Optional[list[float]]:
|
||||
"""Return the cached vector if *content_digest* matches, else ``None``."""
|
||||
entry = self._data.get(slug)
|
||||
if entry is not None and entry.get("digest") == content_digest:
|
||||
self._hits += 1
|
||||
return entry["vector"]
|
||||
self._misses += 1
|
||||
return None
|
||||
|
||||
def put(self, slug: str, content_digest: str, vector: list[float]) -> None:
|
||||
"""Store or overwrite the embedding for *slug*."""
|
||||
self._data[slug] = {"digest": content_digest, "vector": vector}
|
||||
|
||||
def save(self) -> None:
|
||||
"""Write cache to disk."""
|
||||
self._path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._path.write_text(json.dumps(self._data, separators=(",", ":")))
|
||||
|
||||
def stats(self) -> dict:
|
||||
"""Return cache statistics."""
|
||||
return {
|
||||
"entries": len(self._data),
|
||||
"hits": self._hits,
|
||||
"misses": self._misses,
|
||||
}
|
||||
|
||||
def _load(self) -> None:
|
||||
"""Read cache from disk if it exists."""
|
||||
if self._path.is_file():
|
||||
try:
|
||||
self._data = json.loads(self._path.read_text())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
self._data = {}
|
||||
Reference in New Issue
Block a user