107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Invocation-local text and completion checks for compatible chat streams."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
def needs_qwen_flash_next_integrity(*model_names: str | None) -> bool:
|
|
"""Match the affected model id or its local alias, without broad Qwen rules."""
|
|
return any(
|
|
name.strip().lower().replace("_", "-").rsplit("/", 1)[-1] == "qwen3.8-flash-next"
|
|
for name in model_names
|
|
if isinstance(name, str)
|
|
)
|
|
|
|
|
|
class IncompleteChatStreamError(RuntimeError):
|
|
"""A provider stream ended before a complete response could be committed.
|
|
|
|
Only a fixed reason is exposed. Provider text, URLs and error bodies must
|
|
never appear in an exception that the gateway may send to a browser.
|
|
"""
|
|
|
|
def __init__(self, reason: str) -> None:
|
|
self.reason = reason
|
|
super().__init__(f"Incomplete OpenAI-compatible chat stream: {reason}")
|
|
|
|
|
|
@dataclass
|
|
class ChatStreamState:
|
|
"""Track one request, never a model instance or a persisted conversation."""
|
|
|
|
observed_chat_frame: bool = False
|
|
terminal_received: bool = False
|
|
content: dict[int, str] = field(default_factory=dict)
|
|
reasoning: dict[int, str] = field(default_factory=dict)
|
|
|
|
def observe(self, choices: list[dict[str, Any]]) -> None:
|
|
self.observed_chat_frame = True
|
|
# LangChain consumes the first choice, so another choice's finish
|
|
# reason must not make an unfinished first choice look complete.
|
|
if choices and choices[0].get("finish_reason"):
|
|
self.terminal_received = True
|
|
|
|
def ensure_finished(self) -> None:
|
|
# Responses API and adapter-generated chunks do not pass through the
|
|
# chat-completions converter and have their own completion semantics.
|
|
if self.observed_chat_frame and not self.terminal_received:
|
|
raise IncompleteChatStreamError("missing_terminal_frame")
|
|
|
|
def normalize_text(
|
|
self,
|
|
*,
|
|
index: int,
|
|
delta: dict[str, Any],
|
|
original_delta: Mapping[str, Any] | None,
|
|
message: Mapping[str, Any] | None,
|
|
) -> None:
|
|
"""Convert explicit full-message snapshots to suffixes.
|
|
|
|
A real delta always wins and is appended verbatim, including repeated
|
|
words. Only text supplied through ``message`` is treated as a full
|
|
snapshot; prefix guessing on ordinary deltas would lose real tokens.
|
|
"""
|
|
for fields, accumulated in (
|
|
(("content",), self.content),
|
|
(("reasoning_content", "reasoning"), self.reasoning),
|
|
):
|
|
source = original_delta or {}
|
|
current = accumulated.get(index, "")
|
|
incoming = _first_text(source, fields)
|
|
if incoming is not None:
|
|
accumulated[index] = current + incoming[1]
|
|
continue
|
|
# Structured content and vendor reasoning objects are opaque.
|
|
# Preserve them instead of replacing them with a text snapshot.
|
|
if any(source.get(name) is not None and not isinstance(source[name], str) for name in fields):
|
|
continue
|
|
|
|
snapshot = _first_text(message or {}, fields)
|
|
if snapshot is None:
|
|
continue
|
|
name, text = snapshot
|
|
if text.startswith(current):
|
|
suffix = text[len(current):]
|
|
accumulated[index] = text
|
|
elif current.startswith(text):
|
|
# A duplicate/stale snapshot adds nothing to the live text.
|
|
suffix = ""
|
|
else:
|
|
# A delta cannot retract already-yielded bytes. Retry before
|
|
# committing contradictory text into the conversation.
|
|
raise IncompleteChatStreamError("inconsistent_message_snapshot")
|
|
for alias in fields:
|
|
delta.pop(alias, None)
|
|
delta[name] = suffix
|
|
|
|
|
|
def _first_text(source: Mapping[str, Any], fields: tuple[str, ...]) -> tuple[str, str] | None:
|
|
for name in fields:
|
|
value = source.get(name)
|
|
if isinstance(value, str) and value:
|
|
return name, value
|
|
return None
|