From 53bcf431721083ae30b670cf9656bb8c411dcbdb Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 14 Sep 2026 00:56:54 -0400 Subject: [PATCH] feat: redefine generation_ms as net generation time with stall attribution CallStats gains stall_ms/stall_count (None = no stall observation); winner sample triple (wall, stall_ms, stall_count) converted once in _attempt success branch; net = max(0, wall - stall) converged into the shared _record_winner_sample used by both the single-path guard and hedge winner registration. Two-fail hedge rounds record nothing (no winner to attribute). Design section 4.2; issues #27. --- src/polygateway/middleware/retry.py | 57 +++++++++++++++++++++-------- src/polygateway/types.py | 37 +++++++++++++++++-- tests/unit/test_hedge.py | 21 +++++++++++ tests/unit/test_retry.py | 48 +++++++++++++++++++++++- tests/unit/test_types.py | 30 +++++++++++++++ 5 files changed, 173 insertions(+), 20 deletions(-) diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 0350d37..5c01c29 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -58,6 +58,7 @@ if TYPE_CHECKING: RetryPolicy, SourceConfig, TransportResult, + _CallContext, ) @@ -165,6 +166,11 @@ def _is_rate_limited(outcome: LLMResponse | _Failed) -> bool: _HEDGE_LOSER_ATTR = "_polygateway_hedge_loser" + +# 生成样本三元组(1.3.8 设计 §4.2 N8): (墙钟 ms, 沉默总额 ms, 超阈值沉默段数)。 +# 取整换算全库只在 _attempt 成功分支发生一次,消费侧(_record_generation/ +# _register_hedge)只做净口径换算与透传,不再取整。 +_GenerationSample = tuple[int, int | None, int | None] """编排在 cancel() 之前给输家任务置位的标记;_attempt 读它选遥测标签。""" @@ -262,9 +268,10 @@ class RetryMW: await self._admission.on_no_runnable(gate_rejections, reasons, clock) continue async with clock.attempting() as attempt: - # 每轮一个 sink: 裸生成时间由编排裁定归属(T2 单路 = 成功那轮; - # T3 对冲 = 赢家那一路),`_attempt` 只负责把本次 transport 耗时投进来 - generation_sink: list[int] = [] + # 每轮一个 sink: 生成样本(墙钟+沉默归因)由编排裁定归属 + # (单路 = 成功那轮;对冲 = 赢家那一路),`_attempt` 只负责把本次 + # transport 耗时与沉默观测投进来 + generation_sink: list[_GenerationSample] = [] if self._hedge_after_s is None: # 默认关闭: 逐字 1.3.6 单路路径(默认关闭回归门据此成立) outcome = await self._attempt( @@ -284,10 +291,9 @@ class RetryMW: # 免了重试预算就得进 stall 账,否则这段耗时无人治理(见 StallClock) attempt.refund() if isinstance(outcome, LLMResponse): - # 与 register_attempt 同款 None 守卫: 库内现场构造的请求跳过登记; # 对冲轮由 _attempt_hedged 按赢家归属登记,此处外层 sink 恒为空 - if request.call_context is not None and generation_sink: - request.call_context.record_generation(generation_sink[0], accumulate=False) + # (守卫在 _record_generation 内: 上下文缺失/空 sink 均跳过) + self._record_generation(request, generation_sink) return outcome if not rate_limited: fails += 1 @@ -314,7 +320,7 @@ class RetryMW: *, first_token_event: asyncio.Event | None, content_event: asyncio.Event | None, - generation_sink: list[int], + generation_sink: list[_GenerationSample], ) -> LLMResponse | _Failed: call_id = str(uuid.uuid4()) started = self._now() @@ -328,8 +334,9 @@ class RetryMW: if request.call_context is not None: request.call_context.register_attempt() try: - # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身,起点紧贴调用前、 - # 终点为返回后首句(中间无 await);取消落进来时 transport 未返回,不计。 + # 净生成时间(1.3.8): 此处只量 transport 墙钟(起点紧贴调用前、 + # 终点为返回后首句,中间无 await),沉默扣减在消费侧统一换算; + # 取消落进来时 transport 未返回,不计。 # 与该链路 total_latency_ms 同一只注入钟,差值(波动开销)才有意义 gen_started = self._now() result = await self._transport.complete( @@ -348,7 +355,12 @@ class RetryMW: # T4 才在启用 stall 对冲的流式原路上接真事件(设计 §5.2 N3③) content_event=content_event, ) - generation_sink.append(int((self._now() - gen_started) * 1000)) + # 取整全库唯一处(设计 §4.2 N8): transport 的 float 沉默值在此 + # 一次性转 int;None(非流式/增量 chunk<2)透传,消费侧不退减 + wall_ms = int((self._now() - gen_started) * 1000) + stall_ms_i = None if result.stall_ms is None else int(result.stall_ms) + stall_count_i = None if result.stall_count is None else int(result.stall_count) + generation_sink.append((wall_ms, stall_ms_i, stall_count_i)) if result.usage_source == "unavailable": # 用量不可得时按入场预扣量结算(delta==0),否则押金会被整笔退回, # 对"从不返回 usage 帧"的源等于 TPM 闸失效(设计 §3.2 #9) @@ -429,7 +441,7 @@ class RetryMW: # Phase 1 启动原路: 事件与 sink 每轮新建(局部状态,严禁实例属性) source, permit, entry = picked first_token: asyncio.Event = asyncio.Event() - sink_p: list[int] = [] + sink_p: list[_GenerationSample] = [] primary = asyncio.create_task( self._attempt( request, @@ -480,7 +492,7 @@ class RetryMW: return outcome # Phase 4 启动对冲路(v1 单路,H5): 对冲路恒传 first_token_event=None, # 不再触发梯次对冲;content_event 同恒 None(N3③,v1 不梯次) - sink_h: list[int] = [] + sink_h: list[_GenerationSample] = [] hedge = asyncio.create_task( self._attempt( request, @@ -579,14 +591,27 @@ class RetryMW: ) @staticmethod - def _record_generation(request: ChatRequest, sink: list[int]) -> None: + def _record_generation(request: ChatRequest, sink: list[_GenerationSample]) -> None: """record_generation 的共享守卫: 上下文缺失(库内现场构造)或空 sink 均跳过。""" if request.call_context is not None and sink: - request.call_context.record_generation(sink[0], accumulate=False) + RetryMW._record_winner_sample(request.call_context, sink[0]) + + @staticmethod + def _record_winner_sample(context: _CallContext, sample: _GenerationSample) -> None: + """净口径唯一换算(设计 §4.2): net = max(0, 墙钟 − 沉默),clamp 防御(N4)。 + + stall 两值透传: None(非流式/增量 chunk<2)时 net=墙钟不退减、快照归因 + 保持 None(排除条款——None 透传天然达成,无需分支)。 + """ + wall_ms, stall_ms, stall_count = sample + net_ms = max(0, wall_ms - (stall_ms or 0)) + context.record_generation( + net_ms, accumulate=False, stall_ms=stall_ms, stall_count=stall_count + ) @staticmethod def _register_hedge( - request: ChatRequest, *, hedge_won: bool, winner_sink: list[int] | None + request: ChatRequest, *, hedge_won: bool, winner_sink: list[_GenerationSample] | None ) -> None: """对冲登记(赢家裁定后一次性;同 register_attempt 的 None 守卫)。 @@ -597,7 +622,7 @@ class RetryMW: if context is None: return if winner_sink: - context.record_generation(winner_sink[0], accumulate=False) + RetryMW._record_winner_sample(context, winner_sink[0]) context.register_hedge(hedge_won=hedge_won) async def _on_rejected( diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 536e044..63e04c0 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -320,9 +320,20 @@ class CallStats: hedges: int = 0 """本次逻辑调用实际并发发出的对冲路数(触发但准入失败静默不计);1.3.6 及以前恒 0。""" generation_ms: int = 0 - """裸生成时间: 赢家/成功那次 transport 调用的墙钟时长(口径见设计 §4.5 H8)。""" + """净生成时间(1.3.8 起重定义,设计 §4.2): 赢家/成功那次 transport 墙钟 + − Σ(超阈值中段沉默);仍含 TTFT 与尾部(末 chunk→流结束)。 + + 非流式/增量 chunk<2 无沉默观测、不退减(= 原墙钟,此时 stall 两值 None)。 + 墙钟还原 = generation_ms + stall_ms。1.3.7 及以前本字段是完整墙钟, + 口径变化见 CHANGELOG 行为变更段。""" hedge_won: bool = False """赢家是否为对冲路;无对冲恒 False。""" + stall_ms: int | None = None + """赢家路超阈值中段沉默总额(净口径归因;None = 非流式/chunk<2/无赢家)。 + + 墙钟 = generation_ms + stall_ms;None 是"无沉默观测",不是 0。""" + stall_count: int | None = None + """超阈值相邻间隔个数(归因成对报告,Gregg 原则,调研 §B2)。""" class _CallContext: @@ -340,6 +351,8 @@ class _CallContext: "_hedge_won", "_hedges", "_now", + "_stall_count", + "_stall_ms", "_started", "_terminal_claimed", "logical_call_id", @@ -353,6 +366,8 @@ class _CallContext: self._generation_ms = 0 self._hedges = 0 self._hedge_won = False + self._stall_ms: int | None = None + self._stall_count: int | None = None self._terminal_claimed = False def register_attempt(self) -> None: @@ -363,9 +378,23 @@ class _CallContext: """ self._attempts += 1 - def record_generation(self, elapsed_ms: int, *, accumulate: bool) -> None: - """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。""" + def record_generation( + self, + elapsed_ms: int, + *, + accumulate: bool, + stall_ms: int | None = None, + stall_count: int | None = None, + ) -> None: + """chat/OCR 覆盖(结构化重问最后一轮为准);embedding 分批累加。 + + stall 两值(1.3.8 净口径归因): chat 赢家路传入;embedding/ocr 既有调用 + 恒 None = "无沉默观测"(不是"沉默为 0")——它们是各自链路的墙钟,无中段 + 沉默概念。与 generation 同一次登记覆盖,不单独累加。 + """ self._generation_ms = self._generation_ms + elapsed_ms if accumulate else elapsed_ms + self._stall_ms = stall_ms + self._stall_count = stall_count def register_hedge(self, *, hedge_won: bool) -> None: """对冲路实际发出即计数;赢家裁定后一次性登记。""" @@ -381,6 +410,8 @@ class _CallContext: hedges=self._hedges, generation_ms=self._generation_ms, hedge_won=self._hedge_won, + stall_ms=self._stall_ms, + stall_count=self._stall_count, ) def claim_terminal(self) -> bool: diff --git a/tests/unit/test_hedge.py b/tests/unit/test_hedge.py index 3fafc94..7db4980 100644 --- a/tests/unit/test_hedge.py +++ b/tests/unit/test_hedge.py @@ -569,6 +569,27 @@ class TestHedgeFailureCombination: # 两败轮次同样登记对冲路数(设计 §4.5: 实际并发发出即计) assert ctx.snapshot().hedges == 2 + async def test_stall_attribution_none_when_no_winner(self): + """两败轮次无赢家(1.3.8 排除条款): generation 不记、stall 归因恒 None。 + + 对冲已实际发出(hedges 照登),但无赢家路 → 裸生成时间与沉默归因都 + 无归属;两路各自的沉默归属各自 attempt 遥测行(那是 T3 的事)。 + """ + transport = HedgeTransport( + { + "s1": [("fail_after", 0.1, lambda: TransientError("p", source_name="s1"))], + "s2": [("fail_after", 0.12, lambda: TransientError("h", source_name="s2"))], + } + ) + mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1) + ctx = _ctx() + with pytest.raises(AllSourcesExhausted): + await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + stats = ctx.snapshot() + assert stats.hedges == 1 and stats.hedge_won is False + assert stats.generation_ms == 0 + assert stats.stall_ms is None and stats.stall_count is None + async def test_both_429_refund_no_budget(self): """两路皆 429: 免预算且耗时退 stall 账——小 stall 窗下终局 stalled 而非耗尽。""" diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 62bdc41..7c9f5b3 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -107,12 +107,15 @@ class _GenClockTransport: generation_ms 的口径是"只计 transport 调用本身",故推进必须发生在被包 transport 内部;退避耗时由用例自带的 sleep 闭包推进,与本包装无关。 + `stalls`(1.3.8 计划 §3.3): 逐次自控上报的 (stall_ms, stall_count), + fake 自己报数、不经真实计时;None(默认)= 该次调用无沉默观测。 """ - def __init__(self, script, advances, clock): + def __init__(self, script, advances, clock, stalls=None): self._inner = FakeTransport(script) self._advances = list(advances) self._clock = clock + self._stalls = list(stalls) if stalls is not None else None @property def calls(self): @@ -131,6 +134,7 @@ class _GenClockTransport: content_event, ): advance = self._advances.pop(0) + stall = self._stalls.pop(0) if self._stalls is not None else None result = await self._inner.complete( messages=messages, source=source, @@ -141,6 +145,8 @@ class _GenClockTransport: first_token_event=first_token_event, content_event=content_event, ) + if stall is not None: + result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1]) self._clock.advance(advance) return result @@ -1017,6 +1023,46 @@ class TestGenerationMs: stats = ctx.snapshot() assert stats.hedges == 0 and stats.hedge_won is False + async def test_generation_ms_net_of_stall(self): + """净口径(1.3.8 设计 §4.2): generation_ms = 墙钟 − Σ(超阈值中段沉默)。 + + FakeClock 推进 5.0s 墙钟、fake 自报 stall 3200ms/2 段 → 净 1800ms; + 归因两值进快照,墙钟还原公式 generation_ms + stall_ms == 5000 成立。 + 推进量取二进制可精确表示值(同类既有用例惯例,防 int 截断抖动)。 + """ + clock = FakeClock() + transport = _GenClockTransport([_ok()], [5.0], clock, stalls=[(3200.0, 2)]) + mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport) + ctx = self._ctx(clock) + resp = await mw(dataclasses.replace(_REQ, call_context=ctx)) + assert resp.content == "ok" + stats = ctx.snapshot() + assert stats.generation_ms == 1800 + assert stats.stall_ms == 3200 and stats.stall_count == 2 + assert stats.generation_ms + stats.stall_ms == 5000 # 墙钟还原公式 + + async def test_generation_ms_clamps_negative_to_zero(self): + """沉默上报超过墙钟(异常数据/钟偏差防御) → 净口径 clamp 到 0(N4),归因原样保留。""" + clock = FakeClock() + transport = _GenClockTransport([_ok()], [5.0], clock, stalls=[(6000.0, 3)]) + mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport) + ctx = self._ctx(clock) + await mw(dataclasses.replace(_REQ, call_context=ctx)) + stats = ctx.snapshot() + assert stats.generation_ms == 0 + assert stats.stall_ms == 6000 and stats.stall_count == 3 + + async def test_non_stream_stall_none_no_deduction(self): + """stall None(非流式/增量 chunk<2,排除条款) → 净口径不退减(=墙钟),归因 None。""" + clock = FakeClock() + transport = _GenClockTransport([_ok()], [5.0], clock) # 不报 stall = 无观测 + mw, *_ = _harness([_src("a")], [], clock=clock, transport=transport) + ctx = self._ctx(clock) + await mw(dataclasses.replace(_REQ, call_context=ctx)) + stats = ctx.snapshot() + assert stats.generation_ms == 5000 + assert stats.stall_ms is None and stats.stall_count is None + class TestLogicalAttemptCounting: """尝试登记在 transport 调用**之前**(1.3.5 设计 §4)。 diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index 55bbc40..3172361 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -690,6 +690,36 @@ class TestCallStatsAndContext: stats = ctx.snapshot() assert (stats.hedges, stats.generation_ms, stats.hedge_won) == (1, 200, True) + def test_callstats_stall_fields_default_none(self): + """1.3.8 归因两字段默认 None: 仅旧参数构造不炸(1.3.7 及以前兼容形态)。""" + from polygateway.types import CallStats + + stats = CallStats(logical_call_id="lc-1", attempts=2, total_latency_ms=15) + assert stats.stall_ms is None + assert stats.stall_count is None + explicit = CallStats( + logical_call_id="lc-2", + attempts=2, + total_latency_ms=15, + stall_ms=3200, + stall_count=2, + ) + assert (explicit.stall_ms, explicit.stall_count) == (3200, 2) + + def test_callcontext_record_generation_with_stall_and_snapshot(self): + """stall 两值经 record_generation 进 snapshot;不传恒 None(embedding/ocr 形态)。""" + from polygateway.types import _CallContext + + ctx = _CallContext(now=_FakeMonotonic()) + ctx.record_generation(1800, accumulate=False, stall_ms=3200, stall_count=2) + stats = ctx.snapshot() + assert (stats.generation_ms, stats.stall_ms, stats.stall_count) == (1800, 3200, 2) + # 不传(embedding/ocr 既有调用形态): 归因恒 None,None 是"无观测"不是 0 + plain = _CallContext(now=_FakeMonotonic()) + plain.record_generation(500, accumulate=False) + plain_stats = plain.snapshot() + assert plain_stats.stall_ms is None and plain_stats.stall_count is None + def test_context_counts_attempts_and_freezes_elapsed(self): """快照是同步冻结的时间切片: 登记两次尝试后耗时按注入钟折算成毫秒。""" from polygateway.types import _CallContext