""" The AI coach: morning briefing, trend attribution, and the Copilot chat. Division of labour with `insights.py`: every number quoted here was already computed there. This module only turns a structured context into a prompt and turns the reply back into a structured answer. Nothing asks the model to do arithmetic, because a model asked to derive a z-score from a CSV gets it wrong often enough that the briefing would quote figures the charts contradict. Each feature has a rule-based counterpart. A model round-trip through the gateway costs minutes (its primary upstream is a large reasoning model), and a health screen that shows nothing when an upstream is rate-limited is worse than one that shows a plainer answer — so `meta.source` says which one answered rather than the failure being invisible. """ import json from services import ai as ai_svc from services import insights SYSTEM = """# 角色 你是一名资深运动生理学专家与佳明（Garmin）数据分析教练。你解读用户的可穿戴设备 数据，输出严谨、精炼、无废话的生理状态解读与行动指导。 # 生理学原则 1. 训练准备度综合睡眠分数、HRV 状态、恢复时间、急性负荷与压力历史。 2. HRV 反映副交感神经活跃度；HRV 高且静息心率低通常代表恢复良好。 3. 身体电量的充电量受睡眠质量与深睡/REM 比例影响：深睡负责肌肉与体力恢复， REM 负责认知与精神修复。 4. 强度分钟与运动记录代表急性负荷；负荷骤增后 HRV 短暂下降属正常应激反应。 # 数据纪律 - 只使用输入 JSON 中出现的数值，禁止编造或估算任何未给出的数字。 - 字段为 null 表示该项未采集，要么略过，要么明确说明"未采集"，不要当作 0。 - z 值（z）是该指标相对用户自身近 28 天基线的偏离程度，已经算好，直接引用即可， 不要自行重算。|z| < 1 属正常波动，不要渲染成异常。 - 你不是医生，不做医疗诊断；只从运动恢复、疲劳管理与作息角度给建议。发现明显 异常时提示用户咨询专业医师。 # 输出 - 简体中文。 - 逻辑严谨、直接明确，禁止客套、禁止情绪化修辞。 - 最终答案必须是一个 JSON 对象，且是你整段输出中最后出现的 JSON。 JSON 之外的任何文字都会被丢弃。""" BRIEFING_SCHEMA = """{ "status": "对整体恢复状态的定性，不超过 8 字，例如 '恢复良好' / '中等偏上' / '疲劳累积'", "headline": "一句话总结今日身体状态，不超过 40 字", "diagnosis": [ {"title": "维度名，如 睡眠结构 / 自主神经 / 电量与就绪度", "detail": "该维度的判断与依据，引用具体数值，不超过 60 字"} ], "shortfall": "今日最主要的短板，一句话；若无明显短板则写 '无明显短板'", "prescription": { "intensity": "今日运动强度上限，如 低 / 中等 / 中等偏高 / 高", "hrZone": "建议心率区间，如 'Zone 2~Zone 3'；无法判断填 null", "suggestion": "具体运动处方，含项目与时长，不超过 40 字", "durationMin": 建议时长的分钟数（整数）或 null, "avoid": "今日应避免的内容，不超过 20 字；无则填 null" }, "actions": ["今日可执行的具体行动，2~4 条，每条不超过 30 字"] }""" TREND_SCHEMA = """{ "summary": "这段区间内该指标发生了什么，一句话，不超过 50 字", "drivers": [ {"factor": "关联因素名", "detail": "它与该指标的关系及依据，引用数值，不超过 60 字"} ], "caution": "需要留意的风险或误读；没有则填 null", "confidence": "high|medium|low —— 取决于样本量与关联证据强度" }""" def _payload(context): """The context as compact JSON. `ensure_ascii=False` matters for size as much as readability: escaping Chinese labels to \\uXXXX roughly triples their token cost. """ return json.dumps(context, ensure_ascii=False, separators=(",", ":")) def briefing_messages(context): return [ {"role": "system", "content": SYSTEM}, { "role": "user", "content": ( "以下是我的健康数据快照。deviations 中的 z 值是相对我自身近 28 天\n" "基线的偏离，trends 是长周期走势，activityShift 是近 7 天与之前的\n" "活动量对比。\n\n" f"```json\n{_payload(context)}\n```\n\n" "请给出今日晨间简报与运动处方，严格按以下 JSON 结构输出：\n\n" f"{BRIEFING_SCHEMA}" ), }, ] def trend_messages(window): return [ {"role": "system", "content": SYSTEM}, { "role": "user", "content": ( f"以下是我 {window['label']} 指标在 {window['start']} ~ {window['end']}\n" "区间的数据，companions 是同区间内其它指标的均值，activities 是该区间\n" "内的运动记录，baselineBefore 是该区间之前的基线。\n\n" f"```json\n{_payload(window)}\n```\n\n" "请解释这段区间内该指标的变化及其可能的驱动因素，严格按以下 JSON\n" f"结构输出：\n\n{TREND_SCHEMA}" ), }, ] COPILOT_SYSTEM = SYSTEM.replace( """# 输出 - 简体中文。 - 逻辑严谨、直接明确，禁止客套、禁止情绪化修辞。 - 最终答案必须是一个 JSON 对象，且是你整段输出中最后出现的 JSON。 JSON 之外的任何文字都会被丢弃。""", """# 输出 - 简体中文，Markdown 格式。 - 逻辑严谨、直接明确，禁止客套、禁止情绪化修辞。 - 控制在 300 字以内，先给结论再给依据。 - 引用数值时写明是哪一天或哪个区间的值。 - 问题超出所给数据能回答的范围时，直接说明数据里没有，不要猜。""", ) def copilot_messages(context, history, question): """Chat turns for the Copilot. The health context rides in the system turn rather than being prepended to the user's question: it stays out of the visible transcript, and the same snapshot governs every turn instead of being re-sent (and re-charged) with each follow-up. """ messages = [ {"role": "system", "content": COPILOT_SYSTEM}, { "role": "system", "content": ( "以下是提问者的健康数据快照，回答时以它为唯一事实来源：\n" f"```json\n{_payload(context)}\n```" ), }, ] # Filtered first, then capped: capping first lets a single unusable entry # in the tail — a tool frame, an empty message — silently cost the model a # remembered turn. usable = [ {"role": t["role"], "content": (t.get("content") or "").strip()[:2000]} for t in history if t.get("role") in ("user", "assistant") and (t.get("content") or "").strip() ] messages.extend(usable[-8:]) messages.append({"role": "user", "content": question[:2000]}) return messages # --- reply validation ------------------------------------------------------- def _text(value, limit): if value is None: return None text = str(value).strip() return text[:limit] if text else None def parse_briefing(reply): data = ai_svc.extract_json(reply) if not isinstance(data, dict): raise ai_svc.AIError("模型未返回 JSON 对象") prescription = data.get("prescription") if not isinstance(prescription, dict): prescription = {} duration = prescription.get("durationMin") try: duration = int(duration) if duration is not None else None except (TypeError, ValueError): duration = None diagnosis = [] for item in data.get("diagnosis") or []: if isinstance(item, dict): title = _text(item.get("title"), 20) detail = _text(item.get("detail"), 200) else: title, detail = None, _text(item, 200) if detail: diagnosis.append({"title": title or "综合", "detail": detail}) actions = [ _text(a, 60) for a in (data.get("actions") or []) if _text(a, 60) ] out = { "status": _text(data.get("status"), 20) or "状态未定性", "headline": _text(data.get("headline"), 120), "diagnosis": diagnosis[:5], "shortfall": _text(data.get("shortfall"), 120), "prescription": { "intensity": _text(prescription.get("intensity"), 20), "hrZone": _text(prescription.get("hrZone"), 40), "suggestion": _text(prescription.get("suggestion"), 120), "durationMin": duration, "avoid": _text(prescription.get("avoid"), 60), }, "actions": actions[:4], } # A briefing with neither a headline nor any diagnosis is an empty card; # rejecting it here lets the caller fall back to the rule engine instead # of rendering blank space. if not out["headline"] and not out["diagnosis"]: raise ai_svc.AIError("模型返回的简报没有可用内容") return out def parse_trend_insight(reply): data = ai_svc.extract_json(reply) if not isinstance(data, dict): raise ai_svc.AIError("模型未返回 JSON 对象") drivers = [] for item in data.get("drivers") or []: if isinstance(item, dict): factor = _text(item.get("factor"), 30) detail = _text(item.get("detail"), 200) else: factor, detail = None, _text(item, 200) if detail: drivers.append({"factor": factor or "关联因素", "detail": detail}) confidence = str(data.get("confidence", "medium")).lower() if confidence not in ("high", "medium", "low"): confidence = "medium" summary = _text(data.get("summary"), 200) if not summary and not drivers: raise ai_svc.AIError("模型返回的归因没有可用内容") return { "summary": summary, "drivers": drivers[:5], "caution": _text(data.get("caution"), 200), "confidence": confidence, } # --- rule-based counterparts ------------------------------------------------ def rule_briefing(context): """A briefing assembled from the computed features alone. Deliberately quotes the same numbers the AI version would, so a fallback reads as a plainer answer rather than a different one. """ today = context["todayMetrics"] sleep = today["sleep"] or {} nervous = today["autonomicNervous"] recovery = today["recovery"] activity = today["activityToday"] by_metric = {d["metric"]: d for d in context["deviations"]} diagnosis = [] concerns = [] duration = sleep.get("durationHours") if duration is not None: target = sleep.get("targetHours") or insights.SLEEP_TARGET_HOURS parts = [f"睡眠 {duration:.1f} 小时（目标 {target:g}）"] rem = sleep.get("remPercent") if rem is not None: low, high = insights.REM_REFERENCE_PCT parts.append(f"REM {rem:g}%{'（偏低）' if rem < low else ''}") deep = sleep.get("deepPercent") if deep is not None: low, _ = insights.DEEP_REFERENCE_PCT parts.append(f"深睡 {deep:g}%{'（偏低）' if deep < low else '（达标）'}") diagnosis.append({"title": "睡眠结构", "detail": "，".join(parts) + "。"}) if duration < target: concerns.append(f"睡眠比目标少 {target - duration:.1f} 小时") hrv, rhr = nervous.get("hrvMs"), nervous.get("restingHr") if hrv is not None or rhr is not None: parts = [] if hrv is not None: base = by_metric.get("heartRateVariability", {}).get("baselineMean") parts.append( f"HRV {hrv:g} ms" + (f"（基线 {base:g}）" if base is not None else "") ) if rhr is not None: base = by_metric.get("heartRate", {}).get("baselineMean") parts.append( f"静息心率 {rhr:g} bpm" + (f"（基线 {base:g}）" if base is not None else "") ) diagnosis.append({"title": "自主神经", "detail": "，".join(parts) + "。"}) readiness = recovery.get("trainingReadiness") battery = recovery.get("bodyBatteryPeak") if readiness is not None or battery is not None: parts = [] if readiness is not None: parts.append(f"训练准备度 {readiness:g}/100") if battery is not None: parts.append(f"身体电量充至 {battery:g}") diagnosis.append({"title": "恢复与就绪度", "detail": "，".join(parts) + "。"}) # Readiness is Garmin's own composite of sleep, HRV, recovery time and # acute load, so it drives the prescription wherever it exists; the # sleep/HRV fallback below is only for watches that do not report it. if readiness is not None: if readiness >= 75: intensity, zone, suggestion = "高", "Zone 3~Zone 4", "可安排高强度或长时间训练" elif readiness >= 50: intensity, zone, suggestion = "中等", "Zone 2~Zone 3", "30-45 分钟中低强度有氧" else: intensity, zone, suggestion = "低", "Zone 1~Zone 2", "以走路或拉伸为主，优先恢复" elif duration is not None and duration < (sleep.get("targetHours") or 7): intensity, zone, suggestion = "中等偏低", "Zone 2", "30 分钟低强度有氧，避免加练" else: intensity, zone, suggestion = "中等", "Zone 2~Zone 3", "30-45 分钟中低强度有氧" actions = [] steps, goal = activity.get("steps"), activity.get("stepGoal") if steps is not None and goal and steps < goal: actions.append(f"步数 {steps:,} / 目标 {goal:,}，补一段快走") elif steps is not None and steps < 6000: actions.append(f"今日步数 {steps:,}，偏低，安排一次散步") if concerns: actions.append("提前 30 分钟入睡，补回睡眠缺口") sedentary = activity.get("sedentaryHours") if sedentary and sedentary >= 8: actions.append(f"久坐 {sedentary:g} 小时，每小时起身活动 3 分钟") shift = context.get("activityShift", {}).get("steps") if shift and shift.get("changePct") is not None and shift["changePct"] <= -20: actions.append(f"近 7 天步数较此前下降 {abs(shift['changePct']):g}%，注意活动量") if not actions: actions.append("各项指标处于常态，保持当前作息与训练安排") notable = [ d for d in context["deviations"] if d.get("z") is not None and abs(d["z"]) >= insights.Z_NOTABLE ] if notable: top = notable[0] status = "存在偏离" headline = ( f"{top['label']} {top['value']:g}{top['unit']}，" f"偏离近 {top['baselineDays']} 天基线 {abs(top['z']):.1f} 个标准差。" ) else: status = "状态平稳" headline = "各项指标均在个人基线的正常波动范围内。" return { "status": status, "headline": headline, "diagnosis": diagnosis, "shortfall": "；".join(concerns) if concerns else "无明显短板", "prescription": { "intensity": intensity, "hrZone": zone, "suggestion": suggestion, "durationMin": None, "avoid": None, }, "actions": actions[:4], } def rule_trend_insight(window): """Trend attribution without a model: direction, size, and co-movement.""" slope = window.get("slopePer30d") label, unit = window["label"], window["unit"] if slope is None: summary = f"{window['start']} ~ {window['end']} 区间内 {label} 样本不足，无法判断趋势。" else: direction = "上升" if slope > 0 else ("下降" if slope < 0 else "基本持平") summary = ( f"{label} 在该区间{direction}，拟合斜率约 {slope:g}{unit}/30 天，" f"均值 {window['mean']:g}{unit}。" ) drivers = [] baseline = window.get("baselineBefore") if baseline and window.get("mean") is not None: delta = window["mean"] - baseline["mean"] drivers.append({ "factor": "区间前基线", "detail": ( f"区间前 {baseline['days']} 天均值 {baseline['mean']:g}{unit}，" f"区间内{'高出' if delta >= 0 else '低于'} {abs(delta):.2f}{unit}。" ), }) activities = window.get("activities") or [] if activities: minutes = sum(a.get("durationMin") or 0 for a in activities) drivers.append({ "factor": "运动负荷", "detail": f"该区间共 {len(activities)} 次运动，合计约 {minutes} 分钟。", }) return { "summary": summary, "drivers": drivers, "caution": "该结论由规则计算得出，未经模型归因，仅描述相关性而非因果。", "confidence": "low", } # --- per-screen insights ---------------------------------------------------- # One prompt for every screen. The screen-specific part is entirely in the # context `scopes.py` builds, so a new screen needs a builder and nothing here. SCOPE_SCHEMA = """{ "headline": "针对这个页面的一句话结论，不超过 45 字", "points": [ {"title": "维度名，不超过 8 字", "detail": "该维度的判断与依据，引用具体数值，不超过 70 字"} ], "actions": ["可执行的建议，1~3 条，每条不超过 30 字；没有值得建议的就给空数组"], "caution": "需要留意的风险或容易误读之处；没有就填 null", "confidence": "high|medium|low —— 取决于样本量与证据强度" }""" def scope_messages(context): """Prompt for one screen's insight. `highlights` goes in ahead of the raw data on purpose: they are the facts already computed from it, and leading with them is what stops the model re-deriving (and mis-deriving) numbers that are sitting right there. """ highlights = "\n".join( f"- {h['title']}：{h['detail']}" for h in context.get("highlights") or [] ) return [ {"role": "system", "content": SYSTEM}, { "role": "user", "content": ( f"我正在看「{context.get('label')}」这个页面。以下是已经算好的要点：\n\n" f"{highlights}\n\n" "完整数据如下（deviations 的 z 值是相对我自身近 28 天基线的偏离，" "trends 是长周期走势，两者都已算好，直接引用即可）：\n\n" f"```json\n{_payload(context)}\n```\n\n" "请针对这个页面给出解读与建议，只谈这个页面涉及的内容，" f"严格按以下 JSON 结构输出：\n\n{SCOPE_SCHEMA}" ), }, ] def parse_scope_insight(reply): data = ai_svc.extract_json(reply) if not isinstance(data, dict): raise ai_svc.AIError("模型未返回 JSON 对象") points = [] for item in data.get("points") or []: if isinstance(item, dict): title = _text(item.get("title"), 20) detail = _text(item.get("detail"), 220) else: title, detail = None, _text(item, 220) if detail: points.append({"title": title or "要点", "detail": detail}) actions = [_text(a, 60) for a in (data.get("actions") or []) if _text(a, 60)] confidence = str(data.get("confidence", "medium")).lower() if confidence not in ("high", "medium", "low"): confidence = "medium" headline = _text(data.get("headline"), 140) # A card with no headline and no points is blank space; rejecting it lets # the caller fall back to the rule engine rather than render nothing. if not headline and not points: raise ai_svc.AIError("模型返回的解读没有可用内容") return { "headline": headline, "points": points[:5], "actions": actions[:3], "caution": _text(data.get("caution"), 200), "confidence": confidence, } def rule_scope_insight(context): """The computed highlights, rendered as-is. No interpretation, and it says so: the honest fallback is the facts without the reading of them, not a guess at what a model would have said. """ highlights = context.get("highlights") or [] return { # The first highlight leads and is then dropped from the list: showing # it in both places printed the same sentence twice. # # This is why every builder's first `detail` has to read on its own — # here the `title` beside it is not rendered, and "395 天内由 4376 到 # 4905（改善）" without naming 耐力分 is a sentence about nothing. "headline": highlights[0]["detail"] if highlights else None, "points": [dict(h) for h in highlights[1:6]], "actions": [], "caution": "以下为直接计算结果，尚未经过模型解读。", "confidence": "low", }