ericwyuan/GarminHealthLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[阶段4.3] 接入自建 AI 网关，修复多模型层的四个真实缺陷

Browse Source 
改用甲骨文机上已有的 ai-gateway (129.146.203.203:5100)：它本身就
OpenAI 兼容，内部串联 nvidia/gemini/ollama 并轮换 4 个 Gemini key，
比在客户端自己串联更能吸收单厂商的配额和超时。回包里的 provider
字段透传为 meta.upstream，网关侧发生降级时前端也看得见。

fix(ai): 目录里两个 NVIDIA 模型 id 根本不存在
- qwen/qwen2.5-72b-instruct 和 deepseek-ai/deepseek-r1 是我凭印象写的，
  实际 GET /v1/models 里没有，调用一律 404
- 改为该账号清单里确实存在的 nemotron-49b / mistral-large，
  并在注释里写明 id 必须取自实时清单、不能猜

fix(ai): 请求被本机代理劫持导致网关不可达
- requests 默认读 HTTP_PROXY/ALL_PROXY，把发往甲骨文公网 IP 的请求
  也塞进了 127.0.0.1:7897，120s 后超时
- 按 provider 区分：境外厂商(Gemini/NVIDIA)仍走代理，自建网关直连
  (session.trust_env=False)

fix(ai): 承诺的按模型裁剪从未实现
- 模块注释写着 payload 按 (模型窗口, 天数预算) 取小者裁剪，但实际是
  用全局预算构建一次 prompt 发给链上所有模型；365 天数据对 Gemini
  的 1M 窗口无碍，却会撑爆 128k 的模型
- 新增 max_days_for()，在循环内按各模型窗口分别构建 prompt

fix(ai): 推理模型的思考过程吃光输出预算
- 网关首选 nemotron-3-ultra-550b 是推理模型，回答前先输出一段
  chain-of-thought；默认 1024 tokens 全被思考占用，JSON 还没开始
  就被截断
- max_tokens 改为可按 provider 声明，网关条目给 3000

fix(ai): 配置在 import 时被冻结
- DEFAULT_CHAIN/TIMEOUT/DAY_BUDGET 是模块级常量，改环境变量不生效，
  且让开发机 .env 泄漏进测试进程（测试会读到真实 key 和链配置）
- 改为 default_chain()/default_timeout()/default_day_budget() 按调用读取
- conftest 增加 autouse fixture 清空全部 AI_* 变量，测试不再继承 .env

测试 (184 passed, 1 skipped):
- 新增 TestGatewayProvider: 透传 upstream、目标 URL/鉴权头、
  token 失效时继续降级
- 新增 TestProxyPolicy: 境外厂商与自建端点的代理策略相反
- 新增 TestPerModelSizing: 128k 模型收到的 prompt 必须小于 1M 模型
- 新增 TestMaxTokens: 推理端点预算大于默认，且真正写进两种 payload
- 新增 TestLazyConfig: 改环境变量立即生效
- mock 目标从 requests.post 改为 requests.Session.post

实测: 网关链路可返回合法 JSON，但 nemotron-550B 排队较久（约 160s），
故 AI_TIMEOUT_SECONDS 默认调到 180。

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
ericwyuan
2026-08-23 17:44:51 +08:00
parent 8616a13525
commit 8882bf44a4
8 changed files with 672 additions and 2019 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
backend/.env.example
View File

@@ -1,5 +1,7 @@
# --- Server ---
PORT=5000
# BACKEND_PORT takes precedence over PORT. Prefer it: many tools inject PORT
# for the frontend, and Flask would otherwise take the React dev server's port.
BACKEND_PORT=5000
# --- Database: sqlite (default) or mariadb ---
DB_TYPE=sqlite
@@ -24,19 +26,34 @@ 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.
# Any model whose credentials are absent is skipped automatically.
# Google AI Studio -> the "gemini-flash" model id
# Self-hosted AI gateway (model id "gateway"). OpenAI-compatible; it fans out
# over nvidia/gemini/ollama itself and rotates several Gemini keys, so it
# absorbs single-vendor quota limits. Reached directly, bypassing any local
# HTTP proxy. NOTE: its NVIDIA upstream is a large reasoning model — replies
# can take 2-3 minutes, so set AI_TIMEOUT_SECONDS accordingly.
AI_GATEWAY_BASE_URL=http://129.146.203.203:5100/v1
AI_GATEWAY_TOKEN=
AI_GATEWAY_MODEL=ai-gateway-auto
# Google AI Studio -> "gemini-flash". Free-tier quota is small; 429s are common.
GEMINI_API_KEY=
# NVIDIA NIM (OpenAI-compatible) -> "llama-70b", "qwen-72b", "deepseek-r1"
# NVIDIA NIM -> "llama-70b", "nemotron-49b", "mistral-large".
# Model ids come from that account's live GET /v1/models — do not guess them.
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
# out, the next is tried. Read per request, so changes need no restart.
AI_MODEL_CHAIN=gateway,gemini-flash,llama-70b
# Max days of history sent to the model (CSV-encoded, ~4 chars/day).
# Max days of history sent (CSV-encoded). Trimmed further per model so the
# payload always fits that model's own context window.
AI_DAY_BUDGET=365
AI_TIMEOUT_SECONDS=45
AI_TIMEOUT_SECONDS=180
# Output cap. Reasoning models spend part of it thinking before they answer;
# entries that need more declare their own budget in services/ai.py.
AI_MAX_TOKENS=1024

