diff --git a/src/polygateway/__init__.py b/src/polygateway/__init__.py index c76eca6..1f7c436 100644 --- a/src/polygateway/__init__.py +++ b/src/polygateway/__init__.py @@ -50,6 +50,7 @@ from polygateway.types import ( OpenAIChatCompletion, ResponseMode, SourceConfig, + StreamEnd, TelemetryStatus, ThinkingObservation, ) @@ -73,6 +74,7 @@ __all__ = [ "GovernanceBackendError", "LLMResponse", "ResponseMode", + "StreamEnd", "OpenAIChatCompletion", "ModelPrice", "OcrClient", diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 58ee7fd..6268a84 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -425,6 +425,13 @@ class _CallContext: ResponseMode = Literal["text", "openai"] +StreamEnd = Literal["done", "eof_after_finish", "salvaged"] +"""流式结束方式(1.3.10, #31): `done` = 收到 `[DONE]`;`eof_after_finish` = 未发 +`[DONE]` 但 finish_reason 与其后的实测 usage 齐全、连接干净关闭;`salvaged` = 文本 +模式按 `missing_done=salvage` 打捞的截断结果。非流式/缓存命中/失败行为 `None`。""" + +FinishClass = Literal["natural", "truncated", "filtered", "upstream_error", "unknown", "absent"] + @dataclass(frozen=True) class OpenAIChatCompletion: @@ -438,43 +445,228 @@ class OpenAIChatCompletion: def __post_init__(self) -> None: """校验协议快照；transport 翻译错误，缓存坏条目按 miss。""" message = self.message - if not isinstance(message, dict) or message.get("role") != "assistant": - raise ValueError("completion message 必须含 assistant role") - for key in ("content", "reasoning_content"): - if message.get(key) is not None and not isinstance(message[key], str): - raise ValueError(f"completion {key} 必须为字符串或 null") - for value in (self.finish_reason, self.response_id): - if value is not None and not isinstance(value, str): - raise ValueError("completion 身份/结束原因必须为字符串或 null") - if self.usage is not None and not isinstance(self.usage, dict): - raise ValueError("completion usage 必须为对象或 null") - if "tool_calls" in message: - tools = message["tool_calls"] - if not isinstance(tools, list): - raise ValueError("completion tool_calls 必须为数组") - seen: set[str] = set() - for tool in tools: - if not isinstance(tool, dict): - raise ValueError("completion tool 必须为对象") - identity = tool.get("id") - fn = tool.get("function") - if ( - not isinstance(identity, str) - or not identity - or identity in seen - or tool.get("type") != "function" - or not isinstance(fn, dict) - or not isinstance(fn.get("name"), str) - or not fn["name"] - or not isinstance(fn.get("arguments"), str) - ): - raise ValueError("completion tool 身份/函数字段缺失、冲突或非法") - seen.add(identity) + _check_completion_shape(message, self.finish_reason, self.response_id, self.usage) json.dumps({"message": message, "usage": self.usage}, allow_nan=False) object.__setattr__(self, "message", copy.deepcopy(message)) object.__setattr__(self, "usage", copy.deepcopy(self.usage)) +def _check_completion_shape(message: Any, finish_reason: Any, response_id: Any, usage: Any) -> None: + """快照字段形态校验;违约抛 ValueError(由 transport/缓存各自翻译)。""" + if not isinstance(message, dict) or message.get("role") != "assistant": + raise ValueError("completion message 必须含 assistant role") + if not all(_str_or_none(message.get(key)) for key in ("content", "reasoning_content")): + raise ValueError("completion content/reasoning_content 必须为字符串或 null") + if not (_str_or_none(finish_reason) and _str_or_none(response_id)): + raise ValueError("completion 身份/结束原因必须为字符串或 null") + if not isinstance(usage, dict | None): + raise ValueError("completion usage 必须为对象或 null") + if not isinstance(message.get("tool_calls", []), list | None): + raise ValueError("completion tool_calls 必须为数组或 null") + if not _tool_snapshot_ok(message): + raise ValueError("completion tool 身份/函数字段缺失、冲突或非法") + + +def _str_or_none(value: Any) -> bool: + """协议字符串字段的形态判据: 字符串或 null。""" + return value is None or isinstance(value, str) + + +def _tool_ok(tool: Any, seen: set[str]) -> bool: + """单个工具条目是否满足快照不变式;成立时登记其 id 供查重。""" + if not isinstance(tool, dict): + return False + identity, fn = tool.get("id"), tool.get("function") + ok = ( + isinstance(identity, str) + and bool(identity) + and identity not in seen + and tool.get("type") == "function" + and isinstance(fn, dict) + and isinstance(fn.get("name"), str) + and bool(fn["name"]) + and isinstance(fn.get("arguments"), str) + ) + if ok: + seen.add(identity) + return ok + + +def _tool_snapshot_ok(message: Mapping[str, Any]) -> bool: + """工具快照不变式(1.3.9 规则): 缺省/null 视无工具;null 元素跳过。 + + 抽成纯判据而非留在快照构造里: 结局判定(#30)要在构造**之前**知道工具是否 + 成立——截断的半截工具归 ResultInvalid,构造时抛 Transient 就到不了那一行。 + """ + tools = message.get("tool_calls") + if tools is None: + return True + if not isinstance(tools, list): + return False + seen: set[str] = set() + return all(tool is None or _tool_ok(tool, seen) for tool in tools) + + +_FINISH_CLASSES: Mapping[str, FinishClass] = MappingProxyType( + { + "stop": "natural", + "tool_calls": "natural", + "function_call": "natural", + "length": "truncated", + "content_filter": "filtered", + "sensitive": "filtered", # GLM: 安全审核拦截 + "network_error": "upstream_error", # GLM: 模型推理异常 + "error": "upstream_error", + } +) + + +def _classify_finish(value: str | None) -> FinishClass: + """finish_reason → 类别;大小写敏感(上游原值),未登记非空值为 unknown。""" + if value is None: + return "absent" + return _FINISH_CLASSES.get(value, "unknown") + + +def _stream_is_complete( + *, done: bool, finish_reason: str | None, usage_after_finish: bool, usage_measured: bool +) -> StreamEnd | None: + """流完整性(设计 §4.2): None = 截断。 + + 缺 `[DONE]` 时只认"上游声明结束 + 其后到达的实测 usage": 前者证明生成结束, + 后者证明流走到了 include_usage 的自然末尾(finish 之前的累计 usage 不算)。 + """ + if done: + return "done" + if finish_reason is not None and usage_after_finish and usage_measured: + return "eof_after_finish" + return None + + +@dataclass(frozen=True) +class _OutputSummary: + """一次结果的产物摘要(设计 §4.3 定义),判定的唯一输入形态。""" + + any_output: bool + body_output: bool + tools_present: bool + tools_ok: bool + + +def _summarize_text(content: str, thinking: str) -> _OutputSummary: + """文本模式: 产物只看投影正文与思考;空白也算产物,正文须非空白。""" + return _OutputSummary( + any_output=content != "" or thinking != "", + body_output=content.strip() != "", + tools_present=False, + tools_ok=True, + ) + + +def _has_string(value: Any) -> bool: + """递归查找任一非空字符串片段。""" + if isinstance(value, str): + return value != "" + if isinstance(value, dict): + return any(_has_string(item) for item in value.values()) + if isinstance(value, list): + return any(_has_string(item) for item in value) + return False + + +def _summarize_message(message: Mapping[str, Any], projected_content: str) -> _OutputSummary: + """保真模式: 任一非空字符串字段(role 除外)或工具条目即产物。""" + tools = message.get("tool_calls") + tools_present = isinstance(tools, list) and any(tool is not None for tool in tools) + tools_ok = _tool_snapshot_ok(message) + fields = {k: v for k, v in message.items() if k not in ("role", "tool_calls")} + return _OutputSummary( + any_output=tools_present or _has_string(fields), + body_output=projected_content.strip() != "" or (tools_present and tools_ok), + tools_present=tools_present, + tools_ok=tools_ok, + ) + + +@dataclass(frozen=True) +class _Verdict: + """结局判定;`deliver` 为真时另两字段为 None。""" + + deliver: bool + error_kind: Literal["transient", "result_invalid"] | None + code: str | None + + +_DELIVER = _Verdict(True, None, None) +_EMPTY = _Verdict(False, "transient", "empty_completion") +_WITHOUT_CONTENT: Mapping[str, str] = MappingProxyType( + {"truncated": "length_without_content", "filtered": "filtered_without_content"} +) + + +def _judge_broken( + *, mode: ResponseMode, summary: _OutputSummary, finish: FinishClass +) -> _Verdict | None: + """行 1-2: 上游自报失败、坏工具——先于产物判断,正文不得掩盖。""" + if finish == "upstream_error": + return _Verdict(False, "transient", "upstream_finish_error") + if mode == "openai" and summary.tools_present and not summary.tools_ok: + if finish == "truncated": + return _Verdict(False, "result_invalid", "tool_truncated") + return _Verdict(False, "transient", "tool_incomplete") + return None + + +def _judge_outcome(*, mode: ResponseMode, summary: _OutputSummary, finish: FinishClass) -> _Verdict: + """结局分类(设计 §4.3),自上而下首个匹配行生效。 + + 确定性结局(length/过滤)在文本模式归 ResultInvalid 而非 Transient: 重抽只会 + 再花一份钱,还会把健康源推向熔断;保真模式原样交回,由下游按 finish 决定。 + """ + broken = _judge_broken(mode=mode, summary=summary, finish=finish) + if broken is not None: + return broken + if summary.body_output: + return _DELIVER + if finish in _WITHOUT_CONTENT: # 行 3/6/7: 确定性结局 + if mode == "openai": + return _DELIVER + return _Verdict(False, "result_invalid", _WITHOUT_CONTENT[finish]) + # 行 4/8/9: 零产物恒为抖动;有产物无正文时,只有保真且上游明确声明结束才交回 + if summary.any_output and mode == "openai" and finish != "absent": + return _DELIVER + return _EMPTY + + +def _effective_finish(finish_reason: Any, completion: OpenAIChatCompletion | None) -> str | None: + """缓存读写共用的 finish 口径: 顶层优先,缺省回退协议快照(1.3.9 条目无顶层)。""" + if finish_reason is None and completion is not None: + return completion.finish_reason + return finish_reason + + +def _is_cacheable( + *, + mode: ResponseMode, + content: str, + finish_reason: str | None, + stream_end: str | None, + completion: OpenAIChatCompletion | None, +) -> bool: + """可缓存 ⊂ 可交付(设计 §6): 截断、空正文、非自然结束一律不进 TTL。 + + 缓存回放的是"同一请求的确定答案";length/过滤/未知结束带随机性或不可证明 + 是自然结束,固化进 TTL 会把一次偶然变成必然(无缓存毒化铁律)。 + """ + if stream_end not in (None, "done", "eof_after_finish"): + return False + if _classify_finish(finish_reason) not in ("natural", "absent"): + return False + if mode == "openai": + return completion is not None and _has_completion_output(content, completion) + return content.strip() != "" + + def _has_completion_output(content: str, completion: OpenAIChatCompletion | None) -> bool: """成功须有投影正文或完整工具；协议原文为空时不信任缓存旧投影。 @@ -557,6 +749,13 @@ class LLMResponse: completion: OpenAIChatCompletion | None = None + finish_reason: str | None = None + """上游结束原因原值(1.3.10, #30/#31),两模式同口径;`None` = 未上报或不可观测。""" + + stream_end: StreamEnd | None = None + """流式结束方式(1.3.10, #31),见 `StreamEnd`。与 `usage_source` 分工: 那个答 + "账单可信吗",这个答"内容完整吗";`eof_after_finish` 必为 measured。""" + @dataclass(frozen=True) class ChatRequest: @@ -701,6 +900,13 @@ class TransportResult: completion: OpenAIChatCompletion | None = None + finish_reason: str | None = None + """上游结束原因原值(1.3.10, #30/#31),两模式同口径;`None` = 未上报或不可观测。""" + + stream_end: StreamEnd | None = None + """流式结束方式(1.3.10, #31),见 `StreamEnd`。与 `usage_source` 分工: 那个答 + "账单可信吗",这个答"内容完整吗";`eof_after_finish` 必为 measured。""" + @dataclass(frozen=True) class SourceConfig: diff --git a/tests/unit/test_completion_verdict.py b/tests/unit/test_completion_verdict.py new file mode 100644 index 0000000..50234e2 --- /dev/null +++ b/tests/unit/test_completion_verdict.py @@ -0,0 +1,246 @@ +"""统一完成判定(#30/#31): 内核纯函数真值表。""" + +import pytest + +from polygateway.types import ( + LLMResponse, + OpenAIChatCompletion, + TransportResult, + _classify_finish, + _effective_finish, + _is_cacheable, + _judge_outcome, + _OutputSummary, + _stream_is_complete, + _summarize_message, + _summarize_text, + _tool_snapshot_ok, +) + +_TOOL = {"id": "t1", "type": "function", "function": {"name": "f", "arguments": "{}"}} + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("stop", "natural"), + ("tool_calls", "natural"), + ("function_call", "natural"), + ("length", "truncated"), + ("content_filter", "filtered"), + ("sensitive", "filtered"), + ("network_error", "upstream_error"), + ("error", "upstream_error"), + ("end_turn", "unknown"), + ("STOP", "unknown"), + (None, "absent"), + ], +) +def test_classify_finish(value, expected): + assert _classify_finish(value) == expected + + +@pytest.mark.parametrize( + ("done", "finish", "after", "measured", "expected"), + [ + (True, None, False, False, "done"), + (True, "stop", True, True, "done"), + (False, "stop", True, True, "eof_after_finish"), + (False, "length", True, True, "eof_after_finish"), + (False, None, True, True, None), # 无 finish 即截断 + (False, "stop", False, True, None), # usage 早于 finish(累计 usage) + (False, "stop", True, False, None), # finish 后 usage 不可用 + ], +) +def test_stream_is_complete(done, finish, after, measured, expected): + got = _stream_is_complete( + done=done, finish_reason=finish, usage_after_finish=after, usage_measured=measured + ) + assert got == expected + + +def test_tool_snapshot_ok(): + assert _tool_snapshot_ok({"role": "assistant"}) + assert _tool_snapshot_ok({"tool_calls": None}) + assert _tool_snapshot_ok({"tool_calls": [None, _TOOL]}) + assert not _tool_snapshot_ok({"tool_calls": {}}) + assert not _tool_snapshot_ok({"tool_calls": [{"id": "t1", "type": "function"}]}) + assert not _tool_snapshot_ok({"tool_calls": [_TOOL, _TOOL]}) # 重复 id + + +def test_summarize_text(): + assert _summarize_text("", "") == _OutputSummary(False, False, False, True) + assert _summarize_text(" ", "") == _OutputSummary(True, False, False, True) + assert _summarize_text("", "think") == _OutputSummary(True, False, False, True) + assert _summarize_text("hi", "") == _OutputSummary(True, True, False, True) + + +@pytest.mark.parametrize( + ("message", "projected", "expected"), + [ + ({"role": "assistant"}, "", _OutputSummary(False, False, False, True)), + ({"role": "assistant", "content": None}, "", _OutputSummary(False, False, False, True)), + ({"role": "assistant", "refusal": "no"}, "", _OutputSummary(True, False, False, True)), + ( + {"role": "assistant", "reasoning_content": "r"}, + "", + _OutputSummary(True, False, False, True), + ), + ({"role": "assistant", "content": " "}, " ", _OutputSummary(True, False, False, True)), + ( + {"role": "assistant", "x": {"y": ["", "z"]}}, + "", + _OutputSummary(True, False, False, True), + ), + ( + {"role": "assistant", "tool_calls": [None]}, + "", + _OutputSummary(False, False, False, True), + ), + ( + {"role": "assistant", "tool_calls": [_TOOL]}, + "", + _OutputSummary(True, True, True, True), + ), + ( + {"role": "assistant", "tool_calls": [{"id": "t"}]}, + "", + _OutputSummary(True, False, True, False), + ), + ({"role": "assistant", "content": "a"}, "a", _OutputSummary(True, True, False, True)), + ], +) +def test_summarize_message(message, projected, expected): + assert _summarize_message(message, projected) == expected + + +_NONE = _OutputSummary(False, False, False, True) +_THINK = _OutputSummary(True, False, False, True) +_BODY = _OutputSummary(True, True, False, True) +_BAD_TOOL = _OutputSummary(True, True, True, False) + + +@pytest.mark.parametrize( + ("mode", "summary", "finish", "expected"), + [ + # 行 1: upstream_error 优先于一切 + ("openai", _BODY, "upstream_error", ("transient", "upstream_finish_error")), + ("text", _BODY, "upstream_error", ("transient", "upstream_finish_error")), + # 行 2: 坏工具(仅保真),正文不得掩盖 + ("openai", _BAD_TOOL, "truncated", ("result_invalid", "tool_truncated")), + ("openai", _BAD_TOOL, "natural", ("transient", "tool_incomplete")), + ("text", _BAD_TOOL, "natural", None), + # 行 3: 无产物 + truncated/filtered + ("openai", _NONE, "truncated", None), + ("openai", _NONE, "filtered", None), + ("text", _NONE, "truncated", ("result_invalid", "length_without_content")), + ("text", _NONE, "filtered", ("result_invalid", "filtered_without_content")), + # 行 4: 零增量抖动(2026-07-20 裁决) + ("openai", _NONE, "natural", ("transient", "empty_completion")), + ("openai", _NONE, "absent", ("transient", "empty_completion")), + ("openai", _NONE, "unknown", ("transient", "empty_completion")), + ("text", _NONE, "absent", ("transient", "empty_completion")), + # 行 5: 有正文 + ("openai", _BODY, "truncated", None), + ("text", _BODY, "filtered", None), + ("text", _BODY, "absent", None), + # 行 6-9: 有产物无正文 + ("openai", _THINK, "truncated", None), + ("text", _THINK, "truncated", ("result_invalid", "length_without_content")), + ("openai", _THINK, "filtered", None), + ("text", _THINK, "filtered", ("result_invalid", "filtered_without_content")), + ("openai", _THINK, "natural", None), + ("openai", _THINK, "unknown", None), + ("text", _THINK, "natural", ("transient", "empty_completion")), + ("openai", _THINK, "absent", ("transient", "empty_completion")), + ("text", _THINK, "absent", ("transient", "empty_completion")), + ], +) +def test_judge_outcome_rows(mode, summary, finish, expected): + verdict = _judge_outcome(mode=mode, summary=summary, finish=finish) + if expected is None: + assert verdict.deliver and verdict.error_kind is None and verdict.code is None + else: + assert not verdict.deliver + assert (verdict.error_kind, verdict.code) == expected + + +def _completion(content="a", finish="stop", tools=None): + message = {"role": "assistant", "content": content} + if tools is not None: + message["tool_calls"] = tools + return OpenAIChatCompletion(message, finish, None, None) + + +@pytest.mark.parametrize( + ("kwargs", "expected"), + [ + ({"mode": "text", "content": "a", "finish_reason": "stop", "stream_end": "done"}, True), + ({"mode": "text", "content": "a", "finish_reason": None, "stream_end": None}, True), + ( + { + "mode": "text", + "content": "a", + "finish_reason": "stop", + "stream_end": "eof_after_finish", + }, + True, + ), + ({"mode": "text", "content": "a", "finish_reason": None, "stream_end": "salvaged"}, False), + ({"mode": "text", "content": " ", "finish_reason": "stop", "stream_end": "done"}, False), + ({"mode": "text", "content": "a", "finish_reason": "length", "stream_end": None}, False), + ({"mode": "text", "content": "a", "finish_reason": "sensitive", "stream_end": None}, False), + ({"mode": "text", "content": "a", "finish_reason": "end_turn", "stream_end": None}, False), + ({"mode": "text", "content": "a", "finish_reason": "stop", "stream_end": "bogus"}, False), + ], +) +def test_is_cacheable_text(kwargs, expected): + assert _is_cacheable(completion=None, **kwargs) is expected + + +def test_is_cacheable_openai(): + ok = _completion() + assert _is_cacheable( + mode="openai", content="a", finish_reason="stop", stream_end="done", completion=ok + ) + assert not _is_cacheable( + mode="openai", content="a", finish_reason="stop", stream_end="done", completion=None + ) + # 原文为空且无工具: 不信任旧投影 + assert not _is_cacheable( + mode="openai", + content="a", + finish_reason="stop", + stream_end="done", + completion=_completion(content=""), + ) + tool_only = _completion(content=None, finish="tool_calls", tools=[_TOOL]) + assert _is_cacheable( + mode="openai", content="", finish_reason="tool_calls", stream_end=None, completion=tool_only + ) + + +def test_effective_finish(): + assert _effective_finish("stop", _completion(finish="length")) == "stop" + assert _effective_finish(None, _completion(finish="length")) == "length" + assert _effective_finish(None, None) is None + + +def test_completion_accepts_null_tool_calls(): + snap = OpenAIChatCompletion({"role": "assistant", "tool_calls": None}, "stop", None, None) + assert "tool_calls" in snap.message and snap.message["tool_calls"] is None + with pytest.raises(ValueError): + OpenAIChatCompletion({"role": "assistant", "tool_calls": {}}, "stop", None, None) + + +def test_response_fields_default_none(): + result = TransportResult("a", "", 0, 0, "unavailable", None, None, {}) + assert result.finish_reason is None and result.stream_end is None + response = LLMResponse("a", "", "m", "p", 0, 0, 0, None, None, False, "c") + assert response.finish_reason is None and response.stream_end is None + + +def test_stream_end_exported(): + import polygateway + + assert "StreamEnd" in polygateway.__all__