[阶段4.4] AI 建议结果缓存 - 页面不再阻塞等待 160 秒
网关首选的推理模型一次生成约 160 秒，每次打开建议页都重跑不可用。 结果落库缓存，页面读缓存，用户想要新的再手动触发。 db.py: - 新增 ai_recommendations 表，每用户一行（重新生成是替换不是累积） - fingerprint 列记录这条建议是基于哪份数据算出来的 services/analysis.py: - _fingerprint() 对全部每日指标 + 运动条数取 sha256，任何一次同步 新增或修正了数值都会让摘要变化，从而使缓存失效 - TTL 默认 24 小时（AI_CACHE_TTL_HOURS 可调） - 指定 model 参数时绕过缓存：点名某个模型意味着想要那个模型的答案 - 降级到规则引擎的结果不写缓存，避免把兜底答案当成 AI 结果存下来 - 缓存写入失败只打日志，不影响本次请求返回 routes: ?refresh=1 强制重新生成 前端: - "重新生成" 按钮走 refresh，并提示需要 1-3 分钟、可以离开本页 - meta 栏显示是否为缓存结果及生成时间，以及网关的上游厂商 - axios 该请求超时放宽到 240s（冷生成远超默认超时） tests/test_ai_cache.py (20 通过): - 第二次调用不再打模型 - 新增一天数据 / 修正某天数值 / 新增一条运动记录，三种情况都失效 - TTL 边界两侧各一条（刚过期重算、未过期沿用） - 缓存按用户隔离，A 的结果不会答给 B - payload 损坏时重新生成而不是抛异常 - 规则兜底结果和无数据用户都不落缓存 Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
@@ -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) ----------------------------------------------------
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -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])
|
||||
|
||||
240
backend/tests/test_ai_cache.py
Normal file
240
backend/tests/test_ai_cache.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user