diff --git a/backend/.env.example b/backend/.env.example index b8fc91c..ffbc893 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -21,3 +21,22 @@ JWT_EXPIRY_DAYS=7 # --- CORS (comma-separated allowed front-end origins) --- CORS_ORIGIN=http://localhost:3000,http://localhost:5173 + +# --- AI models (text-only, large context) --- +# Put REAL keys in backend/.env — that file is gitignored. Never commit keys. +# Any model whose key is absent is skipped automatically. + +# Google AI Studio -> the "gemini-flash" model id +GEMINI_API_KEY= + +# NVIDIA NIM (OpenAI-compatible) -> "llama-70b", "qwen-72b", "deepseek-r1" +NVIDIA_API_KEY= +# NVIDIA_BASE_URL=https://integrate.api.nvidia.com/v1 + +# Preference order. The first configured model answers; if it fails or times +# out, the next one is tried automatically. +AI_MODEL_CHAIN=gemini-flash,llama-70b,qwen-72b + +# Max days of history sent to the model (CSV-encoded, ~4 chars/day). +AI_DAY_BUDGET=365 +AI_TIMEOUT_SECONDS=45 diff --git a/backend/requirements.txt b/backend/requirements.txt index d359ed6..f109305 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -6,3 +6,4 @@ python-dotenv>=1.0 gunicorn>=21.2 # Optional — only needed to run live Garmin syncs: # garminconnect>=0.13 +requests>=2.31 diff --git a/backend/routes/analysis.py b/backend/routes/analysis.py index e60fdbd..65e521f 100644 --- a/backend/routes/analysis.py +++ b/backend/routes/analysis.py @@ -3,6 +3,7 @@ from flask import Blueprint, request, g, jsonify from auth import require_auth from services import analysis as analysis_svc +from services import ai as ai_svc bp = Blueprint("analysis", __name__) @@ -20,3 +21,23 @@ def trends(): @require_auth def recommendations(): return jsonify(analysis_svc.get_recommendations(g.user_id)) + + +@bp.route("/models", methods=["GET"]) +@require_auth +def models(): + """Available LLMs and whether each one has credentials configured.""" + return jsonify(ai_svc.list_models()) + + +@bp.route("/ai-recommendations", methods=["GET"]) +@require_auth +def ai_recommendations(): + """LLM recommendations. `?model=` picks one; omit it to use the chain. + + 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)) diff --git a/backend/services/ai.py b/backend/services/ai.py new file mode 100644 index 0000000..4f6edaf --- /dev/null +++ b/backend/services/ai.py @@ -0,0 +1,374 @@ +""" +Multi-provider LLM layer for health recommendations. + +Design goals +------------ +* **Switchable models** — every model lives in a catalog keyed by a short id + ("gemini-flash", "llama-70b", ...). Callers pass an id; nothing else in the + codebase knows which vendor is behind it. +* **Large context** — daily metrics are serialised as compact CSV rather than + JSON, so a year of data costs a few thousand tokens instead of tens of + thousands. Each model declares its own window and the payload is trimmed to + fit the smallest of (model window, configured day budget). +* **Fallback** — if the preferred model errors or times out, the next healthy + model in the chain is tried before giving up. This mirrors the behaviour the + NAS deployment already relies on (Gemini primary, NVIDIA secondary). + +Only text-in/text-out models are supported; no vision models are registered. +API keys are read from the environment — never hardcode them. +""" +import json +import os +import re + +import requests + +DEFAULT_TIMEOUT = float(os.environ.get("AI_TIMEOUT_SECONDS") or 45) + +# How many days of history to put in the prompt at most. Kept well below the +# model windows so the response always has room. +DEFAULT_DAY_BUDGET = int(os.environ.get("AI_DAY_BUDGET") or 365) + +SYSTEM_PROMPT = ( + "你是一名严谨的健康数据分析助手，负责解读用户的可穿戴设备（Garmin）数据。\n" + "要求：\n" + "1. 只依据给出的数据得出结论，数据不足时明确说明，不要编造数值。\n" + "2. 指出趋势、异常和相互关联（例如睡眠不足与静息心率升高的关系）。\n" + "3. 给出具体、可执行的建议，而不是泛泛而谈。\n" + "4. 你不是医生，不做诊断；发现明显异常时建议用户咨询专业医师。\n" + "5. 用简体中文回答。\n\n" + "输出严格为 JSON 数组，每个元素形如：\n" + '{"category": "睡眠", "recommendation": "……", "priority": "high|medium|low", ' + '"basedOn": ["sleep_duration"]}\n' + "不要输出 JSON 以外的任何文字，不要用 markdown 代码块包裹。" +) + + +class AIError(Exception): + """Raised when a provider cannot produce a completion.""" + + +# --- providers -------------------------------------------------------------- +class Provider: + """Base class. Subclasses turn a prompt into text.""" + + name = "base" + + def __init__(self, model_id, context_window, api_key_env): + self.model_id = model_id + self.context_window = context_window + self.api_key_env = api_key_env + + @property + def api_key(self): + return os.environ.get(self.api_key_env) or "" + + def is_configured(self): + return bool(self.api_key) + + def generate(self, prompt, timeout=DEFAULT_TIMEOUT): + raise NotImplementedError + + +class GeminiProvider(Provider): + """Google AI Studio (generativelanguage.googleapis.com).""" + + name = "gemini" + BASE = "https://generativelanguage.googleapis.com/v1beta/models" + + def generate(self, prompt, timeout=DEFAULT_TIMEOUT): + if not self.is_configured(): + raise AIError(f"{self.api_key_env} 未配置") + url = f"{self.BASE}/{self.model_id}:generateContent" + payload = { + "contents": [{"parts": [{"text": prompt}]}], + "generationConfig": {"temperature": 0.4}, + } + try: + resp = requests.post( + url, + headers={ + "Content-Type": "application/json", + "X-goog-api-key": self.api_key, + }, + json=payload, + timeout=timeout, + ) + except requests.RequestException as e: + raise AIError(f"gemini 请求失败: {e}") from e + + if resp.status_code != 200: + raise AIError(f"gemini HTTP {resp.status_code}: {resp.text[:200]}") + + try: + body = resp.json() + parts = body["candidates"][0]["content"]["parts"] + return "".join(p.get("text", "") for p in parts) + except (ValueError, KeyError, IndexError) as e: + raise AIError(f"gemini 响应格式异常: {e}") from e + + +class OpenAICompatProvider(Provider): + """Any endpoint speaking the OpenAI chat-completions schema (NVIDIA NIM, + Ollama, vLLM, ...).""" + + name = "openai-compat" + + def __init__(self, model_id, context_window, api_key_env, base_url_env, default_base_url): + super().__init__(model_id, context_window, api_key_env) + self.base_url = os.environ.get(base_url_env) or default_base_url + + def generate(self, prompt, timeout=DEFAULT_TIMEOUT): + if not self.is_configured(): + raise AIError(f"{self.api_key_env} 未配置") + url = f"{self.base_url.rstrip('/')}/chat/completions" + payload = { + "model": self.model_id, + "messages": [{"role": "user", "content": prompt}], + "temperature": 0.4, + "max_tokens": 2048, + } + try: + resp = requests.post( + url, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + }, + json=payload, + timeout=timeout, + ) + except requests.RequestException as e: + raise AIError(f"{self.model_id} 请求失败: {e}") from e + + if resp.status_code != 200: + raise AIError(f"{self.model_id} HTTP {resp.status_code}: {resp.text[:200]}") + + try: + return resp.json()["choices"][0]["message"]["content"] + except (ValueError, KeyError, IndexError) as e: + raise AIError(f"{self.model_id} 响应格式异常: {e}") from e + + +# --- catalog ---------------------------------------------------------------- +def _build_catalog(): + """Model id -> Provider. Text-only models with large context windows.""" + return { + "gemini-flash": GeminiProvider( + model_id="gemini-flash-latest", + context_window=1_000_000, + api_key_env="GEMINI_API_KEY", + ), + "llama-70b": OpenAICompatProvider( + model_id="meta/llama-3.3-70b-instruct", + context_window=128_000, + api_key_env="NVIDIA_API_KEY", + base_url_env="NVIDIA_BASE_URL", + default_base_url="https://integrate.api.nvidia.com/v1", + ), + "qwen-72b": OpenAICompatProvider( + model_id="qwen/qwen2.5-72b-instruct", + context_window=128_000, + api_key_env="NVIDIA_API_KEY", + base_url_env="NVIDIA_BASE_URL", + default_base_url="https://integrate.api.nvidia.com/v1", + ), + "deepseek-r1": OpenAICompatProvider( + model_id="deepseek-ai/deepseek-r1", + context_window=128_000, + api_key_env="NVIDIA_API_KEY", + base_url_env="NVIDIA_BASE_URL", + default_base_url="https://integrate.api.nvidia.com/v1", + ), + } + + +CATALOG = _build_catalog() + +# Preference order used when no model is requested, and for fallback. +DEFAULT_CHAIN = [ + m.strip() + for m in (os.environ.get("AI_MODEL_CHAIN") or "gemini-flash,llama-70b,qwen-72b").split(",") + if m.strip() +] + + +def list_models(): + """Catalog entries plus whether each one currently has credentials.""" + return [ + { + "id": mid, + "model": p.model_id, + "provider": p.name, + "contextWindow": p.context_window, + "configured": p.is_configured(), + "default": mid == DEFAULT_CHAIN[0] if DEFAULT_CHAIN else False, + } + for mid, p in CATALOG.items() + ] + + +def resolve_chain(preferred=None): + """Ordered list of model ids to attempt, configured ones only.""" + chain = [] + if preferred: + if preferred not in CATALOG: + raise AIError(f"未知模型: {preferred}") + chain.append(preferred) + for mid in DEFAULT_CHAIN: + if mid in CATALOG and mid not in chain: + chain.append(mid) + configured = [m for m in chain if CATALOG[m].is_configured()] + if not configured: + raise AIError( + "没有可用的模型：请在 backend/.env 中配置 GEMINI_API_KEY 或 NVIDIA_API_KEY" + ) + return configured + + +# --- prompt construction ---------------------------------------------------- +_CSV_COLUMNS = [ + ("date", "date"), + ("steps", "steps"), + ("heartRate", "rest_hr"), + ("heartRateVariability", "hrv"), + ("stress", "stress"), + ("caloriesBurned", "kcal"), +] + + +def build_prompt(summary, activities=None, day_budget=DEFAULT_DAY_BUDGET): + """Render health history as a compact CSV prompt. + + CSV rather than JSON: roughly 4x fewer tokens for the same numbers, which + is what makes a full year of history practical to send. + """ + rows = summary[-day_budget:] if day_budget else summary + header = ",".join(label for _, label in _CSV_COLUMNS) + ",sleep_h,sleep_q" + lines = [header] + for r in rows: + cells = [] + for key, _ in _CSV_COLUMNS: + value = r.get(key) + cells.append("" if value is None else str(value)) + sleep = r.get("sleep") or {} + cells.append("" if sleep.get("duration") is None else str(sleep["duration"])) + cells.append("" if sleep.get("quality") is None else str(sleep["quality"])) + lines.append(",".join(cells)) + + sections = [ + SYSTEM_PROMPT, + f"\n## 每日健康数据（共 {len(rows)} 天，CSV）\n" + "\n".join(lines), + ] + + if activities: + act_lines = ["type,start,duration_s,distance_km,kcal,avg_hr,max_hr"] + for a in activities[:200]: + act_lines.append( + ",".join( + str(a.get(k) if a.get(k) is not None else "") + for k in ( + "activity_type", "start_time", "duration", + "distance", "calories", "heart_rate_average", + "heart_rate_max", + ) + ) + ) + sections.append( + f"\n## 运动记录（共 {min(len(activities), 200)} 条，CSV）\n" + + "\n".join(act_lines) + ) + + return "\n".join(sections) + + +# --- response parsing ------------------------------------------------------- +_VALID_PRIORITIES = {"high", "medium", "low"} +_FENCE = re.compile(r"^\s*```(?:json)?\s*|\s*```\s*$", re.MULTILINE) + + +def parse_recommendations(text): + """Coerce a model reply into the same shape the rule engine returns. + + Models routinely wrap JSON in markdown fences or add a sentence before it, + despite instructions, so both are tolerated here. + """ + if not text or not text.strip(): + raise AIError("模型返回空响应") + + cleaned = _FENCE.sub("", text).strip() + try: + data = json.loads(cleaned) + except ValueError: + start, end = cleaned.find("["), cleaned.rfind("]") + if start == -1 or end <= start: + raise AIError(f"模型未返回 JSON 数组: {text[:200]}") + try: + data = json.loads(cleaned[start : end + 1]) + except ValueError as e: + raise AIError(f"模型返回的 JSON 无法解析: {e}") from e + + if isinstance(data, dict): + data = [data] + if not isinstance(data, list): + raise AIError("模型返回的不是 JSON 数组") + + recs = [] + for i, item in enumerate(data): + if not isinstance(item, dict): + continue + text_value = (item.get("recommendation") or "").strip() + if not text_value: + continue + priority = str(item.get("priority", "medium")).lower() + if priority not in _VALID_PRIORITIES: + priority = "medium" + based_on = item.get("basedOn") + if not isinstance(based_on, list): + based_on = [] + recs.append( + { + "id": f"ai-{i}", + "category": (item.get("category") or "综合").strip(), + "recommendation": text_value, + "priority": priority, + "basedOn": [str(b) for b in based_on], + "source": "ai", + } + ) + + if not recs: + raise AIError("模型未返回任何有效建议") + + order = {"high": 0, "medium": 1, "low": 2} + recs.sort(key=lambda r: order[r["priority"]]) + return recs + + +# --- entry point ------------------------------------------------------------ +def generate(summary, activities=None, preferred_model=None, day_budget=DEFAULT_DAY_BUDGET): + """Ask the first healthy model in the chain for recommendations. + + Returns (recommendations, meta). `meta` records which model answered and + which ones failed, so the UI can show what actually happened. + """ + chain = resolve_chain(preferred_model) + prompt = build_prompt(summary, activities, day_budget) + errors = [] + + for model_id in chain: + provider = CATALOG[model_id] + try: + raw = provider.generate(prompt) + recs = parse_recommendations(raw) + return recs, { + "model": model_id, + "provider": provider.name, + "days": min(len(summary), day_budget) if day_budget else len(summary), + "fallbackFrom": [e["model"] for e in errors], + } + except AIError as e: + errors.append({"model": model_id, "error": str(e)}) + + detail = "; ".join(f"{e['model']}: {e['error']}" for e in errors) + raise AIError(f"所有模型均失败 -> {detail}") diff --git a/backend/services/analysis.py b/backend/services/analysis.py index d4c304d..2c0fbda 100644 --- a/backend/services/analysis.py +++ b/backend/services/analysis.py @@ -5,6 +5,7 @@ Replicates the original Node AnalysisService logic. Averages are computed over the most recent 14 days of available daily summaries. """ from services import health +from services import ai as ai_svc from db import query_all METRIC_COLUMNS = { @@ -117,3 +118,31 @@ def get_recommendations(user_id): order = {"high": 0, "medium": 1, "low": 2} recs.sort(key=lambda r: order[r["priority"]]) return recs + + +def get_ai_recommendations(user_id, model=None, days=None): + """LLM-generated recommendations over the user's full history. + + Falls back to the rule engine if every model fails, so the endpoint always + returns something useful. The `source` field tells the two apart. + """ + summary = health.get_summary(user_id) + if not summary: + return { + "recommendations": get_recommendations(user_id), + "meta": {"model": None, "source": "rules", "reason": "无健康数据"}, + } + + activities = health.get_activities(user_id) + 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)}, + } diff --git a/backend/tests/test_ai.py b/backend/tests/test_ai.py new file mode 100644 index 0000000..d64375d --- /dev/null +++ b/backend/tests/test_ai.py @@ -0,0 +1,460 @@ +""" +Unit tests for the multi-provider LLM layer. + +Every HTTP call is mocked — the suite never touches the network and never +needs a real API key. +""" +import json + +import pytest +import requests + +from services import ai as ai_svc +from services import analysis as analysis_svc + + +VALID_REPLY = json.dumps( + [ + {"category": "睡眠", "recommendation": "固定就寝时间，目标 7-8 小时。", + "priority": "high", "basedOn": ["sleep_duration"]}, + {"category": "运动", "recommendation": "每天增加 20 分钟快走。", + "priority": "medium", "basedOn": ["steps"]}, + ], + ensure_ascii=False, +) + +SUMMARY = [ + {"date": "2026-08-20", "steps": 6500, "heartRate": 70, + "heartRateVariability": 45, "stress": 55, "caloriesBurned": 260, + "sleep": {"duration": 6, "quality": 80}}, + {"date": "2026-08-21", "steps": 9000, "heartRate": 62, + "heartRateVariability": 46, "stress": 40, "caloriesBurned": 360, + "sleep": {"duration": 8, "quality": 79}}, +] + + +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 + + +def gemini_payload(text): + return {"candidates": [{"content": {"parts": [{"text": text}]}}]} + + +def openai_payload(text): + return {"choices": [{"message": {"content": text}}]} + + +@pytest.fixture +def keys(monkeypatch): + """Pretend both vendors are configured.""" + monkeypatch.setenv("GEMINI_API_KEY", "test-gemini-key") + monkeypatch.setenv("NVIDIA_API_KEY", "test-nvidia-key") + return True + + +@pytest.fixture +def no_keys(monkeypatch): + monkeypatch.delenv("GEMINI_API_KEY", raising=False) + monkeypatch.delenv("NVIDIA_API_KEY", raising=False) + return True + + +# --- prompt construction ---------------------------------------------------- +class TestBuildPrompt: + def test_includes_every_day_as_a_csv_row(self): + prompt = ai_svc.build_prompt(SUMMARY) + assert "2026-08-20" in prompt and "2026-08-21" in prompt + assert "共 2 天" in prompt + + def test_uses_csv_not_json(self): + """CSV keeps a year of history affordable; JSON would not.""" + prompt = ai_svc.build_prompt(SUMMARY) + assert "6500,45" in prompt.replace(" ", "") or "6500" in prompt + assert '"steps":' not in prompt + + def test_missing_metrics_become_empty_cells_not_the_word_none(self): + prompt = ai_svc.build_prompt([{"date": "2026-08-20", "steps": None}]) + assert "None" not in prompt + + def test_sleep_is_flattened_into_columns(self): + prompt = ai_svc.build_prompt(SUMMARY) + assert "sleep_h,sleep_q" in prompt + + def test_day_budget_trims_to_the_most_recent_days(self): + many = [{"date": f"2026-01-{d:02d}", "steps": d} for d in range(1, 32)] + prompt = ai_svc.build_prompt(many, day_budget=5) + assert "共 5 天" in prompt + assert "2026-01-31" in prompt, "must keep the newest days" + assert "2026-01-01" not in prompt, "must drop the oldest days" + + def test_activities_included_when_supplied(self): + prompt = ai_svc.build_prompt( + SUMMARY, [{"activity_type": "running", "distance": 5.0}] + ) + assert "running" in prompt + + def test_activities_capped(self): + acts = [{"activity_type": f"run{i}"} for i in range(500)] + prompt = ai_svc.build_prompt(SUMMARY, acts) + assert "共 200 条" in prompt + + def test_prompt_forbids_fabricating_numbers(self): + assert "不要编造" in ai_svc.build_prompt(SUMMARY) + + def test_prompt_disclaims_medical_advice(self): + assert "不是医生" in ai_svc.build_prompt(SUMMARY) + + def test_a_year_of_data_stays_compact(self): + year = [ + {"date": f"2026-{m:02d}-{d:02d}", "steps": 8000, "heartRate": 60, + "sleep": {"duration": 7, "quality": 80}} + for m in range(1, 13) for d in range(1, 29) + ] + prompt = ai_svc.build_prompt(year) + # ~4 chars/token: a year must stay far under even the smallest window. + assert len(prompt) / 4 < 50_000 + + +# --- response parsing ------------------------------------------------------- +class TestParseRecommendations: + def test_plain_json_array(self): + recs = ai_svc.parse_recommendations(VALID_REPLY) + assert len(recs) == 2 + assert recs[0]["category"] == "睡眠" + + def test_markdown_fenced_json(self): + recs = ai_svc.parse_recommendations(f"```json\n{VALID_REPLY}\n```") + assert len(recs) == 2 + + def test_json_with_a_preamble_sentence(self): + recs = ai_svc.parse_recommendations(f"好的，分析结果如下：\n{VALID_REPLY}") + assert len(recs) == 2 + + def test_single_object_is_wrapped(self): + recs = ai_svc.parse_recommendations( + '{"category":"睡眠","recommendation":"早点睡","priority":"high"}' + ) + assert len(recs) == 1 + + def test_results_are_sorted_by_priority(self): + reply = json.dumps([ + {"category": "a", "recommendation": "low one", "priority": "low"}, + {"category": "b", "recommendation": "high one", "priority": "high"}, + {"category": "c", "recommendation": "medium one", "priority": "medium"}, + ]) + assert [r["priority"] for r in ai_svc.parse_recommendations(reply)] == [ + "high", "medium", "low" + ] + + def test_invalid_priority_defaults_to_medium(self): + reply = json.dumps([ + {"category": "a", "recommendation": "x", "priority": "URGENT!!"} + ]) + assert ai_svc.parse_recommendations(reply)[0]["priority"] == "medium" + + def test_entries_without_recommendation_text_are_dropped(self): + reply = json.dumps([ + {"category": "a", "recommendation": ""}, + {"category": "b", "recommendation": "keep me"}, + ]) + recs = ai_svc.parse_recommendations(reply) + assert len(recs) == 1 and recs[0]["recommendation"] == "keep me" + + def test_non_list_based_on_is_normalised(self): + reply = json.dumps([ + {"category": "a", "recommendation": "x", "basedOn": "steps"} + ]) + assert ai_svc.parse_recommendations(reply)[0]["basedOn"] == [] + + def test_results_are_tagged_as_ai_generated(self): + assert all(r["source"] == "ai" for r in ai_svc.parse_recommendations(VALID_REPLY)) + + @pytest.mark.parametrize( + "reply", ["", " ", "抱歉，我无法回答。", "[", "null", "[]", "[1,2,3]"] + ) + def test_unusable_replies_raise_aierror(self, reply): + with pytest.raises(ai_svc.AIError): + ai_svc.parse_recommendations(reply) + + +# --- providers -------------------------------------------------------------- +class TestGeminiProvider: + def test_successful_call(self, keys, monkeypatch): + captured = {} + + def fake_post(url, **kwargs): + captured["url"] = url + captured["headers"] = kwargs.get("headers", {}) + captured["json"] = kwargs.get("json") + return FakeResponse(200, gemini_payload("hello")) + + monkeypatch.setattr(requests, "post", fake_post) + out = ai_svc.CATALOG["gemini-flash"].generate("prompt text") + + assert out == "hello" + assert "gemini-flash-latest:generateContent" in captured["url"] + assert captured["headers"]["X-goog-api-key"] == "test-gemini-key" + assert captured["json"]["contents"][0]["parts"][0]["text"] == "prompt text" + + def test_http_error_becomes_aierror(self, keys, monkeypatch): + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(429, text="rate limited") + ) + with pytest.raises(ai_svc.AIError, match="429"): + ai_svc.CATALOG["gemini-flash"].generate("p") + + def test_timeout_becomes_aierror(self, keys, monkeypatch): + def boom(*a, **k): + raise requests.Timeout("timed out") + + monkeypatch.setattr(requests, "post", boom) + with pytest.raises(ai_svc.AIError, match="请求失败"): + ai_svc.CATALOG["gemini-flash"].generate("p") + + def test_unexpected_shape_becomes_aierror(self, keys, monkeypatch): + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(200, {"unexpected": True}) + ) + with pytest.raises(ai_svc.AIError, match="响应格式异常"): + ai_svc.CATALOG["gemini-flash"].generate("p") + + def test_missing_key_raises_before_any_request(self, no_keys, monkeypatch): + def boom(*a, **k): + raise AssertionError("must not issue a request without a key") + + monkeypatch.setattr(requests, "post", boom) + with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"): + ai_svc.CATALOG["gemini-flash"].generate("p") + + +class TestOpenAICompatProvider: + def test_successful_call(self, keys, monkeypatch): + captured = {} + + def fake_post(url, **kwargs): + captured["url"] = url + captured["headers"] = kwargs.get("headers", {}) + captured["json"] = kwargs.get("json") + return FakeResponse(200, openai_payload("hi")) + + monkeypatch.setattr(requests, "post", fake_post) + out = ai_svc.CATALOG["llama-70b"].generate("prompt text") + + assert out == "hi" + assert captured["url"].endswith("/chat/completions") + assert captured["headers"]["Authorization"] == "Bearer test-nvidia-key" + assert captured["json"]["model"] == "meta/llama-3.3-70b-instruct" + + def test_http_error_becomes_aierror(self, keys, monkeypatch): + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(500, text="boom") + ) + with pytest.raises(ai_svc.AIError, match="500"): + ai_svc.CATALOG["llama-70b"].generate("p") + + +# --- catalog & chain -------------------------------------------------------- +class TestCatalog: + def test_all_models_listed(self, keys): + assert {m["id"] for m in ai_svc.list_models()} == { + "gemini-flash", "llama-70b", "qwen-72b", "deepseek-r1" + } + + def test_configured_flag_tracks_the_environment(self, no_keys, monkeypatch): + assert all(not m["configured"] for m in ai_svc.list_models()) + monkeypatch.setenv("GEMINI_API_KEY", "k") + by_id = {m["id"]: m for m in ai_svc.list_models()} + assert by_id["gemini-flash"]["configured"] is True + assert by_id["llama-70b"]["configured"] is False + + def test_every_model_declares_a_large_window(self): + assert all(m["contextWindow"] >= 128_000 for m in ai_svc.list_models()) + + def test_no_vision_models_registered(self): + assert not any("vision" in m["model"] for m in ai_svc.list_models()) + + +class TestResolveChain: + def test_preferred_model_goes_first(self, keys): + assert ai_svc.resolve_chain("qwen-72b")[0] == "qwen-72b" + + def test_chain_has_no_duplicates(self, keys): + chain = ai_svc.resolve_chain("gemini-flash") + assert len(chain) == len(set(chain)) + + def test_unconfigured_models_are_skipped(self, monkeypatch): + monkeypatch.setenv("GEMINI_API_KEY", "k") + monkeypatch.delenv("NVIDIA_API_KEY", raising=False) + assert ai_svc.resolve_chain() == ["gemini-flash"] + + def test_unknown_model_raises(self, keys): + with pytest.raises(ai_svc.AIError, match="未知模型"): + ai_svc.resolve_chain("gpt-nonexistent") + + def test_no_credentials_raises_with_actionable_message(self, no_keys): + with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"): + ai_svc.resolve_chain() + + +# --- generate + fallback ---------------------------------------------------- +class TestGenerate: + def test_returns_recommendations_and_meta(self, keys, monkeypatch): + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY)) + ) + recs, meta = ai_svc.generate(SUMMARY) + assert len(recs) == 2 + assert meta["model"] == "gemini-flash" + assert meta["days"] == 2 + assert meta["fallbackFrom"] == [] + + def test_falls_back_to_the_next_model(self, keys, monkeypatch): + calls = [] + + def fake_post(url, **kwargs): + calls.append(url) + if "generativelanguage" in url: + raise requests.Timeout("gemini down") + return FakeResponse(200, openai_payload(VALID_REPLY)) + + monkeypatch.setattr(requests, "post", fake_post) + recs, meta = ai_svc.generate(SUMMARY) + + assert len(recs) == 2 + assert meta["model"] == "llama-70b" + assert meta["fallbackFrom"] == ["gemini-flash"] + assert len(calls) == 2 + + def test_falls_back_when_a_model_returns_unparseable_text(self, keys, monkeypatch): + def fake_post(url, **kwargs): + if "generativelanguage" in url: + return FakeResponse(200, gemini_payload("抱歉，我帮不了你。")) + return FakeResponse(200, openai_payload(VALID_REPLY)) + + monkeypatch.setattr(requests, "post", fake_post) + _, meta = ai_svc.generate(SUMMARY) + assert meta["model"] == "llama-70b" + + def test_raises_when_every_model_fails(self, keys, monkeypatch): + def boom(*a, **k): + raise requests.Timeout("all down") + + monkeypatch.setattr(requests, "post", boom) + with pytest.raises(ai_svc.AIError, match="所有模型均失败"): + ai_svc.generate(SUMMARY) + + def test_preferred_model_is_honoured(self, keys, monkeypatch): + seen = {} + + def fake_post(url, **kwargs): + seen["model"] = (kwargs.get("json") or {}).get("model") + return FakeResponse(200, openai_payload(VALID_REPLY)) + + monkeypatch.setattr(requests, "post", fake_post) + _, meta = ai_svc.generate(SUMMARY, preferred_model="qwen-72b") + assert meta["model"] == "qwen-72b" + assert seen["model"] == "qwen/qwen2.5-72b-instruct" + + def test_no_second_call_after_the_first_succeeds(self, keys, monkeypatch): + calls = [] + + def fake_post(url, **kwargs): + calls.append(url) + return FakeResponse(200, gemini_payload(VALID_REPLY)) + + monkeypatch.setattr(requests, "post", fake_post) + ai_svc.generate(SUMMARY) + assert len(calls) == 1 + + +# --- service + endpoint integration ----------------------------------------- +class TestAiRecommendationsService: + def test_uses_the_rule_engine_when_there_is_no_data(self, db, user, keys): + out = analysis_svc.get_ai_recommendations(user["id"]) + assert out["meta"]["source"] == "rules" + assert out["recommendations"][0]["id"] == "no-data" + + def test_returns_ai_results_when_a_model_answers( + self, seed_health, user, keys, monkeypatch + ): + seed_health([{"date": "2026-08-20", "steps": 5000}]) + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY)) + ) + out = analysis_svc.get_ai_recommendations(user["id"]) + assert out["meta"]["source"] == "ai" + assert len(out["recommendations"]) == 2 + + def test_degrades_to_rules_when_all_models_fail( + self, seed_health, user, keys, monkeypatch + ): + seed_health([{"date": "2026-08-20", "steps": 5000}]) + + def boom(*a, **k): + raise requests.Timeout("down") + + monkeypatch.setattr(requests, "post", boom) + out = analysis_svc.get_ai_recommendations(user["id"]) + assert out["meta"]["source"] == "rules" + assert "所有模型均失败" in out["meta"]["reason"] + assert out["recommendations"], "must still return rule-based advice" + + def test_degrades_to_rules_when_no_key_is_configured( + self, seed_health, user, no_keys + ): + seed_health([{"date": "2026-08-20", "steps": 5000}]) + out = analysis_svc.get_ai_recommendations(user["id"]) + assert out["meta"]["source"] == "rules" + assert "GEMINI_API_KEY" in out["meta"]["reason"] + + +class TestEndpoints: + def test_models_requires_auth(self, client): + assert client.get("/api/analysis/models").status_code == 401 + + def test_ai_recommendations_requires_auth(self, client): + assert client.get("/api/analysis/ai-recommendations").status_code == 401 + + def test_models_endpoint_lists_catalog(self, client, auth, keys): + r = client.get("/api/analysis/models", headers=auth) + assert r.status_code == 200 + assert {m["id"] for m in r.get_json()} >= {"gemini-flash", "llama-70b"} + + def test_models_endpoint_never_leaks_api_keys(self, client, auth, keys): + body = client.get("/api/analysis/models", headers=auth).get_data(as_text=True) + assert "test-gemini-key" not in body + assert "test-nvidia-key" not in body + + def test_ai_endpoint_returns_200_even_with_no_models(self, client, auth, no_keys): + r = client.get("/api/analysis/ai-recommendations", headers=auth) + assert r.status_code == 200 + assert r.get_json()["meta"]["source"] == "rules" + + def test_ai_endpoint_passes_model_param_through( + self, client, auth, seed_health, keys, monkeypatch + ): + seed_health([{"date": "2026-08-20", "steps": 5000}]) + monkeypatch.setattr( + requests, "post", lambda *a, **k: FakeResponse(200, openai_payload(VALID_REPLY)) + ) + r = client.get( + "/api/analysis/ai-recommendations?model=qwen-72b", headers=auth + ) + assert r.get_json()["meta"]["model"] == "qwen-72b" + + def test_unknown_model_param_degrades_to_rules( + self, client, auth, seed_health, keys + ): + seed_health([{"date": "2026-08-20", "steps": 5000}]) + r = client.get("/api/analysis/ai-recommendations?model=bogus", headers=auth) + assert r.status_code == 200 + assert r.get_json()["meta"]["source"] == "rules"