diff --git a/src/polygateway/errors.py b/src/polygateway/errors.py new file mode 100644 index 0000000..57d8aa0 --- /dev/null +++ b/src/polygateway/errors.py @@ -0,0 +1,121 @@ +"""错误四分类与 scope 级不可用语义(M1 设计 §3;ARCH §6)。 + +分类决定治理行为(重试/换源/熔断计数),库内禁止绕过分类做 ad-hoc 判断。 +构造形态承 CHS `app/domain/errors.py` 的 ProviderError 一族。 +""" + +SCOPE_REASONS = frozenset( + {"circuit_open", "retry_exhausted", "stalled", "quota_exhausted", "no_sources"} +) +SOURCE_REASONS = frozenset( + {"network_error", "timeout", "rate_limited", "source_dead", "circuit_open", "cooldown"} +) + + +class PolyGatewayError(Exception): + """库内一切领域错误的基类,携带来源上下文便于遥测与日志定位。""" + + def __init__( + self, + message: str, + *, + source_name: str | None = None, + status_code: int | None = None, + operation: str | None = None, + ) -> None: + super().__init__(message) + self.source_name = source_name + self.status_code = status_code + self.operation = operation + + +class TransientError(PolyGatewayError): + """瞬时错误(超时/5xx/429/网络抖动/SSE 异常): 退避后可重试、可换源、计熔断。""" + + def __init__(self, message: str, *, retry_after_s: float | None = None, **kwargs) -> None: + super().__init__(message, **kwargs) + self.retry_after_s = retry_after_s + + +class SourceDeadError(PolyGatewayError): + """源死亡(401/403/欠费): 不重试,立即换源,该源 force_open。""" + + +class RequestRejectedError(PolyGatewayError): + """请求被拒(400/坏输入): 不重试不换源,直接上抛。""" + + +class ResultInvalidError(PolyGatewayError): + """坏结果 ≠ 坏服务: 调用成功但内容不可解析;熔断记成功,不入 transport 重试。""" + + def __init__( + self, + message: str, + *, + raw_text: str = "", + repair_error: str | None = None, + validation_errors: tuple[str, ...] = (), + **kwargs, + ) -> None: + super().__init__(message, **kwargs) + self.raw_text = raw_text + self.repair_error = repair_error + self.validation_errors = tuple(validation_errors) + + +class GatewayUnavailableError(PolyGatewayError): + """scope 级暂时不可用;业务侧 catch 本类做延期重投(CHS arq 模式)。 + + `retry_after_s` 非可选(0 = 可立即重试),承 CHS ProviderUnavailableError。 + """ + + def __init__( + self, + *, + scope: str, + reason: str, + retry_after_s: float, + per_source_reasons: dict[str, str] | None = None, + source_name: str | None = None, + ) -> None: + if not scope.strip(): + raise ValueError("scope 不能为空") + if reason not in SCOPE_REASONS: + raise ValueError(f"未知 scope 级 reason: {reason!r}(允许: {sorted(SCOPE_REASONS)})") + if retry_after_s < 0: + raise ValueError("retry_after_s 不能为负") + reasons = dict(per_source_reasons or {}) + for src, src_reason in reasons.items(): + if src_reason not in SOURCE_REASONS: + raise ValueError(f"源 {src!r} 的 reason 非法: {src_reason!r}(允许: {sorted(SOURCE_REASONS)})") + super().__init__(f"{scope.lower()} 网关暂时不可用: {reason}", source_name=source_name) + self.scope = scope.lower() + self.reason = reason + self.retry_after_s = retry_after_s + self.per_source_reasons = reasons + + +class CircuitOpenError(GatewayUnavailableError): + """全部候选源被熔断门拒绝;reason 恒为 circuit_open。""" + + def __init__( + self, + *, + scope: str, + retry_after_s: float, + per_source_reasons: dict[str, str] | None = None, + ) -> None: + super().__init__( + scope=scope, + reason="circuit_open", + retry_after_s=retry_after_s, + per_source_reasons=per_source_reasons, + ) + + +class AllSourcesExhausted(GatewayUnavailableError): # noqa: N818 — ARCH §6.1 冻结的公共名 + """重试预算耗尽 / 无可用源 / 配额 fail-fast 等 scope 级失败。""" + + +class GovernanceBackendError(PolyGatewayError): + """限流/熔断状态后端自身故障: 必须报错而非放行(防击穿网关,降级方向铁律)。""" diff --git a/src/polygateway/types.py b/src/polygateway/types.py new file mode 100644 index 0000000..80cd328 --- /dev/null +++ b/src/polygateway/types.py @@ -0,0 +1,186 @@ +"""核心冻结类型(M1 设计 §2;最内层,禁止 import 任何实现)。 + +`LLMResponse` 前 11 个字段与三参考项目逐字保序——它们的测试按位置构造 +fake,字段顺序即公共承诺;新增字段只增不删且必带默认值。 +""" + +from dataclasses import dataclass, field +from typing import Any + +_MISSING_DONE_DOMAIN = frozenset({"retry", "salvage"}) + + +@dataclass(frozen=True) +class LLMResponse: + """一次治理调用的统一响应(与三项目超集兼容,ARCH §5.1)。""" + + content: str + thinking: str + model: str + provider: str + prompt_tokens: int + completion_tokens: int + latency_ms: int + ttft_ms: float | None + max_inter_token_ms: float | None + cache_hit: bool + call_id: str + # —— 库新增(只增不删,必带默认值;迁移兼容硬约束)—— + source_name: str = "" + cost: float | None = None + usage_source: str = "measured" + structured_data: Any | None = None + + +@dataclass(frozen=True) +class ChatRequest: + """洋葱内部流转的不可变请求;中间件用 dataclasses.replace 派生,禁止原地修改。""" + + messages: list[dict[str, Any]] + session_id: str | None = None + parent_call_id: str | None = None + cache_salt: str | None = None + cache_namespace: str | None = None + structured: Any | None = None + stream: bool = True + overlay: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class Usage: + """token 用量;OCR 等无计费调用填 0。""" + + prompt_tokens: int + completion_tokens: int + usage_source: str = "measured" + + +@dataclass(frozen=True) +class SourceStats: + """限流后端回读的单源即时指标(CHS ports.py 同款)。""" + + inflight: int + rpm_used: int + tpm_used: int + + +@dataclass(frozen=True) +class TransportResult: + """transport 单次原始调用的产物;治理字段由 RetryMW 补齐为 LLMResponse。""" + + content: str + thinking: str + prompt_tokens: int + completion_tokens: int + usage_source: str + ttft_ms: float | None + max_inter_token_ms: float | None + raw: dict[str, Any] + + +@dataclass(frozen=True) +class SourceConfig: + """单个模型源的完整配置(CHS config.py 超集;不变式在构造期报错)。 + + 限额闸 0 表示不启用;`enable_thinking` 三态: None=不注入(模型默认)、 + True=注入开启参数、False=注入关闭参数(统一 VT 与 CHS 相反的现状)。 + """ + + name: str + provider: str + base_url: str + api_key: str + model: str + timeout_s: float + max_concurrency: int = 0 + rpm: int = 0 + tpm: int = 0 + est_tokens: int = 0 + ttft_timeout_s: float | None = None + inter_token_timeout_s: float | None = None + enable_thinking: bool | None = None + missing_done: str = "retry" + trust_env: bool = True + + def __post_init__(self) -> None: + self._validate_identity() + self._validate_gates() + self._validate_watchdog() + + def _validate_identity(self) -> None: + for attr in ("name", "provider", "base_url", "api_key", "model"): + if not getattr(self, attr).strip(): + raise ValueError(f"SourceConfig.{attr} 不能为空") + if self.missing_done not in _MISSING_DONE_DOMAIN: + raise ValueError(f"missing_done 必须是 {sorted(_MISSING_DONE_DOMAIN)}: {self.missing_done!r}") + + def _validate_gates(self) -> None: + if self.timeout_s <= 0: + raise ValueError("timeout_s 必须 > 0") + for attr in ("max_concurrency", "rpm", "tpm", "est_tokens"): + if getattr(self, attr) < 0: + raise ValueError(f"SourceConfig.{attr} 不能为负(0 表示不启用)") + if self.tpm > 0 and self.est_tokens <= 0: + raise ValueError("启用 TPM 闸时 est_tokens 必须 > 0(入场预扣依据)") + + def _validate_watchdog(self) -> None: + # CHS config.py:66-82: 流式看门狗成对配置且 0 < inter < ttft < timeout_s + if (self.ttft_timeout_s is None) != (self.inter_token_timeout_s is None): + raise ValueError("ttft_timeout_s 与 inter_token_timeout_s 必须同时设置或同时缺省") + if self.ttft_timeout_s is not None and not ( + 0 < self.inter_token_timeout_s < self.ttft_timeout_s < self.timeout_s + ): + raise ValueError("看门狗不变式要求 0 < inter_token < ttft < timeout_s") + + +@dataclass(frozen=True) +class RetryPolicy: + """重试策略;max_attempts = 总尝试次数(含首次,M1 设计 §2.3 统一语义)。""" + + max_attempts: int + backoff_base_s: float + backoff_max_s: float + + def __post_init__(self) -> None: + if self.max_attempts < 1: + raise ValueError("max_attempts 必须 ≥ 1(含首次尝试)") + if self.backoff_base_s <= 0 or self.backoff_max_s < self.backoff_base_s: + raise ValueError("退避参数要求 0 < backoff_base_s ≤ backoff_max_s") + + +@dataclass(frozen=True) +class BreakerConfig: + """熔断配置;probe_ttl_s 是半开探针租约时长(持有者死亡后自动回收)。""" + + fail_threshold: int + cooldown_s: float + probe_ttl_s: float + + def __post_init__(self) -> None: + if self.fail_threshold < 1 or self.cooldown_s <= 0 or self.probe_ttl_s <= 0: + raise ValueError("熔断配置要求 fail_threshold ≥ 1 且 cooldown_s/probe_ttl_s > 0") + + +@dataclass(frozen=True) +class BackpressurePolicy: + """背压配置;M1 仅使用 poll_interval_s,stall 判定 M2 启用。""" + + stall_window_s: float + poll_interval_s: float + + def __post_init__(self) -> None: + if self.stall_window_s <= 0 or self.poll_interval_s <= 0: + raise ValueError("背压参数必须 > 0") + + +@dataclass(frozen=True) +class GlobalLimits: + """scope 级全局限额;0 表示该闸不启用。""" + + max_concurrency: int + rpm: int + tpm: int + + def __post_init__(self) -> None: + if self.max_concurrency < 0 or self.rpm < 0 or self.tpm < 0: + raise ValueError("全局限额不能为负(0 表示不启用)") diff --git a/tests/unit/test_errors.py b/tests/unit/test_errors.py new file mode 100644 index 0000000..52e004f --- /dev/null +++ b/tests/unit/test_errors.py @@ -0,0 +1,85 @@ +"""errors.py 错误四分类与 scope 级不可用语义测试(M1 设计 §3)。""" + +import pytest + +from polygateway.errors import ( + AllSourcesExhausted, + CircuitOpenError, + GatewayUnavailableError, + GovernanceBackendError, + PolyGatewayError, + RequestRejectedError, + ResultInvalidError, + SourceDeadError, + TransientError, +) + + +class TestBaseShape: + def test_base_carries_source_context(self): + exc = PolyGatewayError("boom", source_name="qwen_1", status_code=502, operation="chat") + assert exc.source_name == "qwen_1" + assert exc.status_code == 502 + assert exc.operation == "chat" + + def test_four_way_taxonomy_inherits_base(self): + for cls in (TransientError, SourceDeadError, RequestRejectedError, ResultInvalidError): + assert issubclass(cls, PolyGatewayError) + + def test_transient_retry_after_optional(self): + assert TransientError("429").retry_after_s is None + assert TransientError("429", retry_after_s=2.5).retry_after_s == 2.5 + + +class TestResultInvalid: + def test_carries_diagnosis(self): + exc = ResultInvalidError( + "bad json", raw_text="{oops", repair_error="unterminated", + validation_errors=("field x missing",), + ) + assert exc.raw_text == "{oops" + assert exc.repair_error == "unterminated" + assert exc.validation_errors == ("field x missing",) + + +class TestGatewayUnavailable: + def test_fields_and_inheritance(self): + exc = AllSourcesExhausted( + scope="LLM", reason="retry_exhausted", retry_after_s=4.0, + per_source_reasons={"qwen_1": "timeout"}, + ) + assert isinstance(exc, GatewayUnavailableError) + assert exc.scope == "llm" # CHS 同款: scope 归一化小写 + assert exc.reason == "retry_exhausted" + assert exc.retry_after_s == 4.0 + assert exc.per_source_reasons == {"qwen_1": "timeout"} + + def test_circuit_open_reason_fixed(self): + exc = CircuitOpenError(scope="LLM", retry_after_s=30.0) + assert exc.reason == "circuit_open" + assert isinstance(exc, GatewayUnavailableError) + + def test_scope_reason_domain_enforced(self): + with pytest.raises(ValueError): + AllSourcesExhausted(scope="LLM", reason="bad_reason", retry_after_s=0.0) + + def test_per_source_reason_domain_enforced(self): + with pytest.raises(ValueError): + AllSourcesExhausted( + scope="LLM", reason="no_sources", retry_after_s=0.0, + per_source_reasons={"qwen_1": "weird"}, + ) + + def test_retry_after_non_negative_and_scope_non_empty(self): + with pytest.raises(ValueError): + AllSourcesExhausted(scope="LLM", reason="stalled", retry_after_s=-1.0) + with pytest.raises(ValueError): + AllSourcesExhausted(scope=" ", reason="stalled", retry_after_s=0.0) + + +class TestBackendFailure: + def test_governance_backend_error_is_not_transient(self): + """限流/熔断后端故障必须报错不放行,且不落入可重试分类。""" + exc = GovernanceBackendError("redis down") + assert isinstance(exc, PolyGatewayError) + assert not isinstance(exc, TransientError) diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py new file mode 100644 index 0000000..c258255 --- /dev/null +++ b/tests/unit/test_types.py @@ -0,0 +1,150 @@ +"""types.py 冻结签名的行为测试(M1 设计 §2)。""" + +import dataclasses + +import pytest + +from polygateway.types import ( + BackpressurePolicy, + BreakerConfig, + ChatRequest, + GlobalLimits, + LLMResponse, + RetryPolicy, + SourceConfig, + TransportResult, + Usage, +) + + +def _make_source(**overrides): + """构造最小合法 SourceConfig,单点覆盖便于逐条触发不变式。""" + base = dict( + name="qwen_1", + provider="qwen", + base_url="https://gw.example/v1", + api_key="sk-test", + model="qwen-max", + timeout_s=120.0, + ) + base.update(overrides) + return SourceConfig(**base) + + +class TestLLMResponse: + def test_eleven_legacy_fields_positional(self): + """三项目 fake 的 11 参位置构造必须零改动成立(迁移兼容硬约束)。""" + resp = LLMResponse( + "content", "thinking", "qwen-max", "qwen", 10, 20, 300, 50.0, 80.0, False, "cid" + ) + assert resp.content == "content" + assert resp.call_id == "cid" + + def test_new_fields_have_defaults(self): + resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid") + assert resp.source_name == "" + assert resp.cost is None + assert resp.usage_source == "measured" + assert resp.structured_data is None + + def test_frozen(self): + resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid") + with pytest.raises(dataclasses.FrozenInstanceError): + resp.content = "x" + + +class TestChatRequest: + def test_defaults_and_frozen(self): + req = ChatRequest(messages=[{"role": "user", "content": "hi"}]) + assert req.session_id is None + assert req.parent_call_id is None + assert req.cache_salt is None + assert req.cache_namespace is None + assert req.structured is None + assert req.stream is True + assert req.overlay == {} + with pytest.raises(dataclasses.FrozenInstanceError): + req.stream = False + + def test_replace_derivation(self): + """中间件用 dataclasses.replace 派生新请求(设计 §4.1 方案 A1)。""" + req = ChatRequest(messages=[{"role": "user", "content": "hi"}]) + derived = dataclasses.replace(req, overlay={"response_format": {"type": "json_object"}}) + assert derived.overlay and req.overlay == {} + + +class TestSourceConfig: + def test_minimal_valid(self): + src = _make_source() + assert src.max_concurrency == 0 and src.rpm == 0 and src.tpm == 0 + assert src.est_tokens == 0 + assert src.enable_thinking is None + assert src.missing_done == "retry" + assert src.trust_env is True + + @pytest.mark.parametrize("field", ["name", "provider", "base_url", "api_key", "model"]) + def test_identity_fields_must_be_non_empty(self, field): + with pytest.raises(ValueError): + _make_source(**{field: " "}) + + def test_timeout_must_be_positive(self): + with pytest.raises(ValueError): + _make_source(timeout_s=0) + + def test_tpm_requires_est_tokens(self): + with pytest.raises(ValueError): + _make_source(tpm=10000, est_tokens=0) + assert _make_source(tpm=10000, est_tokens=800).est_tokens == 800 + + def test_negative_gate_rejected(self): + with pytest.raises(ValueError): + _make_source(rpm=-1) + + def test_watchdog_invariant_chain(self): + """CHS config.py:66-82: 0 < inter < ttft < timeout_s。""" + ok = _make_source(ttft_timeout_s=30.0, inter_token_timeout_s=15.0) + assert ok.ttft_timeout_s == 30.0 + with pytest.raises(ValueError): + _make_source(ttft_timeout_s=30.0, inter_token_timeout_s=40.0) + with pytest.raises(ValueError): + _make_source(ttft_timeout_s=200.0, inter_token_timeout_s=15.0) # ttft >= timeout + with pytest.raises(ValueError): + _make_source(ttft_timeout_s=30.0) # 只设其一 + + def test_missing_done_domain(self): + assert _make_source(missing_done="salvage").missing_done == "salvage" + with pytest.raises(ValueError): + _make_source(missing_done="ignore") + + +class TestResilienceConfigs: + def test_retry_policy_validation(self): + assert RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0) + with pytest.raises(ValueError): + RetryPolicy(max_attempts=0, backoff_base_s=2.0, backoff_max_s=30.0) + with pytest.raises(ValueError): + RetryPolicy(max_attempts=3, backoff_base_s=30.0, backoff_max_s=2.0) + + def test_breaker_config_validation(self): + assert BreakerConfig(fail_threshold=5, cooldown_s=60.0, probe_ttl_s=120.0) + with pytest.raises(ValueError): + BreakerConfig(fail_threshold=0, cooldown_s=60.0, probe_ttl_s=120.0) + + def test_backpressure_and_global_limits(self): + assert BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.5) + with pytest.raises(ValueError): + BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0) + assert GlobalLimits(max_concurrency=8, rpm=60, tpm=100000) + with pytest.raises(ValueError): + GlobalLimits(max_concurrency=-1, rpm=0, tpm=0) + + +class TestAuxTypes: + def test_usage_and_stats(self): + u = Usage(prompt_tokens=10, completion_tokens=20, usage_source="estimated") + assert u.prompt_tokens == 10 + s = TransportResult( + content="c", thinking="", prompt_tokens=1, completion_tokens=2, + usage_source="measured", ttft_ms=12.5, max_inter_token_ms=30.0, raw={"id": "x"}, + ) + assert s.raw["id"] == "x"