LuminousRuoxi
94a26227c1
把对话页从固定状态机换成 skill 驱动的 Agent：顾问说什么都原样发给 agent， 由 SKILL.md + 工具决定该聊还是该调工具。工作台（workbenchView）未改动。 后端 - llm.py：新增 stream_chat()（流式 + tools），并带 reasoning 端点自动降级 （gemini-3.7-flash 拒绝 effort:none，首次 400 后锁定重试） - agent.py：最小对话底座内核——skill 注入 + 工具循环 + SSE 事件 - tools.py：8 个工具包住既有能力（analyze/diagnose/scaffold/reference/ translate/recheck/record_note/confirm），延迟 import main 复用已验证的 endpoint 处理器与测试接缝，prompts.py/schemas.py/evidence.py 未改一行 - main.py：POST /api/chat/stream（SSE） skill - skills/hvr-rewrite/SKILL.md：分诊规则、首轮盘点、AI 味清单对齐 blader/humanizer 的 A–E 分类（含强度校准与反误报）、诚实性硬约束 前端 - 流式正文 + 工具卡片 + 选项按钮；工具 payload 直接喂既有渲染函数 （renderUnderstanding/renderDiagnosis/renderRecheck），没有第二套 UI - 删除随 agent 化失效的 runDiagnose / applyUnderstandingCorrection 死代码 验证（真实模型，非 mock） - 65 单测全绿（test_agent.py 30 + test_demo.py 35） - 逐轮实跑：盘点 → 确认 → 诊断 → 复检，SSE 事件序列与 payload kind 均符合契约 - 前端 readSSE 用真实响应字节按 7/64/全量三种切块回放，事件序列一致 - 实测修掉两个只在真跑时暴露的问题：模型调完分析直接调确认工具导致正文为空 （SKILL.md 补「正文先行」硬规则）；伪标题 `**N. 标题**` 让列表判定失败、 短横线漏成字面字符（前端改逐行分组渲染） 未验证：浏览器人工走查（无浏览器自动化环境），仅到「真实 HTTP + 真实字节回放」这一层。
435 lines
19 KiB
Python
435 lines
19 KiB
Python
"""OpenRouter chat-completions client (S2 seam).
|
||
|
||
OpenAI-compatible direct client — zero heavy deps (httpx only). Key / base URL
|
||
/ model all come from the environment; the client is injectable for tests
|
||
(transport or a stub subclass). No call-quota logic by design (CONTRACT C3:
|
||
no artificial limits on real calls; only ordinary robustness retries).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
from typing import Any, Callable
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger("hvr.llm")
|
||
|
||
DEFAULT_MODEL = "google/gemini-3.7-flash"
|
||
DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
|
||
|
||
_JSON_FENCE = re.compile(r"^```(?:json)?\s*|\s*```$")
|
||
|
||
|
||
class LLMError(Exception):
|
||
"""User-readable upstream failure (HTTP/network/parse-after-retry)."""
|
||
|
||
|
||
def resolve_env(env: dict[str, str] | None = None) -> dict[str, str]:
|
||
"""Resolve runtime config. Key stays external (CONTRACT C2)."""
|
||
env = env or dict(os.environ)
|
||
api_key = env.get("OPENROUTER_API_KEY") or env.get("PRODREAM_BACKEND_OPENROUTER_API_KEY") or ""
|
||
base_url = (env.get("OPENROUTER_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
|
||
# 正向代理（含 Basic Auth），如后端 .env 的 PRODREAM_BACKEND_OPENROUTER_PROXY_URL
|
||
proxy_url = env.get("OPENROUTER_PROXY_URL") or env.get("PRODREAM_BACKEND_OPENROUTER_PROXY_URL") or ""
|
||
model = env.get("HVR_LLM_MODEL") or DEFAULT_MODEL
|
||
return {"api_key": api_key, "base_url": base_url, "proxy_url": proxy_url, "model": model}
|
||
|
||
|
||
def _join_content_parts(content: Any) -> str:
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, list):
|
||
parts: list[str] = []
|
||
for part in content:
|
||
if isinstance(part, str):
|
||
parts.append(part)
|
||
elif isinstance(part, dict):
|
||
parts.append(str(part.get("text") or ""))
|
||
return "".join(parts)
|
||
return ""
|
||
|
||
|
||
def message_text(data: dict[str, Any]) -> str:
|
||
"""Read assistant text from an OpenAI-compatible completion payload.
|
||
|
||
DeepSeek V4 / OpenRouter reasoning models often put thinking in
|
||
``reasoning`` / ``reasoning_content`` and leave ``content`` empty when
|
||
the shared output budget is exhausted.
|
||
"""
|
||
try:
|
||
choice = data["choices"][0]
|
||
message = choice.get("message") or {}
|
||
except (KeyError, IndexError, TypeError) as exc:
|
||
raise LLMError("大模型响应缺少内容字段") from exc
|
||
text = _join_content_parts(message.get("content"))
|
||
if text.strip():
|
||
return text
|
||
finish = str(choice.get("finish_reason") or "")
|
||
has_reasoning = bool(
|
||
message.get("reasoning")
|
||
or message.get("reasoning_content")
|
||
or message.get("reasoning_details")
|
||
)
|
||
extra: list[str] = []
|
||
if finish:
|
||
extra.append(f"finish_reason={finish}")
|
||
if has_reasoning:
|
||
extra.append("思考过程占满了输出额度")
|
||
suffix = "；".join(extra) if extra else "可能被截断或拒绝"
|
||
raise LLMError(f"模型未返回内容（{suffix}）")
|
||
|
||
|
||
def reasoning_config(disable_reasoning: bool = False) -> dict[str, Any]:
|
||
"""OpenRouter rejects payloads that set both ``effort`` and ``max_tokens``."""
|
||
if disable_reasoning:
|
||
return {"effort": "none", "exclude": True}
|
||
# 思考预算实测（2026-08-25 两次）：
|
||
# 2048 → 单次 diagnose 309.8s，模型烧满全部思考 token；DeepSeek 思考
|
||
# token 生成速度远慢于正文（约 1/10），5 分钟几乎全花在想。
|
||
# 512 → 同文书同 prompt 实测 70.2s，输出完整合规（4 briefs / 5 patterns）。
|
||
# 早期 2048→1024 曾因旧版冗长 prompt 引入 502（思考受限→输出不合规→校验
|
||
# 失败）；prompt 压紧凑后 512 可行，JSON 校验失败仍有 complete_json 重试兜底。
|
||
return {"max_tokens": 512}
|
||
|
||
|
||
def extract_json(content: str | None) -> dict[str, Any]:
|
||
"""Parse model output as JSON. Strips ```json fences and trims to the
|
||
outermost balanced {...} region."""
|
||
if not content or not content.strip():
|
||
raise LLMError("模型未返回内容（可能被截断或拒绝）")
|
||
text = _JSON_FENCE.sub("", content.strip()).strip()
|
||
try:
|
||
parsed = json.loads(text)
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except json.JSONDecodeError:
|
||
pass
|
||
start = text.find("{")
|
||
end = text.rfind("}")
|
||
if start != -1 and end > start:
|
||
try:
|
||
parsed = json.loads(text[start : end + 1])
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
except json.JSONDecodeError:
|
||
pass
|
||
raise LLMError("模型输出不是有效的 JSON（解析失败）")
|
||
|
||
|
||
def _readable_error(resp: httpx.Response) -> str:
|
||
"""从上游错误响应提取一行用户可读信息（优先 error.message），
|
||
不把完整 JSON body 甩给页面（PRD §11 失败兜底）。"""
|
||
try:
|
||
body = resp.json()
|
||
if isinstance(body, dict):
|
||
msg = (body.get("error") or {}).get("message", "") if isinstance(body.get("error"), dict) else ""
|
||
if msg:
|
||
return msg[:300]
|
||
except Exception:
|
||
pass
|
||
return resp.text[:200]
|
||
|
||
|
||
class LlmClient:
|
||
"""Thin OpenAI-compatible client. `transport`/`client` injectable for tests."""
|
||
|
||
def __init__(
|
||
self,
|
||
api_key: str | None = None,
|
||
base_url: str | None = None,
|
||
model: str | None = None,
|
||
proxy: str | None = None,
|
||
timeout: float = 180.0,
|
||
# 实测（2026-08-25）：诊断正常输出 ~2.5k tokens（prompt 已压紧凑）；
|
||
# 模型偶发啰嗦打满 16000 → 截断 → JSON 失败 → 重试又 258s，页面卡 5 分钟。
|
||
# 8000 给正常输出 3 倍余量，打满时更快失败并走既有重试兜底。
|
||
max_tokens: int = 8000,
|
||
temperature: float = 0.3,
|
||
transport: httpx.BaseTransport | None = None,
|
||
) -> None:
|
||
cfg = resolve_env()
|
||
self.api_key = api_key or cfg["api_key"]
|
||
self.base_url = base_url or cfg["base_url"]
|
||
self.model = model or cfg["model"]
|
||
self.proxy = proxy if proxy is not None else cfg["proxy_url"]
|
||
self.timeout = timeout
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
if not self.api_key:
|
||
raise LLMError("缺少 OpenRouter Key：请通过 local_start.sh 启动（从 prodream_backend/.env 读取）")
|
||
self._transport = transport
|
||
# 端点是否强制开思考（gemini-3.7-flash 是）——首次 400 后锁定，见 stream_chat
|
||
self._reasoning_locked = False
|
||
|
||
# -- internal ----------------------------------------------------------
|
||
def _post(self, payload: dict[str, Any]) -> httpx.Response:
|
||
kwargs: dict[str, Any] = {"timeout": self.timeout}
|
||
if self._transport is not None:
|
||
kwargs["transport"] = self._transport
|
||
# trust_env=False + 显式 proxy：直连配置的 base_url，不继承本机 Surge 等系统代理；
|
||
# 区域代理（PRODREAM_BACKEND_OPENROUTER_PROXY_URL）作为正向代理传入
|
||
with httpx.Client(trust_env=False, proxy=self.proxy or None, **kwargs) as client:
|
||
resp = client.post(
|
||
f"{self.base_url}/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json=payload,
|
||
)
|
||
return resp
|
||
|
||
def _complete(
|
||
self,
|
||
system: str,
|
||
user: str,
|
||
json_reminder: bool = False,
|
||
disable_reasoning: bool = False,
|
||
) -> str:
|
||
messages = [{"role": "system", "content": system}]
|
||
if json_reminder:
|
||
messages.append(
|
||
{"role": "system", "content": "只输出符合要求的 JSON，不要输出任何其它文字、解释或代码块标记。"}
|
||
)
|
||
messages.append({"role": "user", "content": user})
|
||
payload: dict[str, Any] = {
|
||
"model": self.model,
|
||
"messages": messages,
|
||
"temperature": self.temperature,
|
||
"max_tokens": self.max_tokens,
|
||
"response_format": {"type": "json_object"},
|
||
}
|
||
# DeepSeek V4 Flash 的思考 token 与正文共用输出额度；不限制时
|
||
# diagnose 这类长 JSON 经常只返回空 content（页面上就是 502）。
|
||
# OpenRouter：effort 与 max_tokens 只能二选一。
|
||
payload["reasoning"] = reasoning_config(disable_reasoning)
|
||
# 观测：每次 LLM 调用都打耗时 + usage，慢链路（同事反馈 diagnose 卡 5 分钟）靠它定位
|
||
t0 = time.monotonic()
|
||
try:
|
||
resp = self._post(payload)
|
||
except (httpx.HTTPError, ValueError, TypeError) as exc: # 网络/超时；代理 URL 或配置错误（如 .env 解析出整行变量名）
|
||
dt = time.monotonic() - t0
|
||
logger.warning("llm http-error dt=%.1fs %s", dt, exc.__class__.__name__)
|
||
if isinstance(exc, httpx.HTTPError):
|
||
raise LLMError(f"调用大模型失败（网络/超时）：{exc.__class__.__name__}") from exc
|
||
# 2026-08-25 实测：代理 URL 带变量名前缀 → httpx 构造期 ValueError → 裸 500
|
||
# （此前只捕 httpx.HTTPError）。配置类错误同样转可读 LLMError（PRD §11 失败兜底）
|
||
raise LLMError(f"调用大模型失败（配置错误，请检查代理/地址设置）：{exc}") from exc
|
||
dt = time.monotonic() - t0
|
||
if resp.status_code != 200:
|
||
logger.warning("llm http %s dt=%.1fs retry-hint=%s", resp.status_code, dt, resp.headers.get("retry-after", ""))
|
||
raise LLMError(f"调用大模型失败（HTTP {resp.status_code}）：{_readable_error(resp)}")
|
||
try:
|
||
body = resp.json()
|
||
except Exception as exc:
|
||
raise LLMError("模型未返回内容（可能被截断或拒绝）") from exc
|
||
usage = body.get("usage") or {}
|
||
# uvicorn 默认只给 uvicorn.* logger 配 handler，root 的 info 会被吞；
|
||
# 观测行用 warning 级别保证进入 stderr 日志（慢链路定位依据）
|
||
logger.warning(
|
||
"llm ok model=%s dt=%.1fs prompt=%s completion=%s reasoning=%s total_time_ms=%s",
|
||
self.model, dt,
|
||
usage.get("prompt_tokens"), usage.get("completion_tokens"),
|
||
usage.get("reasoning_tokens"), usage.get("total_time"),
|
||
)
|
||
return message_text(body)
|
||
|
||
def _stream_post(self, payload: dict[str, Any]) -> Any:
|
||
"""流式 POST：返回已进入响应体的 stream context（调用方负责关闭）。
|
||
|
||
与 _post 分开的原因：httpx 的流式响应必须在 with 块内消费完，
|
||
不能像 _post 那样把 Response 交出去——连接会被提前关闭。"""
|
||
kwargs: dict[str, Any] = {"timeout": self.timeout}
|
||
if self._transport is not None:
|
||
kwargs["transport"] = self._transport
|
||
client = httpx.Client(trust_env=False, proxy=self.proxy or None, **kwargs)
|
||
return client, client.stream(
|
||
"POST",
|
||
f"{self.base_url}/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json=payload,
|
||
)
|
||
|
||
# -- public ------------------------------------------------------------
|
||
def stream_chat(
|
||
self,
|
||
messages: list[dict[str, Any]],
|
||
tools: list[dict[str, Any]] | None = None,
|
||
disable_reasoning: bool = True,
|
||
max_tokens: int | None = None,
|
||
) -> Any:
|
||
"""流式对话（对话层专用）。工具层仍走 complete_json。
|
||
|
||
存在的理由：对话层要「边想边说」+ 能调工具，而 complete_json 强制
|
||
``response_format=json_object`` 且一次性返回，两者不能共存——分层后
|
||
工具层保留已验证的重试兜底，对话层只负责流式与工具编排。
|
||
|
||
逐条 yield（dict）：
|
||
{"type": "text", "text": "..."} 正文增量
|
||
{"type": "thinking", "text": "..."} 思考增量（display 用，可为空）
|
||
{"type": "tool_calls", "tool_calls": [{"id", "name", "arguments"}]}
|
||
流结束时一次性给出（增量已合并）
|
||
{"type": "done", "finish_reason": "..."}
|
||
"""
|
||
payload: dict[str, Any] = {
|
||
"model": self.model,
|
||
"messages": messages,
|
||
"temperature": self.temperature,
|
||
"max_tokens": max_tokens or self.max_tokens,
|
||
"stream": True,
|
||
}
|
||
if tools:
|
||
payload["tools"] = tools
|
||
payload["tool_choice"] = "auto"
|
||
# 对话层默认关掉思考：用户等的是第一句话，不是想清楚再开口；
|
||
# 真正需要深想的活都在工具里（工具层另有 512 预算）
|
||
# 2026-09-11 实测：google/gemini-3.7-flash 端点「强制思考」，传
|
||
# effort=none 直接 400「Reasoning is mandatory for this endpoint」。
|
||
# 锁一次后本客户端不再尝试关思考（只第一个请求付代价）。
|
||
if self._reasoning_locked:
|
||
disable_reasoning = False
|
||
payload["reasoning"] = reasoning_config(disable_reasoning)
|
||
return self._iter_stream(payload)
|
||
|
||
def _iter_stream(self, payload: dict[str, Any], _retry: bool = True) -> Any:
|
||
"""消费 SSE 流并归并增量。工具调用的 id/name/arguments 是分片下发的，
|
||
必须按 index 累积到流结束才能拼出完整参数（OpenAI 兼容协议）。"""
|
||
t0 = time.monotonic()
|
||
partial: dict[int, dict[str, str]] = {}
|
||
finish_reason = ""
|
||
client = None
|
||
try:
|
||
client, stream_ctx = self._stream_post(payload)
|
||
with stream_ctx as resp:
|
||
if resp.status_code != 200:
|
||
resp.read()
|
||
detail = _readable_error(resp)
|
||
logger.warning(
|
||
"llm stream http %s dt=%.1fs", resp.status_code, time.monotonic() - t0
|
||
)
|
||
# 该模型不支持关思考 → 锁定并原地重试一次（此时尚未 yield 任何
|
||
# 内容，重试不会造成重复输出）
|
||
if (
|
||
_retry
|
||
and resp.status_code == 400
|
||
and "mandatory" in detail.lower()
|
||
and payload.get("reasoning", {}).get("effort") == "none"
|
||
):
|
||
self._reasoning_locked = True
|
||
payload["reasoning"] = reasoning_config(False)
|
||
client.close()
|
||
yield from self._iter_stream(payload, _retry=False)
|
||
return
|
||
raise LLMError(
|
||
f"调用大模型失败（HTTP {resp.status_code}）：{detail}"
|
||
)
|
||
for line in resp.iter_lines():
|
||
if not line:
|
||
continue
|
||
if not line.startswith("data:"):
|
||
continue
|
||
data = line[5:].strip()
|
||
if data == "[DONE]":
|
||
break
|
||
try:
|
||
chunk = json.loads(data)
|
||
except json.JSONDecodeError:
|
||
continue # 上游偶发的心跳/注释行，跳过不影响正文
|
||
choices = chunk.get("choices") or []
|
||
if not choices:
|
||
continue
|
||
choice = choices[0]
|
||
delta = choice.get("delta") or {}
|
||
if choice.get("finish_reason"):
|
||
finish_reason = str(choice["finish_reason"])
|
||
text = _join_content_parts(delta.get("content"))
|
||
if text:
|
||
yield {"type": "text", "text": text}
|
||
think = _join_content_parts(
|
||
delta.get("reasoning") or delta.get("reasoning_content")
|
||
)
|
||
if think:
|
||
yield {"type": "thinking", "text": think}
|
||
for call in delta.get("tool_calls") or []:
|
||
idx = int(call.get("index") or 0)
|
||
slot = partial.setdefault(idx, {"id": "", "name": "", "arguments": ""})
|
||
if call.get("id"):
|
||
slot["id"] = str(call["id"])
|
||
fn = call.get("function") or {}
|
||
if fn.get("name"):
|
||
slot["name"] = str(fn["name"])
|
||
if fn.get("arguments"):
|
||
slot["arguments"] += str(fn["arguments"])
|
||
except LLMError:
|
||
raise
|
||
except (httpx.HTTPError, ValueError, TypeError) as exc:
|
||
logger.warning(
|
||
"llm stream error dt=%.1fs %s", time.monotonic() - t0, exc.__class__.__name__
|
||
)
|
||
if isinstance(exc, httpx.HTTPError):
|
||
raise LLMError(f"调用大模型失败（网络/超时）：{exc.__class__.__name__}") from exc
|
||
raise LLMError(f"调用大模型失败（配置错误，请检查代理/地址设置）：{exc}") from exc
|
||
finally:
|
||
if client is not None:
|
||
client.close()
|
||
logger.warning(
|
||
"llm stream ok model=%s dt=%.1fs tool_calls=%s finish=%s",
|
||
self.model, time.monotonic() - t0, len(partial), finish_reason,
|
||
)
|
||
if partial:
|
||
yield {
|
||
"type": "tool_calls",
|
||
"tool_calls": [partial[i] for i in sorted(partial)],
|
||
}
|
||
yield {"type": "done", "finish_reason": finish_reason or "stop"}
|
||
|
||
def complete_json(
|
||
self,
|
||
system: str,
|
||
user: str,
|
||
validate: Callable[[dict[str, Any]], str | None] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Structured completion with robustness retries (SEAMS default:
|
||
any LLMError (parse failure / empty content / transient upstream
|
||
failure) is retried once, the retry carrying a JSON-only reminder
|
||
and turning reasoning off so the answer is not crowded out).
|
||
No quota logic.
|
||
|
||
`validate` is an optional semantic check on the parsed JSON: return
|
||
None when acceptable, or a one-line Chinese explanation when not
|
||
(e.g. per-paragraph arrays must match the input paragraph count).
|
||
A failing check is treated exactly like a parse failure — logged,
|
||
retried once, only surfaced after both attempts fail."""
|
||
last_error: LLMError | None = None
|
||
for attempt in range(2):
|
||
try:
|
||
content = self._complete(
|
||
system,
|
||
user,
|
||
json_reminder=(attempt == 1),
|
||
disable_reasoning=(attempt == 1),
|
||
)
|
||
parsed = extract_json(content)
|
||
if validate is not None:
|
||
problem = validate(parsed)
|
||
if problem:
|
||
raise LLMError(problem)
|
||
if attempt == 1:
|
||
logger.warning("llm retry succeeded after first failure: %s", last_error)
|
||
return parsed
|
||
except LLMError as exc:
|
||
last_error = exc
|
||
if attempt == 0:
|
||
logger.warning("llm call failed, retrying once: %s", exc)
|
||
assert last_error is not None
|
||
logger.error("llm call failed after 2 attempts: %s", last_error)
|
||
raise last_error
|