From 4c2a148db9217cef2cb2991c468a146f98c1ed89 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 21 Jul 2026 09:55:08 -0400 Subject: [PATCH] feat: feed selector health and demote in-call failed sources RetryMW keeps a per-call failure map (local, never instance state): a source failing twice in one call yields to the next candidate. Attempt outcomes feed OutcomeAwareSelector behind a swallow-and-warn guard; ResultInvalid and provider-rejected paths record success with count_attempt=False so the breaker window stays clean. Same accounting applied in EmbeddingClient. --- src/polygateway/embedding.py | 3 +- src/polygateway/middleware/breaker.py | 6 +- src/polygateway/middleware/retry.py | 45 ++++++++++--- src/polygateway/ports.py | 4 +- tests/unit/test_retry.py | 94 ++++++++++++++++++++++++++- 5 files changed, 139 insertions(+), 13 deletions(-) diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index 78d32ed..d9512be 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -303,7 +303,8 @@ class EmbeddingClient: """终态异常的门控写回(与 RetryMW 同口径): 坏结果/网关健康拒绝 ≠ 坏服务 → 记成功;网关没响应的拒绝若持探针则归还。""" if isinstance(exc, ResultInvalidError) or exc.status_code is not None: - await self._record_quietly(self._breaker.record_success(entry)) + # M2.5 §3.1: 记成功但不计失败率窗口样本(坏结果/坏请求 ≠ 坏服务) + await self._record_quietly(self._breaker.record_success(entry, count_attempt=False)) elif entry.is_probe: await self._record_quietly(self._breaker.release_probe(entry)) diff --git a/src/polygateway/middleware/breaker.py b/src/polygateway/middleware/breaker.py index 3e20a18..d737b50 100644 --- a/src/polygateway/middleware/breaker.py +++ b/src/polygateway/middleware/breaker.py @@ -25,9 +25,11 @@ class BreakerGate: except Exception as exc: raise GovernanceBackendError(f"熔断后端故障(try_enter): {exc}") from exc - async def record_success(self, entry: GateDecision) -> GateUpdate: + async def record_success( + self, entry: GateDecision, *, count_attempt: bool = True + ) -> GateUpdate: try: - return await self._gate.record_success(entry) + return await self._gate.record_success(entry, count_attempt=count_attempt) except GovernanceBackendError: raise except Exception as exc: diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index b892ac2..c12ae53 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -32,6 +32,7 @@ from polygateway.errors import ( ) from polygateway.middleware.breaker import BreakerGate from polygateway.middleware.ratelimit import QuotaGate +from polygateway.ports import OutcomeAwareSelector from polygateway.sources import SourceCooldownMemo from polygateway.streaming import StreamLivenessTimeout from polygateway.types import LLMResponse @@ -72,6 +73,16 @@ def backoff_delay( return max(delay, retry_after) +def _demote_call_failures( + ordered: list[SourceConfig], attempt_fails: dict[str, int] +) -> list[SourceConfig]: + """调用内降权(设计 §3.3): 本调用失败 ≥2 次的源移尾;全失败则保持原序。""" + demoted = [s for s in ordered if attempt_fails.get(s.name, 0) >= 2] + if not demoted or len(demoted) == len(ordered): + return ordered + return [s for s in ordered if attempt_fails.get(s.name, 0) < 2] + demoted + + def _failure_reason(exc: PolyGatewayError) -> str: """失败原因归类(CHS governance.py:169 同款)。""" if isinstance(exc, SourceDeadError): @@ -124,6 +135,8 @@ class RetryMW: self._bp = backpressure self._quota_full = quota_full self._memo = cooldown_memo or SourceCooldownMemo(now=now) + # M2.5: 选源器可选健康喂数端口,构造期 isinstance 判定一次(设计 §3.2) + self._outcome_sink = selector if isinstance(selector, OutcomeAwareSelector) else None self._emitter = emitter self._now = now self._sleep = sleep @@ -135,13 +148,15 @@ class RetryMW: raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0) fails = 0 reasons: dict[str, str] = {} + # 调用内失败计数(设计 §3.3): 局部状态,调用结束即弃;严禁实例属性(并发共享) + attempt_fails: dict[str, int] = {} entered_at = self._now() # 调用级累计计时,循环内不重置(CHS governance.py:207) while True: - picked, gate_rejections = await self._pick_runnable(reasons) + picked, gate_rejections = await self._pick_runnable(reasons, attempt_fails) if picked is None: await self._on_no_runnable(gate_rejections, reasons, entered_at) continue - outcome = await self._attempt(request, *picked, reasons) + outcome = await self._attempt(request, *picked, reasons, attempt_fails) if isinstance(outcome, LLMResponse): return outcome fails += 1 @@ -158,11 +173,12 @@ class RetryMW: # —— 选源与准入(CHS _pick_runnable 120-167)—— async def _pick_runnable( - self, reasons: dict[str, str] + self, reasons: dict[str, str], attempt_fails: dict[str, int] ) -> tuple[tuple[SourceConfig, Permit, GateDecision] | None, int]: stats = {s.name: await self._quota.stats(s) for s in self._sources} gate_rejections = 0 - for cand in self._selector.order(self._sources, stats): + ordered = _demote_call_failures(self._selector.order(self._sources, stats), attempt_fails) + for cand in ordered: if self._memo.active(cand.name): # 冷却备忘跳过也计入拒绝数,保住 circuit_open 判据(CHS 同款) gate_rejections += 1 @@ -228,6 +244,7 @@ class RetryMW: permit: Permit, entry: GateDecision, reasons: dict[str, str], + attempt_fails: dict[str, int], ) -> LLMResponse | _Failed: call_id = str(uuid.uuid4()) started = self._now() @@ -243,6 +260,7 @@ class RetryMW: actual = result.prompt_tokens + result.completion_tokens await self._record_quietly(self._breaker.record_success(entry)) await self._record_quietly(self._quota.mark_progress()) + self._feed_outcome(source.name, ok=True) response = self._build_response(source, result, call_id, started) await self._emit(request, source, call_id, started, response=response) return response @@ -251,8 +269,8 @@ class RetryMW: await self._emit(request, source, call_id, started, error=exc) raise except ResultInvalidError as exc: - # 坏结果 ≠ 坏服务: 熔断记成功,异常上抛消耗业务失败预算(§6.3) - await self._record_quietly(self._breaker.record_success(entry)) + # 坏结果 ≠ 坏服务: 熔断记成功但不计窗口样本,亦不喂健康分(M2.5 §3.1) + await self._record_quietly(self._breaker.record_success(entry, count_attempt=False)) await self._emit(request, source, call_id, started, error=exc) raise except asyncio.CancelledError: @@ -264,6 +282,8 @@ class RetryMW: dead = isinstance(exc, SourceDeadError) reason = _failure_reason(exc) reasons[source.name] = reason + attempt_fails[source.name] = attempt_fails.get(source.name, 0) + 1 + self._feed_outcome(source.name, ok=False) await self._record_quietly(self._breaker.record_failure(entry, reason, dead)) if not dead: actual = source.est_tokens # 保守: 失败请求可能已被网关计费(CHS 同款) @@ -277,8 +297,8 @@ class RetryMW: ) -> None: provider_responded = exc.source_name == source.name and exc.status_code is not None if provider_responded: - # 网关健康地拒了坏请求 - await self._record_quietly(self._breaker.record_success(entry)) + # 网关健康地拒了坏请求: 记成功但不计窗口样本(M2.5 §3.1) + await self._record_quietly(self._breaker.record_success(entry, count_attempt=False)) elif entry.is_probe: await self._record_quietly(self._breaker.release_probe(entry)) @@ -296,6 +316,15 @@ class RetryMW: except GovernanceBackendError as exc: logger.warning("治理记账写回降级(不冒泡): {}", exc) + def _feed_outcome(self, source_name: str, ok: bool) -> None: + """健康喂数降级执行: 选源器异常不得打断真实成功/失败的主路径(设计 §4)。""" + if self._outcome_sink is None: + return + try: + self._outcome_sink.record_outcome(source_name, ok) + except Exception as exc: + logger.warning("选源健康喂数失败(降级不冒泡): {}", exc) + # —— 辅助 —— def _backoff_delay(self, fails: int, exc: PolyGatewayError) -> float: diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index aa35f0a..a188e3d 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -152,7 +152,9 @@ class ProviderGate(Protocol): async def try_enter(self, source_name: str, owner: str) -> GateDecision: ... - async def record_success(self, entry: GateDecision) -> GateUpdate: ... + async def record_success( + self, entry: GateDecision, *, count_attempt: bool = True + ) -> GateUpdate: ... async def record_failure( self, entry: GateDecision, reason: str, force_open: bool diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 13a3293..52713dd 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -97,6 +97,7 @@ def _harness( quota_full="wait", global_limits=_NO_GLOBAL, rng=lambda: 0.0, + selector=None, ): clock = clock or FakeClock() limiter = InMemoryLimiter( @@ -112,7 +113,7 @@ def _harness( mw = RetryMW( scope="llm", sources=sources, - selector=RoundRobinSelector(), + selector=selector if selector is not None else RoundRobinSelector(), limiter=limiter, gate=gate, transport=transport, @@ -351,3 +352,94 @@ class TestCancellation: # 探针租约已归还: 下一 caller 立即拿到探针而非等租约过期 nxt = await gate.try_enter("a", "w2") assert nxt.allowed and nxt.is_probe + + +class StaticSelector: + """固定配置序,隔离测试调用内降权(不带 record_outcome)。""" + + def order(self, sources, stats): + return list(sources) + + +class RecordingSelector(StaticSelector): + def __init__(self): + self.outcomes = [] + + def record_outcome(self, source_name, ok): + self.outcomes.append((source_name, ok)) + + +class ExplodingSelector(StaticSelector): + def record_outcome(self, source_name, ok): + raise RuntimeError("sink boom") + + +class TestM25Orchestration: + """M2.5 设计 §3.3: 调用内失败降权 + 健康喂数(先红后绿)。""" + + async def test_failed_source_demoted_after_two_strikes(self): + # 失败 1 次仍首选(原地退避重试);失败 2 次让位次优源 + mw, _, _, transport, _, _ = _harness( + [_src("a"), _src("b")], + [TransientError("1"), TransientError("2"), _ok()], + selector=StaticSelector(), + ) + resp = await mw(_REQ) + assert [n for n, _ in transport.calls] == ["a", "a", "b"] + assert resp.source_name == "b" + + async def test_attempt_fails_reset_between_calls(self): + mw, _, _, transport, _, _ = _harness( + [_src("a"), _src("b")], + [TransientError("1"), TransientError("2"), _ok(), _ok()], + selector=StaticSelector(), + ) + await mw(_REQ) + await mw(_REQ) # 新调用状态清零: 回到首选 a + assert [n for n, _ in transport.calls] == ["a", "a", "b", "a"] + + async def test_outcome_feeding_success_and_transient(self): + sel = RecordingSelector() + mw, _, _, _, _, _ = _harness( + [_src("a"), _src("b")], [TransientError("1"), _ok()], selector=sel + ) + await mw(_REQ) + assert sel.outcomes == [("a", False), ("a", True)] + + async def test_outcome_feeding_source_dead(self): + sel = RecordingSelector() + mw, _, _, _, _, _ = _harness( + [_src("a"), _src("b")], [SourceDeadError("401"), _ok()], selector=sel + ) + await mw(_REQ) + assert sel.outcomes == [("a", False), ("b", True)] + + async def test_result_invalid_and_rejected_not_fed(self): + sel = RecordingSelector() + mw, _, _, _, _, _ = _harness( + [_src("a")], [ResultInvalidError("bad", raw_text="x")], selector=sel + ) + with pytest.raises(ResultInvalidError): + await mw(_REQ) + sel2 = RecordingSelector() + mw2, _, _, _, _, _ = _harness( + [_src("a")], + [RequestRejectedError("400", source_name="a", status_code=400)], + selector=sel2, + ) + with pytest.raises(RequestRejectedError): + await mw2(_REQ) + assert sel.outcomes == [] and sel2.outcomes == [] + + async def test_outcome_sink_exception_swallowed(self): + mw, _, _, _, _, _ = _harness([_src("a")], [_ok()], selector=ExplodingSelector()) + resp = await mw(_REQ) + assert resp.content == "ok" # 喂数异常不得打断真实成功返回 + + async def test_result_invalid_gate_window_untouched(self): + # 坏结果 ≠ 坏服务: count_attempt=False,失败率窗口 attempts 不得增长 + mw, _, gate, _, _, _ = _harness([_src("a")], [ResultInvalidError("bad", raw_text="x")]) + with pytest.raises(ResultInvalidError): + await mw(_REQ) + g = gate._gates["a"] + assert g.a0 + g.a1 == 0