Spaces:
Sleeping
Sleeping
File size: 5,018 Bytes
ce11d27 | 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 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """
LLM response cache backed by SQLite.
Avoids redundant API calls by caching responses keyed on
SHA-256(model_name || system_prompt || user_prompt).
"""
from __future__ import annotations
import sqlite3
import time
from typing import Optional
from tracescope.utils.hashing import sha256_key
class LLMResponseCache:
"""SQLite-backed LLM response cache.
Args:
db_path: Path to the SQLite database file.
enabled: If False, all lookups miss and stores are skipped.
"""
def __init__(self, db_path: str, enabled: bool = True):
self.db_path = db_path
self.enabled = enabled
if enabled:
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_table()
else:
self._conn = None
def _init_table(self):
self._conn.execute(
"""
CREATE TABLE IF NOT EXISTS llm_cache (
hash TEXT PRIMARY KEY,
model_name TEXT NOT NULL,
response TEXT NOT NULL,
created_at REAL NOT NULL
)
"""
)
self._conn.commit()
def get(self, model_name: str, system_prompt: str, user_prompt: str) -> Optional[str]:
"""Look up a cached response. Returns None on miss."""
if not self.enabled:
return None
key = sha256_key(model_name, system_prompt, user_prompt)
row = self._conn.execute(
"SELECT response FROM llm_cache WHERE hash = ?", (key,)
).fetchone()
return row[0] if row else None
def put(self, model_name: str, system_prompt: str, user_prompt: str, response: str):
"""Store a response in the cache."""
if not self.enabled:
return
key = sha256_key(model_name, system_prompt, user_prompt)
self._conn.execute(
"""
INSERT OR REPLACE INTO llm_cache (hash, model_name, response, created_at)
VALUES (?, ?, ?, ?)
""",
(key, model_name, response, time.time()),
)
self._conn.commit()
def clear(self):
"""Clear all cached responses."""
if self._conn:
self._conn.execute("DELETE FROM llm_cache")
self._conn.commit()
def close(self):
if self._conn:
self._conn.close()
class ResultCache:
"""SQLite-backed cache for expensive ML computation results.
Caches clustering, dimension reduction, and velocity grid results
keyed on embeddings fingerprint (matching Android's DimensionReducerAdapter).
Uses the same SQLite database as LLMResponseCache but a separate table.
Args:
db_path: Path to the SQLite database file.
enabled: If False, all lookups miss and stores are skipped.
"""
def __init__(self, db_path: str, enabled: bool = True):
self.db_path = db_path
self.enabled = enabled
if enabled:
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_table()
else:
self._conn = None
def _init_table(self):
self._conn.execute(
"""
CREATE TABLE IF NOT EXISTS result_cache (
key TEXT PRIMARY KEY,
step TEXT NOT NULL,
data BLOB NOT NULL,
created_at REAL NOT NULL
)
"""
)
self._conn.commit()
def get_result(self, step: str, fingerprint: str) -> Optional[bytes]:
"""Look up a cached computation result.
Args:
step: Pipeline step name (e.g. "clustering", "dim_reduction", "velocity_grid").
fingerprint: Embeddings fingerprint (SHA-256 hex digest).
Returns:
Cached data as bytes, or None on miss.
"""
if not self.enabled:
return None
key = sha256_key(step, fingerprint)
row = self._conn.execute(
"SELECT data FROM result_cache WHERE key = ?", (key,)
).fetchone()
return row[0] if row else None
def put_result(self, step: str, fingerprint: str, data: bytes):
"""Store a computation result in the cache.
Args:
step: Pipeline step name.
fingerprint: Embeddings fingerprint.
data: Result data as bytes (JSON-encoded or raw numpy).
"""
if not self.enabled:
return
key = sha256_key(step, fingerprint)
self._conn.execute(
"""
INSERT OR REPLACE INTO result_cache (key, step, data, created_at)
VALUES (?, ?, ?, ?)
""",
(key, step, data, time.time()),
)
self._conn.commit()
def clear(self):
"""Clear all cached results."""
if self._conn:
self._conn.execute("DELETE FROM result_cache")
self._conn.commit()
def close(self):
if self._conn:
self._conn.close()
|