iomgaa/PolyGateway
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 17 Wiki Activity

feat: preserve governed OpenAI tool messages in 1.4.0

Browse Source 
Add faithful single-choice chat results, tool stream activity, error-frame retries, and isolated protocol caching.

Reject empty cached completions with the same value predicate as transport, add regression coverage, and adopt Apache-2.0.
This commit is contained in:
iomgaa
2026-10-01 14:32:19 -04:00
parent 0e29deac1a
commit 9185eab8c9
42 changed files with 2227 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/polygateway/transports/openai_compat.py
+99 -41
View File
@@ -38,11 +38,19 @@ from polygateway.thinking import (
validate_thinking_raw,
)
from polygateway.transports._http_errors import compose_message, summarize_body
from polygateway.transports._openai_completion import (
ChatCompletionAccumulator,
ChatDelta,
parse_chat_completion,
)
from polygateway.types import (
Effort,
EmbeddingTransportResult,
OpenAIChatCompletion,
ResponseMode,
SourceConfig,
TransportResult,
_has_completion_output,
)
if TYPE_CHECKING:
@@ -62,29 +70,6 @@ def _sse_data_payload(raw: str) -> str | None:
return line[len("data:") :].strip()
def _sse_delta(chunk: dict[str, Any], usage_sink: dict[str, Any]) -> tuple[bool, str] | None:
"""从 chunk 提取增量: (True, content) 或 (False, reasoning);usage 帧旁路进 sink。"""
if chunk.get("usage"):
usage_sink["usage"] = chunk["usage"]
if "model" not in usage_sink:
# 首个**有效**值即固定: 末帧的异常值不得覆盖它;但首帧报空串也不能锁死
# sink——否则后续真实版本会丢(issue #3)
reported = _coerce_model_reported(chunk.get("model"))
if reported is not None:
usage_sink["model"] = reported
choices = chunk.get("choices") or []
if not choices:
return None
delta = choices[0].get("delta") or {}
content = delta.get("content")
if content:
return (True, content)
reasoning = delta.get("reasoning_content")
if reasoning:
return (False, reasoning)
return None
def _stall_from_chunk_times(
times: list[float], threshold_ms: int
) -> tuple[float | None, float | None]:
@@ -108,10 +93,25 @@ def _stall_from_chunk_times(
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]
) -> AsyncIterator[tuple[bool, str]]:
"""逐行解析 SSE 流;[DONE] 置 sink["done"];畸形 JSON 帧 → 瞬时错误(可重试)。"""
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:
@@ -122,11 +122,16 @@ async def _iter_sse_deltas(
try:
chunk = json.loads(data)
except json.JSONDecodeError as exc:
raise TransientError(
f"SSE 帧畸形(malformed_json): {data[:80]!r}", operation="chat"
) from exc
delta = _sse_delta(chunk, usage_sink)
if delta is not None:
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
@@ -458,6 +463,7 @@ class OpenAICompatTransport:
reasoning_effort: Effort | None,
first_token_event: asyncio.Event | None,
content_event: asyncio.Event | None,
response_mode: ResponseMode,
) -> TransportResult:
"""一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。
@@ -468,6 +474,16 @@ class OpenAICompatTransport:
`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(
@@ -492,12 +508,30 @@ class OpenAICompatTransport:
try:
if stream:
result = await self._complete_stream(
client, url, payload, source, profile, first_token_event, content_event
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
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:
@@ -585,6 +619,7 @@ class OpenAICompatTransport:
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:
@@ -594,8 +629,9 @@ class OpenAICompatTransport:
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),
_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,
@@ -606,7 +642,7 @@ class OpenAICompatTransport:
last = started
max_gap = 0.0
chunk_times: list[float] = []
async for is_content, text in guarded:
async for delta in guarded:
now = time.monotonic()
# 增量 chunk 时间戳(1.3.8 沉默计量): 循环体只会被真实增量触发——
# 空 chunk/usage 帧/SSE keepalive 已在 _sse_data_payload/_sse_delta
@@ -625,10 +661,18 @@ class OpenAICompatTransport:
else:
max_gap = max(max_gap, (now - last) * 1000)
last = now
(content_parts if is_content else thinking_parts).append(text)
if delta.content:
content_parts.append(delta.content)
if delta.reasoning_content:
thinking_parts.append(delta.reasoning_content)
if response_mode == "openai" and not sink.get("done"):
raise TransientError(
"保真响应缺 [DONE]，禁止打捞", source_name=source.name, operation="chat"
)
completion_result = accumulator.finish() if response_mode == "openai" else None
salvaged = self._check_done(sink, content_parts, thinking_parts, source)
content, thinking = self._finalize_text(content_parts, thinking_parts, profile)
self._reject_empty_completion(content, source)
self._reject_empty_completion(content, completion_result, source)
prompt, completion, usage_source = _resolve_stream_usage(sink, salvaged)
reasoning_tokens = _coerce_reasoning_tokens(sink.get("usage"))
stall_ms, stall_count = _stall_from_chunk_times(chunk_times, source.stall_threshold_ms)
@@ -643,6 +687,7 @@ class OpenAICompatTransport:
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,
@@ -670,14 +715,16 @@ class OpenAICompatTransport:
raise TransientError(f"{source.name} SSE missing_done: 截断且无 [DONE]", **ctx)
return True
def _reject_empty_completion(self, content: str, source: SourceConfig) -> None:
def _reject_empty_completion(
self, content: str, completion: OpenAIChatCompletion | None, source: SourceConfig
) -> None:
"""空补全 → 瞬时错误(2026-07-20 人类裁决,M1 验证发现)。
服务 200 且流程完整([DONE]/usage 正常)但 content 为空——MiniMax 等
网关的间歇异常形态。视为服务抖动: 退避重试/换源,**绝不缓存空响应**;
承 CHS "VLM 零 content"归瞬时的先例(invokers.py:309)。
"""
if not content.strip():
if not _has_completion_output(content, completion):
raise TransientError(
f"{source.name} 空补全(empty_completion): 流程完整但零内容",
source_name=source.name,
@@ -704,6 +751,7 @@ class OpenAICompatTransport:
profile: ProviderProfile,
first_token_event: asyncio.Event | None,
content_event: asyncio.Event | None,
response_mode: ResponseMode,
) -> TransportResult:
"""非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。
@@ -723,16 +771,25 @@ class OpenAICompatTransport:
raise TransientError(
f"{source.name} 非流式响应非法 JSON", source_name=source.name, operation="chat"
) from exc
_check_chat_error(body)
completion_result = parse_chat_completion(body) if response_mode == "openai" else None
choices = body.get("choices") or []
if not choices:
raise TransientError(
f"{source.name} 响应缺 choices", source_name=source.name, operation="chat"
)
message = choices[0].get("message") or {}
if not isinstance(choices, list) or not isinstance(choices[0], dict):
raise TransientError("chat choices 形态非法", operation="chat")
message = choices[0].get("message", {})
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")
content, thinking = self._finalize_text(
[message.get("content") or ""], [message.get("reasoning_content") or ""], profile
)
self._reject_empty_completion(content, source)
self._reject_empty_completion(content, completion_result, source)
prompt, completion, usage_source = _resolve_usage(body.get("usage") or {})
reasoning_tokens = _coerce_reasoning_tokens(body.get("usage"))
return TransportResult(
@@ -744,6 +801,7 @@ class OpenAICompatTransport:
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,