feat(ai): AI 教练 —— 晨间简报、运动处方、趋势归因与 Copilot

数值全部在服务端算好再交给模型，模型只做解读。让模型从 CSV 里自己推 z 分数，它算错的次数足以让简报引用图表反驳它的数字。 - services/insights.py：z 分数（28 天个人基线，且**排除当天**——用一个值参与算出来的均值去衡量它自己，会把真实离群点摊平）、13 个月趋势斜率（按序数日期最小二乘，手表放充电器上一周不会压缩 x 轴）、近 7 天活动量对比。 - services/coach.py：三套提示词 + 回复解析，每套都配一个规则引擎版本。网关一次生成要几分钟，上游被限流时给一个朴素的答案，好过给一张空卡片。 - services/ai.py：多轮 chat()、SSE stream()、complete()/stream_chat()，以及 extract_json()——上游是推理模型，可见输出以思维链开头，所以从末尾倒着找最后一个配平的 JSON（字符串感知，扛得住引号里的 } 和转义引号）。 - 接口 briefing / trend-insight / copilot(SSE)，缓存表 ai_insights。 - 前端：今日页晨报卡（后台生成 + 轮询升级）、全局 Copilot 浮窗、指标详情页归因面板。features.ai 打开。实测（对着自建 ai-gateway）：晨报一次 273 秒，缓存命中 18 毫秒——所以简报绝不能同步阻塞首屏。网关的流式通道比阻塞通道更不可靠：同一条提示词流式 139 秒后返回「所有模型均不可用」，阻塞则成功，因此 stream_chat() 在流式零输出时对同一模型退回非流式重试。Copilot 实测 TTFB 9ms、全程 40 秒。顺带修两处：refresh 原来只跳过缓存读、不删行，导致「重新生成」后的轮询读到旧行、看到 cached 就停了，用户一直盯着他刚要求替换掉的那段字；基线零方差时原来返回 z=0.0，把「和每一条观测都不同」标成「完全正常」，改为 z=null。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-01 13:57:35 +08:00
parent a746327560
commit c57c930949
21 changed files with 3677 additions and 22 deletions
--- a/backend/services/analysis.py
+++ b/backend/services/analysis.py
@@ -8,9 +8,12 @@ import datetime
 import hashlib
 import json
 import os
+import threading

 from services import health
 from services import ai as ai_svc
+from services import coach
+from services import insights
 from db import query_all, query_one, execute
 from config import DB_TYPE

@@ -256,3 +259,261 @@ def get_ai_recommendations(user_id, model=None, days=None, refresh=False):

 def clear_ai_cache(user_id):
    execute("DELETE FROM ai_recommendations WHERE user_id = ?", [user_id])
