Semantic cache for LLM apps

#!/usr/bin/env python3
"""
Minimal VCAL integration (no SDK required)

Flow:
1) Create an embedding vector for the user question (Ollama embeddings).
2) Ask VCAL Server /v1/qa with that vector.
3) If cache miss: call your LLM (placeholder), then /v1/upsert to store the answer.
"""
import os
import hashlib
import requests

VCAL_BASE = os.getenv("VCAL_URL", "http://127.0.0.1:8080").rstrip("/")
VCAL_KEY  = os.getenv("VCAL_API_KEY", "")

OLLAMA_URL  = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/")
EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text")

SIM_THRESHOLD = float(os.getenv("VCAL_SIM_THR", "0.85"))
EF_SEARCH     = int(os.getenv("VCAL_EF_SEARCH", "128"))

HEADERS = {"Content-Type": "application/json"}
if VCAL_KEY:
    HEADERS["X-VCAL-Key"] = VCAL_KEY

def embed_text(text: str) -> list[float]:
    r = requests.post(f"{OLLAMA_URL}/api/embeddings",
                      json={"model": EMBED_MODEL, "prompt": text},
                      timeout=30)
    r.raise_for_status()
    vec = r.json().get("embedding")
    if not isinstance(vec, list) or not vec:
        raise RuntimeError("Bad embedding response")
    return vec

def ext_id_for(q: str) -> int:
    h = hashlib.blake2b(q.strip().lower().encode("utf-8"), digest_size=8).digest()
    return int.from_bytes(h, "big", signed=False)

def vcal_qa(vec: list[float]) -> dict:
    r = requests.post(f"{VCAL_BASE}/v1/qa",
                      headers=HEADERS,
                      json={"query": vec, "k": 1, "ef": EF_SEARCH, "sim_threshold": SIM_THRESHOLD},
                      timeout=30)
    r.raise_for_status()
    return r.json()

def vcal_upsert(ext_id: int, vec: list[float], answer: str) -> None:
    r = requests.post(f"{VCAL_BASE}/v1/upsert",
                      headers=HEADERS,
                      json={"ext_id": ext_id, "vector": vec, "answer": answer},
                      timeout=30)
    r.raise_for_status()

def call_llm_fallback(question: str) -> str:
    return f"(LLM fallback) Answer to: {question}"

def main():
    q = input("Ask a question: ").strip()
    if not q:
        return

    vec = embed_text(q)
    qa = vcal_qa(vec)

    if qa.get("hit") and qa.get("answer"):
        print("VCAL HIT ✅")
        print(qa["answer"])
        return

    print("VCAL MISS ❌ -> calling LLM fallback…")
    ans = call_llm_fallback(q)
    vcal_upsert(ext_id_for(q), vec, ans)
    print("Stored in VCAL ✅")
    print(ans)

if __name__ == "__main__":
    main()