Persistent embedding cache

Completion

Task: Implement the core caching logic with float32 round-trip

You are given a partially written function get_cached_embedding(text, model, cache) that should mimic the per-text caching logic used in cached_text_embeddings from the repository source. Your job is to complete the two missing parts:

Compute the cache key using embed_cache_key(text, model=model) — this generates a SHA‑256 hash of "{model}\x1f{text}" (as seen in the source).
Round‑trip the embedding vector through float32 before storing it in the cache. This ensures that a freshly embedded (miss) vector returns exactly the same float values as a later cache hit that reads from the packed float32 BLOB — avoiding the miss/hit precision asymmetry described in the comment.

The scaffold provides a mock embed_model that returns a dummy vector, a simple dictionary cache (like the SQLite store), and the embed_cache_key definition from the repository. Fill in the # TODO: lines.

Important: Use the exact same functions and variable names as in the real source: hashlib.sha256, array.array('f'), .tobytes(), .frombytes(), and the key construction with \x1f (chr(31)).

Your code

import hashlib
import array

def embed_cache_key(text: str, *, model: str | None = None, namespace: str = "") -> str:
    """Cache key = sha256(model \x1f text). Copied from the source."""
    h = hashlib.sha256()
    if namespace:
        payload = f"{model or 'BAAI/bge-small-en-v1.5'}\x1f{namespace}\x1f{text}"
    else:
        payload = f"{model or 'BAAI/bge-small-en-v1.5'}\x1f{text}"
    h.update(payload.encode("utf-8"))
    return h.hexdigest()

def mock_embed(text: str) -> list[float]:
    """Mock embedding returning a fixed vector (for testing)."""
    return [0.1, 0.2, 0.3, 0.4]

def get_cached_embedding(text: str, model: str, cache: dict) -> list[float]:
    # Compute the cache key using the real embed_cache_key function
    # TODO 1: compute key
    key = embed_cache_key(text, model=model)

if key in cache:
        return cache[key]

# Miss: call the embedding model
    vec = mock_embed(text)

# Round‑trip through float32 to match the stored BLOB precision
    # TODO 2: round-trip vec through float32 and assign the result to cache[key]
    blob = array.array('f', vec).tobytes()
    a = array.array('f')
    a.frombytes(blob)
    cache[key] = a.tolist()

return cache[key]

# Test the function
if __name__ == "__main__":
    cache = {}
    res1 = get_cached_embedding("hello", "test-model", cache)
    print("First call (miss):", res1)
    res2 = get_cached_embedding("hello", "test-model", cache)
    print("Second call (hit):", res2)
    assert res1 == res2, "Hit and miss should return identical values"
    print("Precision symmetry confirmed.")

Sources

roadmap-kg/kg/memory_common.py:643-685
roadmap-kg/kg/memory_common.py:757-804
roadmap-kg/kg/ground_content.py:264-308