iomgaa
abcdf6930e
- fidelity cacheability also requires a non-empty projected body, so think-tag-only answers are no longer cached - a no-argument tool sent with arguments null only is a deterministic dialect: ResultInvalid tool_arguments_null instead of retried - text mode treats choice index null as 0 and no longer lets False pose as 0 - null-only tool_calls elements no longer produce an empty tool_calls list; non-stream fidelity checks role before the verdict; warning paths are length-bounded - docs: ARCHITECTURE 7.8 field counts, README cache/watchdog wording, CHANGELOG notes, review record in the design, hook note in the plan Red: 8 new tests failed before this change (e.g. test_openai_cacheable_requires_projected_body, test_null_arguments_tool_is_result_invalid_stream, test_text_mode_choice_index_null_is_zero[None-True]).
957 lines
42 KiB
Python
957 lines
42 KiB
Python
"""OpenAI 兼容 transport(D2 默认): 手写 httpx + SSE 解析 + 错误翻译。
|
|
|
|
SSE 纯函数移植 VT `adapters/llm.py:51-124`;错误翻译移植 CHS
|
|
`app/providers/invokers.py:127-227`。职责只到"一次原始调用"——重试/限流/
|
|
缓存归中间件。看门狗活性口径: content 与 reasoning_content 增量都作为流
|
|
元素产出,思考流天然刷新计时(CHS 迁移约束 R1)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass, field, replace
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import httpx
|
|
from loguru import logger
|
|
|
|
from polygateway.errors import (
|
|
PolyGatewayError,
|
|
RequestRejectedError,
|
|
ResultInvalidError,
|
|
SourceDeadError,
|
|
TransientError,
|
|
)
|
|
from polygateway.providers import ProviderProfile, get_provider
|
|
from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
|
|
from polygateway.thinking import (
|
|
ThinkingCapability,
|
|
ThinkingUnsupportedError,
|
|
effective_effort,
|
|
get_capability,
|
|
observe_thinking,
|
|
reconcile_thinking,
|
|
resolve_thinking,
|
|
validate_thinking_raw,
|
|
)
|
|
from polygateway.transports._http_errors import compose_message, summarize_body
|
|
from polygateway.transports._openai_completion import (
|
|
ChatCompletionAccumulator,
|
|
ChatDelta,
|
|
build_completion,
|
|
check_completion_body,
|
|
)
|
|
from polygateway.types import (
|
|
Effort,
|
|
EmbeddingTransportResult,
|
|
ResponseMode,
|
|
SourceConfig,
|
|
StreamEnd,
|
|
TransportResult,
|
|
_classify_finish,
|
|
_judge_outcome,
|
|
_stream_is_complete,
|
|
_summarize_message,
|
|
_summarize_text,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
import asyncio
|
|
from collections.abc import AsyncIterator, Callable, Mapping
|
|
|
|
_THINK_PATTERN = re.compile(r"<think>(.*?)</think>", re.DOTALL)
|
|
# 方言告警节流上限与 finish 值截断(设计 §5.4): 键由上游决定,必须有界
|
|
_WARN_CAP = 1024
|
|
_FINISH_WARN_CHARS = 64
|
|
_PATH_WARN_CHARS = 128
|
|
|
|
# —— SSE 纯函数(VT llm.py 同款)——
|
|
|
|
|
|
def _sse_data_payload(raw: str) -> str | None:
|
|
"""提取 SSE data 行载荷;ping(: 开头)/空行/非 data 行返回 None 跳过。"""
|
|
line = raw.strip()
|
|
if not line or line.startswith(":") or not line.startswith("data:"):
|
|
return None
|
|
return line[len("data:") :].strip()
|
|
|
|
|
|
def _stall_from_chunk_times(
|
|
times: list[float], threshold_ms: int
|
|
) -> tuple[float | None, float | None]:
|
|
"""增量 chunk 时间戳序列 → (stall_ms, stall_count);chunk <2 → (None, None)。
|
|
|
|
沉默 = 相邻间隔**严格大于**阈值(恰好 = 阈值不减,pi-tps 口径);
|
|
尾部(末 chunk → 流结束)不计——墙钟基底口径,尾部归活跃侧(设计 §4.2 N1)。
|
|
无超阈段但 chunk ≥2 时返回 (0.0, 0.0) 而非 None——"观测过且零沉默"与
|
|
"无从观测"是两回事,下游预算公式以 `stall_ms is None` 为分支条件(设计 §4.2 N2)。
|
|
"""
|
|
if len(times) < 2:
|
|
return None, None
|
|
total = 0.0
|
|
count = 0
|
|
# strict=False: 两序列恒差 1(times[1:] 少末尾),B905 要求显式表态——不等长是本设计的本意
|
|
for prev, cur in zip(times, times[1:], strict=False):
|
|
gap_ms = (cur - prev) * 1000
|
|
if gap_ms > threshold_ms:
|
|
total += gap_ms
|
|
count += 1
|
|
return (total, float(count)) if count else (0.0, 0.0)
|
|
|
|
|
|
def _check_chat_error(chunk: Any) -> None:
|
|
"""所有模式拒绝顶层 error；诊断复用有界摘要。"""
|
|
if not isinstance(chunk, dict):
|
|
raise TransientError("chat 响应必须为 JSON 对象", operation="chat")
|
|
if "error" in chunk:
|
|
summary = summarize_body(json.dumps({"error": chunk["error"]}, ensure_ascii=False))
|
|
raise TransientError(
|
|
compose_message("上游 chat error 帧", summary), operation="chat", body_text=summary
|
|
)
|
|
if chunk.get("usage") is not None and not isinstance(chunk["usage"], dict):
|
|
raise TransientError("chat usage 必须为对象或 null", operation="chat")
|
|
|
|
|
|
async def _iter_sse_deltas(
|
|
lines: AsyncIterator[str],
|
|
usage_sink: dict[str, Any],
|
|
accumulator: ChatCompletionAccumulator,
|
|
) -> AsyncIterator[ChatDelta]:
|
|
"""解析 SSE，一帧只产生一次进展事件。"""
|
|
async for raw in lines:
|
|
data = _sse_data_payload(raw)
|
|
if data is None:
|
|
continue
|
|
if data == "[DONE]":
|
|
usage_sink["done"] = True
|
|
return
|
|
try:
|
|
chunk = json.loads(data)
|
|
except json.JSONDecodeError as exc:
|
|
raise TransientError("SSE 帧畸形(malformed_json)", operation="chat") from exc
|
|
_check_chat_error(chunk)
|
|
if chunk.get("usage") is not None:
|
|
usage_sink["usage"] = chunk["usage"]
|
|
if "model" not in usage_sink:
|
|
model = _coerce_model_reported(chunk.get("model"))
|
|
if model is not None:
|
|
usage_sink["model"] = model
|
|
delta = accumulator.add_chunk(chunk)
|
|
if delta.progress:
|
|
yield delta
|
|
|
|
|
|
@dataclass
|
|
class _StreamTiming:
|
|
"""一次流式尝试的计时与文本片段(每尝试新建,不跨重试/对冲共享)。"""
|
|
|
|
content_parts: list[str] = field(default_factory=list)
|
|
reasoning_parts: list[str] = field(default_factory=list)
|
|
alias_parts: list[str] = field(default_factory=list)
|
|
chunk_times: list[float] = field(default_factory=list)
|
|
ttft_ms: float | None = None
|
|
max_gap: float = 0.0
|
|
|
|
@property
|
|
def has_reasoning(self) -> bool:
|
|
"""出现过非空 reasoning_content: 别名 `reasoning` 让位(设计 §4.4)。"""
|
|
return bool(self.reasoning_parts)
|
|
|
|
|
|
async def _consume_deltas(
|
|
guarded: AsyncIterator[ChatDelta],
|
|
started: float,
|
|
first_token_event: asyncio.Event | None,
|
|
content_event: asyncio.Event | None,
|
|
) -> _StreamTiming:
|
|
"""消费进展事件: 计时、置位对冲事件、收集文本片段。"""
|
|
timing = _StreamTiming()
|
|
last = started
|
|
async for delta in guarded:
|
|
now = time.monotonic()
|
|
# 增量 chunk 时间戳(1.3.8 沉默计量): 循环体只会被真实增量触发——
|
|
# 空 chunk/usage 帧/SSE keepalive 已在聚合器被过滤
|
|
timing.chunk_times.append(now)
|
|
if content_event is not None:
|
|
# 每个增量 chunk 置位一次: 中段沉默对冲的活性复位信号
|
|
content_event.set()
|
|
if timing.ttft_ms is None:
|
|
timing.ttft_ms = (now - started) * 1000
|
|
# 首个增量即对冲语义上的"首 token"(思考增量同样是存活证据,
|
|
# 与看门狗活性口径一致);None = 调用方未启用对冲,零分支成本
|
|
if first_token_event is not None:
|
|
first_token_event.set()
|
|
else:
|
|
timing.max_gap = max(timing.max_gap, (now - last) * 1000)
|
|
last = now
|
|
for parts, text in (
|
|
(timing.content_parts, delta.content),
|
|
(timing.reasoning_parts, delta.reasoning_content),
|
|
(timing.alias_parts, delta.reasoning),
|
|
):
|
|
if text:
|
|
parts.append(text)
|
|
return timing
|
|
|
|
|
|
def _raise_unless_deliverable(
|
|
mode: ResponseMode,
|
|
message: dict[str, Any] | None,
|
|
content: str,
|
|
thinking: str,
|
|
finish_reason: str | None,
|
|
source: SourceConfig,
|
|
) -> None:
|
|
"""结局判定(设计 §4.3)的唯一翻译点: 判定值 → 领域错误。
|
|
|
|
原因码以固定格式进 message(遥测 error 列可检索);不回显推理/工具内容。
|
|
"""
|
|
if mode == "openai" and message is not None:
|
|
summary = _summarize_message(message, content)
|
|
else:
|
|
summary = _summarize_text(content, thinking)
|
|
verdict = _judge_outcome(mode=mode, summary=summary, finish=_classify_finish(finish_reason))
|
|
if verdict.deliver:
|
|
return
|
|
text = (
|
|
f"{source.name} {verdict.code}: finish_reason={finish_reason!r}, "
|
|
f"thinking_len={len(thinking)}"
|
|
)
|
|
if verdict.error_kind == "result_invalid":
|
|
raise ResultInvalidError(text, raw_text="", source_name=source.name, operation="chat")
|
|
raise TransientError(text, source_name=source.name, operation="chat")
|
|
|
|
|
|
def _text_choice(body: dict[str, Any], source: SourceConfig) -> tuple[Any, str | None]:
|
|
"""文本模式取首个 choice 的 message 与 finish(非字符串 finish 记 None)。"""
|
|
choices = body.get("choices") or []
|
|
if not choices:
|
|
raise TransientError(
|
|
f"{source.name} 响应缺 choices", source_name=source.name, operation="chat"
|
|
)
|
|
if not isinstance(choices, list) or not isinstance(choices[0], dict):
|
|
raise TransientError("chat choices 形态非法", operation="chat")
|
|
finish = choices[0].get("finish_reason")
|
|
return choices[0].get("message", {}), finish if isinstance(finish, str) else None
|
|
|
|
|
|
def _nonstream_message(
|
|
body: dict[str, Any], mode: ResponseMode, source: SourceConfig
|
|
) -> tuple[dict[str, Any], str | None]:
|
|
"""非流式取单 choice 的原始 message 与 finish;只校验形态,不构造快照。"""
|
|
if mode == "openai":
|
|
message, finish = check_completion_body(body)
|
|
else:
|
|
message, finish = _text_choice(body, source)
|
|
if not isinstance(message, dict) or any(
|
|
message.get(key) is not None and not isinstance(message[key], str)
|
|
for key in ("content", "reasoning_content")
|
|
):
|
|
raise TransientError("chat message 形态非法", operation="chat")
|
|
return message, finish
|
|
|
|
|
|
def _reasoning_of(message: dict[str, Any]) -> str:
|
|
"""非流式思考投影: 非空 reasoning_content 优先,否则 `reasoning` 别名。"""
|
|
primary = message.get("reasoning_content")
|
|
if isinstance(primary, str) and primary:
|
|
return primary
|
|
alias = message.get("reasoning")
|
|
return alias if isinstance(alias, str) else ""
|
|
|
|
|
|
# —— 错误翻译(CHS invokers.py 同款)——
|
|
|
|
|
|
def _parse_retry_after(raw: str | None, *, source_name: str) -> float | None:
|
|
"""解析 Retry-After 头;仅支持秒数形态,HTTP-date 返回 None(CHS 同款)。
|
|
|
|
**非有限值必须当作"无提示"**(issue F1): `"inf"` / `"1e999"` 能被 `float()`
|
|
成功解析,又能通过 `seconds > 0`,于是一路变成 `retry_after_s=inf`——而
|
|
`backoff_delay` 的 `max(delay, retry_after)` 取大之后就是一次**永不醒来**的
|
|
退避 sleep(库刻意不拿 `backoff_max_s` 去夹它,见设计 §6.2)。
|
|
|
|
`source_name` 是**必填** keyword-only 参数: 本函数是私有的,不给默认值,
|
|
漏传即 `TypeError`,免得将来新增调用点静默丢掉源标识(告警定位不到是哪个源)。
|
|
`nan` 不新增分支,沿用既有 `seconds > 0` 恒假的值语义。
|
|
"""
|
|
if raw is None:
|
|
return None
|
|
try:
|
|
seconds = float(raw.strip())
|
|
except ValueError:
|
|
return None
|
|
if math.isinf(seconds):
|
|
# 只写源名与判据词: 429 风暴下回显原始头会把日志淹掉,也无助于定位
|
|
logger.warning(
|
|
"{} 的 Retry-After 非有限值,按无提示处理(retry_after_not_finite)", source_name
|
|
)
|
|
return None
|
|
return seconds if seconds > 0 else None
|
|
|
|
|
|
def _translate_429(
|
|
source: SourceConfig, body_text: str, headers: Mapping[str, str], ctx: dict[str, Any]
|
|
) -> Exception:
|
|
"""429 细分。`body_text` 必须是**未截断的原文**——`ctx["body_text"]` 是摘要,
|
|
头尾保留会破坏 JSON 结构,拿它解析会让超长 body 的配额耗尽退化成普通限速
|
|
(该源不再 force_open),把一个诊断改进变成治理 bug(issue #10 实现红线)。
|
|
"""
|
|
try:
|
|
err_type = json.loads(body_text).get("error", {}).get("type", "")
|
|
except (json.JSONDecodeError, AttributeError):
|
|
err_type = ""
|
|
summary = ctx["body_text"]
|
|
if err_type == "insufficient_quota":
|
|
return SourceDeadError(
|
|
compose_message(f"{source.name} 配额耗尽(insufficient_quota)", summary), **ctx
|
|
)
|
|
return TransientError(
|
|
compose_message(f"{source.name} 限速: 429", summary),
|
|
retry_after_s=_parse_retry_after(headers.get("retry-after"), source_name=source.name),
|
|
**ctx,
|
|
)
|
|
|
|
|
|
def _classify(status: int) -> tuple[type[PolyGatewayError], str]:
|
|
"""状态码 → (错误类, message 标签);映射与 ARCH §6.2 逐条相同,本次零变更。"""
|
|
if status in (401, 403):
|
|
return SourceDeadError, "凭据失效/欠费"
|
|
if status == 400:
|
|
return RequestRejectedError, "请求被拒"
|
|
if status >= 500:
|
|
return TransientError, "瞬时错误"
|
|
return RequestRejectedError, "客户端错误"
|
|
|
|
|
|
def _status_to_error(
|
|
source: SourceConfig,
|
|
status: int,
|
|
body_text: str,
|
|
headers: Mapping[str, str],
|
|
*,
|
|
operation: str,
|
|
) -> Exception:
|
|
"""非 2xx → 领域错误,**全部分支**携带响应体摘要(issue #10)。
|
|
|
|
摘要只算一次,message 与 `body_text` 共用同一份串: 两份不同长度会让"遥测里
|
|
看到的"与"下游 catch 到的"对不上,排查时反而多一层困惑。
|
|
|
|
`operation` 是 **HTTP 子操作**词表(`chat` / `embedding` / `ocr_text` / ...),由
|
|
调用点显式给定。它曾被硬编码成 `"chat"`,而 `embed()` 的非 200 分支也走它
|
|
——于是现存所有 embedding HTTP 失败的 `exc.operation` 都是错的(1.3.5 设计 §5 I1)。
|
|
注意它与遥测新列 `operation`(公开方法四值)是**两个语义**,不做自动转换。
|
|
"""
|
|
summary = summarize_body(body_text)
|
|
ctx: dict[str, Any] = {
|
|
"source_name": source.name,
|
|
"status_code": status,
|
|
"operation": operation,
|
|
"body_text": summary,
|
|
}
|
|
if status == 429:
|
|
return _translate_429(source, body_text, headers, ctx)
|
|
cls, label = _classify(status)
|
|
return cls(compose_message(f"{source.name} {label}: {status}", summary), **ctx)
|
|
|
|
|
|
def _strip_think(content: str) -> tuple[str, str]:
|
|
"""剥离 <think> 标签(qwen 系),返回 (正文, 思考流)。VT llm.py:147-164 同款。"""
|
|
match = _THINK_PATTERN.search(content)
|
|
if match is None:
|
|
return content, ""
|
|
return _THINK_PATTERN.sub("", content).strip(), match.group(1).strip()
|
|
|
|
|
|
def _resolve_usage(usage: dict[str, Any]) -> tuple[int, int, str]:
|
|
"""usage 帧读取;缺失/非法记 0/0 并标 unavailable(est_tokens 解耦设计 §3.2 #3)。
|
|
|
|
不再拿 `est_tokens` 兜底: 它按 CHS 定义是"最坏情形上界",拿上界当实测值
|
|
只会系统性高估账单;宁可把用量记成显式的"不可得"(cost 随之为 NULL),
|
|
让缺口可被统计,也不编一个看似有效的数字。用量口径自此不依赖源配置,
|
|
故不再收 `SourceConfig`。
|
|
"""
|
|
prompt, completion = usage.get("prompt_tokens"), usage.get("completion_tokens")
|
|
if isinstance(prompt, int) and isinstance(completion, int) and prompt + completion > 0:
|
|
return prompt, completion, "measured"
|
|
return 0, 0, "unavailable"
|
|
|
|
|
|
def _coerce_cached_tokens(usage: Any) -> int | None:
|
|
"""取 usage.prompt_tokens_details.cached_tokens(issue #3);形态异常一律 None。
|
|
|
|
`0` 与 `None` 必须可区分: 前者是"该源上报了一次真实零命中",后者是"该源
|
|
不报这个数",下游对两者的处置不同(后者不可做缓存成本校正)。故只把
|
|
**负数与非整数**归 None,`0` 如实保留。`bool` 显式排除——isinstance(True, int)
|
|
在 Python 里为真,放行会把 `True` 记成 1 个命中 token。
|
|
"""
|
|
if not isinstance(usage, dict):
|
|
return None
|
|
details = usage.get("prompt_tokens_details")
|
|
if not isinstance(details, dict):
|
|
return None
|
|
cached = details.get("cached_tokens")
|
|
if isinstance(cached, bool) or not isinstance(cached, int) or cached < 0:
|
|
return None
|
|
return cached
|
|
|
|
|
|
def _coerce_reasoning_tokens(usage: Any) -> int | None:
|
|
"""取 usage.completion_tokens_details.reasoning_tokens(issue #6);形态异常一律 None。
|
|
|
|
与 `_coerce_cached_tokens` 逐条同构(两者是 OpenAI 兼容 usage 里对称的一对):
|
|
`0` 如实保留、负数与非整数归 None、`bool` 显式排除。差别只在语义——本字段
|
|
的 None 是"**本次调用**未上报"而非"该源不上报": 中转在上游不返回 usage 时
|
|
会本地补算并整体替换 usage 对象,把 details 一并吃掉(findings §4c)。
|
|
"""
|
|
if not isinstance(usage, dict):
|
|
return None
|
|
details = usage.get("completion_tokens_details")
|
|
if not isinstance(details, dict):
|
|
return None
|
|
reasoning = details.get("reasoning_tokens")
|
|
if isinstance(reasoning, bool) or not isinstance(reasoning, int) or reasoning < 0:
|
|
return None
|
|
return reasoning
|
|
|
|
|
|
def _coerce_model_reported(value: Any) -> str | None:
|
|
"""取响应体的 model 字段(issue #3);非 str 或空白串一律 None,收口时去空白。
|
|
|
|
去空白不是洁癖: 下游拿这个串做实验快照的 key,`" m "` 与 `"m"` 会造成假分叉。
|
|
"""
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
return value.strip()
|
|
|
|
|
|
def _resolve_stream_usage(sink: dict[str, Any], salvaged: bool) -> tuple[int, int, str]:
|
|
"""流式用量口径: 打捞路径把 measured 降级为 estimated,unavailable 原样保留。
|
|
|
|
前置条件不可省(解耦设计 §3.2 #4): usage 帧本就缺失时 `0/0` 会被洗成
|
|
`estimated`,进而按 token 换算出一个假的 `0.0` 成本。
|
|
"""
|
|
prompt, completion, usage_source = _resolve_usage(sink.get("usage") or {})
|
|
if salvaged and usage_source == "measured":
|
|
# 收到 usage 帧但流被截断: 数字真实、可信度降级(M1 设计 §6)
|
|
usage_source = "estimated"
|
|
return prompt, completion, usage_source
|
|
|
|
|
|
def _extract_vectors(
|
|
data: dict[str, Any], source: SourceConfig, expected_count: int, ctx: dict[str, Any]
|
|
) -> list[list[float]]:
|
|
"""按 data[].index 重排提取向量并校验条数/维度一致性。"""
|
|
try:
|
|
vectors = [
|
|
[float(x) for x in item["embedding"]]
|
|
for item in sorted(data["data"], key=lambda it: int(it["index"]))
|
|
]
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ResultInvalidError(f"{source.name} embedding 响应形态异常: {exc}", **ctx) from exc
|
|
if len(vectors) != expected_count:
|
|
raise ResultInvalidError(
|
|
f"{source.name} 返回 {len(vectors)} 条向量,与输入 {expected_count} 条不符", **ctx
|
|
)
|
|
if len({len(v) for v in vectors}) != 1 or not vectors[0]:
|
|
raise ResultInvalidError(
|
|
f"{source.name} 向量维度异常: {sorted({len(v) for v in vectors})}", **ctx
|
|
)
|
|
return vectors
|
|
|
|
|
|
def _resolve_embedding_usage(data: dict[str, Any]) -> tuple[int, str]:
|
|
"""usage 读取;缺失/非法记 0 并标 unavailable(与 chat 同口径,设计 §3.2 #3)。"""
|
|
prompt = (data.get("usage") or {}).get("prompt_tokens")
|
|
if isinstance(prompt, int) and prompt > 0:
|
|
return prompt, "measured"
|
|
return 0, "unavailable"
|
|
|
|
|
|
def _parse_embedding_payload(
|
|
resp: httpx.Response, source: SourceConfig, expected_count: int
|
|
) -> EmbeddingTransportResult:
|
|
"""解析 /embeddings 响应;一切形态异常归 ResultInvalidError(坏结果≠坏服务)。"""
|
|
ctx: dict[str, Any] = {"source_name": source.name, "operation": "embedding"}
|
|
try:
|
|
data = resp.json()
|
|
except json.JSONDecodeError as exc:
|
|
raise ResultInvalidError(f"{source.name} embedding 响应非 JSON: {exc}", **ctx) from exc
|
|
vectors = _extract_vectors(data, source, expected_count, ctx)
|
|
prompt_tokens, usage_source = _resolve_embedding_usage(data)
|
|
return EmbeddingTransportResult(
|
|
vectors=vectors,
|
|
dim=len(vectors[0]),
|
|
prompt_tokens=prompt_tokens,
|
|
usage_source=usage_source,
|
|
raw={"id": data.get("id")},
|
|
)
|
|
|
|
|
|
def _default_client_factory(source: SourceConfig) -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(
|
|
headers={"Authorization": f"Bearer {source.api_key}"},
|
|
timeout=httpx.Timeout(source.timeout_s),
|
|
trust_env=source.trust_env,
|
|
)
|
|
|
|
|
|
class OpenAICompatTransport:
|
|
"""默认 transport: 每源一个预配 httpx client,懒创建,aclose 统一释放。"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
registry: Mapping[str, ProviderProfile] | None = None,
|
|
capabilities: Mapping[str, ThinkingCapability] | None = None,
|
|
client_factory: Callable[[SourceConfig], httpx.AsyncClient] | None = None,
|
|
) -> None:
|
|
self._registry = registry
|
|
self._capabilities = capabilities
|
|
# 未登记模型只喊一次: 装配期已喊过,逐次调用再喊是日志洪水。
|
|
# 实例级而非模块级 —— 模块级可变状态违反纯 asyncio 中立铁律
|
|
self._warned_models: set[str] = set()
|
|
# 对账告警独立节流,**不复用** `_warned_models`: 两者语义不同(那个 set 记
|
|
# 的是"未登记能力已告警过",这个记的是"某源某方向的矛盾已告警过"),共用
|
|
# 一个容器会让两种告警的生命周期纠缠在一起——将来任一侧想加清空/过期策略,
|
|
# 都会连带改掉另一侧的行为。(键空间恰好不相交,故当下**不会**互相压制;
|
|
# 分开维护的理由是语义,不是碰撞)
|
|
self._warned_mismatches: set[tuple[str, str, Effort | None]] = set()
|
|
# 方言告警(1.3.10)另立集合: 冲突删键/未知 finish 与推理对账语义无关
|
|
self._warned_fields: set[tuple[str, str]] = set()
|
|
self._warn_saturated = False
|
|
self._client_factory = client_factory or _default_client_factory
|
|
self._clients: dict[str, httpx.AsyncClient] = {}
|
|
|
|
def _client_for(self, source: SourceConfig) -> httpx.AsyncClient:
|
|
client = self._clients.get(source.name)
|
|
if client is None:
|
|
client = self._client_factory(source)
|
|
self._clients[source.name] = client
|
|
return client
|
|
|
|
def _build_payload(
|
|
self,
|
|
*,
|
|
messages: list[dict[str, Any]],
|
|
source: SourceConfig,
|
|
profile: ProviderProfile,
|
|
stream: bool,
|
|
overlay: dict[str, Any],
|
|
reasoning_effort: Effort | None,
|
|
) -> tuple[dict[str, Any], Effort | None]:
|
|
"""组装请求体,并交回本次**实际**发出去的档位(`None` = 未表态,不注入)。
|
|
|
|
返回二元组而非只返回 payload: 实际档在 `nearest` 映射后与请求档分叉,而
|
|
除本函数外没有第二处知道映射结果——不交出去,遥测就只能事后再算一遍,
|
|
算出来必是请求档。
|
|
"""
|
|
payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream}
|
|
if stream:
|
|
payload["stream_options"] = {"include_usage": True} # 强制 usage 帧(三项目同款)
|
|
# 形态(provider 级)与能力(model 级)在此相遇;不可满足时 ValueError,
|
|
# 由 complete() 翻译为四分类之一(issue #5)
|
|
capability = get_capability(source.model, table=self._capabilities)
|
|
first_time = source.model not in self._warned_models
|
|
self._warned_models.add(source.model)
|
|
# 三层优先级在此汇合: 请求级 > 源级 > enable_thinking 语法糖(设计 §4.2)。
|
|
# 判定与装配守卫共用同一个纯函数,两处分叉就会变成"装配期放行、运行期报错"
|
|
effort = effective_effort(
|
|
request_effort=reasoning_effort,
|
|
source_effort=source.reasoning_effort,
|
|
enable_thinking=source.enable_thinking,
|
|
)
|
|
resolution = resolve_thinking(
|
|
profile,
|
|
capability,
|
|
effort,
|
|
model=source.model,
|
|
# 源级 `EFFORT_FALLBACK` 必须真的走到这里: 硬编码 "error" 会让人类明确
|
|
# 要求实现的 `nearest` 在零告警下变成死代码(2026-09-05 独立验证查出)
|
|
fallback=source.effort_fallback,
|
|
warn_unregistered=first_time,
|
|
)
|
|
for raw, origin in ((source.extra_body, "source extra_body"), (overlay, "request overlay")):
|
|
validate_thinking_raw(raw, effort=effort, wire=profile.thinking, origin=origin)
|
|
payload.update(resolution.payload)
|
|
# 顺序即优先级(issue #4 设计决策 A): 配置级 extra_body 在前,调用级
|
|
# overlay(含结构化注入)在后覆盖之。两行不可调换
|
|
payload.update(source.extra_body)
|
|
payload.update(overlay)
|
|
return payload, resolution.applied_effort
|
|
|
|
async def complete(
|
|
self,
|
|
*,
|
|
messages: list[dict[str, Any]],
|
|
source: SourceConfig,
|
|
stream: bool,
|
|
overlay: dict[str, Any],
|
|
call_id: str,
|
|
reasoning_effort: Effort | None,
|
|
first_token_event: asyncio.Event | None,
|
|
content_event: asyncio.Event | None,
|
|
response_mode: ResponseMode,
|
|
) -> TransportResult:
|
|
"""一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。
|
|
|
|
`reasoning_effort` 是**请求级**档位(`None` = 不表态);它与源级配置的优先级
|
|
在 `_build_payload` 里由 `effective_effort` 裁定,本层只负责把它送到。
|
|
`first_token_event` 为对冲处置位: `None` = 不观测首 token;仅流式路径
|
|
(`_complete_stream`)在首个增量到达时置位,非流式路径收它但永不置位。
|
|
`content_event` 为中段沉默对冲处置位(1.3.8): 同款约定,流式路径**每个**
|
|
增量 chunk 置位一次,非流式路径收它但永不置位。
|
|
"""
|
|
if response_mode not in ("text", "openai"):
|
|
raise RequestRejectedError(
|
|
"非法 response_mode", source_name=source.name, operation="chat"
|
|
)
|
|
if response_mode == "openai":
|
|
for layer in (source.extra_body, overlay):
|
|
if "n" in layer and (type(layer["n"]) is not int or layer["n"] != 1):
|
|
raise RequestRejectedError(
|
|
"openai 模式仅支持 n=1", source_name=source.name, operation="chat"
|
|
)
|
|
profile = get_provider(source.provider, registry=self._registry)
|
|
try:
|
|
payload, applied_effort = self._build_payload(
|
|
messages=messages,
|
|
source=source,
|
|
profile=profile,
|
|
stream=stream,
|
|
overlay=overlay,
|
|
reasoning_effort=reasoning_effort,
|
|
)
|
|
except ThinkingUnsupportedError as exc:
|
|
# 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它。只捕这个
|
|
# 专用类型而非宽 catch ValueError —— 后者会把序列化等无关错误误贴标签
|
|
raise RequestRejectedError(
|
|
f"{source.name} 推理开关无法满足: {exc}",
|
|
source_name=source.name,
|
|
operation="chat",
|
|
) from exc
|
|
url = source.base_url.rstrip("/") + "/chat/completions"
|
|
client = self._client_for(source)
|
|
ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"}
|
|
try:
|
|
if stream:
|
|
result = await self._complete_stream(
|
|
client,
|
|
url,
|
|
payload,
|
|
source,
|
|
profile,
|
|
first_token_event,
|
|
content_event,
|
|
response_mode,
|
|
)
|
|
else:
|
|
result = await self._complete_once(
|
|
client,
|
|
url,
|
|
payload,
|
|
source,
|
|
profile,
|
|
first_token_event,
|
|
content_event,
|
|
response_mode,
|
|
)
|
|
except TransientError as exc:
|
|
if exc.source_name is None:
|
|
exc.source_name = source.name
|
|
raise
|
|
except StreamLivenessTimeout as exc:
|
|
raise TransientError(f"{source.name} 流活性超时({exc.kind})", **ctx) from exc
|
|
except httpx.TimeoutException as exc:
|
|
raise TransientError(f"{source.name} 超时: {exc}", **ctx) from exc
|
|
except httpx.TransportError as exc:
|
|
# VT 宽集: 覆盖断连/协议错误/读写失败(设计 §9 行 8)
|
|
raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc
|
|
# 实际发出去的档只有 `_build_payload` 知道,而组装 TransportResult 的两条
|
|
# 路径都在更深一层。在此唯一汇合点补齐,好过给两条路径各加一个参数——那正是
|
|
# 遥测那边被明令禁止的"复制参数列表"形态,两条路径迟早只改一条
|
|
result = replace(result, applied_effort=applied_effort)
|
|
# 此处是唯一同时握有请求档位与响应结果的地方,对账只能落在这里
|
|
self._warn_on_thinking_mismatch(source, result)
|
|
return result
|
|
|
|
def _warn_on_thinking_mismatch(self, source: SourceConfig, result: TransportResult) -> None:
|
|
"""声明与观测矛盾即 warning;按 (source, model, 实际档位) 节流,同组合只喊一次。
|
|
|
|
三段缺一不可。**档位**: 同一模型的 low 与 max 是两个独立的矛盾,共用一个
|
|
键会让第二个永久静音(旧版拿 `enable_thinking` 当第三段,而档位根本不经过
|
|
那个字段,于是同一模型的所有档共用一个键)。**源名**: 多源多账号是本库的
|
|
核心场景,同一 model 跨 N 个源是常态,而每个源背后是独立的账号/网关,一个
|
|
源的行为不代表另一个——漏掉源名,5 个源里第一个出问题的喊完一次,其余四个
|
|
永久静音。逐次调用刷屏会把告警变成噪声,噪声等于没有告警。
|
|
|
|
档位取 `result.applied_effort`(真正发出去的那一档)而非请求档: `nearest`
|
|
映射后二者分叉,而对账问的是"我发出去的要求有没有被满足"——拿一个从未发出
|
|
过的档去对账,文案与键都指向了一次不存在的请求。被映射到同一档的两个请求
|
|
因此共用一个键,这正是它们该有的关系(同一条实际要求,同一个矛盾)。
|
|
|
|
**先判键再对账**: `reconcile_thinking` 会拼含完整 `evidence` 的长字符串,
|
|
而非流式档每次调用都命中这一分支,节流后再拼是纯粹的热路径浪费。
|
|
"""
|
|
key = (source.name, source.model, result.applied_effort)
|
|
if key in self._warned_mismatches:
|
|
return
|
|
message = reconcile_thinking(
|
|
effort=result.applied_effort,
|
|
observation=result.thinking_observation,
|
|
capability=get_capability(source.model, table=self._capabilities),
|
|
model=source.model,
|
|
)
|
|
if message is None:
|
|
return
|
|
self._warned_mismatches.add(key)
|
|
# 源名拼在调用点而不是加进 `reconcile_thinking` 的签名: 那是纯判定函数,
|
|
# 输入只该含判定依据(声明/观测/能力/模型),源名是**定位信息**,进不了判据。
|
|
# 单参数传入 loguru: 文案里带 `thinking:{type:disabled}` 这类字面花括号
|
|
# (能力表 evidence),将来有人给这行加个格式化参数就会炸在成功调用的返回
|
|
# 路径上(与 telemetry/sqlite.py 的缺列告警同一先例)
|
|
logger.warning("源 {} —— {}", source.name, message)
|
|
|
|
async def embed(
|
|
self, *, texts: list[str], source: SourceConfig, call_id: str
|
|
) -> EmbeddingTransportResult:
|
|
"""一次原始 embedding 调用(M2 §7): POST /embeddings,错误翻译同 chat。
|
|
|
|
响应按 data[].index 重排保序(GovDoc embedding.py:149 / VT :164 同款);
|
|
空 data/长度不符/维度不一致 → ResultInvalidError(坏结果不熔断)。
|
|
"""
|
|
if not texts:
|
|
raise ValueError("texts 不能为空(空输入由 EmbeddingClient 短路)")
|
|
url = source.base_url.rstrip("/") + "/embeddings"
|
|
client = self._client_for(source)
|
|
ctx: dict[str, Any] = {"source_name": source.name, "operation": "embedding"}
|
|
try:
|
|
resp = await client.post(url, json={"model": source.model, "input": texts})
|
|
except httpx.TimeoutException as exc:
|
|
raise TransientError(f"{source.name} 超时: {exc}", **ctx) from exc
|
|
except httpx.TransportError as exc:
|
|
raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc
|
|
if resp.status_code != 200:
|
|
# 历史误标修正: 本分支属 `embed()`,与上方 ctx 同为 `"embedding"`
|
|
raise _status_to_error(
|
|
source, resp.status_code, resp.text, resp.headers, operation="embedding"
|
|
)
|
|
return _parse_embedding_payload(resp, source, len(texts))
|
|
|
|
async def _complete_stream(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
url: str,
|
|
payload: dict[str, Any],
|
|
source: SourceConfig,
|
|
profile: ProviderProfile,
|
|
first_token_event: asyncio.Event | None,
|
|
content_event: asyncio.Event | None,
|
|
response_mode: ResponseMode,
|
|
) -> TransportResult:
|
|
started = time.monotonic()
|
|
async with client.stream("POST", url, json=payload) as resp:
|
|
if resp.status_code != 200:
|
|
body = (await resp.aread()).decode("utf-8", errors="replace")
|
|
raise _status_to_error(
|
|
source, resp.status_code, body, resp.headers, operation="chat"
|
|
)
|
|
sink: dict[str, Any] = {}
|
|
accumulator = ChatCompletionAccumulator(response_mode=response_mode)
|
|
guarded = stream_with_liveness_timeouts(
|
|
_iter_sse_deltas(resp.aiter_lines(), sink, accumulator),
|
|
ttft_s=source.ttft_timeout_s or source.timeout_s,
|
|
inter_token_s=source.inter_token_timeout_s or source.timeout_s,
|
|
total_s=source.timeout_s,
|
|
)
|
|
timing = await _consume_deltas(guarded, started, first_token_event, content_event)
|
|
return self._stream_result(sink, accumulator, timing, source, profile, response_mode)
|
|
|
|
def _stream_result(
|
|
self,
|
|
sink: dict[str, Any],
|
|
accumulator: ChatCompletionAccumulator,
|
|
timing: _StreamTiming,
|
|
source: SourceConfig,
|
|
profile: ProviderProfile,
|
|
response_mode: ResponseMode,
|
|
) -> TransportResult:
|
|
"""流结束后的统一判定(设计 §4.2-§4.3): 完整性 → 结局 → 组装。"""
|
|
self._warn_dialect(source, accumulator.conflicts, accumulator.finish_reason)
|
|
thinking_parts = timing.reasoning_parts if timing.has_reasoning else timing.alias_parts
|
|
content, thinking = self._finalize_text(timing.content_parts, thinking_parts, profile)
|
|
stream_end = _stream_is_complete(
|
|
done=bool(sink.get("done")),
|
|
finish_reason=accumulator.finish_reason,
|
|
usage_after_finish=accumulator.usage_after_finish,
|
|
usage_measured=_resolve_usage(sink.get("usage") or {})[2] == "measured",
|
|
)
|
|
if stream_end is None:
|
|
stream_end = self._truncated(timing, thinking_parts, source, response_mode)
|
|
message = accumulator.raw_message() if response_mode == "openai" else None
|
|
_raise_unless_deliverable(
|
|
response_mode, message, content, thinking, accumulator.finish_reason, source
|
|
)
|
|
completion_result = accumulator.finish() if response_mode == "openai" else None
|
|
prompt, completion, usage_source = _resolve_stream_usage(sink, stream_end == "salvaged")
|
|
reasoning_tokens = _coerce_reasoning_tokens(sink.get("usage"))
|
|
stall_ms, stall_count = _stall_from_chunk_times(
|
|
timing.chunk_times, source.stall_threshold_ms
|
|
)
|
|
return TransportResult(
|
|
content=content,
|
|
thinking=thinking,
|
|
prompt_tokens=prompt,
|
|
completion_tokens=completion,
|
|
usage_source=usage_source,
|
|
ttft_ms=timing.ttft_ms,
|
|
max_inter_token_ms=(timing.max_gap if timing.ttft_ms is not None else None),
|
|
stall_ms=stall_ms,
|
|
stall_count=stall_count,
|
|
raw={"usage": sink.get("usage")},
|
|
completion=completion_result,
|
|
cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage")),
|
|
model_reported=_coerce_model_reported(sink.get("model")),
|
|
reasoning_tokens=reasoning_tokens,
|
|
# 两条组装路径必须同口径裁定: 只在一条路径上给结论,下游就得靠
|
|
# "这次是不是流式"去猜可观测性,那正是 issue #16/#17 的根因形态
|
|
thinking_observation=observe_thinking(
|
|
thinking=thinking, reasoning_tokens=reasoning_tokens
|
|
),
|
|
finish_reason=accumulator.finish_reason,
|
|
stream_end=stream_end,
|
|
)
|
|
|
|
def _truncated(
|
|
self,
|
|
timing: _StreamTiming,
|
|
thinking_parts: list[str],
|
|
source: SourceConfig,
|
|
response_mode: ResponseMode,
|
|
) -> StreamEnd:
|
|
"""截断流(设计 §4.2): 保真不打捞;文本按 missing_done,零产物恒 retry。"""
|
|
ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"}
|
|
if response_mode == "openai":
|
|
raise TransientError(f"{source.name} missing_done: 保真响应截断,禁止打捞", **ctx)
|
|
if not any(timing.content_parts) and not any(thinking_parts):
|
|
raise TransientError(f"{source.name} SSE early_eof: 零内容断流", **ctx)
|
|
if source.missing_done == "retry":
|
|
raise TransientError(f"{source.name} SSE missing_done: 截断且无 [DONE]", **ctx)
|
|
return "salvaged"
|
|
|
|
def _warn_dialect(
|
|
self, source: SourceConfig, conflicts: frozenset[str], finish_reason: str | None
|
|
) -> None:
|
|
"""方言告警: 冲突删键与未登记 finish 各按 (源, 路径/值) 只喊一次。"""
|
|
for raw_path in sorted(conflicts):
|
|
path = raw_path[:_PATH_WARN_CHARS] # 键名由上游决定,长度同样须有界
|
|
self._warn_once(
|
|
source.name, f"field:{path}", f"字段 {path} 合并冲突已删除,保真结果缺该字段"
|
|
)
|
|
if finish_reason is not None and _classify_finish(finish_reason) == "unknown":
|
|
value = finish_reason[:_FINISH_WARN_CHARS]
|
|
self._warn_once(
|
|
source.name,
|
|
f"finish:{value}",
|
|
f"未登记的 finish_reason {value!r},按正常结束交付,该值的结果不缓存",
|
|
)
|
|
|
|
def _warn_once(self, source_name: str, key: str, text: str) -> None:
|
|
"""有界节流: 键由上游决定,长跑 worker 不能让集合无限增长(设计 §5.4)。"""
|
|
if (source_name, key) in self._warned_fields:
|
|
return
|
|
if len(self._warned_fields) >= _WARN_CAP:
|
|
if not self._warn_saturated:
|
|
self._warn_saturated = True
|
|
logger.warning("方言告警节流集合已饱和({}),后续同类告警省略", _WARN_CAP)
|
|
return
|
|
self._warned_fields.add((source_name, key))
|
|
# 单参数传入 loguru: 文案可能含上游给的花括号(同 `_warn_on_thinking_mismatch`)
|
|
logger.warning("源 {} —— {}", source_name, text)
|
|
|
|
def _finalize_text(
|
|
self, content_parts: list[str], thinking_parts: list[str], profile: ProviderProfile
|
|
) -> tuple[str, str]:
|
|
content = "".join(content_parts)
|
|
thinking = "".join(thinking_parts)
|
|
if profile.strip_think_tags and "<think>" in content:
|
|
content, tag_thinking = _strip_think(content)
|
|
if tag_thinking:
|
|
thinking = tag_thinking
|
|
return content, thinking
|
|
|
|
async def _complete_once(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
url: str,
|
|
payload: dict[str, Any],
|
|
source: SourceConfig,
|
|
profile: ProviderProfile,
|
|
first_token_event: asyncio.Event | None,
|
|
content_event: asyncio.Event | None,
|
|
response_mode: ResponseMode,
|
|
) -> TransportResult:
|
|
"""非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。
|
|
|
|
接收 `first_token_event` 但**永不置位**: 非流式无中途信号,事件自然退化
|
|
为纯时间阈值(对冲只能靠 `hedge_after_s` 触发)。`content_event`(1.3.8)
|
|
同款: 接收但**永不置位**——非流式无"中段"概念,stall 两值走
|
|
TransportResult 默认 None(排除条款)。
|
|
|
|
无流完整性可言,直接以 body 的 finish_reason 走结局判定;保真模式**先判定
|
|
后构造快照**,否则半截工具在构造时就抛 Transient,到不了 `tool_truncated`。
|
|
"""
|
|
resp = await client.post(url, json=payload)
|
|
if resp.status_code != 200:
|
|
raise _status_to_error(
|
|
source, resp.status_code, resp.text, resp.headers, operation="chat"
|
|
)
|
|
try:
|
|
body = resp.json()
|
|
except json.JSONDecodeError as exc:
|
|
raise TransientError(
|
|
f"{source.name} 非流式响应非法 JSON", source_name=source.name, operation="chat"
|
|
) from exc
|
|
_check_chat_error(body)
|
|
message, finish = _nonstream_message(body, response_mode, source)
|
|
content, thinking = self._finalize_text(
|
|
[message.get("content") or ""], [_reasoning_of(message)], profile
|
|
)
|
|
self._warn_dialect(source, frozenset(), finish)
|
|
raw_message = message if response_mode == "openai" else None
|
|
_raise_unless_deliverable(response_mode, raw_message, content, thinking, finish, source)
|
|
completion_result = (
|
|
build_completion(message, body, finish) if response_mode == "openai" else None
|
|
)
|
|
prompt, completion, usage_source = _resolve_usage(body.get("usage") or {})
|
|
reasoning_tokens = _coerce_reasoning_tokens(body.get("usage"))
|
|
return TransportResult(
|
|
content=content,
|
|
thinking=thinking,
|
|
prompt_tokens=prompt,
|
|
completion_tokens=completion,
|
|
usage_source=usage_source,
|
|
ttft_ms=None,
|
|
max_inter_token_ms=None,
|
|
raw={"usage": body.get("usage")},
|
|
completion=completion_result,
|
|
cached_prompt_tokens=_coerce_cached_tokens(body.get("usage")),
|
|
model_reported=_coerce_model_reported(body.get("model")),
|
|
reasoning_tokens=reasoning_tokens,
|
|
# 本路径的裁定多半落 UNKNOWN(M3 实测: 推理已计费却正文与 details 双
|
|
# 缺)。如实标记"观测不到",好过让下游误读成"没推理"
|
|
thinking_observation=observe_thinking(
|
|
thinking=thinking, reasoning_tokens=reasoning_tokens
|
|
),
|
|
finish_reason=finish,
|
|
)
|
|
|
|
async def aclose(self) -> None:
|
|
"""幂等关闭全部源 client。"""
|
|
clients, self._clients = self._clients, {}
|
|
for client in clients.values():
|
|
await client.aclose()
|