"""OpenAI 单 choice 消息聚合：仅协议值处理，不做 I/O 或治理。 1.3.10(#30): null 一律视为缺省;已知字段沿用 1.3.9 语义;未知字段按 OpenAI 流式 累加惯例无损合并,冲突删键并登记路径——方言是确定性的,拒绝只会让重试重演同一失败。 """ import copy from dataclasses import dataclass from typing import Any from polygateway.errors import TransientError from polygateway.types import OpenAIChatCompletion, ResponseMode _DELTA_KNOWN = frozenset({"role", "content", "reasoning_content", "tool_calls"}) _TOOL_KNOWN = frozenset({"index", "id", "type", "function"}) _FUNCTION_KNOWN = frozenset({"name", "arguments"}) # 标签键: 每帧可能重复下发的身份/格式标记,同值幂等,绝不拼接(拼成 "x-v1x-v1" # 回传上游会被拒);也不算生成进展 _LABEL_KEYS = frozenset({"index", "type", "id", "format"}) def _invalid(reason: str) -> TransientError: """协议形态失败，不回显可能敏感的工具参数。""" return TransientError(f"OpenAI 协议失败: {reason}", operation="chat") def _object(value: Any, name: str) -> dict[str, Any]: """外部对象先校验再读取。""" if not isinstance(value, dict): raise _invalid(f"{name} 必须为对象") return value def _identity(target: dict[str, Any], key: str, value: Any) -> bool: """身份字段只接受同值重复，不把重复完整 ID 当分片拼接。""" if value is None: return False if not isinstance(value, str): raise _invalid(f"{key} 必须为字符串") if key in target and target[key] != value: raise _invalid(f"{key} 冲突") changed = key not in target and bool(value) target[key] = value return changed def _append(target: dict[str, Any], key: str, value: Any) -> str: """字符串片段逐字拼接，保留缺省/null/空串区别。""" if value is None: target.setdefault(key, None) return "" if not isinstance(value, str): raise _invalid(f"{key} 必须为字符串或 null") target[key] = (target.get(key) or "") + value return value def _assemble(message: dict[str, Any], tools: dict[int, dict[str, Any]]) -> dict[str, Any]: """按 index 排序组装工具;`finish()` 与 `raw_message()` 共用,判定与快照同源。""" assembled = copy.deepcopy(message) if tools: assembled["tool_calls"] = [copy.deepcopy(tools[index]) for index in sorted(tools)] return assembled def _completion(message: dict[str, Any], metadata: dict[str, Any]) -> OpenAIChatCompletion: """复用内核最终快照校验并翻译领域错误。""" try: return OpenAIChatCompletion( message, metadata.get("finish_reason"), metadata.get("id"), metadata.get("usage") ) except (ValueError, TypeError) as exc: raise _invalid(str(exc)) from exc # —— 未知字段合并(设计 §5.2) —— class _MergeConflictError(Exception): """合并冲突(内部控制流): 由 `_ExtensionMerger.merge_key` 就地收口为删键。""" def _is_indexed(items: list[Any]) -> bool: """数组元素(忽略 null)均为带整数 index 的对象。""" present = [item for item in items if item is not None] return bool(present) and all( isinstance(item, dict) and type(item.get("index")) is int for item in present ) def _fragment_progress(key: str, value: Any) -> bool: """不落盘的活性判据(文本模式): 非标签键上的非空字符串片段。""" if isinstance(value, str): return key not in _LABEL_KEYS and value != "" if isinstance(value, dict): return any(_fragment_progress(k, v) for k, v in value.items()) if isinstance(value, list): return any(_fragment_progress(key, item) for item in value) return False class _ExtensionMerger: """每尝试私有的未知字段合并器;冲突路径即封禁,后续同路径值丢弃。""" def __init__(self) -> None: self.conflicts: set[str] = set() self._array_modes: dict[str, bool] = {} def merge_key(self, target: dict[str, Any], key: str, value: Any, path: str) -> bool: """合并一个键;返回是否并入了非空字符串片段(活性)。""" if path in self.conflicts: return False if value is None: target.setdefault(key, None) return False try: merged, grew = self._merge(target.get(key), value, key, path) except _MergeConflictError: target.pop(key, None) self.conflicts.add(path) return False target[key] = merged return grew def _merge(self, old: Any, new: Any, key: str, path: str) -> tuple[Any, bool]: if isinstance(new, dict): return self._merge_dict(old, new, path) if isinstance(new, list): return self._merge_list(old, new, key, path) if key in _LABEL_KEYS or not isinstance(new, str): return self._merge_scalar(old, new) if old is not None and not isinstance(old, str): raise _MergeConflictError return (old or "") + new, new != "" @staticmethod def _merge_scalar(old: Any, new: Any) -> tuple[Any, bool]: """标签键与数字/布尔: 同值幂等,不同值冲突(不学 SDK 相加,计数相加会造假)。""" if old is None: return new, False if type(old) is type(new) and old == new: return old, False raise _MergeConflictError def _merge_dict(self, old: Any, new: dict[str, Any], path: str) -> tuple[Any, bool]: if old is not None and not isinstance(old, dict): raise _MergeConflictError target = old if old is not None else {} grew = False for sub_key, sub_value in new.items(): grew |= self.merge_key(target, sub_key, sub_value, f"{path}.{sub_key}") return target, grew def _merge_list(self, old: Any, new: list[Any], key: str, path: str) -> tuple[Any, bool]: if old is not None and not isinstance(old, list): raise _MergeConflictError target = old if old is not None else [] items = [item for item in new if item is not None] if not items: return target, False indexed = _is_indexed(items) if self._array_modes.setdefault(path, indexed) != indexed: raise _MergeConflictError # 形态改变: 首次出现决定合并策略 if indexed: return target, self._merge_indexed(target, items, f"{path}[]") return target, self._merge_appended(target, items, key, f"{path}[]") def _merge_appended(self, target: list[Any], items: list[Any], key: str, path: str) -> bool: grew = False for item in items: merged, item_grew = self._merge(None, item, key, path) target.append(merged) grew |= item_grew return grew def _merge_indexed(self, target: list[Any], items: list[Any], path: str) -> bool: grew = False for item in items: slot = next((e for e in target if e.get("index") == item["index"]), None) if slot is None: slot = {} target.append(slot) grew |= self._merge_dict(slot, item, path)[1] return grew @dataclass(frozen=True) class ChatDelta: """一帧提取结果；progress 表示真实生成进展。 `reasoning` 是 `reasoning` 字段(vLLM/OpenRouter 新名)的片段,投影 thinking 时 作为 `reasoning_content` 缺席时的别名(设计 §4.4)。 """ content: str reasoning_content: str progress: bool reasoning: str = "" class ChatCompletionAccumulator: """每次 transport 尝试独享的聚合器，文本模式不严格校验工具产物。""" def __init__(self, *, response_mode: ResponseMode) -> None: """初始化尝试私有状态。""" self.mode = response_mode self.message: dict[str, Any] = {} self.metadata: dict[str, Any] = {} self.tools: dict[int, dict[str, Any]] = {} self._merger = _ExtensionMerger() self._finish_seen = False self._usage_after_finish = False # —— 只读视图(transport 判定消费) —— @property def finish_reason(self) -> str | None: """choice 0 首个非 null finish_reason;未见为 None。""" return self.metadata.get("finish_reason") @property def finish_seen(self) -> bool: """是否已收到 finish_reason。""" return self._finish_seen @property def usage_after_finish(self) -> bool: """finish 同帧或之后是否收到非 null usage(设计 §4.2)。""" return self._usage_after_finish @property def conflicts(self) -> frozenset[str]: """被删除的冲突键路径(数组下标已归一为 `[]`)。""" return frozenset(self._merger.conflicts) def raw_message(self) -> dict[str, Any]: """未校验的合并消息深拷贝(含排序工具),供构造快照前的结局判定。""" return _assemble(self.message, self.tools) # —— 聚合 —— def add_chunk(self, chunk: dict[str, Any]) -> ChatDelta: """聚合一个已解码帧，一帧至多一个活性事件。""" chunk = _object(chunk, "chunk") if self.mode == "openai": _identity(self.metadata, "id", chunk.get("id")) choice = self._take_choice(chunk.get("choices")) delta = self._take_delta(choice) if choice is not None else ChatDelta("", "", False) self._take_usage(chunk.get("usage")) return delta def _take_usage(self, usage: Any) -> None: """usage 在 finish 处理之后登记,使同帧 finish+usage 算作"之后"。""" if usage is None: return self.metadata["usage"] = _object(usage, "usage") if self._finish_seen: self._usage_after_finish = True def _take_choice(self, choices: Any) -> dict[str, Any] | None: """选出本帧 choice 并登记 finish;usage-only 帧返回 None。""" if choices is None: return None if not isinstance(choices, list): raise _invalid("choices 必须为数组") if not choices: return None if self.mode == "openai": if len(choices) != 1: raise _invalid("仅支持单 choice") choice = _object(choices[0], "choice") if type(choice.get("index")) is not int or choice["index"] != 0: raise _invalid("choice index 必须为 0") _identity(self.metadata, "finish_reason", choice.get("finish_reason")) else: choice = self._text_choice(choices) if choice is None: return None self._finish_seen = self._finish_seen or self.finish_reason is not None return choice def _text_choice(self, choices: list[Any]) -> dict[str, Any] | None: """文本模式取 index 缺省或为 0 的首个 choice,不混入其他 choice(n>1)。""" for raw in choices: choice = _object(raw, "choice") if choice.get("index", 0) == 0: finish = choice.get("finish_reason") if isinstance(finish, str) and "finish_reason" not in self.metadata: self.metadata["finish_reason"] = finish return choice return None def _take_delta(self, choice: dict[str, Any]) -> ChatDelta: """拆分已知与未知字段;已知字段语义逐字沿用 1.3.9。""" raw_delta = choice.get("delta") delta = _object(raw_delta if raw_delta is not None else {}, "delta") if self.mode == "openai" and "role" in delta: self._take_role(delta["role"]) content = self._take_text(delta, "content") reasoning_content = self._take_text(delta, "reasoning_content") alias = delta.get("reasoning") if isinstance(delta.get("reasoning"), str) else "" progress = bool(content or reasoning_content) if "tool_calls" in delta: progress |= self._take_tools(delta["tool_calls"]) progress |= self._take_unknown(delta) return ChatDelta(content, reasoning_content, progress, alias) def _take_role(self, role: Any) -> None: if role is None: self.message.setdefault("role", None) return if role != "assistant": raise _invalid("role 必须为 assistant") self.message["role"] = role def _take_text(self, delta: dict[str, Any], key: str) -> str: if key not in delta: return "" if self.mode == "openai" or delta[key] is not None: return _append(self.message, key, delta[key]) return "" def _take_unknown(self, delta: dict[str, Any]) -> bool: """未知 delta 键: 保真合并进 message;文本只算活性。""" grew = False for key, value in delta.items(): if key in _DELTA_KNOWN: continue if self.mode == "openai": grew |= self._merger.merge_key(self.message, key, value, f"delta.{key}") else: grew |= _fragment_progress(key, value) return grew def _take_tools(self, tools: Any) -> bool: if tools is None: if self.mode == "openai": self.message.setdefault("tool_calls", None) return False if not isinstance(tools, list): if self.mode == "openai": raise _invalid("tool_calls 必须为数组") return False if self.mode == "openai" and self.message.get("tool_calls") is None: self.message["tool_calls"] = [] progress = False for raw in tools: if raw is None: continue if self.mode == "openai": progress |= self._take_tool(_object(raw, "tool")) else: progress |= self._text_tool_progress(raw) return progress def _text_tool_progress(self, raw: Any) -> bool: """文本只消费可识别的工具进展，不承诺工具完整性。""" if not isinstance(raw, dict) or type(raw.get("index")) is not int: return False tool = self.tools.setdefault(raw["index"], {}) progress = False for key in ("id", "type"): if isinstance(raw.get(key), str) and raw[key] and key not in tool: tool[key] = raw[key] progress = True fn = raw.get("function") if isinstance(fn, dict): progress |= any( isinstance(fn.get(k), str) and bool(fn[k]) for k in ("name", "arguments") ) return progress def _take_tool(self, raw: dict[str, Any]) -> bool: index = raw.get("index") if type(index) is not int or index < 0: raise _invalid("tool index 必须为非负整数") tool = self.tools.setdefault(index, {}) progress = _identity(tool, "id", raw.get("id")) progress |= _identity(tool, "type", raw.get("type")) if raw.get("function") is not None: progress |= self._take_function(tool, _object(raw["function"], "function")) for key, value in raw.items(): if key not in _TOOL_KNOWN: progress |= self._merger.merge_key(tool, key, value, f"tool_calls[].{key}") return progress def _take_function(self, tool: dict[str, Any], fn: dict[str, Any]) -> bool: target = tool.setdefault("function", {}) progress = False for key, value in fn.items(): if key in _FUNCTION_KNOWN: progress |= bool(_append(target, key, value)) else: path = f"tool_calls[].function.{key}" progress |= self._merger.merge_key(target, key, value, path) return progress def finish(self) -> OpenAIChatCompletion: """线路完整性由调用者保证；此处只校验最终消息。""" return _completion(self.raw_message(), self.metadata) def _strip_null_tools(message: dict[str, Any]) -> dict[str, Any]: """null 工具元素剔除: 与流式跳过同口径,原样回传 [null] 会被上游拒。""" tools = message.get("tool_calls") if isinstance(tools, list) and None in tools: return {**message, "tool_calls": [tool for tool in tools if tool is not None]} return message def _single_choice(body: dict[str, Any]) -> dict[str, Any]: """恰好一个 choice,index 缺省或为 0;拒绝静默截取多 choice。""" choices = body.get("choices") if not isinstance(choices, list) or len(choices) != 1: raise _invalid("仅支持单 choice") choice = _object(choices[0], "choice") if "index" in choice and (type(choice["index"]) is not int or choice["index"] != 0): raise _invalid("choice index 必须为 0") return choice def check_completion_body(body: Any) -> tuple[dict[str, Any], str | None]: """非流式保真形态校验(不构造快照): 单 choice、index 为 0、finish 为字符串或 null。 与快照构造拆开: 结局判定要先于构造(截断的半截工具须判 ResultInvalid, 构造时抛 Transient 就到不了那一行,设计 §4.3 行 2)。 """ choice = _single_choice(_object(body, "body")) finish = choice.get("finish_reason") if finish is not None and not isinstance(finish, str): raise _invalid("finish_reason 必须为字符串或 null") return _strip_null_tools(_object(choice.get("message"), "message")), finish def build_completion( message: dict[str, Any], body: dict[str, Any], finish: str | None ) -> OpenAIChatCompletion: """非流式快照构造;保留消息扩展字段。""" return _completion(message, {**body, "finish_reason": finish}) def parse_chat_completion(body: dict[str, Any]) -> OpenAIChatCompletion: """非流式完整消息保留扩展字段，拒绝静默截取多 choice。""" message, finish = check_completion_body(body) return build_completion(message, body, finish)