Spaces:
Running
Running
File size: 3,598 Bytes
8186ed2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """Streamlit demo: semantic cache in front of an OpenRouter LLM."""
from __future__ import annotations
import streamlit as st
from semcache import SemanticCache, ask
from semcache._openrouter import get_api_key
st.set_page_config(page_title="Semantic Cache for LLMs", page_icon="⚡", layout="wide")
EXAMPLES = [
"What is the capital of France?",
"Tell me France's capital city", # paraphrase of #1 -> semantic HIT
"How do I reverse a list in Python?",
"What's the Python way to reverse a list?", # paraphrase of #3 -> semantic HIT
"What is the boiling point of water?",
"Explain reciprocal rank fusion in one sentence.",
]
def _cache() -> SemanticCache:
if "cache" not in st.session_state:
st.session_state.cache = SemanticCache(
threshold=0.85, cost_per_1k_tokens=0.50, avg_llm_latency_s=1.5
)
return st.session_state.cache
cache = _cache()
st.title("⚡ Semantic Cache for LLM Responses")
st.caption(
"Caches answers by prompt **embedding** — a semantically similar prompt returns "
"instantly, skipping the LLM call. Reuses free OpenRouter models."
)
with st.sidebar:
st.header("Cache settings")
cache.threshold = st.slider("Similarity threshold", 0.50, 0.99, cache.threshold, 0.01)
cache.max_size = int(st.number_input("Max cache size", 8, 4096, cache.max_size, 8))
cache.cost_per_1k_tokens = st.number_input(
"$ / 1k tokens (for savings est.)", 0.0, 100.0, cache.cost_per_1k_tokens, 0.05
)
st.caption("Models are free-tier; $ is illustrative from this rate.")
if st.button("Seed example prompts"):
for q in EXAMPLES:
try:
ask(q, cache)
except Exception as e: # noqa: BLE001
st.error(f"Seed failed: {e}")
break
else:
st.success(f"Seeded {len(EXAMPLES)} prompts.")
if st.button("Clear cache"):
cache.clear()
cache.stats = type(cache.stats)()
st.session_state.pop("history", None)
if not get_api_key():
st.warning("Set `OPENROUTER_API_KEY` (env or `.streamlit/secrets.toml`). See README.")
s = cache.stats
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("Hit rate", f"{s.hit_rate * 100:.0f}%")
c2.metric("Lookups", s.lookups)
c3.metric("LLM calls saved", s.llm_calls_saved)
c4.metric("Latency saved", f"{s.latency_saved_s:.1f}s")
c5.metric("Est. $ saved", f"${s.cost_saved_est:.4f}")
prompt = st.chat_input("Ask something (then ask it again in different words)…")
if prompt and prompt.strip():
try:
r = ask(prompt, cache)
except Exception as e: # noqa: BLE001
st.error(f"LLM/embedding error: {e}")
else:
st.session_state.setdefault("history", []).append(r)
for r in reversed(st.session_state.get("history", [])):
with st.chat_message("user"):
st.write(r.prompt)
with st.chat_message("assistant"):
if r.hit:
kind = "exact" if r.exact else f"semantic · sim {r.similarity:.3f}"
st.success(f"⚡ CACHE HIT ({kind}) · {r.latency_s * 1000:.0f} ms")
if r.matched_prompt and not r.exact:
st.caption(f"matched cached prompt: “{r.matched_prompt}”")
else:
st.info(f"🌐 MISS → called LLM · {r.latency_s:.2f} s")
st.write(r.response)
with st.expander(f"Cache contents ({len(cache)} entries)"):
rows = [
{"prompt": e.prompt, "hits": e.hits, "chars": len(e.response)}
for e in cache.entries()
]
if rows:
st.table(rows)
else:
st.write("empty")
|