From 463eca380d5bd187152bc89b7033b95d16ee4912 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 10 Sep 2026 13:28:51 -0400 Subject: [PATCH] feat: expose bare generation time and hedge flags in CallStats CallStats gains hedges/generation_ms/hedge_won (all defaulted, appended after total_latency_ms); _CallContext counts them via record_generation (overwrite for chat/OCR, accumulate for embedding batches) and register_hedge, and snapshot carries them out. All three _attempt implementations time only the transport call itself on the same injected clock as total_latency_ms; the chat sink is recorded by the orchestrator so hedge winner attribution stays with T3. Hedge counters stay 0/False until the T3 orchestration lands. Red-green evidence: tests/outputs/137/t2/ (10 new tests AttributeError red, then green; full unit+contracts 1621 passed). --- src/polygateway/embedding.py | 5 ++ src/polygateway/middleware/retry.py | 17 +++++- src/polygateway/ocr.py | 5 ++ src/polygateway/types.py | 37 +++++++++++- tests/unit/test_client.py | 60 ++++++++++++++++++++ tests/unit/test_embedding.py | 14 +++++ tests/unit/test_ocr_client.py | 15 +++++ tests/unit/test_retry.py | 87 ++++++++++++++++++++++++++++- tests/unit/test_types.py | 49 ++++++++++++++++ 9 files changed, 283 insertions(+), 6 deletions(-) diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index a2b0684..a37f794 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -370,7 +370,11 @@ class EmbeddingClient: # 登记在 transport 调用**之前**(同 RetryMW): 失败与取消的尝试也真的发出去了 context.register_attempt() try: + # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身(同 RetryMW 口径), + # 与本链路 total_latency_ms 同一只注入钟;分批累加在成功分支登记 + gen_started = self._now() result = await self._transport.embed(texts=batch, source=source, call_id=call_id) + gen_ms = int((self._now() - gen_started) * 1000) if self._expected_dim is not None and result.dim != self._expected_dim: raise ResultInvalidError( f"{source.name} 维度 {result.dim} 不符期望 {self._expected_dim}", @@ -384,6 +388,7 @@ class EmbeddingClient: actual = result.prompt_tokens # 真实 usage 恰为 0 也是已知事实, 后续取消不得改写成 est settlement_known = True + context.record_generation(gen_ms, accumulate=True) await self._record_quietly(self._breaker.record_success(entry)) await self._record_quietly(self._quota.mark_progress()) latency_ms = int((self._now() - started) * 1000) diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 9a8a782..1c5d81d 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -246,12 +246,20 @@ class RetryMW: await self._admission.on_no_runnable(gate_rejections, reasons, clock) continue async with clock.attempting() as attempt: - outcome = await self._attempt(request, *picked, reasons, attempt_fails) + # 每轮一个 sink: 裸生成时间由编排裁定归属(T2 单路 = 成功那轮; + # T3 对冲 = 赢家那一路),`_attempt` 只负责把本次 transport 耗时投进来 + generation_sink: list[int] = [] + outcome = await self._attempt( + request, *picked, reasons, attempt_fails, generation_sink=generation_sink + ) rate_limited = _is_rate_limited(outcome) if rate_limited: # 免了重试预算就得进 stall 账,否则这段耗时无人治理(见 StallClock) attempt.refund() if isinstance(outcome, LLMResponse): + # 与 register_attempt 同款 None 守卫: 库内现场构造的请求跳过登记 + if request.call_context is not None: + request.call_context.record_generation(generation_sink[0], accumulate=False) return outcome if not rate_limited: fails += 1 @@ -275,6 +283,8 @@ class RetryMW: entry: GateDecision, reasons: dict[str, str], attempt_fails: dict[str, int], + *, + generation_sink: list[int], ) -> LLMResponse | _Failed: call_id = str(uuid.uuid4()) started = self._now() @@ -288,6 +298,10 @@ class RetryMW: if request.call_context is not None: request.call_context.register_attempt() try: + # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身,起点紧贴调用前、 + # 终点为返回后首句(中间无 await);取消落进来时 transport 未返回,不计。 + # 与该链路 total_latency_ms 同一只注入钟,差值(波动开销)才有意义 + gen_started = self._now() result = await self._transport.complete( messages=request.messages, source=source, @@ -300,6 +314,7 @@ class RetryMW: # T3 对冲编排接线前恒为 None: 调用方不观测首 token(计划 §3.1) first_token_event=None, ) + generation_sink.append(int((self._now() - gen_started) * 1000)) if result.usage_source == "unavailable": # 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回, # 对"从不返回 usage 帧"的源等于 TPM 闸失效(设计 §3.2 #9) diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 4c6bbca..a0f862c 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -395,11 +395,16 @@ class OcrClient: # layout 的 POST + ZIP GET 在同一次 `_invoke` 内,故这里只登记 **1** 次 context.register_attempt() try: + # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身(同 RetryMW 口径); + # layout 的 POST + ZIP GET 在同一次 `_invoke` 内,属同一段生成耗时 + gen_started = self._now() result = await self._invoke(kind, image, source, call_id) + gen_ms = int((self._now() - gen_started) * 1000) await self._record_quietly(self._breaker.record_success(entry)) await self._record_quietly(self._quota.mark_progress()) self._feed_outcome(source.name, ok=True) latency_ms = int((self._now() - started) * 1000) + context.record_generation(gen_ms, accumulate=False) await self._emit( kind, operation, diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 8800036..a9499aa 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -317,23 +317,42 @@ class CallStats: 含缓存 IO、退避等待、准入等待、重问、分批与内联记账。 "总耗时减最后一次尝试耗时"**不等于**纯等待(含其他本地工作)。""" + hedges: int = 0 + """本次逻辑调用实际并发发出的对冲路数(触发但准入失败静默不计);1.3.6 及以前恒 0。""" + generation_ms: int = 0 + """裸生成时间: 赢家/成功那次 transport 调用的墙钟时长(口径见设计 §4.5 H8)。""" + hedge_won: bool = False + """赢家是否为对冲路;无对冲恒 False。""" + class _CallContext: """私有可变逻辑调用上下文: 只持计数、单调时钟与终态去重位,不做 I/O。 - **每调用一个实例**的单任务对象: chat 重试、结构化重问、embedding 分批 - 都在同一任务内串行推进,故计数无需锁。**严禁提升为 client 实例属性** + **每逻辑调用一个实例**,可多任务并发登记(对冲);全部方法无 await, + 事件循环内任务安全。**严禁提升为 client 实例属性** ——那会让同一 client 的并发调用互相串掉计数与逻辑 ID(库铁律"纯 asyncio 中立"、 VT `evolve_llm = llm` 教训的同一形态)。 """ - __slots__ = ("_attempts", "_now", "_started", "_terminal_claimed", "logical_call_id") + __slots__ = ( + "_attempts", + "_generation_ms", + "_hedge_won", + "_hedges", + "_now", + "_started", + "_terminal_claimed", + "logical_call_id", + ) def __init__(self, *, now: Callable[[], float]) -> None: self.logical_call_id = str(uuid.uuid4()) self._now = now self._started = now() self._attempts = 0 + self._generation_ms = 0 + self._hedges = 0 + self._hedge_won = False self._terminal_claimed = False def register_attempt(self) -> None: @@ -344,12 +363,24 @@ class _CallContext: """ self._attempts += 1 + def record_generation(self, elapsed_ms: int, *, accumulate: bool) -> None: + """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。""" + self._generation_ms = self._generation_ms + elapsed_ms if accumulate else elapsed_ms + + def register_hedge(self, *, hedge_won: bool) -> None: + """对冲路实际发出即计数;赢家裁定后一次性登记。""" + self._hedges += 1 + self._hedge_won = hedge_won + def snapshot(self) -> CallStats: """同步冻结当前快照;**绝不 await**,可多次调用。""" return CallStats( logical_call_id=self.logical_call_id, attempts=self._attempts, total_latency_ms=int((self._now() - self._started) * 1000), + hedges=self._hedges, + generation_ms=self._generation_ms, + hedge_won=self._hedge_won, ) def claim_terminal(self) -> bool: diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 47f5b0e..063644a 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -144,6 +144,66 @@ class TestChatEndToEnd: await client.chat([{"role": "user", "content": "hi"}], structured="json") +class TestGenerationMsClient: + """裸生成时间的 client 级口径(1.3.7 批次 C2/F)。 + + `_ScriptedGenClockTransport` 在每次 transport 调用内推进注入钟, + 使"时间花在哪"可断言(同 test_backpressure.ClockAdvancingTransport 范式)。 + """ + + _MSG = [{"role": "user", "content": "hi"}] + + async def test_generation_ms_structured_last_round_wins(self): + """结构化重问覆盖而非累加: 首轮坏 JSON 推进 1s,重问轮推进 0.25s → 250。 + + 推进量取二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms。 + """ + from pydantic import BaseModel + + class Answer(BaseModel): + answer: int + + clock = _StatsClock() + transport = _ScriptedGenClockTransport( + [_ok("not json at all"), _ok('{"answer": 1}')], [1.0, 0.25], clock + ) + async with _client(transport=transport, structured_max_retries=1, now=clock) as client: + resp = await client.chat(self._MSG, structured=Answer) + assert resp.content == '{"answer": 1}' and len(transport.calls) == 2 + assert resp.call_stats is not None + assert resp.call_stats.attempts == 2 + assert resp.call_stats.generation_ms == 250 + + async def test_cache_hit_generation_ms_zero(self): + """缓存命中不产生 transport 调用: generation_ms 恒 0(0 是实测,非"未知")。""" + client = _client(cache=InMemoryCache(), cache_namespace="proj", cache_ttl_s=3600) + async with client: + first = await client.chat(self._MSG) + second = await client.chat(self._MSG) + assert first.cache_hit is False and second.cache_hit is True + assert second.call_stats is not None + assert second.call_stats.generation_ms == 0 + assert second.call_stats.attempts == 0 + + +class _ScriptedGenClockTransport: + """脚本化假 transport: 每次成功调用在返回前按脚本推进注入钟(批次 C2/F)。""" + + def __init__(self, results, advances, clock): + self._results = list(results) + self._advances = list(advances) + self._clock = clock + self.calls = [] + + async def complete( + self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + ): + self.calls.append(call_id) + result = self._results.pop(0) + self._clock.advance(self._advances.pop(0)) + return result + + class TestSamplingOverlay: """调用级采样参数入口(issue #4 Task 3)。""" diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py index 430231d..ab12fc1 100644 --- a/tests/unit/test_embedding.py +++ b/tests/unit/test_embedding.py @@ -319,6 +319,20 @@ class TestEmbedBatching: assert resp.cost is None +class TestEmbedGenerationMs: + """裸生成时间按批累加(1.3.7 H8): generation_ms = 各批 transport 耗时之和。""" + + async def test_generation_ms_sums_batch_transports(self): + # 0.25s 为二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms + clock = FakeClock() + transport = _ClockAdvancingEmbedTransport([(0.25, "ok"), (0.25, "ok")], clock) + client, _ = _embed_client([_src()], [], transport=transport, now=clock) + resp = await client.embed(["a", "bb", "ccc", "dddd"]) + assert resp.call_stats is not None + assert resp.call_stats.attempts == 2 + assert resp.call_stats.generation_ms == 500 + + class TestEmbedPostProcess: async def test_normalize_l2(self): raw = EmbeddingTransportResult( diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index a72caae..8660e33 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -196,6 +196,21 @@ class TestSuccessPaths: await client.parse_layout(b"") +class TestOcrGenerationMs: + """OCR 裸生成时间单次覆盖(1.3.7 H8): 计时只包 transport 调用本身。""" + + async def test_generation_ms_single_transport_call(self): + # 0.5s 为二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms + clock = FakeClock() + transport = ClockAdvancingOcrTransport([(0.5, "text")], clock) + client, _, _ = _client([_src()], [], now=clock, transport=transport) + r = await client.recognize_text(b"jpg") + assert r.text == "LINE-1" + assert r.call_stats is not None + assert r.call_stats.attempts == 1 + assert r.call_stats.generation_ms == 500 + + class TestFailover: async def test_transient_retries_with_backoff(self): sleeps = [] diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 309502f..3a2f951 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -93,6 +93,39 @@ class FakeTransport: return action +class _GenClockTransport: + """委托 FakeTransport 的薄包装: 每次调用返回前按脚本推进注入钟(1.3.7 批次 C)。 + + generation_ms 的口径是"只计 transport 调用本身",故推进必须发生在被包 + transport 内部;退避耗时由用例自带的 sleep 闭包推进,与本包装无关。 + """ + + def __init__(self, script, advances, clock): + self._inner = FakeTransport(script) + self._advances = list(advances) + self._clock = clock + + @property + def calls(self): + return self._inner.calls + + async def complete( + self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event + ): + advance = self._advances.pop(0) + result = await self._inner.complete( + messages=messages, + source=source, + stream=stream, + overlay=overlay, + call_id=call_id, + reasoning_effort=reasoning_effort, + first_token_event=first_token_event, + ) + self._clock.advance(advance) + return result + + class HangingGate(InMemoryGate): """在指定记账写回处永久挂起的门控: 把"取消落在某个 await 上"变成确定性事件。 @@ -141,6 +174,8 @@ def _harness( selector=None, pacer=None, gate=None, + transport=None, + sleep=None, ): clock = clock or FakeClock() limiter = InMemoryLimiter( @@ -151,8 +186,8 @@ def _harness( now=clock, ) gate = gate if gate is not None else InMemoryGate(config=_BREAKER, now=clock) - transport = FakeTransport(script) - sleep = FakeSleep() + transport = transport if transport is not None else FakeTransport(script) + sleep = sleep if sleep is not None else FakeSleep() mw = RetryMW( scope="llm", sources=sources, @@ -916,6 +951,54 @@ class TestRateLimitPushback: assert len(transport.calls) == 3 +class TestGenerationMs: + """裸生成时间(1.3.7 H8): 只计 transport 调用本身,不含退避/准入/遥测收尾。""" + + def _ctx(self, clock): + from polygateway.types import _CallContext + + return _CallContext(now=clock) + + async def test_generation_ms_excludes_backoff_and_admission(self): + """[Transient, ok] 脚本: 退避推进 5s、成功次 transport 推进 0.25s。 + + generation_ms 恒等于成功次 transport 的 250ms;若口径混入了退避, + 它会涨到 5250ms 量级——与 total_latency_ms 的下界断言互为对偶。 + 推进量取二进制可精确表示值(0.25/5.0): int 截断下 0.2 之类会因浮点 + 误差落到 199,断言随之抖动(同 test_types 既有用例只用 1.5/2.0 的惯例)。 + """ + clock = FakeClock() + + async def advancing_sleep(seconds): + clock.advance(seconds) + + transport = _GenClockTransport( + [TransientError("boom", source_name="a"), _ok()], [0.0, 0.25], clock + ) + mw, *_ = _harness( + [_src("a")], + [], + clock=clock, + transport=transport, + sleep=advancing_sleep, + rng=lambda: 2.0, # backoff = 2.0 * (0.5 + 2.0) = 5.0s + ) + ctx = self._ctx(clock) + resp = await mw(dataclasses.replace(_REQ, call_context=ctx)) + assert resp.content == "ok" and len(transport.calls) == 2 + stats = ctx.snapshot() + assert stats.generation_ms == 250 + assert stats.total_latency_ms >= 5250 + + async def test_generation_ms_zero_hedge_flags_without_hedging(self): + """无对冲时 hedges/hedge_won 恒 0/False(对冲登记是 T3 的事)。""" + mw, *_ = _harness([_src("a")], [_ok()]) + ctx = self._ctx(FakeClock()) + await mw(dataclasses.replace(_REQ, call_context=ctx)) + stats = ctx.snapshot() + assert stats.hedges == 0 and stats.hedge_won is False + + class TestLogicalAttemptCounting: """尝试登记在 transport 调用**之前**(1.3.5 设计 §4)。 diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index 4459383..d9e1863 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -622,6 +622,55 @@ class TestCallStatsAndContext: with pytest.raises(dataclasses.FrozenInstanceError): stats.attempts = 3 + def test_callstats_hedge_fields_default(self): + """1.3.7 三字段全带默认值: 仅旧三参数构造不炸,无对冲恒 0/0/False。""" + from polygateway.types import CallStats + + stats = CallStats(logical_call_id="lc-1", attempts=2, total_latency_ms=15) + assert stats.hedges == 0 + assert stats.generation_ms == 0 + assert stats.hedge_won is False + explicit = CallStats( + logical_call_id="lc-2", + attempts=2, + total_latency_ms=15, + hedges=1, + generation_ms=42, + hedge_won=True, + ) + assert (explicit.hedges, explicit.generation_ms, explicit.hedge_won) == (1, 42, True) + + def test_callcontext_record_generation_overwrite_and_accumulate(self): + """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。""" + from polygateway.types import _CallContext + + ctx = _CallContext(now=_FakeMonotonic()) + ctx.record_generation(100, accumulate=False) + ctx.record_generation(30, accumulate=False) + assert ctx.snapshot().generation_ms == 30 + ctx.record_generation(50, accumulate=True) + assert ctx.snapshot().generation_ms == 80 + + def test_callcontext_register_hedge_counts(self): + """对冲路实际发出即计数;赢家裁定后一次性登记赢家身份。""" + from polygateway.types import _CallContext + + ctx = _CallContext(now=_FakeMonotonic()) + ctx.register_hedge(hedge_won=False) + ctx.register_hedge(hedge_won=True) + stats = ctx.snapshot() + assert stats.hedges == 2 and stats.hedge_won is True + + def test_snapshot_includes_hedge_fields(self): + """快照把三字段带出: 裸生成时间与对冲计数不停留在内部状态里。""" + from polygateway.types import _CallContext + + ctx = _CallContext(now=_FakeMonotonic()) + ctx.record_generation(200, accumulate=False) + ctx.register_hedge(hedge_won=True) + stats = ctx.snapshot() + assert (stats.hedges, stats.generation_ms, stats.hedge_won) == (1, 200, True) + def test_context_counts_attempts_and_freezes_elapsed(self): """快照是同步冻结的时间切片: 登记两次尝试后耗时按注入钟折算成毫秒。""" from polygateway.types import _CallContext