c57c930949
数值全部在服务端算好再交给模型，模型只做解读。让模型从 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>
811 lines
30 KiB
Python
811 lines
30 KiB
Python
"""
|
||
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
|
||
|
||
# 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
|
||
|
||
|
||
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"
|
||
"要求：\n"
|
||
"1. 只依据给出的数据得出结论，数据不足时明确说明，不要编造数值。\n"
|
||
"2. 指出趋势、异常和相互关联（例如睡眠不足与静息心率升高的关系）。\n"
|
||
"3. 给出具体、可执行的建议，而不是泛泛而谈。\n"
|
||
"4. 你不是医生，不做诊断；发现明显异常时建议用户咨询专业医师。\n"
|
||
"5. 用简体中文回答。\n\n"
|
||
"输出严格为 JSON 数组，最多 5 条，每条 recommendation 不超过 120 字，\n"
|
||
"每个元素形如：\n"
|
||
'{"category": "睡眠", "recommendation": "……", "priority": "high|medium|low", '
|
||
'"basedOn": ["sleep_duration"]}\n'
|
||
"不要输出 JSON 以外的任何文字，不要用 markdown 代码块包裹。"
|
||
)
|
||
|
||
|
||
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.
|
||
|
||
`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"
|
||
|
||
# Whether this endpoint has an incremental transport of its own. False
|
||
# means `stream` is the blocking call in disguise, which `stream_chat`
|
||
# needs to know: retrying such a provider without streaming would just
|
||
# run the same request a second time.
|
||
streaming = False
|
||
|
||
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):
|
||
return os.environ.get(self.api_key_env) or ""
|
||
|
||
def is_configured(self):
|
||
return bool(self.api_key)
|
||
|
||
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 chat(self, messages, timeout=None, max_tokens=None):
|
||
"""Multi-turn completion.
|
||
|
||
`messages` is the OpenAI shape — a list of
|
||
{"role": "system"|"user"|"assistant", "content": str}. Providers whose
|
||
wire format differs translate it themselves.
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
def generate(self, prompt, timeout=None):
|
||
"""Single-turn convenience wrapper, kept for the recommendation path."""
|
||
return self.chat([{"role": "user", "content": prompt}], timeout)
|
||
|
||
def stream(self, messages, timeout=None, max_tokens=None):
|
||
"""Yield Completion deltas as the reply arrives.
|
||
|
||
The base implementation is not incremental: it waits for the whole
|
||
answer and emits it as a single delta. That keeps every entry in the
|
||
catalog streamable from the caller's point of view — a provider with
|
||
no SSE transport produces one late chunk rather than an error, so the
|
||
Copilot route does not need a per-provider branch.
|
||
"""
|
||
yield self.chat(messages, timeout, max_tokens)
|
||
|
||
|
||
class GeminiProvider(Provider):
|
||
"""Google AI Studio (generativelanguage.googleapis.com)."""
|
||
|
||
name = "gemini"
|
||
BASE = "https://generativelanguage.googleapis.com/v1beta/models"
|
||
|
||
def chat(self, messages, timeout=None, max_tokens=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"
|
||
# Gemini splits what OpenAI keeps in one list: system turns move to
|
||
# `systemInstruction`, and the assistant role is spelled "model".
|
||
contents, system = [], []
|
||
for message in messages:
|
||
role = message.get("role")
|
||
if role == "system":
|
||
system.append(message.get("content") or "")
|
||
continue
|
||
contents.append({
|
||
"role": "model" if role == "assistant" else "user",
|
||
"parts": [{"text": message.get("content") or ""}],
|
||
})
|
||
payload = {
|
||
"contents": contents,
|
||
"generationConfig": {
|
||
"temperature": 0.4,
|
||
"maxOutputTokens": max_tokens or self.max_tokens,
|
||
},
|
||
}
|
||
if system:
|
||
payload["systemInstruction"] = {
|
||
"parts": [{"text": "\n\n".join(system)}]
|
||
}
|
||
try:
|
||
resp = self._session().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 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, ...).
|
||
|
||
`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"
|
||
streaming = True
|
||
|
||
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
|
||
|
||
@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 _request(self, messages, timeout, max_tokens, stream):
|
||
if not self.is_configured():
|
||
raise AIError(
|
||
f"{self.api_key_env} 未配置" if self.requires_key
|
||
else f"{self.base_url_env} 未配置"
|
||
)
|
||
url = f"{self.base_url.rstrip('/')}/chat/completions"
|
||
payload = {
|
||
"model": self.model_id,
|
||
"messages": messages,
|
||
"temperature": 0.4,
|
||
"max_tokens": max_tokens or self.max_tokens,
|
||
}
|
||
if stream:
|
||
payload["stream"] = True
|
||
headers = {"Content-Type": "application/json"}
|
||
if self.api_key:
|
||
headers["Authorization"] = f"Bearer {self.api_key}"
|
||
try:
|
||
return self._session().post(
|
||
url, headers=headers, json=payload, timeout=timeout, stream=stream
|
||
)
|
||
except requests.RequestException as e:
|
||
raise AIError(f"{self.model_id} 请求失败: {e}") from e
|
||
|
||
def chat(self, messages, timeout=None, max_tokens=None):
|
||
timeout = timeout or default_timeout()
|
||
resp = self._request(messages, timeout, max_tokens, stream=False)
|
||
|
||
if resp.status_code != 200:
|
||
raise AIError(f"{self.model_id} HTTP {resp.status_code}: {resp.text[:200]}")
|
||
|
||
try:
|
||
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
|
||
|
||
def stream(self, messages, timeout=None, max_tokens=None):
|
||
"""Server-sent chunks in the OpenAI streaming schema.
|
||
|
||
Note what "streaming" buys here in practice: the gateway forwards its
|
||
upstream's chunks, and its primary upstream is a reasoning model that
|
||
emits nothing until it has finished thinking. So this shortens the
|
||
wait to first text on some upstreams and not at all on others — it is
|
||
a transport, not a latency guarantee.
|
||
|
||
A `data:` frame carrying an `error` object is the gateway's way of
|
||
reporting "no upstream answered" mid-stream, so it is raised rather
|
||
than yielded as content.
|
||
"""
|
||
timeout = timeout or default_timeout()
|
||
resp = self._request(messages, timeout, max_tokens, stream=True)
|
||
if resp.status_code != 200:
|
||
raise AIError(f"{self.model_id} HTTP {resp.status_code}: {resp.text[:200]}")
|
||
|
||
try:
|
||
for raw in resp.iter_lines(decode_unicode=True):
|
||
if not raw or not raw.startswith("data:"):
|
||
continue
|
||
data = raw[len("data:"):].strip()
|
||
if data == "[DONE]":
|
||
return
|
||
try:
|
||
chunk = json.loads(data)
|
||
except ValueError:
|
||
continue
|
||
if isinstance(chunk, dict) and chunk.get("error"):
|
||
message = chunk["error"]
|
||
if isinstance(message, dict):
|
||
message = message.get("message") or message
|
||
raise AIError(f"{self.model_id}: {message}")
|
||
choices = chunk.get("choices") or []
|
||
if not choices:
|
||
continue
|
||
text = (choices[0].get("delta") or {}).get("content")
|
||
if text:
|
||
yield Completion(text, chunk.get("provider"))
|
||
except requests.RequestException as e:
|
||
raise AIError(f"{self.model_id} 流式中断: {e}") from e
|
||
finally:
|
||
resp.close()
|
||
|
||
|
||
# --- 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.
|
||
|
||
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": _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.
|
||
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,
|
||
"model": p.model_id,
|
||
"provider": p.name,
|
||
"contextWindow": p.context_window,
|
||
"configured": p.is_configured(),
|
||
"default": mid == head,
|
||
}
|
||
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 ----------------------------------------------------
|
||
# Kept deliberately short: every extra column multiplies by the number of
|
||
# days sent, and the column names double as the vocabulary the model cites
|
||
# back in `basedOn`.
|
||
_CSV_COLUMNS = [
|
||
("date", "date"),
|
||
("steps", "steps"),
|
||
("distanceMeters", "dist_m"),
|
||
("heartRate", "rest_hr"),
|
||
("heartRateMax", "max_hr"),
|
||
("heartRateVariability", "hrv"),
|
||
("stress", "stress"),
|
||
("stressMax", "stress_max"),
|
||
("bodyBatteryHigh", "bb_high"),
|
||
("bodyBatteryLow", "bb_low"),
|
||
("spo2Avg", "spo2"),
|
||
("respirationAvg", "resp"),
|
||
("intensityMinutes", "intensity_min"),
|
||
("caloriesBurned", "kcal"),
|
||
("activeCalories", "active_kcal"),
|
||
("floorsAscended", "floors"),
|
||
("trainingReadiness", "readiness"),
|
||
("enduranceScore", "endurance"),
|
||
]
|
||
|
||
|
||
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,sleep_deep_s,sleep_rem_s,sleep_awake_s"
|
||
)
|
||
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 {}
|
||
for key in ("duration", "quality", "deepSeconds", "remSeconds", "awakeSeconds"):
|
||
value = sleep.get(key)
|
||
cells.append("" if value is None else str(value))
|
||
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 ------------------------------------------------------------
|
||
# 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, 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)
|
||
errors = []
|
||
|
||
for model_id in chain:
|
||
provider = CATALOG[model_id]
|
||
days = max_days_for(provider, day_budget)
|
||
prompt = build_prompt(summary, activities, days)
|
||
try:
|
||
completion = provider.generate(prompt)
|
||
recs = parse_recommendations(completion.text)
|
||
return recs, {
|
||
"model": model_id,
|
||
"provider": provider.name,
|
||
"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)})
|
||
|
||
detail = "; ".join(f"{e['model']}: {e['error']}" for e in errors)
|
||
raise AIError(f"所有模型均失败 -> {detail}")
|
||
|
||
|
||
# --- generic chat entry points ----------------------------------------------
|
||
# Used by the coach (briefing / trend insight / Copilot), which needs multi-turn
|
||
# messages and a bigger output cap than the recommendation list: a reasoning
|
||
# upstream spends part of its budget thinking out loud before the answer, and a
|
||
# briefing is prose rather than five short strings.
|
||
FALLBACK_COACH_MAX_TOKENS = 4000
|
||
|
||
|
||
def coach_max_tokens():
|
||
return int(os.environ.get("AI_COACH_MAX_TOKENS") or FALLBACK_COACH_MAX_TOKENS)
|
||
|
||
|
||
def complete(messages, preferred_model=None, max_tokens=None, timeout=None):
|
||
"""First healthy model in the chain answers. Returns (Completion, meta).
|
||
|
||
Same fallback policy as `generate`, but the caller supplies the whole
|
||
message list and parses the reply itself.
|
||
"""
|
||
chain = resolve_chain(preferred_model)
|
||
max_tokens = max_tokens or coach_max_tokens()
|
||
errors = []
|
||
|
||
for model_id in chain:
|
||
provider = CATALOG[model_id]
|
||
try:
|
||
completion = provider.chat(messages, timeout, max_tokens)
|
||
except AIError as e:
|
||
errors.append({"model": model_id, "error": str(e)})
|
||
continue
|
||
if not (completion.text or "").strip():
|
||
errors.append({"model": model_id, "error": "空响应"})
|
||
continue
|
||
return completion, {
|
||
"model": model_id,
|
||
"provider": provider.name,
|
||
"upstream": completion.upstream,
|
||
"fallbackFrom": [e["model"] for e in errors],
|
||
"errors": errors,
|
||
}
|
||
|
||
detail = "; ".join(f"{e['model']}: {e['error']}" for e in errors)
|
||
raise AIError(f"所有模型均失败 -> {detail}")
|
||
|
||
|
||
def stream_chat(messages, preferred_model=None, max_tokens=None, timeout=None):
|
||
"""Stream a reply, yielding Completion deltas.
|
||
|
||
Two failover rules, and the order matters:
|
||
|
||
1. **Same model, without streaming, before moving on.** Measured against
|
||
the gateway with a real briefing prompt: the streaming request came
|
||
back "所有模型均不可用" after 139s while the identical non-streaming
|
||
request answered in 273s. Its streaming path is simply less reliable
|
||
than its blocking one, so a stream that produces nothing is retried
|
||
blind before the model is written off. The reply then arrives as a
|
||
single late delta rather than not at all.
|
||
2. **No failover once text has been yielded.** By then it has usually
|
||
reached the user's screen, and switching models mid-answer splices two
|
||
different replies together — the exact defect the gateway's own NVIDIA
|
||
adapter has (its README, known issue #1). A half-written answer that
|
||
fails visibly beats one finished in another voice.
|
||
"""
|
||
chain = resolve_chain(preferred_model)
|
||
max_tokens = max_tokens or coach_max_tokens()
|
||
errors = []
|
||
|
||
for model_id in chain:
|
||
provider = CATALOG[model_id]
|
||
started = False
|
||
try:
|
||
for delta in provider.stream(messages, timeout, max_tokens):
|
||
started = True
|
||
yield delta
|
||
except AIError as e:
|
||
if started:
|
||
raise
|
||
errors.append({"model": model_id, "error": f"流式: {e}"})
|
||
if started:
|
||
return
|
||
|
||
if not provider.streaming:
|
||
# Its `stream` was the blocking call already; retrying it here
|
||
# would spend a second full generation on the same failure.
|
||
continue
|
||
|
||
try:
|
||
completion = provider.chat(messages, timeout, max_tokens)
|
||
except AIError as e:
|
||
errors.append({"model": model_id, "error": str(e)})
|
||
continue
|
||
if (completion.text or "").strip():
|
||
yield completion
|
||
return
|
||
errors.append({"model": model_id, "error": "空响应"})
|
||
|
||
detail = "; ".join(f"{e['model']}: {e['error']}" for e in errors)
|
||
raise AIError(f"所有模型均失败 -> {detail}")
|
||
|
||
|
||
# --- JSON extraction --------------------------------------------------------
|
||
def extract_json(text):
|
||
"""Pull the last complete JSON object or array out of a model reply.
|
||
|
||
The gateway's primary upstream is a reasoning model whose visible output
|
||
*begins* with its chain of thought ("The user wants ... so we must ..."),
|
||
with the real answer at the end. Scanning from the front therefore finds
|
||
prose, or a JSON fragment the model was merely considering. Scanning
|
||
backwards from the last closing brace finds the answer it settled on.
|
||
|
||
Brace counting is string-aware, because a Chinese briefing routinely
|
||
contains a quoted `}` or an escaped quote and a naive count breaks on both.
|
||
"""
|
||
if not text or not text.strip():
|
||
raise AIError("模型返回空响应")
|
||
|
||
cleaned = _FENCE.sub("", text).strip()
|
||
try:
|
||
return json.loads(cleaned)
|
||
except ValueError:
|
||
pass
|
||
|
||
for close, opener in (("}", "{"), ("]", "[")):
|
||
end = cleaned.rfind(close)
|
||
while end != -1:
|
||
start = _matching_open(cleaned, end, opener, close)
|
||
if start is not None:
|
||
try:
|
||
return json.loads(cleaned[start : end + 1])
|
||
except ValueError:
|
||
pass
|
||
end = cleaned.rfind(close, 0, end)
|
||
|
||
raise AIError(f"模型未返回可解析的 JSON: {text[-200:]}")
|
||
|
||
|
||
def _matching_open(text, end, opener, close):
|
||
"""Index of the bracket that `text[end]` closes, or None if unbalanced."""
|
||
depth = 0
|
||
in_string = False
|
||
for i in range(end, -1, -1):
|
||
ch = text[i]
|
||
if in_string:
|
||
# Walking backwards, a quote ends the string only when it is not
|
||
# itself escaped — count the run of backslashes before it.
|
||
if ch == '"':
|
||
backslashes = 0
|
||
j = i - 1
|
||
while j >= 0 and text[j] == "\\":
|
||
backslashes += 1
|
||
j -= 1
|
||
if backslashes % 2 == 0:
|
||
in_string = False
|
||
continue
|
||
if ch == '"':
|
||
in_string = True
|
||
continue
|
||
if ch == close:
|
||
depth += 1
|
||
elif ch == opener:
|
||
depth -= 1
|
||
if depth == 0:
|
||
return i
|
||
return None
|