5
backend/config.py
View File

@@ -38,7 +38,10 @@ JWT_SECRET = os.environ.get("JWT_SECRET") or "dev_secret_change_me"
JWT_EXPIRY_DAYS = int(os.environ.get("JWT_EXPIRY_DAYS") or 7)
# --- Server -----------------------------------------------------------------
PORT = int(os.environ.get("PORT") or 5000)
# BACKEND_PORT wins over PORT: `PORT` is set by many dev tools and PaaS
# runtimes for the *frontend*, and letting it through made Flask seize the
# React dev server's port during `npm run dev`.
PORT = int(os.environ.get("BACKEND_PORT") or os.environ.get("PORT") or 5000)
# Comma-separated list of allowed front-end origins (CORS).
_CORS_RAW = os.environ.get("CORS_ORIGIN") or "http://localhost:3000,http://localhost:5173"

288
backend/services/ai.py
View File

@@ -23,11 +23,30 @@ import re
import requests
DEFAULT_TIMEOUT = float(os.environ.get("AI_TIMEOUT_SECONDS") or 45)
# Tunables are read per call rather than captured at import: module-level
# constants freeze whatever the environment held when the module first loaded,
# which both hides live config changes and leaks a developer's .env into tests.
FALLBACK_TIMEOUT = 60.0
FALLBACK_DAY_BUDGET = 365
# 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)
def default_timeout():
return float(os.environ.get("AI_TIMEOUT_SECONDS") or FALLBACK_TIMEOUT)
def default_day_budget():
"""Max days of history to put in a prompt, before per-model trimming."""
return int(os.environ.get("AI_DAY_BUDGET") or FALLBACK_DAY_BUDGET)
# Output cap. Deliberately modest: a long generation is what blows past an
# upstream's own timeout (the self-hosted gateway allows its adapters only
# 30-45s), and the reply here is a short JSON list, not an essay.
FALLBACK_MAX_TOKENS = 1024
def default_max_tokens():
return int(os.environ.get("AI_MAX_TOKENS") or FALLBACK_MAX_TOKENS)
SYSTEM_PROMPT = (
"你是一名严谨的健康数据分析助手，负责解读用户的可穿戴设备（Garmin）数据。\n"
@@ -37,7 +56,8 @@ SYSTEM_PROMPT = (
"3. 给出具体、可执行的建议，而不是泛泛而谈。\n"
"4. 你不是医生，不做诊断；发现明显异常时建议用户咨询专业医师。\n"
"5. 用简体中文回答。\n\n"
"输出严格为 JSON 数组，每个元素形如：\n"
"输出严格为 JSON 数组，最多 5 条，每条 recommendation 不超过 120 字，\n"
"每个元素形如：\n"
'{"category": "睡眠", "recommendation": "……", "priority": "high|medium|low", '
'"basedOn": ["sleep_duration"]}\n'
"不要输出 JSON 以外的任何文字，不要用 markdown 代码块包裹。"
@@ -48,16 +68,54 @@ class AIError(Exception):
"""Raised when a provider cannot produce a completion."""
class Completion:
"""A model reply plus, where the endpoint reports it, the upstream that
actually served the request.
The self-hosted gateway multiplexes over nvidia/gemini/ollama and names
the winner in its response, so `upstream` is what makes a gateway-side
failover visible to the UI instead of silently invisible.
"""
__slots__ = ("text", "upstream")
def __init__(self, text, upstream=None):
self.text = text
self.upstream = upstream
# --- providers --------------------------------------------------------------
class Provider:
"""Base class. Subclasses turn a prompt into text."""
"""Base class. Subclasses turn a prompt into text.
`use_proxy` decides whether HTTP(S)_PROXY / ALL_PROXY from the environment
apply. It matters because the two kinds of endpoint want opposite answers:
overseas vendors (Gemini, NVIDIA) may only be reachable *through* a local
proxy, while a self-hosted box on a public IP is reachable directly and
breaks if forced through one.
"""
name = "base"
def __init__(self, model_id, context_window, api_key_env):
def __init__(
self, model_id, context_window, api_key_env, use_proxy=True, max_tokens=None
):
self.model_id = model_id
self.context_window = context_window
self.api_key_env = api_key_env
self.use_proxy = use_proxy
self._max_tokens = max_tokens
@property
def max_tokens(self):
"""Output cap for this endpoint.
Reasoning models emit a chain-of-thought *before* the answer, so a cap
sized for the answer alone gets spent on the thinking and truncates
before any JSON appears. Those endpoints therefore declare a larger
budget than the default.
"""
return self._max_tokens or default_max_tokens()
@property
def api_key(self):
@@ -66,7 +124,14 @@ class Provider:
def is_configured(self):
return bool(self.api_key)
def generate(self, prompt, timeout=DEFAULT_TIMEOUT):
def _session(self):
session = requests.Session()
# trust_env=False also drops netrc/CA-bundle env lookups, which is the
# intent here: talk to the host directly, exactly as configured.
session.trust_env = self.use_proxy
return session
def generate(self, prompt, timeout=None):
raise NotImplementedError
@@ -76,16 +141,20 @@ class GeminiProvider(Provider):
name = "gemini"
BASE = "https://generativelanguage.googleapis.com/v1beta/models"
def generate(self, prompt, timeout=DEFAULT_TIMEOUT):
def generate(self, prompt, timeout=None):
if not self.is_configured():
raise AIError(f"{self.api_key_env} 未配置")
timeout = timeout or default_timeout()
url = f"{self.BASE}/{self.model_id}:generateContent"
payload = {
"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {"temperature": 0.4},
"generationConfig": {
"temperature": 0.4,
"maxOutputTokens": self.max_tokens,
},
}
try:
resp = requests.post(
resp = self._session().post(
url,
headers={
"Content-Type": "application/json",
@@ -103,40 +172,70 @@ class GeminiProvider(Provider):
try:
body = resp.json()
parts = body["candidates"][0]["content"]["parts"]
return "".join(p.get("text", "") for p in parts)
return Completion("".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, ...)."""
Ollama, vLLM, ...).
`requires_key=False` covers self-hosted runtimes such as Ollama, which
authenticate by network reachability rather than by a token. Those are
opt-in: they count as configured only once their base URL is set, so an
unset OLLAMA_BASE_URL keeps the entry out of the fallback chain.
"""
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 __init__(
self,
model_id,
context_window,
base_url_env,
default_base_url="",
api_key_env=None,
requires_key=True,
use_proxy=True,
max_tokens=None,
):
super().__init__(
model_id, context_window, api_key_env or "", use_proxy, max_tokens
)
self.base_url_env = base_url_env
self.default_base_url = default_base_url
self.requires_key = requires_key
def generate(self, prompt, timeout=DEFAULT_TIMEOUT):
@property
def base_url(self):
return os.environ.get(self.base_url_env) or self.default_base_url
def is_configured(self):
if not self.base_url:
return False
return bool(self.api_key) if self.requires_key else True
def generate(self, prompt, timeout=None):
if not self.is_configured():
raise AIError(f"{self.api_key_env} 未配置")
raise AIError(
f"{self.api_key_env} 未配置" if self.requires_key
else f"{self.base_url_env} 未配置"
)
timeout = timeout or default_timeout()
url = f"{self.base_url.rstrip('/')}/chat/completions"
payload = {
"model": self.model_id,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.4,
"max_tokens": 2048,
"max_tokens": self.max_tokens,
}
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
try:
resp = requests.post(
url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
},
json=payload,
timeout=timeout,
resp = self._session().post(
url, headers=headers, json=payload, timeout=timeout
)
except requests.RequestException as e:
raise AIError(f"{self.model_id} 请求失败: {e}") from e
@@ -145,56 +244,91 @@ class OpenAICompatProvider(Provider):
raise AIError(f"{self.model_id} HTTP {resp.status_code}: {resp.text[:200]}")
try:
return resp.json()["choices"][0]["message"]["content"]
body = resp.json()
# `provider` is a gateway extension, absent from stock OpenAI
# responses — hence the .get rather than an index.
return Completion(
body["choices"][0]["message"]["content"], body.get("provider")
)
except (ValueError, KeyError, IndexError) as e:
raise AIError(f"{self.model_id} 响应格式异常: {e}") from e
# --- catalog ----------------------------------------------------------------
NVIDIA_BASE = "https://integrate.api.nvidia.com/v1"
def _nvidia(model_id, context_window):
return OpenAICompatProvider(
model_id=model_id,
context_window=context_window,
api_key_env="NVIDIA_API_KEY",
base_url_env="NVIDIA_BASE_URL",
default_base_url=NVIDIA_BASE,
)
def _build_catalog():
"""Model id -> Provider. Text-only models with large context windows."""
"""Model id -> Provider. Text-only models with large context windows.
The NVIDIA model strings below were taken from that account's live
`GET /v1/models` listing. Do not guess them: ids that merely look
plausible (`qwen/qwen2.5-72b-instruct`, `deepseek-ai/deepseek-r1`)
return HTTP 404 from this endpoint.
"""
return {
# Preferred entry: the self-hosted gateway on the Oracle box. It
# multiplexes over nvidia/gemini/ollama behind one OpenAI-compatible
# endpoint and rotates several Gemini keys, so it absorbs the quota
# and timeout failures that a single upstream hits on its own. Its
# reply names the upstream that served the request.
"gateway": OpenAICompatProvider(
model_id=os.environ.get("AI_GATEWAY_MODEL") or "ai-gateway-auto",
context_window=128_000,
api_key_env="AI_GATEWAY_TOKEN",
base_url_env="AI_GATEWAY_BASE_URL",
# Self-hosted and directly reachable: a local proxy would only
# add a hop that times out.
use_proxy=False,
# Its primary upstream is a reasoning model that thinks out loud
# before answering; at the default cap the trace consumed the whole
# budget and the reply was truncated before the JSON began.
max_tokens=3000,
),
# Direct upstreams, for pinning one vendor or for running without the
# gateway. These need their own keys in this app's .env.
"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",
),
"llama-70b": _nvidia("meta/llama-3.3-70b-instruct", 128_000),
"nemotron-49b": _nvidia("nvidia/llama-3.3-nemotron-super-49b-v1.5", 128_000),
"mistral-large": _nvidia("mistralai/mistral-large-2-instruct", 128_000),
}
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()
]
FALLBACK_CHAIN = "gateway,gemini-flash,llama-70b"
def default_chain():
"""Preference order, read from the environment on every call.
Deliberately not a module-level constant: it is read at request time so a
changed AI_MODEL_CHAIN takes effect without a restart, and so tests can
set it without reaching into module internals.
"""
raw = os.environ.get("AI_MODEL_CHAIN") or FALLBACK_CHAIN
return [m.strip() for m in raw.split(",") if m.strip()]
def list_models():
"""Catalog entries plus whether each one currently has credentials."""
chain = default_chain()
head = chain[0] if chain else None
return [
{
"id": mid,
@@ -202,7 +336,7 @@ def list_models():
"provider": p.name,
"contextWindow": p.context_window,
"configured": p.is_configured(),
"default": mid == DEFAULT_CHAIN[0] if DEFAULT_CHAIN else False,
"default": mid == head,
}
for mid, p in CATALOG.items()
]
@@ -215,7 +349,7 @@ def resolve_chain(preferred=None):
if preferred not in CATALOG:
raise AIError(f"未知模型: {preferred}")
chain.append(preferred)
for mid in DEFAULT_CHAIN:
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()]
@@ -237,12 +371,13 @@ _CSV_COLUMNS = [
]
def build_prompt(summary, activities=None, day_budget=DEFAULT_DAY_BUDGET):
def build_prompt(summary, activities=None, day_budget=None):
"""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.
"""
day_budget = day_budget if day_budget is not None else default_day_budget()
rows = summary[-day_budget:] if day_budget else summary
header = ",".join(label for _, label in _CSV_COLUMNS) + ",sleep_h,sleep_q"
lines = [header]
@@ -346,26 +481,49 @@ def parse_recommendations(text):
# --- entry point ------------------------------------------------------------
def generate(summary, activities=None, preferred_model=None, day_budget=DEFAULT_DAY_BUDGET):
# One CSV day is ~40 characters ≈ 10 tokens. Half the window is left for the
# system prompt, the activity table and the model's own answer.
_TOKENS_PER_DAY = 10
_WINDOW_UTILISATION = 0.5
def max_days_for(provider, day_budget=None):
"""How many days of history fit in this model's context window.
Models in the chain have windows that differ by more than an order of
magnitude (32k for a local Ollama vs 1M for Gemini), so the payload has to
be sized per model — a prompt that fits Gemini would overflow Ollama.
"""
day_budget = day_budget if day_budget is not None else default_day_budget()
fits = int(provider.context_window * _WINDOW_UTILISATION / _TOKENS_PER_DAY)
return max(1, min(day_budget, fits)) if day_budget else max(1, fits)
def generate(summary, activities=None, preferred_model=None, day_budget=None):
"""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.
Returns (recommendations, meta). `meta` records which model answered, how
much history it actually saw, and every model that failed on the way —
the failures are kept even on success so a silent degradation to a weaker
model is still visible.
"""
chain = resolve_chain(preferred_model)
prompt = build_prompt(summary, activities, day_budget)
errors = []
for model_id in chain:
provider = CATALOG[model_id]
days = max_days_for(provider, day_budget)
prompt = build_prompt(summary, activities, days)
try:
raw = provider.generate(prompt)
recs = parse_recommendations(raw)
completion = provider.generate(prompt)
recs = parse_recommendations(completion.text)
return recs, {
"model": model_id,
"provider": provider.name,
"days": min(len(summary), day_budget) if day_budget else len(summary),
"upstream": completion.upstream,
"days": min(len(summary), days),
"fallbackFrom": [e["model"] for e in errors],
"errors": errors,
}
except AIError as e:
errors.append({"model": model_id, "error": str(e)})

2
backend/services/analysis.py
View File

@@ -134,7 +134,7 @@ def get_ai_recommendations(user_id, model=None, days=None):
}
activities = health.get_activities(user_id)
budget = days or ai_svc.DEFAULT_DAY_BUDGET
budget = days or ai_svc.default_day_budget()
try:
recs, meta = ai_svc.generate(

24
backend/tests/conftest.py
View File

@@ -25,6 +25,30 @@ os.environ.setdefault(
import db as db_module # noqa: E402
from app import create_app # noqa: E402
# config.py calls load_dotenv() at import, so backend/.env leaks into the test
# process — a developer's real AI_MODEL_CHAIN or API keys would silently change
# what the suite exercises (and could bill real API calls). Clear them here;
# individual tests opt back in through the `keys` / `gateway` fixtures.
_AI_ENV_VARS = (
"AI_MODEL_CHAIN",
"AI_DAY_BUDGET",
"AI_TIMEOUT_SECONDS",
"GEMINI_API_KEY",
"NVIDIA_API_KEY",
"NVIDIA_BASE_URL",
"AI_GATEWAY_TOKEN",
"AI_GATEWAY_BASE_URL",
"AI_GATEWAY_MODEL",
"OLLAMA_BASE_URL",
"OLLAMA_MODEL",
)
@pytest.fixture(autouse=True)
def _isolate_ai_env(monkeypatch):
for var in _AI_ENV_VARS:
monkeypatch.delenv(var, raising=False)
@pytest.fixture
def db(tmp_path, monkeypatch):

295
backend/tests/test_ai.py
View File

@@ -55,14 +55,29 @@ def openai_payload(text):
@pytest.fixture
def keys(monkeypatch):
"""Pretend both vendors are configured."""
"""Direct vendor keys configured; the gateway stays out of the chain."""
monkeypatch.setenv("GEMINI_API_KEY", "test-gemini-key")
monkeypatch.setenv("NVIDIA_API_KEY", "test-nvidia-key")
monkeypatch.delenv("AI_GATEWAY_TOKEN", raising=False)
monkeypatch.delenv("AI_GATEWAY_BASE_URL", raising=False)
return True
@pytest.fixture
def no_keys(monkeypatch):
for var in (
"GEMINI_API_KEY", "NVIDIA_API_KEY",
"AI_GATEWAY_TOKEN", "AI_GATEWAY_BASE_URL",
):
monkeypatch.delenv(var, raising=False)
return True
@pytest.fixture
def gateway(monkeypatch):
"""Only the self-hosted gateway is configured."""
monkeypatch.setenv("AI_GATEWAY_TOKEN", "test-gateway-token")
monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
return True
@@ -191,47 +206,47 @@ class TestGeminiProvider:
def test_successful_call(self, keys, monkeypatch):
captured = {}
def fake_post(url, **kwargs):
def fake_post(self, 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)
monkeypatch.setattr(requests.Session, "post", fake_post)
out = ai_svc.CATALOG["gemini-flash"].generate("prompt text")
assert out == "hello"
assert out.text == "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")
requests.Session, "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):
def boom(self, *a, **k):
raise requests.Timeout("timed out")
monkeypatch.setattr(requests, "post", boom)
monkeypatch.setattr(requests.Session, "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})
requests.Session, "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):
def boom(self, *a, **k):
raise AssertionError("must not issue a request without a key")
monkeypatch.setattr(requests, "post", boom)
monkeypatch.setattr(requests.Session, "post", boom)
with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"):
ai_svc.CATALOG["gemini-flash"].generate("p")
@@ -240,33 +255,232 @@ class TestOpenAICompatProvider:
def test_successful_call(self, keys, monkeypatch):
captured = {}
def fake_post(url, **kwargs):
def fake_post(self, 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)
monkeypatch.setattr(requests.Session, "post", fake_post)
out = ai_svc.CATALOG["llama-70b"].generate("prompt text")
assert out == "hi"
assert out.text == "hi"
assert out.upstream is None, "stock OpenAI replies carry no provider field"
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")
requests.Session, "post", lambda *a, **k: FakeResponse(500, text="boom")
)
with pytest.raises(ai_svc.AIError, match="500"):
ai_svc.CATALOG["llama-70b"].generate("p")
class TestGatewayProvider:
"""The self-hosted gateway: OpenAI-compatible, plus a `provider` field
naming whichever upstream actually served the request."""
def test_reports_the_upstream_that_answered(self, gateway, monkeypatch):
payload = {**openai_payload("hi"), "provider": "nvidia"}
monkeypatch.setattr(requests.Session, "post", lambda *a, **k: FakeResponse(200, payload))
out = ai_svc.CATALOG["gateway"].generate("p")
assert out.text == "hi"
assert out.upstream == "nvidia"
def test_targets_the_configured_base_url(self, gateway, monkeypatch):
captured = {}
def fake_post(self, url, **kwargs):
captured["url"] = url
captured["headers"] = kwargs.get("headers", {})
captured["model"] = (kwargs.get("json") or {}).get("model")
return FakeResponse(200, openai_payload("hi"))
monkeypatch.setattr(requests.Session, "post", fake_post)
ai_svc.CATALOG["gateway"].generate("p")
assert captured["url"] == "http://gw.test:5100/v1/chat/completions"
assert captured["headers"]["Authorization"] == "Bearer test-gateway-token"
assert captured["model"] == "ai-gateway-auto"
def test_upstream_surfaces_in_generate_meta(self, gateway, monkeypatch):
payload = {**openai_payload(VALID_REPLY), "provider": "gemini"}
monkeypatch.setattr(requests.Session, "post", lambda *a, **k: FakeResponse(200, payload))
_, meta = ai_svc.generate(SUMMARY)
assert meta["model"] == "gateway"
assert meta["upstream"] == "gemini"
def test_gateway_401_falls_through(self, monkeypatch):
"""A stale gateway token must not strand the request."""
monkeypatch.setenv("AI_GATEWAY_TOKEN", "expired")
monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
monkeypatch.setenv("GEMINI_API_KEY", "k")
def fake_post(self, url, **kwargs):
if "gw.test" in url:
return FakeResponse(401, text="unauthorized")
return FakeResponse(200, gemini_payload(VALID_REPLY))
monkeypatch.setattr(requests.Session, "post", fake_post)
_, meta = ai_svc.generate(SUMMARY)
assert meta["model"] == "gemini-flash"
assert meta["fallbackFrom"] == ["gateway"]
# --- proxy handling ---------------------------------------------------------
class TestProxyPolicy:
"""Overseas vendors may only be reachable through a local proxy, while a
self-hosted box on a public IP breaks when forced through one — so the two
must not share a policy."""
def test_hosted_vendors_honour_environment_proxies(self):
assert ai_svc.CATALOG["gemini-flash"].use_proxy is True
assert ai_svc.CATALOG["llama-70b"].use_proxy is True
def test_self_hosted_gateway_bypasses_proxies(self):
assert ai_svc.CATALOG["gateway"].use_proxy is False
def test_session_trust_env_follows_the_flag(self):
"""Regression: requests picked up ALL_PROXY and routed the gateway
call through a local proxy, which timed out after 120s."""
assert ai_svc.CATALOG["gateway"]._session().trust_env is False
assert ai_svc.CATALOG["gemini-flash"]._session().trust_env is True
# --- output budget ----------------------------------------------------------
class TestMaxTokens:
def test_default_applies_to_ordinary_models(self):
assert ai_svc.CATALOG["gemini-flash"].max_tokens == ai_svc.FALLBACK_MAX_TOKENS
def test_reasoning_endpoint_declares_a_larger_budget(self):
"""Regression: the gateway's primary upstream thinks out loud before
answering; at the default cap the trace consumed the whole budget and
the reply was truncated before any JSON appeared."""
assert ai_svc.CATALOG["gateway"].max_tokens > ai_svc.FALLBACK_MAX_TOKENS
def test_env_overrides_the_default_but_not_an_explicit_budget(self, monkeypatch):
monkeypatch.setenv("AI_MAX_TOKENS", "77")
assert ai_svc.CATALOG["gemini-flash"].max_tokens == 77
assert ai_svc.CATALOG["gateway"].max_tokens == 3000
def test_budget_reaches_the_openai_payload(self, gateway, monkeypatch):
seen = {}
def fake_post(self, url, **kwargs):
seen["max_tokens"] = kwargs["json"]["max_tokens"]
return FakeResponse(200, openai_payload("hi"))
monkeypatch.setattr(requests.Session, "post", fake_post)
ai_svc.CATALOG["gateway"].generate("p")
assert seen["max_tokens"] == 3000
def test_budget_reaches_the_gemini_payload(self, keys, monkeypatch):
seen = {}
def fake_post(self, url, **kwargs):
seen["cap"] = kwargs["json"]["generationConfig"]["maxOutputTokens"]
return FakeResponse(200, gemini_payload("hi"))
monkeypatch.setattr(requests.Session, "post", fake_post)
ai_svc.CATALOG["gemini-flash"].generate("p")
assert seen["cap"] == ai_svc.FALLBACK_MAX_TOKENS
# --- lazily-read configuration ----------------------------------------------
class TestLazyConfig:
"""Regression: these were module-level constants, so they froze whatever
the environment held at import — hiding config changes and letting a
developer's .env leak into the test run."""
def test_chain_reflects_the_current_environment(self, monkeypatch):
monkeypatch.setenv("AI_MODEL_CHAIN", "llama-70b,gemini-flash")
assert ai_svc.default_chain() == ["llama-70b", "gemini-flash"]
monkeypatch.setenv("AI_MODEL_CHAIN", "gateway")
assert ai_svc.default_chain() == ["gateway"]
def test_timeout_reflects_the_current_environment(self, monkeypatch):
monkeypatch.setenv("AI_TIMEOUT_SECONDS", "7")
assert ai_svc.default_timeout() == 7.0
def test_day_budget_reflects_the_current_environment(self, monkeypatch):
monkeypatch.setenv("AI_DAY_BUDGET", "42")
assert ai_svc.default_day_budget() == 42
def test_defaults_apply_when_unset(self):
assert ai_svc.default_timeout() == ai_svc.FALLBACK_TIMEOUT
assert ai_svc.default_day_budget() == ai_svc.FALLBACK_DAY_BUDGET
assert ai_svc.default_chain()[0] == "gateway"
def test_default_flag_tracks_the_chain_head(self, monkeypatch, keys):
monkeypatch.setenv("AI_MODEL_CHAIN", "llama-70b,gemini-flash")
by_id = {m["id"]: m for m in ai_svc.list_models()}
assert by_id["llama-70b"]["default"] is True
assert by_id["gemini-flash"]["default"] is False
# --- context sizing ---------------------------------------------------------
class TestPerModelSizing:
"""Chain members' windows differ by >30x, so the payload is sized per
model rather than once for the whole chain."""
def test_small_window_gets_fewer_days_than_a_large_one(self):
# A budget above what 128k can hold, so the window is what binds.
budget = 100_000
small = ai_svc.max_days_for(ai_svc.CATALOG["llama-70b"], budget) # 128k
large = ai_svc.max_days_for(ai_svc.CATALOG["gemini-flash"], budget) # 1M
assert small < large
def test_budget_binds_when_it_is_the_tighter_limit(self):
"""At the default 365-day budget every model gets the same 365 days —
no window in the catalog is small enough to bind first."""
budget = ai_svc.default_day_budget()
days = {
mid: ai_svc.max_days_for(p, budget) for mid, p in ai_svc.CATALOG.items()
}
assert set(days.values()) == {budget}
def test_never_exceeds_the_configured_budget(self):
assert ai_svc.max_days_for(ai_svc.CATALOG["gemini-flash"], day_budget=30) == 30
def test_always_allows_at_least_one_day(self):
tiny = ai_svc.OpenAICompatProvider(
model_id="tiny", context_window=10,
base_url_env="X", default_base_url="http://x", requires_key=False,
)
assert ai_svc.max_days_for(tiny) >= 1
def test_each_model_gets_a_prompt_sized_for_itself(self, keys, monkeypatch):
"""Regression: one prompt was built for the whole chain, so a payload
sized for Gemini's 1M window was also sent to 128k models.
Needs more days than the 128k window holds (~6.4k) for the trimming to
bite, hence the deliberately oversized history.
"""
history = [{"date": "2026-01-01", "steps": 8000} for _ in range(8000)]
sizes = {}
def fake_post(self, url, **kwargs):
if "generativelanguage" in url:
sizes["gemini"] = len(kwargs["json"]["contents"][0]["parts"][0]["text"])
raise requests.Timeout("force fallback")
sizes["nvidia"] = len(kwargs["json"]["messages"][0]["content"])
return FakeResponse(200, openai_payload(VALID_REPLY))
monkeypatch.setattr(requests.Session, "post", fake_post)
ai_svc.generate(history, day_budget=100_000)
assert sizes["nvidia"] < sizes["gemini"], (
"the 128k model must receive a smaller prompt than the 1M model"
)
# --- 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"
"gateway", "gemini-flash", "llama-70b", "nemotron-49b", "mistral-large"
}
def test_configured_flag_tracks_the_environment(self, no_keys, monkeypatch):
@@ -285,15 +499,14 @@ class TestCatalog:
class TestResolveChain:
def test_preferred_model_goes_first(self, keys):
assert ai_svc.resolve_chain("qwen-72b")[0] == "qwen-72b"
assert ai_svc.resolve_chain("nemotron-49b")[0] == "nemotron-49b"
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):
def test_unconfigured_models_are_skipped(self, no_keys, 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):
@@ -304,12 +517,18 @@ class TestResolveChain:
with pytest.raises(ai_svc.AIError, match="GEMINI_API_KEY"):
ai_svc.resolve_chain()
def test_gateway_needs_both_token_and_base_url(self, no_keys, monkeypatch):
monkeypatch.setenv("AI_GATEWAY_TOKEN", "t")
assert ai_svc.CATALOG["gateway"].is_configured() is False
monkeypatch.setenv("AI_GATEWAY_BASE_URL", "http://gw.test:5100/v1")
assert ai_svc.CATALOG["gateway"].is_configured() is True
# --- 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))
requests.Session, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY))
)
recs, meta = ai_svc.generate(SUMMARY)
assert len(recs) == 2
@@ -320,13 +539,13 @@ class TestGenerate:
def test_falls_back_to_the_next_model(self, keys, monkeypatch):
calls = []
def fake_post(url, **kwargs):
def fake_post(self, 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)
monkeypatch.setattr(requests.Session, "post", fake_post)
recs, meta = ai_svc.generate(SUMMARY)
assert len(recs) == 2
@@ -335,43 +554,43 @@ class TestGenerate:
assert len(calls) == 2
def test_falls_back_when_a_model_returns_unparseable_text(self, keys, monkeypatch):
def fake_post(url, **kwargs):
def fake_post(self, url, **kwargs):
if "generativelanguage" in url:
return FakeResponse(200, gemini_payload("抱歉，我帮不了你。"))
return FakeResponse(200, openai_payload(VALID_REPLY))
monkeypatch.setattr(requests, "post", fake_post)
monkeypatch.setattr(requests.Session, "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):
def boom(self, *a, **k):
raise requests.Timeout("all down")
monkeypatch.setattr(requests, "post", boom)
monkeypatch.setattr(requests.Session, "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):
def fake_post(self, 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"
monkeypatch.setattr(requests.Session, "post", fake_post)
_, meta = ai_svc.generate(SUMMARY, preferred_model="nemotron-49b")
assert meta["model"] == "nemotron-49b"
assert seen["model"] == "nvidia/llama-3.3-nemotron-super-49b-v1.5"
def test_no_second_call_after_the_first_succeeds(self, keys, monkeypatch):
calls = []
def fake_post(url, **kwargs):
def fake_post(self, url, **kwargs):
calls.append(url)
return FakeResponse(200, gemini_payload(VALID_REPLY))
monkeypatch.setattr(requests, "post", fake_post)
monkeypatch.setattr(requests.Session, "post", fake_post)
ai_svc.generate(SUMMARY)
assert len(calls) == 1
@@ -388,7 +607,7 @@ class TestAiRecommendationsService:
):
seed_health([{"date": "2026-08-20", "steps": 5000}])
monkeypatch.setattr(
requests, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY))
requests.Session, "post", lambda *a, **k: FakeResponse(200, gemini_payload(VALID_REPLY))
)
out = analysis_svc.get_ai_recommendations(user["id"])
assert out["meta"]["source"] == "ai"
@@ -399,10 +618,10 @@ class TestAiRecommendationsService:
):
seed_health([{"date": "2026-08-20", "steps": 5000}])
def boom(*a, **k):
def boom(self, *a, **k):
raise requests.Timeout("down")
monkeypatch.setattr(requests, "post", boom)
monkeypatch.setattr(requests.Session, "post", boom)
out = analysis_svc.get_ai_recommendations(user["id"])
assert out["meta"]["source"] == "rules"
assert "所有模型均失败" in out["meta"]["reason"]
@@ -444,12 +663,12 @@ class TestEndpoints:
):
seed_health([{"date": "2026-08-20", "steps": 5000}])
monkeypatch.setattr(
requests, "post", lambda *a, **k: FakeResponse(200, openai_payload(VALID_REPLY))
requests.Session, "post", lambda *a, **k: FakeResponse(200, openai_payload(VALID_REPLY))
)
r = client.get(
"/api/analysis/ai-recommendations?model=qwen-72b", headers=auth
"/api/analysis/ai-recommendations?model=nemotron-49b", headers=auth
)
assert r.get_json()["meta"]["model"] == "qwen-72b"
assert r.get_json()["meta"]["model"] == "nemotron-49b"
def test_unknown_model_param_degrades_to_rules(
self, client, auth, seed_health, keys

2042
package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

2
package.json
View File

@@ -8,7 +8,7 @@
],
"scripts": {
"dev": "concurrently -n backend,client -c blue,green \"npm run dev:backend\" \"npm run dev:client\"",
"dev:backend": "cd backend && .venv/bin/python app.py",
"dev:backend": "cd backend && BACKEND_PORT=5000 .venv/bin/python app.py",
"dev:client": "npm start --workspace=client",
"build": "npm run build --workspace=client",
"typecheck": "npm run typecheck --workspace=client",