diff --git a/backend/db.py b/backend/db.py index 9cf8cd9..3ffbdbe 100644 --- a/backend/db.py +++ b/backend/db.py @@ -83,6 +83,21 @@ CREATE TABLE IF NOT EXISTS sync_status ( updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ); + +-- One cached LLM answer per user. Generating one takes minutes against a +-- large reasoning model, which is far too slow to sit in a page load, so the +-- result is stored and reused until the underlying data changes. +-- `fingerprint` identifies the health data the advice was derived from. +CREATE TABLE IF NOT EXISTS ai_recommendations ( + user_id VARCHAR(64) PRIMARY KEY, + fingerprint VARCHAR(64) NOT NULL, + model VARCHAR(64), + upstream VARCHAR(64), + days INT, + payload TEXT NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (user_id) REFERENCES users(id) +); """ # --- MariaDB pool (lazy) ---------------------------------------------------- diff --git a/backend/routes/analysis.py b/backend/routes/analysis.py index 65e521f..1d34808 100644 --- a/backend/routes/analysis.py +++ b/backend/routes/analysis.py @@ -35,9 +35,15 @@ def models(): def ai_recommendations(): """LLM recommendations. `?model=` picks one; omit it to use the chain. + Served from cache unless `?refresh=1` or an explicit `model` is given — + a fresh generation can take minutes against a large reasoning model. + Always 200: when no model succeeds the rule engine answers instead, and meta.source says which produced the result. """ model = request.args.get("model") or None days = request.args.get("days", type=int) - return jsonify(analysis_svc.get_ai_recommendations(g.user_id, model, days)) + refresh = request.args.get("refresh") in ("1", "true", "yes") + return jsonify( + analysis_svc.get_ai_recommendations(g.user_id, model, days, refresh) + ) diff --git a/backend/services/analysis.py b/backend/services/analysis.py index a2c2003..0b97a82 100644 --- a/backend/services/analysis.py +++ b/backend/services/analysis.py @@ -4,9 +4,15 @@ Analysis service: metric trends + a rule-based recommendation engine. Replicates the original Node AnalysisService logic. Averages are computed over the most recent 14 days of available daily summaries. """ +import datetime +import hashlib +import json +import os + from services import health from services import ai as ai_svc -from db import query_all +from db import query_all, query_one, execute +from config import DB_TYPE METRIC_COLUMNS = { "steps": "steps", @@ -120,11 +126,99 @@ def get_recommendations(user_id): return recs -def get_ai_recommendations(user_id, model=None, days=None): - """LLM-generated recommendations over the user's full history. +CACHE_TTL_HOURS = int(os.environ.get("AI_CACHE_TTL_HOURS") or 24) - Falls back to the rule engine if every model fails, so the endpoint always - returns something useful. The `source` field tells the two apart. + +def _fingerprint(summary, activities): + """Identify the data a cached answer was derived from. + + Cheap and order-independent: the day count, the newest and oldest dates, + and every metric value. Any sync that adds or corrects a value changes the + digest, which is what expires the cache. + """ + parts = [str(len(summary)), str(len(activities))] + for row in summary: + parts.append( + "|".join( + str(row.get(k)) + for k in ("date", "steps", "heartRate", "heartRateVariability", + "stress", "caloriesBurned") + ) + ) + sleep = row.get("sleep") or {} + parts.append(f"{sleep.get('duration')}/{sleep.get('quality')}") + return hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()[:64] + + +def _read_cache(user_id, fingerprint): + row = query_one( + "SELECT * FROM ai_recommendations WHERE user_id = ?", [user_id] + ) + if not row or row["fingerprint"] != fingerprint: + return None + + created = row.get("created_at") + if created: + try: + ts = datetime.datetime.fromisoformat(str(created).replace(" ", "T")) + age = datetime.datetime.utcnow() - ts + if age > datetime.timedelta(hours=CACHE_TTL_HOURS): + return None + except ValueError: + # An unparseable timestamp should not permanently poison the cache. + return None + + try: + recs = json.loads(row["payload"]) + except (ValueError, TypeError): + return None + + return { + "recommendations": recs, + "meta": { + "source": "ai", + "model": row["model"], + "upstream": row["upstream"], + "days": row["days"], + "cached": True, + "generatedAt": created, + }, + } + + +def _write_cache(user_id, fingerprint, recs, meta): + cols = ["user_id", "fingerprint", "model", "upstream", "days", "payload", + "created_at"] + placeholders = ", ".join(["?"] * len(cols)) + if DB_TYPE == "mariadb": + updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "user_id") + sql = ( + f"INSERT INTO ai_recommendations ({', '.join(cols)}) " + f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}" + ) + else: + updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "user_id") + sql = ( + f"INSERT INTO ai_recommendations ({', '.join(cols)}) " + f"VALUES ({placeholders}) ON CONFLICT(user_id) DO UPDATE SET {updates}" + ) + execute(sql, [ + user_id, fingerprint, meta.get("model"), meta.get("upstream"), + meta.get("days"), json.dumps(recs, ensure_ascii=False), + datetime.datetime.utcnow().isoformat(timespec="seconds"), + ]) + + +def get_ai_recommendations(user_id, model=None, days=None, refresh=False): + """LLM recommendations over the user's history, cached. + + A generation costs minutes against a large reasoning model, so a stored + answer is reused until the health data changes (or the TTL lapses). + `refresh=True` and an explicit `model` both bypass the cache — asking for + a specific model means wanting that model's answer, not a stored one. + + Falls back to the rule engine when every model fails, so the endpoint + always returns something useful; `meta.source` tells the two apart. """ summary = health.get_summary(user_id) if not summary: @@ -134,15 +228,31 @@ def get_ai_recommendations(user_id, model=None, days=None): } activities = health.get_activities(user_id) - budget = days or ai_svc.default_day_budget() + fingerprint = _fingerprint(summary, activities) + if not refresh and not model: + cached = _read_cache(user_id, fingerprint) + if cached: + return cached + + budget = days or ai_svc.default_day_budget() try: recs, meta = ai_svc.generate( summary, activities, preferred_model=model, day_budget=budget ) - return {"recommendations": recs, "meta": {**meta, "source": "ai"}} except ai_svc.AIError as e: return { "recommendations": get_recommendations(user_id), "meta": {"model": None, "source": "rules", "reason": str(e)}, } + + try: + _write_cache(user_id, fingerprint, recs, meta) + except Exception as e: # noqa: BLE001 - a cache write must never fail the request + print(f"[analysis] failed to cache recommendations: {e}") + + return {"recommendations": recs, "meta": {**meta, "source": "ai", "cached": False}} + + +def clear_ai_cache(user_id): + execute("DELETE FROM ai_recommendations WHERE user_id = ?", [user_id]) diff --git a/backend/tests/test_ai_cache.py b/backend/tests/test_ai_cache.py new file mode 100644 index 0000000..7ed653b --- /dev/null +++ b/backend/tests/test_ai_cache.py @@ -0,0 +1,240 @@ +""" +Unit tests for the AI recommendation cache. + +A generation costs minutes against a large reasoning model, so the result is +stored and reused. These tests pin when it is reused and — more importantly — +when it must not be. +""" +import datetime +import json + +import pytest +import requests + +from services import analysis as analysis_svc +from services import health as health_svc + + +VALID_REPLY = json.dumps( + [{"category": "睡眠", "recommendation": "早点睡。", "priority": "high", + "basedOn": ["sleep_duration"]}], + ensure_ascii=False, +) + + +class FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload + self.text = text or json.dumps(payload or {}) + + def json(self): + if self._payload is None: + raise ValueError("no json") + return self._payload + + +@pytest.fixture +def keys(monkeypatch): + monkeypatch.setenv("GEMINI_API_KEY", "test-key") + return True + + +@pytest.fixture +def counting_llm(monkeypatch): + """Mock the LLM and count how many times it is actually called.""" + calls = [] + + def fake_post(self, url, **kwargs): + calls.append(url) + return FakeResponse( + 200, {"candidates": [{"content": {"parts": [{"text": VALID_REPLY}]}}]} + ) + + monkeypatch.setattr(requests.Session, "post", fake_post) + return calls + + +@pytest.fixture +def seeded(seed_health, user): + seed_health([{"date": "2026-08-20", "steps": 5000, "sleep_duration": 6}]) + return user + + +class TestCacheHit: + def test_first_call_reaches_the_model(self, seeded, keys, counting_llm): + out = analysis_svc.get_ai_recommendations(seeded["id"]) + assert out["meta"]["source"] == "ai" + assert out["meta"]["cached"] is False + assert len(counting_llm) == 1 + + def test_second_call_is_served_from_cache(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + + assert len(counting_llm) == 1, "the model must not be called twice" + assert out["meta"]["cached"] is True + assert out["meta"]["source"] == "ai" + + def test_cached_result_matches_the_generated_one(self, seeded, keys, counting_llm): + first = analysis_svc.get_ai_recommendations(seeded["id"]) + second = analysis_svc.get_ai_recommendations(seeded["id"]) + assert first["recommendations"] == second["recommendations"] + + def test_cache_records_which_model_answered(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + assert out["meta"]["model"] == "gemini-flash" + + def test_cache_reports_when_it_was_generated(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + assert out["meta"]["generatedAt"] + + +class TestCacheInvalidation: + def test_new_health_data_invalidates(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + health_svc.upsert_health_daily( + seeded["id"], {"date": "2026-08-21", "steps": 9000} + ) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + + assert len(counting_llm) == 2, "a new day of data must trigger a regeneration" + assert out["meta"]["cached"] is False + + def test_corrected_value_invalidates(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + # Same date, different step count — a re-sync correcting a value. + health_svc.upsert_health_daily( + seeded["id"], {"date": "2026-08-20", "steps": 12345, "sleepDuration": 6} + ) + analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 2 + + def test_new_activity_invalidates(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + health_svc.insert_activity( + seeded["id"], + {"activityType": "running", "startTime": "2026-08-20T07:00:00", + "endTime": "2026-08-20T07:30:00"}, + ) + analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 2 + + def test_refresh_bypasses_the_cache(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + out = analysis_svc.get_ai_recommendations(seeded["id"], refresh=True) + assert len(counting_llm) == 2 + assert out["meta"]["cached"] is False + + def test_explicit_model_bypasses_the_cache(self, seeded, keys, counting_llm): + """Asking for a named model means wanting that model's answer.""" + analysis_svc.get_ai_recommendations(seeded["id"]) + analysis_svc.get_ai_recommendations(seeded["id"], model="gemini-flash") + assert len(counting_llm) == 2 + + def test_expired_entry_is_regenerated(self, seeded, keys, counting_llm, db): + analysis_svc.get_ai_recommendations(seeded["id"]) + stale = ( + datetime.datetime.utcnow() + - datetime.timedelta(hours=analysis_svc.CACHE_TTL_HOURS + 1) + ).isoformat(timespec="seconds") + db.execute( + "UPDATE ai_recommendations SET created_at = ? WHERE user_id = ?", + [stale, seeded["id"]], + ) + analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 2 + + def test_entry_just_inside_the_ttl_is_kept(self, seeded, keys, counting_llm, db): + analysis_svc.get_ai_recommendations(seeded["id"]) + fresh = ( + datetime.datetime.utcnow() + - datetime.timedelta(hours=analysis_svc.CACHE_TTL_HOURS - 1) + ).isoformat(timespec="seconds") + db.execute( + "UPDATE ai_recommendations SET created_at = ? WHERE user_id = ?", + [fresh, seeded["id"]], + ) + analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 1 + + def test_clear_cache_forces_regeneration(self, seeded, keys, counting_llm): + analysis_svc.get_ai_recommendations(seeded["id"]) + analysis_svc.clear_ai_cache(seeded["id"]) + analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 2 + + +class TestIsolationAndRobustness: + def test_cache_is_per_user(self, seeded, keys, counting_llm, db, client): + analysis_svc.get_ai_recommendations(seeded["id"]) + + other = client.post( + "/api/auth/register", + json={"email": "other@example.com", "garminEmail": "o@example.com", + "garminPassword": "pw123456"}, + ).get_json() + health_svc.upsert_health_daily( + other["id"], {"date": "2026-08-20", "steps": 5000, "sleepDuration": 6} + ) + + analysis_svc.get_ai_recommendations(other["id"]) + assert len(counting_llm) == 2, "one user's cache must not answer another's" + + def test_only_one_row_per_user(self, seeded, keys, counting_llm, db): + for _ in range(3): + analysis_svc.get_ai_recommendations(seeded["id"], refresh=True) + rows = db.query_all( + "SELECT * FROM ai_recommendations WHERE user_id = ?", [seeded["id"]] + ) + assert len(rows) == 1, "regeneration must replace, not accumulate" + + def test_corrupt_payload_regenerates_instead_of_raising( + self, seeded, keys, counting_llm, db + ): + analysis_svc.get_ai_recommendations(seeded["id"]) + db.execute( + "UPDATE ai_recommendations SET payload = ? WHERE user_id = ?", + ["not json", seeded["id"]], + ) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + assert len(counting_llm) == 2 + assert out["recommendations"] + + def test_rule_fallback_is_not_cached(self, seeded, monkeypatch, db): + """A degraded answer must not be stored as if it were the AI's.""" + def boom(self, *a, **k): + raise requests.Timeout("down") + + monkeypatch.setattr(requests.Session, "post", boom) + out = analysis_svc.get_ai_recommendations(seeded["id"]) + + assert out["meta"]["source"] == "rules" + assert db.query_one( + "SELECT * FROM ai_recommendations WHERE user_id = ?", [seeded["id"]] + ) is None + + def test_no_data_user_is_not_cached(self, user, keys, counting_llm, db): + analysis_svc.get_ai_recommendations(user["id"]) + assert len(counting_llm) == 0 + assert db.query_one( + "SELECT * FROM ai_recommendations WHERE user_id = ?", [user["id"]] + ) is None + + +class TestEndpoint: + def test_second_request_is_cached(self, client, auth, seeded, keys, counting_llm): + client.get("/api/analysis/ai-recommendations", headers=auth) + r = client.get("/api/analysis/ai-recommendations", headers=auth) + assert r.get_json()["meta"]["cached"] is True + assert len(counting_llm) == 1 + + def test_refresh_param_forces_regeneration( + self, client, auth, seeded, keys, counting_llm + ): + client.get("/api/analysis/ai-recommendations", headers=auth) + r = client.get("/api/analysis/ai-recommendations?refresh=1", headers=auth) + assert r.get_json()["meta"]["cached"] is False + assert len(counting_llm) == 2 diff --git a/client/src/pages/Recommendations.tsx b/client/src/pages/Recommendations.tsx index 223cb3f..d29e534 100644 --- a/client/src/pages/Recommendations.tsx +++ b/client/src/pages/Recommendations.tsx @@ -17,15 +17,19 @@ function Recommendations() { const [loading, setLoading] = useState(true); const [error, setError] = useState(''); - const load = useCallback(async (model?: string) => { + const [regenerating, setRegenerating] = useState(false); + + const load = useCallback(async (model?: string, refresh?: boolean) => { setLoading(true); + setRegenerating(Boolean(refresh || model)); setError(''); try { - setResult(await apiClient.getAiRecommendations(model || undefined)); + setResult(await apiClient.getAiRecommendations(model || undefined, refresh)); } catch (err: any) { setError(errorMessage(err, '获取建议失败')); } finally { setLoading(false); + setRegenerating(false); } }, []); @@ -73,13 +77,20 @@ function Recommendations() { + {regenerating && ( +
+ 正在请求大模型重新分析,通常需要 1-3 分钟(推理模型会先推演再作答)。 + 期间可以离开本页,结果会被缓存下来。 +
+ )} + {configured.length === 0 && models.length > 0 && (
尚未配置任何模型密钥。在 backend/.env 中填入 @@ -94,7 +105,16 @@ function Recommendations() {
{meta.source === 'ai' ? ( <> - AI 生成 · 模型 {meta.model} · 分析了 {meta.days} 天数据 + AI 生成 · 模型 {meta.model} + {meta.upstream && `(上游 ${meta.upstream})`} + {' · '}分析了 {meta.days} 天数据 + {meta.cached && ( + + · 缓存结果 + {meta.generatedAt && + `,生成于 ${new Date(meta.generatedAt + 'Z').toLocaleString('zh-CN')}`} + + )} {meta.fallbackFrom && meta.fallbackFrom.length > 0 && ( ({meta.fallbackFrom.join('、')} 失败后自动切换) diff --git a/client/src/services/api.ts b/client/src/services/api.ts index f95228c..211340a 100644 --- a/client/src/services/api.ts +++ b/client/src/services/api.ts @@ -51,9 +51,13 @@ export interface AiRecommendations { source: 'ai' | 'rules'; model: string | null; provider?: string; + upstream?: string | null; days?: number; fallbackFrom?: string[]; reason?: string; + /** True when served from the stored answer rather than freshly generated. */ + cached?: boolean; + generatedAt?: string; }; } @@ -207,10 +211,19 @@ class ApiClient { return data; } - async getAiRecommendations(model?: string, days?: number) { + /** + * Served from the stored answer unless `refresh` is set or a `model` is + * named. A fresh generation can take minutes, so the caller should show a + * long-running state for those two cases. + */ + async getAiRecommendations(model?: string, refresh?: boolean, days?: number) { const { data } = await this.client.get( '/analysis/ai-recommendations', - { params: { model, days } } + { + params: { model, days, ...(refresh ? { refresh: 1 } : {}) }, + // A cold generation runs well past axios's default timeout. + timeout: 240_000, + } ); return data; }