Add semantic-cache example (cache LLM responses by meaning)

moss-live-labs/examples/semantic-cache/.env.example

@@ -0,0 +1,6 @@
+# Moss
+MOSS_PROJECT_ID=Your moss project id here
+MOSS_PROJECT_KEY=Your moss project key here
+
+# OpenAI — the model whose responses we cache
+OPENAI_API_KEY=Your openai api key here

moss-live-labs/examples/semantic-cache/.python-version

@@ -0,0 +1 @@
+3.12

moss-live-labs/examples/semantic-cache/README.md

@@ -0,0 +1,64 @@
+# Semantic cache for LLM responses
+
+Cache LLM answers by **meaning**, not by exact text.
+
+A normal cache keys on the literal request string, so two phrasings of the same
+question miss:
+
+```
+"what are your hours?"   ->  MISS  ->  calls the model
+"when do you open?"      ->  MISS  ->  calls the model again   (paid twice)
+```
+
+A semantic cache embeds the question and looks up the nearest one it has already
+answered. If it's close enough, it returns the stored answer with no model call.
+Repeat questions come back in single-digit-millisecond retrieval, on-device.
+
+```
+"what are your hours?"   ->  MISS  ->  calls the model, stores the answer
+"when do you open?"      ->  HIT   ->  returns the stored answer, no model call
+```
+
+## How it works
+
+Here's the idea, simplified (the runnable `ask()` in
+[`semantic_cache.py`](./semantic_cache.py) also returns whether it was a hit and
+prints timing):
+
+```python
+async def ask(self, question):
+    hit = await self.store.query(question, QueryOptions(top_k=1, alpha=1.0))  # pure semantic
+    if hit.docs and hit.docs[0].score >= THRESHOLD:      # close enough in meaning?
+        return (hit.docs[0].metadata or {}).get("answer")  # cache hit — no LLM call
+    answer = await call_the_model(question)             # miss — ask once
+    await self.store.add_docs(
+        [DocumentInfo(id=question, text=question, metadata={"answer": answer})])
+    return answer
+```
+
+Moss keys the cache on the question's embedding and serves the nearest match
+in <10 ms locally, so the lookup is far cheaper than the model call it avoids.
+The one knob that matters is `THRESHOLD` (the similarity score, 0-1): too low and
+you answer questions people didn't quite ask; too high and you miss obvious matches.
+
+## What you need
+
+- A [Moss](https://moss.dev) account (`MOSS_PROJECT_ID` / `MOSS_PROJECT_KEY`)
+- An OpenAI key (the example uses `gpt-4o-mini` as the model being cached)
+- Python 3.10+
+
+## Run
+
+```bash
+uv sync                       # or: pip install moss openai python-dotenv
+cp .env.example .env          # fill in your keys
+uv run python semantic_cache.py   # runs inside the project venv (plain `python` if you used pip)
+```
+
+Expected: the first question is a `MISS` (calls the model), the paraphrased
+second question is a `HIT` (returns instantly, no model call).
+
+## Resources
+
+- [Docs](https://docs.moss.dev)
+- [GitHub](https://github.com/usemoss/moss)

moss-live-labs/examples/semantic-cache/pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "moss-semantic-cache"
+version = "0.1.0"
+description = "Semantic cache for LLM responses, built on Moss"
+requires-python = ">=3.10"
+dependencies = [
+    "moss>=1.7.1",
+    "openai>=1.0",
+    "python-dotenv>=1.0",
+]

moss-live-labs/examples/semantic-cache/semantic_cache.py

@@ -0,0 +1,112 @@
+"""Semantic cache for LLM responses, built on Moss.
+
+A normal cache keys on the exact text of a request, so two ways of asking the
+same thing ("what are your hours?" / "when do you open?") miss and you pay the
+model twice. A *semantic* cache keys on meaning: embed the question, look up the
+nearest one you've already answered, and if it's close enough, return the stored
+answer without calling the model.
+
+The core is the SemanticCache class below (the `ask` method is the whole idea).
+The one knob that matters is the similarity threshold.
+"""
+
+import asyncio
+import os
+import sys
+import time
+import uuid
+
+from dotenv import load_dotenv
+from openai import AsyncOpenAI
+
+from moss import MossClient, DocumentInfo, QueryOptions
+
+# semantic similarity above which a cached answer is "close enough" to reuse.
+# too low -> you answer questions people didn't quite ask; too high -> you miss.
+THRESHOLD = 0.92
+
+
+class SemanticCache:
+    """A small store of past questions -> answers, looked up by meaning.
+
+    `store` is any Moss index/session with query/add_docs; `llm` is an AsyncOpenAI
+    client. Both are injected so this class stays reusable and free of import-time
+    side effects.
+
+    Note: entries here are keyed by question meaning only. A production cache
+    should also scope by tenant / user / model / prompt version (e.g. a separate
+    index per scope, or a metadata filter) so answers aren't replayed across
+    contexts that merely phrase things similarly.
+    """
+
+    def __init__(self, store, llm, model: str = "gpt-4o-mini"):
+        self.store = store
+        self.llm = llm
+        self.model = model
+
+    async def ask(self, question: str) -> tuple[str, bool]:
+        # 1. look for the closest question we've already answered.
+        #    alpha=1.0 -> pure semantic (embedding) match, so the score reflects
+        #    meaning rather than keyword overlap.
+        hit = await self.store.query(question, QueryOptions(top_k=1, alpha=1.0))
+        if hit.docs and hit.docs[0].score >= THRESHOLD:
+            answer = (hit.docs[0].metadata or {}).get("answer")
+            if answer is not None:
+                return answer, True  # cache hit — no LLM call
+
+        # 2. miss: ask the model once
+        resp = await self.llm.chat.completions.create(
+            model=self.model,
+            messages=[{"role": "user", "content": question}],
+        )
+        answer = resp.choices[0].message.content or ""
+
+        # 3. remember it so any wording of it is instant next time
+        await self.store.add_docs(
+            [DocumentInfo(id=question, text=question, metadata={"answer": answer})]
+        )
+        return answer, False
+
+
+def _require(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        sys.exit(f"Missing {name}. Copy .env.example to .env and fill in your keys.")
+    return value
+
+
+async def main():
+    load_dotenv()
+    # fail fast with a clear message if credentials are missing
+    project_id = _require("MOSS_PROJECT_ID")
+    project_key = _require("MOSS_PROJECT_KEY")
+    _require("OPENAI_API_KEY")  # read by AsyncOpenAI from the environment
+
+    moss = MossClient(project_id=project_id, project_key=project_key)
+
+    # AsyncOpenAI owns an HTTP client; the context manager closes it on exit.
+    async with AsyncOpenAI() as llm:
+        # A Moss session is the cache store. We use a unique name per run so the demo
+        # always starts empty and shows a clean MISS -> HIT (a session auto-loads an
+        # existing cloud index of the same name, which would otherwise make the first
+        # question a HIT). In production, use a stable name and call
+        # `await store.push_index()` to persist the cache across runs and processes.
+        store = await moss.session(index_name=f"qa-cache-demo-{uuid.uuid4().hex[:8]}")
+        cache = SemanticCache(store, llm)
+
+        # the 2nd question means the same as the 1st, phrased differently -> cache hit
+        questions = [
+            "What are your opening hours?",
+            "when do you open?",
+            "How do I reset my password?",
+        ]
+        for q in questions:
+            t = time.perf_counter()
+            answer, hit = await cache.ask(q)
+            ms = (time.perf_counter() - t) * 1000
+            tag = "HIT " if hit else "MISS"
+            print(f"[{tag} {ms:7.1f} ms]  {q}\n   -> {answer.strip()[:90]}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())