+
+
+# --- AI coach: briefing, trend attribution, Copilot -------------------------
+# Same caching rationale as the recommendations above, with one addition: a
+# briefing is the first thing on the 今日 screen, so it can never wait on a
+# generation. The endpoint answers immediately from the rule engine and the
+# model's version replaces it on a later poll.
+_JOB_LOCK = threading.Lock()
+_JOBS = set()
+
+
+def _insight_key(kind, subject):
+    return f"{kind}:{subject}"
+
+
+def _read_insight(user_id, kind, subject, fingerprint):
+    row = query_one(
+        "SELECT * FROM ai_insights WHERE user_id = ? AND kind = ? AND subject = ?",
+        [user_id, kind, subject],
+    )
+    if not row or row["fingerprint"] != fingerprint:
+        return None
+    try:
+        payload = json.loads(row["payload"])
+    except (ValueError, TypeError):
+        return None
+    return payload, {
+        "source": "ai",
+        "model": row["model"],
+        "upstream": row["upstream"],
+        "cached": True,
+        "generatedAt": row.get("created_at"),
+    }
+
+
+def _write_insight(user_id, kind, subject, fingerprint, payload, meta):
+    cols = ["id", "user_id", "kind", "subject", "fingerprint", "model",
+            "upstream", "payload", "created_at"]
+    placeholders = ", ".join(["?"] * len(cols))
+    updatable = [c for c in cols if c != "id"]
+    if DB_TYPE == "mariadb":
+        updates = ", ".join(f"{c}=VALUES({c})" for c in updatable)
+        sql = (f"INSERT INTO ai_insights ({', '.join(cols)}) "
+               f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}")
+    else:
+        updates = ", ".join(f"{c}=excluded.{c}" for c in updatable)
+        sql = (f"INSERT INTO ai_insights ({', '.join(cols)}) "
+               f"VALUES ({placeholders}) ON CONFLICT(id) DO UPDATE SET {updates}")
+    # The id is derived rather than random so a re-generation overwrites the
+    # row it replaces instead of accumulating one per attempt.
+    row_id = hashlib.sha256(
+        f"{user_id}|{kind}|{subject}".encode("utf-8")
+    ).hexdigest()[:64]
+    execute(sql, [
+        row_id, user_id, kind, subject, fingerprint, meta.get("model"),
+        meta.get("upstream"), json.dumps(payload, ensure_ascii=False),
+        datetime.datetime.utcnow().isoformat(timespec="seconds"),
+    ])
+
+
+def _context_fingerprint(context):
+    """Digest of everything the prompt will contain.
+
+    The whole context rather than a chosen subset: a briefing is derived from
+    all of it, so any change to any field — a corrected sleep stage, a newly
+    synced activity — should expire the cached answer.
+    """
+    blob = json.dumps(context, ensure_ascii=False, sort_keys=True)
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:64]
+
+
+def _run_in_background(key, target):
+    """Start `target` once per key; a second caller joins the first one's run.
+
+    Guards against the obvious failure mode of a poll-until-ready endpoint:
+    the client polls every few seconds while a generation takes minutes, and
+    without this every poll would start another one.
+
+    The claim is per-process, not per-deployment: with Gunicorn's two workers
+    each can start one generation for the same key. That is deliberate rather
+    than overlooked — the `job_locks` table would make it exclusive, but the
+    cost here is a duplicate call, not a duplicate row (the cache id is derived
+    from user+kind+subject, so the second write lands on the first one's row).
+    A cross-process lock is worth adding only if the gateway's rate limits
+    start to bite.
+    """
+    with _JOB_LOCK:
+        if key in _JOBS:
+            return False
+        _JOBS.add(key)
+
+    def runner():
+        try:
+            target()
+        except Exception as e:  # noqa: BLE001 - a background job must not die silently
+            print(f"[analysis] background job {key} failed: {e}")
+        finally:
+            with _JOB_LOCK:
+                _JOBS.discard(key)
+
+    threading.Thread(target=runner, name=f"ai-{key}", daemon=True).start()
+    return True
+
+
+def _generating(key):
+    with _JOB_LOCK:
+        return key in _JOBS
+
+
+def generate_briefing(user_id, context, model=None):
+    """Ask a model for the briefing and store it. Returns (briefing, meta)."""
+    completion, meta = ai_svc.complete(coach.briefing_messages(context), model)
+    briefing = coach.parse_briefing(completion.text)
+    _write_insight(
+        user_id, "briefing", context["snapshotDate"],
+        _context_fingerprint(context), briefing, meta,
+    )
+    return briefing, meta
+
+
+def get_briefing(user_id, date=None, model=None, refresh=False, wait=False):
+    """The morning briefing for one day.
+
+    Non-blocking by default: a cached answer is returned if it matches the
+    current data, otherwise the rule-based briefing is returned straight away
+    and a model generation starts in the background. `wait=True` blocks for
+    the model instead — for callers that can afford minutes, such as a manual
+    "regenerate" or a scheduled pre-warm.
+    """
+    context = insights.build_context(user_id, date)
+    if not context:
+        return {
+            "briefing": None,
+            "context": None,
+            "meta": {"source": "none", "reason": "无健康数据"},
+        }
+
+    fingerprint = _context_fingerprint(context)
+    subject = context["snapshotDate"]
+    key = _insight_key("briefing", subject)
+
+    if not refresh:
+        cached = _read_insight(user_id, "briefing", subject, fingerprint)
+        if cached:
+            briefing, meta = cached
+            return {"briefing": briefing, "context": context, "meta": meta}
+    else:
+        # Drop the stored answer, not just skip it. Without this the poll that
+        # follows a regenerate reads the *old* row, sees `cached: true`, and
+        # stops polling — so the user keeps looking at the text they just
+        # asked to replace until something else expires it.
+        _delete_insight(user_id, "briefing", subject)
+
+    if wait:
+        try:
+            briefing, meta = generate_briefing(user_id, context, model)
+            return {
+                "briefing": briefing, "context": context,
+                "meta": {**meta, "source": "ai", "cached": False},
+            }
+        except ai_svc.AIError as e:
+            return {
+                "briefing": coach.rule_briefing(context), "context": context,
+                "meta": {"source": "rules", "reason": str(e)},
+            }
+
+    started = _run_in_background(
+        key, lambda: generate_briefing(user_id, context, model)
+    )
+    return {
+        "briefing": coach.rule_briefing(context),
+        "context": context,
+        "meta": {
+            "source": "rules",
+            # `pending` is what tells the client to poll again: the card it is
+            # showing is the placeholder, not the final answer.
+            "pending": True,
+            "generating": started or _generating(key),
+        },
+    }
+
+
+def get_trend_insight(user_id, metric, start, end, model=None, refresh=False):
+    """Attribution for a user-selected span of one metric (chart brush).
+
+    Blocking, unlike the briefing: this one is requested by an explicit
+    gesture on a chart, so there is a spinner to attach the wait to and no
+    useful placeholder to show in the meantime.
+    """
+    window = insights.window_context(user_id, metric, start, end)
+    if not window:
+        return {"insight": None, "window": None,
+                "meta": {"source": "none", "reason": "所选区间没有数据"}}
+
+    subject = f"{metric}:{start}:{end}"
+    fingerprint = _context_fingerprint(window)
+    if not refresh:
+        cached = _read_insight(user_id, "trend", subject, fingerprint)
+        if cached:
+            insight, meta = cached
+            return {"insight": insight, "window": window, "meta": meta}
+
+    try:
+        completion, meta = ai_svc.complete(coach.trend_messages(window), model)
+        insight = coach.parse_trend_insight(completion.text)
+    except ai_svc.AIError as e:
+        return {
+            "insight": coach.rule_trend_insight(window), "window": window,
+            "meta": {"source": "rules", "reason": str(e)},
+        }
+
+    try:
+        _write_insight(user_id, "trend", subject, fingerprint, insight, meta)
+    except Exception as e:  # noqa: BLE001 - a cache write must never fail the request
+        print(f"[analysis] failed to cache trend insight: {e}")
+    return {"insight": insight, "window": window,
+            "meta": {**meta, "source": "ai", "cached": False}}
+
+
+def copilot_stream(user_id, question, history=None, date=None, model=None):
+    """Stream a Copilot answer, yielding (event, data) pairs.
+
+    A generator rather than a return value so the route can forward each delta
+    as it arrives; the health context is assembled once, here, so the route
+    stays free of feature logic.
+    """
+    context = insights.build_context(user_id, date)
+    if not context:
+        yield "error", {"message": "暂无健康数据，请先同步 Garmin 数据。"}
+        return
+
+    messages = coach.copilot_messages(context, history or [], question)
+    yield "start", {"snapshotDate": context["snapshotDate"]}
+    upstream = None
+    try:
+        for delta in ai_svc.stream_chat(messages, model):
+            upstream = delta.upstream or upstream
+            yield "delta", {"text": delta.text}
+    except ai_svc.AIError as e:
+        yield "error", {"message": str(e)}
+        return
+    yield "done", {"upstream": upstream}
+
+
+def _delete_insight(user_id, kind, subject):
+    execute(
+        "DELETE FROM ai_insights WHERE user_id = ? AND kind = ? AND subject = ?",
+        [user_id, kind, subject],
+    )
+
+
+def clear_insight_cache(user_id, kind=None):
+    if kind:
+        execute(
+            "DELETE FROM ai_insights WHERE user_id = ? AND kind = ?", [user_id, kind]
+        )
+    else:
+        execute("DELETE FROM ai_insights WHERE user_id = ?", [user_id])