From 7b9815f4bc97a7c85533b0454065eda1786c608e Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 20 Jul 2026 07:47:05 -0400 Subject: [PATCH] feat: add gateway client with env-driven assembly Includes config aggregation for multi-source env keys, from_env and from_settings factories with explicit shared-backend injection, gather_bounded, top-level exports, tightened import-linter layers with the gate removed from the Makefile, and the finalized .env.example. --- .env.example | 69 +++-- Makefile | 9 +- pyproject.toml | 7 +- src/polygateway/__init__.py | 44 ++- src/polygateway/backends/memory/breaker.py | 47 ++- src/polygateway/backends/memory/limiter.py | 4 +- src/polygateway/client.py | 302 +++++++++++++++++++ src/polygateway/config.py | 313 ++++++++++++++++++++ src/polygateway/errors.py | 4 +- src/polygateway/middleware/breaker.py | 4 +- src/polygateway/middleware/cache.py | 8 +- src/polygateway/middleware/retry.py | 49 ++- src/polygateway/middleware/telemetry.py | 12 +- src/polygateway/structured/json_repair.py | 4 +- src/polygateway/telemetry/sqlite.py | 22 +- src/polygateway/transports/openai_compat.py | 74 +++-- src/polygateway/types.py | 4 +- tests/contracts/conftest.py | 19 +- tests/unit/test_cache.py | 58 +++- tests/unit/test_client.py | 256 ++++++++++++++++ tests/unit/test_config.py | 166 +++++++++++ tests/unit/test_errors.py | 12 +- tests/unit/test_openai_compat.py | 59 ++-- tests/unit/test_ports.py | 102 +++++-- tests/unit/test_retry.py | 91 ++++-- tests/unit/test_sources.py | 8 +- tests/unit/test_streaming.py | 4 +- tests/unit/test_structured.py | 26 +- tests/unit/test_telemetry.py | 151 +++++++--- tests/unit/test_types.py | 26 +- 30 files changed, 1701 insertions(+), 253 deletions(-) create mode 100644 src/polygateway/client.py create mode 100644 src/polygateway/config.py create mode 100644 tests/unit/test_client.py create mode 100644 tests/unit/test_config.py diff --git a/.env.example b/.env.example index d3a4291..b2d64b6 100644 --- a/.env.example +++ b/.env.example @@ -1,27 +1,54 @@ -# PolyGateway 工程配置模板（复制为 .env 使用；.env 不提交） -# 键名清单在 M1 设计文档定稿后补全；以下为已定的韧性参数键名风格 -# （沿用三个参考项目的习惯，降低迁移改名成本，见 ARCHITECTURE.md §9）。 +# PolyGateway 工程配置模板(复制为 .env 使用;.env 不提交)。 +# 键名清单 = M1 设计文档 §8 定稿;缺关键配置直接报错,不做默认值兜底。 -# ── 多源配置（{SCOPE}__{PROVIDER}__{N}__{FIELD}）── -# LLM__QWEN__1__BASE_URL= -# LLM__QWEN__1__API_KEY= -# LLM__QWEN__1__MODEL= +# ══ 多源配置: {SCOPE}__{PROVIDER}__{N}__{FIELD} ══ +# PROVIDER 必须是注册表键(qwen/deepseek/openai,或 register_provider 注册后经 registry 传入)。 +# 必填: BASE_URL / API_KEY / MODEL / TIMEOUT_S(或用平铺 LLM_TIMEOUT 作缺省)。 +LLM__QWEN__1__BASE_URL= +LLM__QWEN__1__API_KEY= +LLM__QWEN__1__MODEL= +LLM__QWEN__1__TIMEOUT_S=120 +# 可选(0 = 该闸不启用;TPM > 0 时 EST_TOKENS 必填 > 0): +# LLM__QWEN__1__MAX_CONCURRENCY=8 +# LLM__QWEN__1__RPM=60 +# LLM__QWEN__1__TPM=100000 +# LLM__QWEN__1__EST_TOKENS=2000 +# LLM__QWEN__1__TTFT_TIMEOUT_S=30 # 须与 INTER_TOKEN 成对;0 < inter < ttft < timeout +# LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S=15 +# LLM__QWEN__1__ENABLE_THINKING=true # 三态: 缺省=不注入 / true=注入开启 / false=注入关闭 +# LLM__QWEN__1__MISSING_DONE=retry # SSE 缺 [DONE]: retry(默认) | salvage +# LLM__QWEN__1__TRUST_ENV=true # false = 绕过本地代理(LAN 直连) -# ── 韧性参数 ── -# LLM_TIMEOUT=120 -# LLM_MAX_RETRIES=3 -# LLM_RETRY_BASE_DELAY=2.0 -# LLM_RETRY_MAX_DELAY=30.0 -# LLM_CIRCUIT_BREAKER_THRESHOLD=48 -# LLM_CIRCUIT_BREAKER_COOLDOWN=60 -# LLM_TTFT_TIMEOUT=30 +# ══ scope 级全局闸(跨源合计;0/缺省 = 不启用)══ +# LLM__GLOBAL__MAX_CONCURRENCY=8 +# LLM__GLOBAL__RPM=120 +# LLM__GLOBAL__TPM=200000 + +# ══ 韧性参数(平铺键 = 单 scope 简写,沿用三项目习惯;scope 键优先)══ +LLM_MAX_RETRIES=3 # 总尝试次数(含首次);或 LLM__RETRY__MAX_ATTEMPTS +LLM_RETRY_BASE_DELAY=2.0 # 或 LLM__RETRY__BACKOFF_BASE_S +LLM_RETRY_MAX_DELAY=30.0 # 或 LLM__RETRY__BACKOFF_MAX_S +LLM_CIRCUIT_BREAKER_THRESHOLD=5 # 有效阈值自动取 max(此值, 并发×2);或 LLM__BREAKER__FAIL_THRESHOLD +LLM_CIRCUIT_BREAKER_COOLDOWN=60 # 或 LLM__BREAKER__COOLDOWN_S +# LLM_TIMEOUT=120 # 源缺 TIMEOUT_S 时的缺省 +# LLM_TTFT_TIMEOUT=30 # 平铺看门狗缺省(成对生效) # LLM_INTER_TOKEN_TIMEOUT=15 +# LLM__BREAKER__PROBE_TTL_S=240 # 缺省派生: max(2×最大源超时, cooldown) +# LLM__BACKPRESSURE__STALL_WINDOW_S=300 # M2 启用 stall 判定 +# LLM__BACKPRESSURE__POLL_INTERVAL_S=0.05 +# LLM__SELECTOR=round_robin # round_robin(默认) | least_inflight +# LLM__QUOTA_FULL=wait # wait(默认) | fail_fast -# ── 后端选择（命名 M1 定稿）── -# PGW_LIMITER_BACKEND=memory -# PGW_TELEMETRY_BACKEND=sqlite -# PGW_QUOTA_FULL=wait +# ══ 装配选择(PGW_*)══ +PGW_LIMITER_BACKEND=memory # M1 仅 memory;redis 随 M2 +PGW_BREAKER_BACKEND=memory +PGW_CACHE_BACKEND=none # redis | memory | none(必填,显式优于隐式) +PGW_TELEMETRY_BACKEND=none # sqlite | none(postgres 随 M2) +# PGW_TELEMETRY_SQLITE_PATH=logs/telemetry.db # sqlite 时必填 +# PGW_CACHE_NAMESPACE=<项目名或租户前缀> # 缓存启用时必填(防跨项目毒化) +# PGW_CACHE_TTL_S=604800 # 缓存启用时必填,须 > 0 +# PGW_STRUCTURED_MAX_RETRIES=1 # 0 = 解析失败不重问(CHS 策略) +# PGW_LEASE_TTL_S=1500 # permit 租约;须 ≥ 最大源 timeout -# ── Redis（缓存 / 分布式限流熔断）── +# ══ Redis(缓存;M2 起亦供分布式限流/熔断)══ # REDIS_URL=redis://localhost:6379/0 -# REDIS_CACHE_TTL=86400 diff --git a/Makefile b/Makefile index a7d8e8d..5ebf1e7 100644 --- a/Makefile +++ b/Makefile @@ -8,12 +8,9 @@ install: test: conda run -n $(ENV) pytest tests/ --cov=src/polygateway --cov-report=term-missing -# lint-imports 在 M1 模块落地前门控跳过（契约见 pyproject.toml [tool.importlinter]） lint: conda run -n $(ENV) ruff check src/ tests/ --fix - @conda run -n $(ENV) python -c "import polygateway.ports" 2>/dev/null \ - && conda run -n $(ENV) lint-imports \ - || echo "import-linter: M1 模块未创建，跳过（契约已在 pyproject.toml 声明）" + conda run -n $(ENV) lint-imports format: conda run -n $(ENV) ruff format src/ tests/ @@ -21,9 +18,7 @@ format: check: conda run -n $(ENV) ruff format --check src/ tests/ conda run -n $(ENV) ruff check src/ tests/ - @conda run -n $(ENV) python -c "import polygateway.ports" 2>/dev/null \ - && conda run -n $(ENV) lint-imports \ - || echo "import-linter: M1 模块未创建，跳过（契约已在 pyproject.toml 声明）" + conda run -n $(ENV) lint-imports ci: check test diff --git a/pyproject.toml b/pyproject.toml index 97b063a..c90847d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,10 +60,13 @@ known-first-party = ["polygateway"] root_packages = ["polygateway"] [[tool.importlinter.contracts]] -name = "洋葱分层：client → 实现层 → 内核（ports/types/errors/streaming）" +name = "洋葱分层：client → config → middleware → 实现层（互不依赖）→ 叶子 → 内核" type = "layers" layers = [ "polygateway.client", - "polygateway.middleware : polygateway.transports : polygateway.backends : polygateway.telemetry : polygateway.structured : polygateway.providers : polygateway.sources", + "polygateway.config", + "polygateway.middleware", + "polygateway.transports | polygateway.backends | polygateway.telemetry | polygateway.structured", + "polygateway.providers : polygateway.sources", "polygateway.ports : polygateway.types : polygateway.errors : polygateway.streaming", ] diff --git a/src/polygateway/__init__.py b/src/polygateway/__init__.py index 97ea021..14bd9fd 100644 --- a/src/polygateway/__init__.py +++ b/src/polygateway/__init__.py @@ -1,7 +1,45 @@ -"""PolyGateway：实验室统一的大语言模型（LLM/VLM/OCR）调度与中转库。 +"""PolyGateway: 实验室统一的大语言模型(LLM/VLM/OCR)调度与中转库。 -治理单位是一次模型调用：多源选择、限流、错误分类与重试、熔断、响应缓存、 -流式活性看门狗、遥测（含成本）。架构单一事实源见 research-wiki/ARCHITECTURE.md。 +治理单位是一次模型调用: 多源选择、限流、错误分类与重试、熔断、响应缓存、 +流式活性看门狗、遥测。架构单一事实源见 research-wiki/ARCHITECTURE.md。 +顶层导出即公共 API 面: 错误四分类必须从这里引用(ARCH §5.1 约定②)。 """ +from polygateway.client import GatewayClient, gather_bounded +from polygateway.config import GatewaySettings +from polygateway.errors import ( + AllSourcesExhausted, + CircuitOpenError, + GatewayUnavailableError, + GovernanceBackendError, + PolyGatewayError, + RequestRejectedError, + ResultInvalidError, + SourceDeadError, + TransientError, +) +from polygateway.providers import DEFAULT_PROFILES, ProviderProfile, register_provider +from polygateway.types import LLMResponse, SourceConfig + __version__ = "0.1.0" + +__all__ = [ + "DEFAULT_PROFILES", + "AllSourcesExhausted", + "CircuitOpenError", + "GatewayClient", + "GatewaySettings", + "GatewayUnavailableError", + "GovernanceBackendError", + "LLMResponse", + "PolyGatewayError", + "ProviderProfile", + "RequestRejectedError", + "ResultInvalidError", + "SourceConfig", + "SourceDeadError", + "TransientError", + "__version__", + "gather_bounded", + "register_provider", +] diff --git a/src/polygateway/backends/memory/breaker.py b/src/polygateway/backends/memory/breaker.py index 129669c..e9f3fbf 100644 --- a/src/polygateway/backends/memory/breaker.py +++ b/src/polygateway/backends/memory/breaker.py @@ -51,8 +51,13 @@ class InMemoryGate: g.probe_owner = owner g.probe_expires = self._now() + self._cfg.probe_ttl_s return GateDecision( - source_name=source_name, allowed=True, state=GateState.HALF_OPEN, - epoch=g.epoch, is_probe=True, probe_owner=owner, retry_after_s=0.0, + source_name=source_name, + allowed=True, + state=GateState.HALF_OPEN, + epoch=g.epoch, + is_probe=True, + probe_owner=owner, + retry_after_s=0.0, ) async def try_enter(self, source_name: str, owner: str) -> GateDecision: @@ -63,23 +68,36 @@ class InMemoryGate: now = self._now() if g.state is GateState.CLOSED: return GateDecision( - source_name=source_name, allowed=True, state=GateState.CLOSED, - epoch=g.epoch, is_probe=False, probe_owner=None, retry_after_s=0.0, + source_name=source_name, + allowed=True, + state=GateState.CLOSED, + epoch=g.epoch, + is_probe=False, + probe_owner=None, + retry_after_s=0.0, ) if g.state is GateState.OPEN: if now >= g.open_until: return self._grant_probe(g, source_name, owner) return GateDecision( - source_name=source_name, allowed=False, state=GateState.OPEN, - epoch=g.epoch, is_probe=False, probe_owner=None, + source_name=source_name, + allowed=False, + state=GateState.OPEN, + epoch=g.epoch, + is_probe=False, + probe_owner=None, retry_after_s=g.open_until - now, ) # HALF_OPEN: 探针在途;租约过期则接管,否则拒绝(防惊群) if now >= g.probe_expires: return self._grant_probe(g, source_name, owner) return GateDecision( - source_name=source_name, allowed=False, state=GateState.HALF_OPEN, - epoch=g.epoch, is_probe=False, probe_owner=None, + source_name=source_name, + allowed=False, + state=GateState.HALF_OPEN, + epoch=g.epoch, + is_probe=False, + probe_owner=None, retry_after_s=g.probe_expires - now, ) @@ -96,8 +114,13 @@ class InMemoryGate: def _snapshot(self, g: _SourceGate, applied: bool) -> GateUpdate: return GateUpdate( - applied=applied, state=g.state, epoch=g.epoch, failure_count=g.fails, - retry_after_s=max(0.0, g.open_until - self._now()) if g.state is GateState.OPEN else 0.0, + applied=applied, + state=g.state, + epoch=g.epoch, + failure_count=g.fails, + retry_after_s=max(0.0, g.open_until - self._now()) + if g.state is GateState.OPEN + else 0.0, ) def _open(self, g: _SourceGate, reason: str) -> None: @@ -118,7 +141,9 @@ class InMemoryGate: g.probe_expires = 0.0 return self._snapshot(g, applied=True) - async def record_failure(self, entry: GateDecision, reason: str, force_open: bool) -> GateUpdate: + async def record_failure( + self, entry: GateDecision, reason: str, force_open: bool + ) -> GateUpdate: g = self._gate(entry.source_name) if not self._fenced(g, entry): return self._snapshot(g, applied=False) diff --git a/src/polygateway/backends/memory/limiter.py b/src/polygateway/backends/memory/limiter.py index 6e124ce..278bdd8 100644 --- a/src/polygateway/backends/memory/limiter.py +++ b/src/polygateway/backends/memory/limiter.py @@ -27,7 +27,9 @@ _WINDOW_S = 60.0 class _MemoryPermit: """入场许可;release/settle 幂等(CHS _RedisPermit 同款 flag 语义)。""" - def __init__(self, limiter: InMemoryLimiter, source: str, lease_id: str, est: int, window: int) -> None: + def __init__( + self, limiter: InMemoryLimiter, source: str, lease_id: str, est: int, window: int + ) -> None: self._limiter = limiter self._source = source self._lease_id = lease_id diff --git a/src/polygateway/client.py b/src/polygateway/client.py new file mode 100644 index 0000000..ce9fb3e --- /dev/null +++ b/src/polygateway/client.py @@ -0,0 +1,302 @@ +"""GatewayClient: 唯一组装层(D1 组装点)+ from_env/from_settings 工厂 + gather_bounded。 + +90% 用户三行起步: `client = GatewayClient.from_env(); resp = await client.chat(messages)`。 +多逻辑角色 = 每角色调一次 `from_env(scope=...)`;共享限流/熔断状态 = 自建 +后端实例(源表取并集)显式传入多次调用——共享必须显式,禁止隐式全局 +(VT `evolve_llm = llm` 教训;ARCH §7.7 R5)。 +""" + +from __future__ import annotations + +import asyncio +import random +import time +from typing import TYPE_CHECKING, Any, Literal, TypeVar + +from polygateway.backends.memory.breaker import InMemoryGate +from polygateway.backends.memory.cache import InMemoryCache +from polygateway.backends.memory.limiter import InMemoryLimiter +from polygateway.config import GatewaySettings +from polygateway.middleware.base import compose +from polygateway.middleware.cache import CacheMW +from polygateway.middleware.retry import RetryMW +from polygateway.middleware.structured import StructuredMW +from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW +from polygateway.providers import get_provider +from polygateway.sources import LeastInflightSelector, RoundRobinSelector, SourceCooldownMemo +from polygateway.transports.openai_compat import OpenAICompatTransport +from polygateway.types import ChatRequest, LLMResponse + +if TYPE_CHECKING: + from collections.abc import Awaitable, Iterable, Mapping + + from pydantic import BaseModel + + from polygateway.ports import ( + CacheBackend, + Middleware, + ProviderGate, + RateLimiter, + SourceSelector, + StructuredOutputStrategy, + TelemetryRecorder, + Transport, + ) + from polygateway.providers import ProviderProfile + from polygateway.types import ( + BackpressurePolicy, + RetryPolicy, + SourceConfig, + ) + +_T = TypeVar("_T") + + +class GatewayClient: + """统一治理入口;构造函数全量注入(测试/高级),工厂覆盖 90% 场景。""" + + def __init__( + self, + *, + scope: str, + sources: list[SourceConfig], + selector: SourceSelector, + limiter: RateLimiter, + gate: ProviderGate, + transport: Transport, + retry: RetryPolicy, + backpressure: BackpressurePolicy, + quota_full: str = "wait", + telemetry: TelemetryRecorder | None = None, + cache: CacheBackend | None = None, + cache_namespace: str | None = None, + cache_ttl_s: int | None = None, + structured_strategy: StructuredOutputStrategy | None = None, + structured_escalation: StructuredOutputStrategy | None = None, + structured_max_retries: int = 1, + now: Any = time.monotonic, + sleep: Any = asyncio.sleep, + rng: Any = random.random, + ) -> None: + emitter = TelemetryEmitter(telemetry) if telemetry is not None else None + terminal = RetryMW( + scope=scope, + sources=sources, + selector=selector, + limiter=limiter, + gate=gate, + transport=transport, + retry=retry, + backpressure=backpressure, + quota_full=quota_full, + cooldown_memo=SourceCooldownMemo(now=now), + emitter=emitter, + now=now, + sleep=sleep, + rng=rng, + ) + middlewares: list[Middleware] = [] + if emitter is not None: + middlewares.append(TelemetryMW(emitter, now=now)) + if cache is not None: + if cache_namespace is None or cache_ttl_s is None: + raise ValueError("启用缓存必须提供 cache_namespace 与 cache_ttl_s") + # 多源 scope 的 key 身份 = 排序去重的 model 合集;源集合变化 → 一次性冷启动 + fingerprint = ",".join(sorted({s.model for s in sources})) + middlewares.append( + CacheMW( + backend=cache, + model_fingerprint=fingerprint, + default_namespace=cache_namespace, + ttl_s=cache_ttl_s, + strategy=structured_strategy, + ) + ) + if structured_strategy is not None: + middlewares.append( + StructuredMW( + strategy=structured_strategy, + max_retries=structured_max_retries, + escalation=structured_escalation, + ) + ) + self._structured_available = structured_strategy is not None + self._handler = compose(middlewares, terminal) + self._transport = transport + self._telemetry = telemetry + self._cache = cache + self._closed = False + + async def chat( + self, + 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: type[BaseModel] | Literal["json"] | None = None, + stream: bool = True, + ) -> LLMResponse: + """一次治理调用(签名冻结,ARCH §5.2;与三项目 LLMProvider 协议兼容)。""" + if structured is not None and not self._structured_available: + raise ImportError( + "结构化输出未启用: 安装 pip install 'polygateway[structured]' 后重新装配" + ) + request = ChatRequest( + messages=messages, + session_id=session_id, + parent_call_id=parent_call_id, + cache_salt=cache_salt, + cache_namespace=cache_namespace, + structured=structured, + stream=stream, + ) + return await self._handler(request) + + async def aclose(self) -> None: + """幂等释放: transport 连接池、遥测连接、缓存客户端。""" + if self._closed: + return + self._closed = True + transport_aclose = getattr(self._transport, "aclose", None) + if transport_aclose is not None: + await transport_aclose() + telemetry_close = getattr(self._telemetry, "close", None) + if telemetry_close is not None: + telemetry_close() + cache_aclose = getattr(self._cache, "aclose", None) + if cache_aclose is not None: + await cache_aclose() + + async def __aenter__(self) -> GatewayClient: + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.aclose() + + # —— 工厂 —— + + @classmethod + def from_settings( + cls, + settings: GatewaySettings, + *, + limiter: RateLimiter | None = None, + gate: ProviderGate | None = None, + cache: CacheBackend | None = None, + telemetry: TelemetryRecorder | None = None, + registry: Mapping[str, ProviderProfile] | None = None, + ) -> GatewayClient: + """按配置装配;显式传入的后端实例即共享(None 项按配置自建私有实例)。""" + sources = list(settings.sources) + profiles = [get_provider(s.provider, registry=registry) for s in sources] + strategy, escalation = _build_structured(profiles) + return cls( + scope=settings.scope, + sources=sources, + selector=_build_selector(settings.selector), + limiter=limiter + or InMemoryLimiter( + scope=settings.scope, + sources={s.name: s for s in sources}, + global_limits=settings.global_limits, + lease_ttl_s=settings.lease_ttl_s, + ), + gate=gate or InMemoryGate(config=settings.breaker), + transport=OpenAICompatTransport(registry=registry), + retry=settings.retry, + backpressure=settings.backpressure, + quota_full=settings.quota_full, + telemetry=telemetry if telemetry is not None else _build_telemetry(settings), + cache=cache if cache is not None else _build_cache(settings), + cache_namespace=settings.cache_namespace, + cache_ttl_s=settings.cache_ttl_s, + structured_strategy=strategy, + structured_escalation=escalation, + structured_max_retries=settings.structured_max_retries, + ) + + @classmethod + def from_env( + cls, + scope: str = "LLM", + *, + limiter: RateLimiter | None = None, + gate: ProviderGate | None = None, + cache: CacheBackend | None = None, + telemetry: TelemetryRecorder | None = None, + registry: Mapping[str, ProviderProfile] | None = None, + env: Mapping[str, str] | None = None, + ) -> GatewayClient: + """从 .env/环境变量装配一个 scope 的 client(键名清单见 .env.example)。""" + return cls.from_settings( + GatewaySettings.from_env(scope, env=env), + limiter=limiter, + gate=gate, + cache=cache, + telemetry=telemetry, + registry=registry, + ) + + +def _build_selector(name: str) -> SourceSelector: + return RoundRobinSelector() if name == "round_robin" else LeastInflightSelector() + + +def _build_cache(settings: GatewaySettings) -> CacheBackend | None: + if settings.cache_backend == "none": + return None + if settings.cache_backend == "memory": + return InMemoryCache() + from polygateway.backends.redis_cache import RedisCache + + assert settings.redis_url is not None # 内部不变量: config 已校验 + return RedisCache.from_url(settings.redis_url) + + +def _build_telemetry(settings: GatewaySettings) -> TelemetryRecorder | None: + if settings.telemetry_backend == "none": + return None + from polygateway.telemetry.sqlite import SQLiteRecorder + + assert settings.telemetry_sqlite_path is not None # 内部不变量: config 已校验 + return SQLiteRecorder(settings.telemetry_sqlite_path) + + +def _build_structured( + profiles: list[ProviderProfile], +) -> tuple[StructuredOutputStrategy | None, StructuredOutputStrategy | None]: + """按注册表能力选策略(设计 §5): 全员支持原生 schema 才用 NativeSchema。 + + json_repair 缺失(未装 structured extra)→ 返回 (None, None), + chat(structured=...) 时显式报缺 extra,绝不静默跳过校验。 + """ + try: + from polygateway.structured.json_repair import JsonRepairStrategy + from polygateway.structured.native_schema import NativeSchemaStrategy + except ImportError: + return None, None + try: + native_all = all(p.supports_native_schema for p in profiles) + if native_all: + return NativeSchemaStrategy(), NativeSchemaStrategy() + return JsonRepairStrategy(), None + except ImportError: + return None, None + + +async def gather_bounded(aws: Iterable[Awaitable[_T]], *, concurrency: int) -> list[_T]: + """有界并发 gather(D5 便利函数,替代 VT 手搓 semaphore+gather 样板)。 + + 语义与 `asyncio.gather` 默认一致: 结果保序、首个异常上抛;仅增加并发上限。 + """ + if concurrency < 1: + raise ValueError("concurrency 必须 ≥ 1") + sem = asyncio.Semaphore(concurrency) + + async def _run(aw: Awaitable[_T]) -> _T: + async with sem: + return await aw + + return await asyncio.gather(*(_run(aw) for aw in aws)) diff --git a/src/polygateway/config.py b/src/polygateway/config.py new file mode 100644 index 0000000..92662b8 --- /dev/null +++ b/src/polygateway/config.py @@ -0,0 +1,313 @@ +"""配置聚合与装配期校验(M1 设计 §8;ARCH §9)。 + +键族: 多源 `{SCOPE}__{PROVIDER}__{N}__{FIELD}`、scope 韧性键、平铺 LLM_* +简写(单 scope 迁移零改名)、PGW_* 装配键。缺关键配置直接报错,严禁默认 +值兜底;scope 键与平铺键并存时 scope 键优先。 + +实现说明: 多源键族是动态键名,pydantic-settings 的静态字段模型无法表达, +故用其底层 python-dotenv 读 `.env` 并与 os.environ 合并(环境变量优先), +fail-loud 校验语义与 pydantic-settings 一致。 +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from dotenv import dotenv_values + +from polygateway.types import ( + BackpressurePolicy, + BreakerConfig, + GlobalLimits, + RetryPolicy, + SourceConfig, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + +# FIELD → (SourceConfig 属性, 类型);CHS config.py:95-104 全集 + M1 新增 +_SOURCE_FIELDS: dict[str, tuple[str, str]] = { + "BASE_URL": ("base_url", "str"), + "API_KEY": ("api_key", "str"), + "MODEL": ("model", "str"), + "MAX_CONCURRENCY": ("max_concurrency", "int"), + "RPM": ("rpm", "int"), + "TPM": ("tpm", "int"), + "EST_TOKENS": ("est_tokens", "int"), + "TIMEOUT_S": ("timeout_s", "float"), + "TTFT_TIMEOUT_S": ("ttft_timeout_s", "float"), + "INTER_TOKEN_TIMEOUT_S": ("inter_token_timeout_s", "float"), + "ENABLE_THINKING": ("enable_thinking", "bool"), + "MISSING_DONE": ("missing_done", "str"), + "TRUST_ENV": ("trust_env", "bool"), +} +_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"}) +_SELECTORS = frozenset({"round_robin", "least_inflight"}) +_QUOTA_FULL = frozenset({"wait", "fail_fast"}) +# 背压默认(M1 仅 poll 生效;CHS _BACKOFF_S=0.05 同源) +_DEFAULT_STALL_WINDOW_S = 300.0 +_DEFAULT_POLL_INTERVAL_S = 0.05 +_DEFAULT_LEASE_TTL_S = 1500.0 # CHS _DEFAULT_LEASE_TTL_MS 同源 + + +def _cast(raw: str, kind: str, key: str) -> object: + try: + if kind == "int": + return int(raw) + if kind == "float": + return float(raw) + if kind == "bool": + lowered = raw.strip().lower() + if lowered in ("1", "true", "yes", "on"): + return True + if lowered in ("0", "false", "no", "off"): + return False + raise ValueError(f"非法布尔值: {raw!r}") + return raw + except ValueError as exc: + raise ValueError(f"配置 {key} 解析失败: {exc}") from exc + + +def _first(env: Mapping[str, str], *keys: str) -> tuple[str, str] | None: + for key in keys: + raw = env.get(key) + if raw is not None and raw != "": + return key, raw + return None + + +def _require(env: Mapping[str, str], *keys: str) -> tuple[str, str]: + found = _first(env, *keys) + if found is None: + raise ValueError(f"缺关键配置: 需设置 {' 或 '.join(keys)}") + return found + + +@dataclass(frozen=True) +class GatewaySettings: + """一个 scope 的完整装配配置;构造经 from_env 聚合并通过全部守卫。""" + + scope: str + sources: tuple[SourceConfig, ...] + global_limits: GlobalLimits + retry: RetryPolicy + breaker: BreakerConfig + backpressure: BackpressurePolicy + selector: str + quota_full: str + limiter_backend: str + breaker_backend: str + cache_backend: str + cache_namespace: str | None + cache_ttl_s: int | None + telemetry_backend: str + telemetry_sqlite_path: str | None + redis_url: str | None + structured_max_retries: int + lease_ttl_s: float + + @classmethod + def from_env( + cls, + scope: str = "LLM", + env: Mapping[str, str] | None = None, + *, + env_file: str = ".env", + ) -> GatewaySettings: + """聚合 env(缺省 .env + os.environ,后者优先)并执行装配守卫。""" + if env is None: + env = { + k: v for k, v in {**dotenv_values(env_file), **os.environ}.items() if v is not None + } + scope_u = scope.upper() + sources = _load_sources(scope_u, env) + global_limits = _load_global_limits(scope_u, env) + retry = _load_retry(scope_u, env) + breaker = _load_breaker(scope_u, env, sources, global_limits) + settings = cls( + scope=scope_u.lower(), + sources=tuple(sources), + global_limits=global_limits, + retry=retry, + breaker=breaker, + backpressure=_load_backpressure(scope_u, env), + selector=_load_choice(env, f"{scope_u}__SELECTOR", _SELECTORS, "round_robin"), + quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), + **_load_pgw(env), + ) + _guard_lease(settings) + return settings + + +def _load_sources(scope: str, env: Mapping[str, str]) -> list[SourceConfig]: + specs: dict[tuple[str, str], dict[str, object]] = {} + for key, raw in env.items(): + parts = key.split("__") + if len(parts) != 4 or parts[0] != scope or parts[1] in _RESERVED_SEGMENTS: + continue + _, provider, index, field = parts + if field not in _SOURCE_FIELDS: + raise ValueError(f"未知源配置字段 {field}(键 {key});允许: {sorted(_SOURCE_FIELDS)}") + attr, kind = _SOURCE_FIELDS[field] + specs.setdefault((provider.lower(), index), {})[attr] = _cast(raw, kind, key) + if not specs: + raise ValueError(f"scope {scope} 未配置任何源({scope}__{{PROVIDER}}__{{N}}__{{FIELD}})") + return [ + _build_source(scope, provider, index, spec, env) + for (provider, index), spec in sorted(specs.items()) + ] + + +def _build_source( + scope: str, provider: str, index: str, spec: dict[str, object], env: Mapping[str, str] +) -> SourceConfig: + prefix = f"{scope}__{provider.upper()}__{index}" + for field in ("BASE_URL", "API_KEY", "MODEL"): + attr, _ = _SOURCE_FIELDS[field] + if attr not in spec: + raise ValueError(f"缺关键配置: {prefix}__{field}") + if "timeout_s" not in spec: + key, raw = _require(env, f"{prefix}__TIMEOUT_S", "LLM_TIMEOUT") + spec["timeout_s"] = _cast(raw, "float", key) + # 平铺看门狗键作为缺省(成对才生效,遵守 SourceConfig 不变式) + if "ttft_timeout_s" not in spec and "inter_token_timeout_s" not in spec: + flat = _first(env, "LLM_TTFT_TIMEOUT") + flat_inter = _first(env, "LLM_INTER_TOKEN_TIMEOUT") + if flat and flat_inter: + spec["ttft_timeout_s"] = _cast(flat[1], "float", flat[0]) + spec["inter_token_timeout_s"] = _cast(flat_inter[1], "float", flat_inter[0]) + return SourceConfig(name=f"{provider}_{index}", provider=provider, **spec) + + +def _load_global_limits(scope: str, env: Mapping[str, str]) -> GlobalLimits: + def read(field: str) -> int: + found = _first(env, f"{scope}__GLOBAL__{field}") + return int(_cast(found[1], "int", found[0])) if found else 0 + + return GlobalLimits(max_concurrency=read("MAX_CONCURRENCY"), rpm=read("RPM"), tpm=read("TPM")) + + +def _load_retry(scope: str, env: Mapping[str, str]) -> RetryPolicy: + key_a, attempts = _require(env, f"{scope}__RETRY__MAX_ATTEMPTS", "LLM_MAX_RETRIES") + key_b, base = _require(env, f"{scope}__RETRY__BACKOFF_BASE_S", "LLM_RETRY_BASE_DELAY") + key_m, max_d = _require(env, f"{scope}__RETRY__BACKOFF_MAX_S", "LLM_RETRY_MAX_DELAY") + return RetryPolicy( + max_attempts=int(_cast(attempts, "int", key_a)), + backoff_base_s=float(_cast(base, "float", key_b)), + backoff_max_s=float(_cast(max_d, "float", key_m)), + ) + + +def _load_breaker( + scope: str, env: Mapping[str, str], sources: list[SourceConfig], global_limits: GlobalLimits +) -> BreakerConfig: + key_t, thr = _require(env, f"{scope}__BREAKER__FAIL_THRESHOLD", "LLM_CIRCUIT_BREAKER_THRESHOLD") + key_c, cool = _require(env, f"{scope}__BREAKER__COOLDOWN_S", "LLM_CIRCUIT_BREAKER_COOLDOWN") + threshold = int(_cast(thr, "int", key_t)) + cooldown_s = float(_cast(cool, "float", key_c)) + # 有效阈值 = max(配置值, 并发×2)——三项目 .env 注释的手动约定入库(设计 §9 行 4) + concurrency = global_limits.max_concurrency or max( + (s.max_concurrency for s in sources), default=0 + ) + if concurrency > 0: + threshold = max(threshold, concurrency * 2) + probe = _first(env, f"{scope}__BREAKER__PROBE_TTL_S") + if probe is not None: + probe_ttl_s = float(_cast(probe[1], "float", probe[0])) + else: + # 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期 + probe_ttl_s = max(2 * max(s.timeout_s for s in sources), cooldown_s) + return BreakerConfig(fail_threshold=threshold, cooldown_s=cooldown_s, probe_ttl_s=probe_ttl_s) + + +def _load_backpressure(scope: str, env: Mapping[str, str]) -> BackpressurePolicy: + stall = _first(env, f"{scope}__BACKPRESSURE__STALL_WINDOW_S") + poll = _first(env, f"{scope}__BACKPRESSURE__POLL_INTERVAL_S") + return BackpressurePolicy( + stall_window_s=float(_cast(stall[1], "float", stall[0])) + if stall + else _DEFAULT_STALL_WINDOW_S, + poll_interval_s=float(_cast(poll[1], "float", poll[0])) + if poll + else _DEFAULT_POLL_INTERVAL_S, + ) + + +def _load_choice(env: Mapping[str, str], key: str, allowed: frozenset[str], default: str) -> str: + found = _first(env, key) + value = found[1] if found else default + if value not in allowed: + raise ValueError(f"配置 {key} 非法值 {value!r};允许: {sorted(allowed)}") + return value + + +def _load_pgw(env: Mapping[str, str]) -> dict[str, object]: + limiter_backend = _load_choice( + env, "PGW_LIMITER_BACKEND", frozenset({"memory", "redis"}), "memory" + ) + breaker_backend = _load_choice( + env, "PGW_BREAKER_BACKEND", frozenset({"memory", "redis"}), "memory" + ) + if "redis" in (limiter_backend, breaker_backend): + raise ValueError("限流/熔断 Redis 后端在 M2 交付;M1 仅支持 memory") + _, cache_backend = _require(env, "PGW_CACHE_BACKEND") + _, telemetry_backend = _require(env, "PGW_TELEMETRY_BACKEND") + if cache_backend not in ("redis", "memory", "none"): + raise ValueError(f"PGW_CACHE_BACKEND 非法值 {cache_backend!r}") + if telemetry_backend not in ("sqlite", "none"): + raise ValueError(f"PGW_TELEMETRY_BACKEND 非法值 {telemetry_backend!r}(postgres 在 M2)") + redis_url = env.get("REDIS_URL") or None + return { + "limiter_backend": limiter_backend, + "breaker_backend": breaker_backend, + "cache_backend": cache_backend, + **_load_cache_keys(env, cache_backend, redis_url), + "telemetry_backend": telemetry_backend, + "telemetry_sqlite_path": _require(env, "PGW_TELEMETRY_SQLITE_PATH")[1] + if telemetry_backend == "sqlite" + else None, + "redis_url": redis_url, + "structured_max_retries": _load_structured_retries(env), + "lease_ttl_s": _load_lease_ttl(env), + } + + +def _load_cache_keys( + env: Mapping[str, str], cache_backend: str, redis_url: str | None +) -> dict[str, object]: + if cache_backend == "none": + return {"cache_namespace": None, "cache_ttl_s": None} + _, namespace = _require(env, "PGW_CACHE_NAMESPACE") + key, ttl_raw = _require(env, "PGW_CACHE_TTL_S") + ttl = int(_cast(ttl_raw, "int", key)) + if ttl <= 0: + raise ValueError("PGW_CACHE_TTL_S(TTL)必须 > 0,禁止永不过期") + if cache_backend == "redis" and redis_url is None: + raise ValueError("缺关键配置: PGW_CACHE_BACKEND=redis 需设置 REDIS_URL") + return {"cache_namespace": namespace, "cache_ttl_s": ttl} + + +def _load_structured_retries(env: Mapping[str, str]) -> int: + found = _first(env, "PGW_STRUCTURED_MAX_RETRIES") + value = int(_cast(found[1], "int", found[0])) if found else 1 + if value < 0: + raise ValueError("PGW_STRUCTURED_MAX_RETRIES 不能为负") + return value + + +def _guard_lease(settings: GatewaySettings) -> None: + """装配守卫: 调用超时须 ≤ permit 租约 TTL,防租约先于请求过期(ARCH §7.3)。""" + slowest = max(s.timeout_s for s in settings.sources) + if slowest > settings.lease_ttl_s: + raise ValueError( + f"源最大 timeout_s({slowest})超过 permit 租约 TTL({settings.lease_ttl_s});" + f"调大 PGW_LEASE_TTL_S 或调小超时" + ) + + +def _load_lease_ttl(env: Mapping[str, str]) -> float: + found = _first(env, "PGW_LEASE_TTL_S") + return float(_cast(found[1], "float", found[0])) if found else _DEFAULT_LEASE_TTL_S diff --git a/src/polygateway/errors.py b/src/polygateway/errors.py index 57d8aa0..e3cd4de 100644 --- a/src/polygateway/errors.py +++ b/src/polygateway/errors.py @@ -87,7 +87,9 @@ class GatewayUnavailableError(PolyGatewayError): 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)})") + 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 diff --git a/src/polygateway/middleware/breaker.py b/src/polygateway/middleware/breaker.py index 2061aba..3e20a18 100644 --- a/src/polygateway/middleware/breaker.py +++ b/src/polygateway/middleware/breaker.py @@ -33,7 +33,9 @@ class BreakerGate: except Exception as exc: raise GovernanceBackendError(f"熔断后端故障(record_success): {exc}") from exc - async def record_failure(self, entry: GateDecision, reason: str, force_open: bool) -> GateUpdate: + async def record_failure( + self, entry: GateDecision, reason: str, force_open: bool + ) -> GateUpdate: try: return await self._gate.record_failure(entry, reason, force_open) except GovernanceBackendError: diff --git a/src/polygateway/middleware/cache.py b/src/polygateway/middleware/cache.py index c9f8255..252474c 100644 --- a/src/polygateway/middleware/cache.py +++ b/src/polygateway/middleware/cache.py @@ -111,8 +111,12 @@ class CacheMW: fields = {k: v for k, v in data.items() if k in _RESPONSE_FIELDS} structured_data = self._rebuild_structured(fields.get("content", ""), request) fields.update( - cache_hit=True, latency_ms=0, ttft_ms=None, max_inter_token_ms=None, - call_id=str(uuid.uuid4()), structured_data=structured_data, + cache_hit=True, + latency_ms=0, + ttft_ms=None, + max_inter_token_ms=None, + call_id=str(uuid.uuid4()), + structured_data=structured_data, ) return LLMResponse(**fields) except Exception as exc: diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index c119d97..e691c4b 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -129,7 +129,8 @@ class RetryMW: fails += 1 if fails >= self._retry.max_attempts: raise AllSourcesExhausted( - scope=self._scope, reason="retry_exhausted", + scope=self._scope, + reason="retry_exhausted", retry_after_s=self._retry.backoff_base_s, per_source_reasons=reasons, ) from outcome.exc @@ -177,8 +178,10 @@ class RetryMW: ) if self._quota_full == "fail_fast": raise AllSourcesExhausted( - scope=self._scope, reason="quota_exhausted", - retry_after_s=self._bp.poll_interval_s, per_source_reasons=reasons, + scope=self._scope, + reason="quota_exhausted", + retry_after_s=self._bp.poll_interval_s, + per_source_reasons=reasons, ) await self._sleep(self._bp.poll_interval_s) @@ -197,8 +200,11 @@ class RetryMW: actual = 0 try: result = await self._transport.complete( - messages=request.messages, source=source, - stream=request.stream, overlay=request.overlay, call_id=call_id, + messages=request.messages, + source=source, + stream=request.stream, + overlay=request.overlay, + call_id=call_id, ) actual = result.prompt_tokens + result.completion_tokens await self._breaker.record_success(entry) @@ -254,13 +260,20 @@ class RetryMW: self, source: SourceConfig, result: TransportResult, call_id: str, started: float ) -> LLMResponse: return LLMResponse( - content=result.content, thinking=result.thinking, - model=source.model, provider=source.provider, - prompt_tokens=result.prompt_tokens, completion_tokens=result.completion_tokens, + content=result.content, + thinking=result.thinking, + model=source.model, + provider=source.provider, + prompt_tokens=result.prompt_tokens, + completion_tokens=result.completion_tokens, latency_ms=int((self._now() - started) * 1000), - ttft_ms=result.ttft_ms, max_inter_token_ms=result.max_inter_token_ms, - cache_hit=False, call_id=call_id, - source_name=source.name, cost=None, usage_source=result.usage_source, + ttft_ms=result.ttft_ms, + max_inter_token_ms=result.max_inter_token_ms, + cache_hit=False, + call_id=call_id, + source_name=source.name, + cost=None, + usage_source=result.usage_source, ) async def _settle_and_release(self, permit: Permit, actual: int) -> None: @@ -276,15 +289,23 @@ class RetryMW: logger.warning("permit 结算/释放失败(不掩盖主异常): {}", exc) async def _emit( - self, request: ChatRequest, source: SourceConfig, call_id: str, started: float, - *, response: LLMResponse | None = None, error: object | None = None, + self, + request: ChatRequest, + source: SourceConfig, + call_id: str, + started: float, + *, + response: LLMResponse | None = None, + error: object | None = None, ) -> None: """逐次遥测(经注入的单一 Emitter);遥测失败不得影响调用(铁律)。""" if self._emitter is None: return try: await self._emitter.emit_attempt( - request=request, source=source, call_id=call_id, + request=request, + source=source, + call_id=call_id, latency_ms=int((self._now() - started) * 1000), response=response, error=None if error is None else str(error), diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index aeda520..a247b22 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -166,15 +166,19 @@ class TelemetryMW: response = await call_next(request) except (GatewayUnavailableError, GovernanceBackendError) as exc: await self._emitter.emit_terminal_failure( - request=request, call_id=str(uuid.uuid4()), - latency_ms=int((self._now() - started) * 1000), error=str(exc), + request=request, + call_id=str(uuid.uuid4()), + latency_ms=int((self._now() - started) * 1000), + error=str(exc), ) raise except asyncio.CancelledError: # 尽力而为: 取消也留痕(§5.1 约定④);随后立即重抛 await self._emitter.emit_terminal_failure( - request=request, call_id=str(uuid.uuid4()), - latency_ms=int((self._now() - started) * 1000), error="cancelled", + request=request, + call_id=str(uuid.uuid4()), + latency_ms=int((self._now() - started) * 1000), + error="cancelled", ) raise if response.cache_hit: diff --git a/src/polygateway/structured/json_repair.py b/src/polygateway/structured/json_repair.py index 3b4b7e1..2a84dba 100644 --- a/src/polygateway/structured/json_repair.py +++ b/src/polygateway/structured/json_repair.py @@ -49,9 +49,7 @@ class JsonRepairStrategy: try: data = json.loads(repair_json(stripped)) except (json.JSONDecodeError, ValueError) as exc: - raise ResultInvalidError( - "JSON 修复失败", raw_text=text, repair_error=str(exc) - ) from exc + raise ResultInvalidError("JSON 修复失败", raw_text=text, repair_error=str(exc)) from exc if self._normalize is not None: data = self._normalize(data) return data diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 802b7ab..0b5d58a 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -39,10 +39,24 @@ CREATE TABLE IF NOT EXISTS llm_calls ( """ _COLUMNS = ( - "call_id", "parent_call_id", "session_id", "model", "provider", "source_name", - "messages", "response", "thinking", "prompt_tokens", "completion_tokens", - "usage_source", "latency_ms", "ttft_ms", "max_inter_token_ms", "cache_hit", - "error", "cost", + "call_id", + "parent_call_id", + "session_id", + "model", + "provider", + "source_name", + "messages", + "response", + "thinking", + "prompt_tokens", + "completion_tokens", + "usage_source", + "latency_ms", + "ttft_ms", + "max_inter_token_ms", + "cache_hit", + "error", + "cost", ) _INSERT = ( diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index 511a70f..a50c453 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -67,7 +67,9 @@ async def _iter_sse_deltas( try: chunk = json.loads(data) except json.JSONDecodeError as exc: - raise TransientError(f"SSE 帧畸形(malformed_json): {data[:80]!r}", operation="chat") from exc + raise TransientError( + f"SSE 帧畸形(malformed_json): {data[:80]!r}", operation="chat" + ) from exc delta = _sse_delta(chunk, usage_sink) if delta is not None: yield delta @@ -95,12 +97,16 @@ def _translate_429(source: SourceConfig, body_text: str, headers: Mapping[str, s if err_type == "insufficient_quota": return SourceDeadError( f"{source.name} 配额耗尽(insufficient_quota)", - source_name=source.name, status_code=429, operation="chat", + source_name=source.name, + status_code=429, + operation="chat", ) return TransientError( f"{source.name} 限速: 429", retry_after_s=_parse_retry_after(headers.get("retry-after")), - source_name=source.name, status_code=429, operation="chat", + source_name=source.name, + status_code=429, + operation="chat", ) @@ -164,8 +170,13 @@ class OpenAICompatTransport: return client def _build_payload( - self, *, messages: list[dict[str, Any]], source: SourceConfig, - profile: ProviderProfile, stream: bool, overlay: dict[str, Any], + self, + *, + messages: list[dict[str, Any]], + source: SourceConfig, + profile: ProviderProfile, + stream: bool, + overlay: dict[str, Any], ) -> dict[str, Any]: payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream} if stream: @@ -178,8 +189,13 @@ class OpenAICompatTransport: return payload async def complete( - self, *, messages: list[dict[str, Any]], source: SourceConfig, - stream: bool, overlay: dict[str, Any], call_id: str, + self, + *, + messages: list[dict[str, Any]], + source: SourceConfig, + stream: bool, + overlay: dict[str, Any], + call_id: str, ) -> TransportResult: """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。""" profile = get_provider(source.provider, registry=self._registry) @@ -202,8 +218,12 @@ class OpenAICompatTransport: raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc async def _complete_stream( - self, client: httpx.AsyncClient, url: str, payload: dict[str, Any], - source: SourceConfig, profile: ProviderProfile, + self, + client: httpx.AsyncClient, + url: str, + payload: dict[str, Any], + source: SourceConfig, + profile: ProviderProfile, ) -> TransportResult: started = time.monotonic() async with client.stream("POST", url, json=payload) as resp: @@ -236,15 +256,22 @@ class OpenAICompatTransport: if salvaged: usage_source = "estimated" # 打捞路径强制 estimated(设计 §6) return TransportResult( - content=content, thinking=thinking, prompt_tokens=prompt, - completion_tokens=completion, usage_source=usage_source, - ttft_ms=ttft_ms, max_inter_token_ms=(max_gap if ttft_ms is not None else None), + content=content, + thinking=thinking, + prompt_tokens=prompt, + completion_tokens=completion, + usage_source=usage_source, + ttft_ms=ttft_ms, + max_inter_token_ms=(max_gap if ttft_ms is not None else None), raw={"usage": sink.get("usage")}, ) def _check_done( - self, sink: dict[str, Any], content_parts: list[str], - thinking_parts: list[str], source: SourceConfig, + self, + sink: dict[str, Any], + content_parts: list[str], + thinking_parts: list[str], + source: SourceConfig, ) -> bool: """缺 [DONE] 语义(设计 §6): 零内容恒 retry;有内容按 missing_done 策略。""" if sink.get("done"): @@ -268,8 +295,12 @@ class OpenAICompatTransport: return content, thinking async def _complete_once( - self, client: httpx.AsyncClient, url: str, payload: dict[str, Any], - source: SourceConfig, profile: ProviderProfile, + self, + client: httpx.AsyncClient, + url: str, + payload: dict[str, Any], + source: SourceConfig, + profile: ProviderProfile, ) -> TransportResult: """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。""" resp = await client.post(url, json=payload) @@ -292,9 +323,14 @@ class OpenAICompatTransport: ) prompt, completion, usage_source = _resolve_usage(body.get("usage") or {}, source) return TransportResult( - content=content, thinking=thinking, prompt_tokens=prompt, - completion_tokens=completion, usage_source=usage_source, - ttft_ms=None, max_inter_token_ms=None, raw={"usage": body.get("usage")}, + content=content, + thinking=thinking, + prompt_tokens=prompt, + completion_tokens=completion, + usage_source=usage_source, + ttft_ms=None, + max_inter_token_ms=None, + raw={"usage": body.get("usage")}, ) async def aclose(self) -> None: diff --git a/src/polygateway/types.py b/src/polygateway/types.py index 80cd328..c3f96ab 100644 --- a/src/polygateway/types.py +++ b/src/polygateway/types.py @@ -112,7 +112,9 @@ class SourceConfig: 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}") + raise ValueError( + f"missing_done 必须是 {sorted(_MISSING_DONE_DOMAIN)}: {self.missing_done!r}" + ) def _validate_gates(self) -> None: if self.timeout_s <= 0: diff --git a/tests/contracts/conftest.py b/tests/contracts/conftest.py index de6e87b..cb0de99 100644 --- a/tests/contracts/conftest.py +++ b/tests/contracts/conftest.py @@ -25,10 +25,14 @@ class FakeClock: def make_source(name: str = "s1", **overrides) -> SourceConfig: - base = dict( - name=name, provider="openai", base_url="https://gw.example/v1", - api_key="sk-test", model="m", timeout_s=10.0, - ) + base = { + "name": name, + "provider": "openai", + "base_url": "https://gw.example/v1", + "api_key": "sk-test", + "model": "m", + "timeout_s": 10.0, + } base.update(overrides) return SourceConfig(**base) @@ -44,8 +48,11 @@ def limiter_factory(request, clock): def make(sources: list[SourceConfig], global_limits: GlobalLimits, lease_ttl_s: float = 100.0): return InMemoryLimiter( - scope="llm", sources={s.name: s for s in sources}, - global_limits=global_limits, lease_ttl_s=lease_ttl_s, now=clock, + scope="llm", + sources={s.name: s for s in sources}, + global_limits=global_limits, + lease_ttl_s=lease_ttl_s, + now=clock, ) return make diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index a8b2c80..db64ae0 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -15,12 +15,21 @@ _MSGS = [{"role": "user", "content": "hi"}] def _resp(content="cached", **overrides): - base = dict( - content=content, thinking="", model="m", provider="p", - prompt_tokens=1, completion_tokens=2, latency_ms=30, - ttft_ms=5.0, max_inter_token_ms=2.0, cache_hit=False, call_id="orig", - source_name="s1", usage_source="measured", - ) + base = { + "content": content, + "thinking": "", + "model": "m", + "provider": "p", + "prompt_tokens": 1, + "completion_tokens": 2, + "latency_ms": 30, + "ttft_ms": 5.0, + "max_inter_token_ms": 2.0, + "cache_hit": False, + "call_id": "orig", + "source_name": "s1", + "usage_source": "measured", + } base.update(overrides) return LLMResponse(**base) @@ -46,21 +55,33 @@ class TestKeyFormula: def test_multimodal_part_digested_not_inlined(self): big_b64 = "data:image/png;base64," + "A" * 1_000_000 - messages = [{"role": "user", "content": [ - {"type": "image_url", "image_url": {"url": big_b64}}, - {"type": "text", "text": "describe"}, - ]}] + messages = [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": big_b64}}, + {"type": "text", "text": "describe"}, + ], + } + ] digested = digest_messages(messages) payload = json.dumps(digested, ensure_ascii=False) assert len(payload) < 500 # 大图不进 canonical_json expected = hashlib.sha256(big_b64.encode()).hexdigest() assert expected in payload # 但字节变化仍改变 key # 图像字节变化 → key 变 - messages2 = [{"role": "user", "content": [ - {"type": "image_url", "image_url": {"url": big_b64[:-1] + "B"}}, - {"type": "text", "text": "describe"}, - ]}] - assert build_cache_key("m", messages, "p", None) != build_cache_key("m", messages2, "p", None) + messages2 = [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": big_b64[:-1] + "B"}}, + {"type": "text", "text": "describe"}, + ], + } + ] + assert build_cache_key("m", messages, "p", None) != build_cache_key( + "m", messages2, "p", None + ) class _Terminal: @@ -76,7 +97,12 @@ class _Terminal: def _mw(backend, **kwargs): - defaults = dict(backend=backend, model_fingerprint="m", default_namespace="proj", ttl_s=3600) + defaults = { + "backend": backend, + "model_fingerprint": "m", + "default_namespace": "proj", + "ttl_s": 3600, + } defaults.update(kwargs) return CacheMW(**defaults) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py new file mode 100644 index 0000000..7203871 --- /dev/null +++ b/tests/unit/test_client.py @@ -0,0 +1,256 @@ +"""GatewayClient 装配与端到端(fake 后端 + MockTransport)测试(设计 §2.4)。""" + +import asyncio +import json +import sys +from pathlib import Path + +import httpx +import pytest + +from polygateway import ( + AllSourcesExhausted, + GatewayClient, + GatewaySettings, + gather_bounded, +) +from polygateway.backends.memory.breaker import InMemoryGate +from polygateway.backends.memory.cache import InMemoryCache +from polygateway.backends.memory.limiter import InMemoryLimiter +from polygateway.sources import RoundRobinSelector +from polygateway.structured.json_repair import JsonRepairStrategy +from polygateway.transports.openai_compat import OpenAICompatTransport +from polygateway.types import ( + BackpressurePolicy, + BreakerConfig, + GlobalLimits, + RetryPolicy, + SourceConfig, +) + +_REPO = Path(__file__).resolve().parents[2] + +_ENV = { + "LLM__QWEN__1__BASE_URL": "https://gw.example/v1", + "LLM__QWEN__1__API_KEY": "sk-a", + "LLM__QWEN__1__MODEL": "qwen-max", + "LLM__QWEN__1__TIMEOUT_S": "120", + "LLM_MAX_RETRIES": "3", + "LLM_RETRY_BASE_DELAY": "2.0", + "LLM_RETRY_MAX_DELAY": "30.0", + "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", + "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", +} + + +def _sse(content='{"answer": 1}'): + chunk = json.dumps({"choices": [{"delta": {"content": content}}]}) + usage = json.dumps({"choices": [], "usage": {"prompt_tokens": 3, "completion_tokens": 4}}) + body = f"data: {chunk}\n\ndata: {usage}\n\ndata: [DONE]\n\n" + return httpx.Response(200, content=body.encode(), headers={"content-type": "text/event-stream"}) + + +def _source(name="qwen_1", **overrides): + base = { + "name": name, + "provider": "qwen", + "base_url": "https://gw.example/v1", + "api_key": "sk", + "model": "qwen-max", + "timeout_s": 10.0, + } + base.update(overrides) + return SourceConfig(**base) + + +def _client(sources=None, handler=None, *, limiter=None, quota_full="wait", **overrides): + sources = sources or [_source()] + handler = handler or (lambda request: _sse()) + transport = OpenAICompatTransport( + client_factory=lambda src: httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) + defaults = { + "scope": "llm", + "sources": sources, + "selector": RoundRobinSelector(), + "limiter": limiter + or InMemoryLimiter( + scope="llm", + sources={s.name: s for s in sources}, + global_limits=GlobalLimits(0, 0, 0), + ), + "gate": InMemoryGate(config=BreakerConfig(5, 60.0, 120.0)), + "transport": transport, + "retry": RetryPolicy(3, 2.0, 30.0), + "backpressure": BackpressurePolicy(300.0, 0.01), + "quota_full": quota_full, + "structured_strategy": JsonRepairStrategy(), + } + defaults.update(overrides) + return GatewayClient(**defaults) + + +class TestChatEndToEnd: + async def test_plain_chat(self): + async with _client() as client: + resp = await client.chat([{"role": "user", "content": "hi"}]) + assert resp.content == '{"answer": 1}' + assert resp.source_name == "qwen_1" and resp.prompt_tokens == 3 + + async def test_structured_json_tier(self): + async with _client() as client: + resp = await client.chat([{"role": "user", "content": "hi"}], structured="json") + assert resp.structured_data == {"answer": 1} + + async def test_cache_hit_roundtrip(self): + client = _client(cache=InMemoryCache(), cache_namespace="proj", cache_ttl_s=3600) + async with client: + first = await client.chat([{"role": "user", "content": "hi"}]) + second = await client.chat([{"role": "user", "content": "hi"}]) + assert first.cache_hit is False and second.cache_hit is True + + async def test_structured_unavailable_fails_loudly(self): + async with _client(structured_strategy=None) as client: + with pytest.raises(ImportError, match="structured"): + await client.chat([{"role": "user", "content": "hi"}], structured="json") + + +class TestFactories: + def test_from_env_assembles(self): + client = GatewayClient.from_env("LLM", env=_ENV) + assert isinstance(client, GatewayClient) + + def test_from_env_unknown_provider_fails_at_assembly(self): + env = dict(_ENV) + for key in list(env): + if key.startswith("LLM__QWEN__"): + env[key.replace("QWEN", "GLM")] = env.pop(key) + with pytest.raises(ValueError, match="glm"): + GatewayClient.from_env("LLM", env=env) + + def test_from_settings_respects_injection(self): + settings = GatewaySettings.from_env("LLM", env=_ENV) + shared = InMemoryLimiter( + scope="shared", + sources={s.name: s for s in settings.sources}, + global_limits=GlobalLimits(0, 0, 0), + ) + client = GatewayClient.from_settings(settings, limiter=shared) + assert isinstance(client, GatewayClient) + + +class TestSharedBackend: + async def test_two_clients_share_global_concurrency_gate(self): + """VT R5: 两个逻辑角色显式注入同一 limiter → 共享全局并发闸。""" + src_a, src_b = _source("role_a_1"), _source("role_b_1") + shared = InMemoryLimiter( + scope="shared", + sources={"role_a_1": src_a, "role_b_1": src_b}, # 共享后端持源并集 + global_limits=GlobalLimits(max_concurrency=1, rpm=0, tpm=0), + ) + started = asyncio.Event() + + async def slow_handler(request): + started.set() + await asyncio.sleep(0.2) + return _sse() + + client_a = _client([src_a], slow_handler, limiter=shared) + client_b = _client([src_b], limiter=shared, quota_full="fail_fast") + task = asyncio.ensure_future(client_a.chat([{"role": "user", "content": "x"}])) + await started.wait() + with pytest.raises(AllSourcesExhausted) as ei: + await client_b.chat([{"role": "user", "content": "y"}]) + assert ei.value.reason == "quota_exhausted" # 全局闸被 A 占满 → B 立即失败 + await task + + async def test_aclose_idempotent(self): + client = _client() + await client.aclose() + await client.aclose() + + +class TestGatherBounded: + async def test_order_preserved_and_concurrency_capped(self): + peak = {"now": 0, "max": 0} + + async def work(i): + peak["now"] += 1 + peak["max"] = max(peak["max"], peak["now"]) + await asyncio.sleep(0.01) + peak["now"] -= 1 + return i + + results = await gather_bounded((work(i) for i in range(10)), concurrency=3) + assert results == list(range(10)) + assert peak["max"] <= 3 + + async def test_exception_propagates(self): + async def boom(): + raise RuntimeError("x") + + async def ok(): + return 1 + + with pytest.raises(RuntimeError): + await gather_bounded([ok(), boom()], concurrency=2) + + async def test_invalid_concurrency(self): + with pytest.raises(ValueError): + await gather_bounded([], concurrency=0) + + +def _load_reference_protocol(insert_path: str, module: str): + sys.path.insert(0, str(_REPO / insert_path)) + try: + import importlib + + return importlib.import_module(module).LLMProvider + finally: + sys.path.pop(0) + + +class TestReferenceProtocolCompat: + """结构兼容断言(只读 import reference;失败即公共承诺破裂)。""" + + def test_satisfies_govdoc_llm_provider(self): + try: + proto = _load_reference_protocol( + "reference/GovDoc-SaaS/packages/docagent-core/src", "docagent_core.protocols" + ) + except ImportError: + # 兜底: 按 protocols.py:15-25 逐字复制的结构断言 + from typing import Any, Protocol, runtime_checkable + + @runtime_checkable + class proto(Protocol): # noqa: N801 — 复制自 GovDoc protocols.py:15-25 + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ): ... + + assert isinstance(_client(), proto) + + def test_satisfies_videotree_llm_provider(self): + try: + proto = _load_reference_protocol("reference/Video-Tree-TRM5", "core.protocols") + except ImportError: + from typing import Any, Protocol, runtime_checkable + + @runtime_checkable + class proto(Protocol): # noqa: N801 — 复制自 VT core/protocols.py:18-29 + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + cache_salt: str | None = None, + ): ... + + assert isinstance(_client(), proto) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..9908277 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,166 @@ +"""config.py 配置聚合测试(设计 §8): 多源命名、键优先级、缺失报错。""" + +import pytest + +from polygateway.config import GatewaySettings + +_BASE_ENV = { + "LLM__QWEN__1__BASE_URL": "https://gw-a.example/v1", + "LLM__QWEN__1__API_KEY": "sk-a", + "LLM__QWEN__1__MODEL": "qwen-max", + "LLM__QWEN__1__TIMEOUT_S": "120", + "LLM_MAX_RETRIES": "3", + "LLM_RETRY_BASE_DELAY": "2.0", + "LLM_RETRY_MAX_DELAY": "30.0", + "LLM_CIRCUIT_BREAKER_THRESHOLD": "5", + "LLM_CIRCUIT_BREAKER_COOLDOWN": "60", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", +} + + +def _env(**overrides): + env = dict(_BASE_ENV) + env.update({k: v for k, v in overrides.items() if v is not None}) + for k, v in overrides.items(): + if v is None: + env.pop(k, None) + return env + + +class TestSourceAggregation: + def test_single_source_parsed(self): + s = GatewaySettings.from_env("LLM", env=_env()) + assert len(s.sources) == 1 + src = s.sources[0] + assert src.name == "qwen_1" and src.provider == "qwen" + assert src.base_url == "https://gw-a.example/v1" and src.timeout_s == 120.0 + + def test_multi_source_and_optional_fields(self): + env = _env( + **{ + "LLM__DEEPSEEK__2__BASE_URL": "https://gw-b.example/v1", + "LLM__DEEPSEEK__2__API_KEY": "sk-b", + "LLM__DEEPSEEK__2__MODEL": "deepseek-chat", + "LLM__DEEPSEEK__2__TIMEOUT_S": "90", + "LLM__DEEPSEEK__2__RPM": "60", + "LLM__DEEPSEEK__2__ENABLE_THINKING": "true", + "LLM__DEEPSEEK__2__MISSING_DONE": "salvage", + } + ) + s = GatewaySettings.from_env("LLM", env=env) + by_name = {src.name: src for src in s.sources} + assert set(by_name) == {"qwen_1", "deepseek_2"} + ds = by_name["deepseek_2"] + assert ds.rpm == 60 and ds.enable_thinking is True and ds.missing_done == "salvage" + assert by_name["qwen_1"].enable_thinking is None # 未配置 = 三态 None + + def test_other_scope_keys_ignored(self): + env = _env( + **{ + "OCR__MONKEY__1__BASE_URL": "http://lan/parse", + "OCR__MONKEY__1__API_KEY": "x", + "OCR__MONKEY__1__MODEL": "monkey", + "OCR__MONKEY__1__TIMEOUT_S": "60", + } + ) + s = GatewaySettings.from_env("LLM", env=env) + assert len(s.sources) == 1 + + def test_flat_timeout_is_source_default(self): + env = _env(LLM_TIMEOUT="300", **{"LLM__QWEN__1__TIMEOUT_S": None}) + s = GatewaySettings.from_env("LLM", env=env) + assert s.sources[0].timeout_s == 300.0 + + @pytest.mark.parametrize("missing", ["BASE_URL", "API_KEY", "MODEL"]) + def test_missing_required_source_field_fails(self, missing): + with pytest.raises(ValueError, match=missing): + GatewaySettings.from_env("LLM", env=_env(**{f"LLM__QWEN__1__{missing}": None})) + + def test_unknown_field_fails_loudly(self): + with pytest.raises(ValueError, match="TEMPRATURE"): + GatewaySettings.from_env("LLM", env=_env(**{"LLM__QWEN__1__TEMPRATURE": "0.7"})) + + def test_no_sources_fails(self): + env = {k: v for k, v in _BASE_ENV.items() if not k.startswith("LLM__")} + with pytest.raises(ValueError, match="源"): + GatewaySettings.from_env("LLM", env=env) + + +class TestResilienceKeys: + def test_flat_legacy_keys(self): + s = GatewaySettings.from_env("LLM", env=_env()) + assert s.retry.max_attempts == 3 and s.retry.backoff_base_s == 2.0 + assert s.breaker.fail_threshold == 5 and s.breaker.cooldown_s == 60.0 + + def test_scope_keys_override_flat(self): + env = _env(**{"LLM__RETRY__MAX_ATTEMPTS": "7", "LLM__BREAKER__COOLDOWN_S": "15"}) + s = GatewaySettings.from_env("LLM", env=env) + assert s.retry.max_attempts == 7 + assert s.breaker.cooldown_s == 15.0 + assert s.breaker.fail_threshold == 5 # 未覆盖的仍取平铺键 + + def test_missing_retry_config_fails(self): + with pytest.raises(ValueError, match="MAX_RETRIES|MAX_ATTEMPTS"): + GatewaySettings.from_env("LLM", env=_env(LLM_MAX_RETRIES=None)) + + def test_probe_ttl_derived_when_absent(self): + s = GatewaySettings.from_env("LLM", env=_env()) + # 派生规则: max(2 × 最大源 timeout, cooldown) + assert s.breaker.probe_ttl_s == max(2 * 120.0, 60.0) + s2 = GatewaySettings.from_env("LLM", env=_env(**{"LLM__BREAKER__PROBE_TTL_S": "45"})) + assert s2.breaker.probe_ttl_s == 45.0 + + def test_selector_and_quota_full(self): + s = GatewaySettings.from_env("LLM", env=_env()) + assert s.selector == "round_robin" and s.quota_full == "wait" + s2 = GatewaySettings.from_env( + "LLM", env=_env(**{"LLM__SELECTOR": "least_inflight", "LLM__QUOTA_FULL": "fail_fast"}) + ) + assert s2.selector == "least_inflight" and s2.quota_full == "fail_fast" + with pytest.raises(ValueError): + GatewaySettings.from_env("LLM", env=_env(**{"LLM__SELECTOR": "random"})) + + def test_global_limits(self): + env = _env(**{"LLM__GLOBAL__MAX_CONCURRENCY": "8", "LLM__GLOBAL__RPM": "120"}) + s = GatewaySettings.from_env("LLM", env=env) + assert s.global_limits.max_concurrency == 8 and s.global_limits.rpm == 120 + assert s.global_limits.tpm == 0 + + +class TestAssemblyGuards: + def test_cache_requires_namespace_and_ttl(self): + env = _env(PGW_CACHE_BACKEND="memory") + with pytest.raises(ValueError, match="NAMESPACE"): + GatewaySettings.from_env("LLM", env=env) + env2 = _env(PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="0") + with pytest.raises(ValueError, match="TTL"): + GatewaySettings.from_env("LLM", env=env2) + env3 = _env(PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600") + s = GatewaySettings.from_env("LLM", env=env3) + assert s.cache_namespace == "proj" and s.cache_ttl_s == 3600 + + def test_redis_cache_requires_url(self): + env = _env(PGW_CACHE_BACKEND="redis", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600") + with pytest.raises(ValueError, match="REDIS_URL"): + GatewaySettings.from_env("LLM", env=env) + + def test_sqlite_telemetry_requires_path(self): + env = _env(PGW_TELEMETRY_BACKEND="sqlite") + with pytest.raises(ValueError, match="SQLITE_PATH"): + GatewaySettings.from_env("LLM", env=env) + + def test_timeout_must_fit_lease_ttl(self): + env = _env(PGW_LEASE_TTL_S="60", **{"LLM__QWEN__1__TIMEOUT_S": "120"}) + with pytest.raises(ValueError, match="租约|lease"): + GatewaySettings.from_env("LLM", env=env) + + def test_effective_breaker_threshold_auto_raised(self): + env = _env(**{"LLM__QWEN__1__MAX_CONCURRENCY": "8"}) + s = GatewaySettings.from_env("LLM", env=env) + # 有效阈值 = max(配置值 5, 并发 8 × 2) = 16(.env 注释约定入库) + assert s.breaker.fail_threshold == 16 + + def test_m1_only_memory_governance_backends(self): + with pytest.raises(ValueError, match="M2"): + GatewaySettings.from_env("LLM", env=_env(PGW_LIMITER_BACKEND="redis")) diff --git a/tests/unit/test_errors.py b/tests/unit/test_errors.py index 52e004f..15719ee 100644 --- a/tests/unit/test_errors.py +++ b/tests/unit/test_errors.py @@ -34,7 +34,9 @@ class TestBaseShape: class TestResultInvalid: def test_carries_diagnosis(self): exc = ResultInvalidError( - "bad json", raw_text="{oops", repair_error="unterminated", + "bad json", + raw_text="{oops", + repair_error="unterminated", validation_errors=("field x missing",), ) assert exc.raw_text == "{oops" @@ -45,7 +47,9 @@ class TestResultInvalid: class TestGatewayUnavailable: def test_fields_and_inheritance(self): exc = AllSourcesExhausted( - scope="LLM", reason="retry_exhausted", retry_after_s=4.0, + scope="LLM", + reason="retry_exhausted", + retry_after_s=4.0, per_source_reasons={"qwen_1": "timeout"}, ) assert isinstance(exc, GatewayUnavailableError) @@ -66,7 +70,9 @@ class TestGatewayUnavailable: def test_per_source_reason_domain_enforced(self): with pytest.raises(ValueError): AllSourcesExhausted( - scope="LLM", reason="no_sources", retry_after_s=0.0, + scope="LLM", + reason="no_sources", + retry_after_s=0.0, per_source_reasons={"qwen_1": "weird"}, ) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index 9440225..3aee2d7 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -22,10 +22,14 @@ from polygateway.types import SourceConfig def _source(**overrides): - base = dict( - name="qwen_1", provider="qwen", base_url="https://gw.example/v1", - api_key="sk-test", model="qwen-max", timeout_s=5.0, - ) + base = { + "name": "qwen_1", + "provider": "qwen", + "base_url": "https://gw.example/v1", + "api_key": "sk-test", + "model": "qwen-max", + "timeout_s": 5.0, + } base.update(overrides) return SourceConfig(**base) @@ -47,9 +51,7 @@ _USAGE = {"prompt_tokens": 11, "completion_tokens": 7} def _sse_stream(*frames, done=True): text = "".join(frames) + ("data: [DONE]\n\n" if done else "") - return httpx.Response( - 200, content=text.encode(), headers={"content-type": "text/event-stream"} - ) + return httpx.Response(200, content=text.encode(), headers={"content-type": "text/event-stream"}) def _transport_for(handler): @@ -61,8 +63,11 @@ def _transport_for(handler): async def _complete(transport, source, *, stream=True, overlay=None): return await transport.complete( - messages=[{"role": "user", "content": "hi"}], source=source, - stream=stream, overlay=overlay or {}, call_id="cid-1", + messages=[{"role": "user", "content": "hi"}], + source=source, + stream=stream, + overlay=overlay or {}, + call_id="cid-1", ) @@ -76,8 +81,13 @@ class TestSsePureFunctions: async def test_iter_deltas_yields_and_flags_done(self): async def lines(): - for raw in [_chunk(content="he"), ": ping", _chunk(reasoning="think"), - _chunk(usage=_USAGE), "data: [DONE]"]: + for raw in [ + _chunk(content="he"), + ": ping", + _chunk(reasoning="think"), + _chunk(usage=_USAGE), + "data: [DONE]", + ]: for line in raw.splitlines(): yield line @@ -97,8 +107,12 @@ class TestSsePureFunctions: class TestStreamHappyPath: async def test_full_stream_with_usage(self): def handler(request): - return _sse_stream(_chunk(reasoning="ponder"), _chunk(content="hello"), - _chunk(content=" world"), _chunk(usage=_USAGE)) + return _sse_stream( + _chunk(reasoning="ponder"), + _chunk(content="hello"), + _chunk(content=" world"), + _chunk(usage=_USAGE), + ) result = await _complete(_transport_for(handler), _source()) assert result.content == "hello world" @@ -152,10 +166,13 @@ class TestNonStreamFastPath: def handler(request): body = json.loads(request.content) assert body.get("stream") is False and "stream_options" not in body - return httpx.Response(200, json={ - "choices": [{"message": {"content": "42", "reasoning_content": "count"}}], - "usage": _USAGE, - }) + return httpx.Response( + 200, + json={ + "choices": [{"message": {"content": "42", "reasoning_content": "count"}}], + "usage": _USAGE, + }, + ) result = await _complete(_transport_for(handler), _source(), stream=False) assert result.content == "42" and result.thinking == "count" @@ -189,7 +206,8 @@ class TestRequestShaping: return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE)) await _complete( - _transport_for(handler), _source(), + _transport_for(handler), + _source(), overlay={"response_format": {"type": "json_object"}}, ) assert seen["response_format"] == {"type": "json_object"} @@ -222,8 +240,9 @@ class TestErrorTranslation: async def test_retry_after_http_date_ignored(self): def handler(request): - return httpx.Response(429, content=b"{}", - headers={"retry-after": "Wed, 21 Oct 2026 07:28:00 GMT"}) + return httpx.Response( + 429, content=b"{}", headers={"retry-after": "Wed, 21 Oct 2026 07:28:00 GMT"} + ) with pytest.raises(TransientError) as ei: await _complete(_transport_for(handler), _source()) diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index 730a3cc..bb275e2 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -31,48 +31,89 @@ class _DummyPermit: class _DummyLimiter: - async def try_acquire(self, source_key: str, est_tokens: int): return _DummyPermit() - async def acquire(self, source_key: str, est_tokens: int): return _DummyPermit() - async def source_stats(self, source_key: str): return SourceStats(0, 0, 0) + async def try_acquire(self, source_key: str, est_tokens: int): + return _DummyPermit() + + async def acquire(self, source_key: str, est_tokens: int): + return _DummyPermit() + + async def source_stats(self, source_key: str): + return SourceStats(0, 0, 0) + async def mark_progress(self) -> None: ... - async def progress_age_s(self) -> float: return 0.0 + async def progress_age_s(self) -> float: + return 0.0 class _DummyGate: - async def try_enter(self, source_name: str, owner: str): raise NotImplementedError - async def record_success(self, entry): raise NotImplementedError - async def record_failure(self, entry, reason: str, force_open: bool): raise NotImplementedError - async def release_probe(self, entry): raise NotImplementedError - async def retry_after_s(self, sources): return 0.0 + async def try_enter(self, source_name: str, owner: str): + raise NotImplementedError + + async def record_success(self, entry): + raise NotImplementedError + + async def record_failure(self, entry, reason: str, force_open: bool): + raise NotImplementedError + + async def release_probe(self, entry): + raise NotImplementedError + + async def retry_after_s(self, sources): + return 0.0 class _DummyMw: - async def __call__(self, request, call_next): return await call_next(request) + async def __call__(self, request, call_next): + return await call_next(request) class _DummyTransport: - async def complete(self, *, messages, source, stream, overlay, call_id): raise NotImplementedError + async def complete(self, *, messages, source, stream, overlay, call_id): + raise NotImplementedError class _DummyCache: - async def get(self, key: str): return None + async def get(self, key: str): + return None + async def set(self, key: str, value: str, ttl_s: int) -> None: ... class _DummySelector: - def order(self, sources, stats): return list(sources) + def order(self, sources, stats): + return list(sources) class _DummyStrategy: - def request_overlay(self, schema): return {} - def parse(self, text: str) -> Any: return {} + def request_overlay(self, schema): + return {} + + def parse(self, text: str) -> Any: + return {} class _DummyRecorder: async def record_llm_call( - self, *, call_id, parent_call_id, session_id, model, provider, source_name, - messages, response, thinking, prompt_tokens, completion_tokens, usage_source, - latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error, cost, + self, + *, + call_id, + parent_call_id, + session_id, + model, + provider, + source_name, + messages, + response, + thinking, + prompt_tokens, + completion_tokens, + usage_source, + latency_ms, + ttft_ms, + max_inter_token_ms, + cache_hit, + error, + cost, ) -> None: ... @@ -95,10 +136,15 @@ def test_protocols_are_runtime_checkable(impl, protocol): def _decision(**overrides) -> GateDecision: - base = dict( - source_name="qwen_1", allowed=True, state=GateState.CLOSED, - epoch=0, is_probe=False, probe_owner=None, retry_after_s=0.0, - ) + base = { + "source_name": "qwen_1", + "allowed": True, + "state": GateState.CLOSED, + "epoch": 0, + "is_probe": False, + "probe_owner": None, + "retry_after_s": 0.0, + } base.update(overrides) return GateDecision(**base) @@ -141,9 +187,15 @@ class TestGateDecisionInvariants: class TestGateUpdate: def test_bounds(self): - u = GateUpdate(applied=True, state=GateState.CLOSED, epoch=0, failure_count=0, retry_after_s=0.0) + u = GateUpdate( + applied=True, state=GateState.CLOSED, epoch=0, failure_count=0, retry_after_s=0.0 + ) assert u.applied with pytest.raises(ValueError): - GateUpdate(applied=True, state=GateState.CLOSED, epoch=-1, failure_count=0, retry_after_s=0.0) + GateUpdate( + applied=True, state=GateState.CLOSED, epoch=-1, failure_count=0, retry_after_s=0.0 + ) with pytest.raises(ValueError): - GateUpdate(applied=True, state=GateState.CLOSED, epoch=0, failure_count=-1, retry_after_s=0.0) + GateUpdate( + applied=True, state=GateState.CLOSED, epoch=0, failure_count=-1, retry_after_s=0.0 + ) diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 677a1c0..346c5cb 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -19,7 +19,6 @@ from polygateway.errors import ( TransientError, ) from polygateway.middleware.retry import RetryMW -from polygateway.ports import GateState from polygateway.sources import RoundRobinSelector, SourceCooldownMemo from polygateway.types import ( BackpressurePolicy, @@ -37,18 +36,28 @@ _NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0) def _src(name, **overrides): - base = dict( - name=name, provider="openai", base_url="https://gw.example/v1", - api_key="sk", model="m", timeout_s=10.0, - ) + base = { + "name": name, + "provider": "openai", + "base_url": "https://gw.example/v1", + "api_key": "sk", + "model": "m", + "timeout_s": 10.0, + } base.update(overrides) return SourceConfig(**base) def _ok(content="ok"): return TransportResult( - content=content, thinking="", prompt_tokens=10, completion_tokens=5, - usage_source="measured", ttft_ms=12.0, max_inter_token_ms=3.0, raw={}, + content=content, + thinking="", + prompt_tokens=10, + completion_tokens=5, + usage_source="measured", + ttft_ms=12.0, + max_inter_token_ms=3.0, + raw={}, ) @@ -79,23 +88,42 @@ class FakeSleep: self.delays.append(seconds) -def _harness(sources, script, *, clock=None, max_attempts=3, quota_full="wait", - global_limits=_NO_GLOBAL, rng=lambda: 0.0): +def _harness( + sources, + script, + *, + clock=None, + max_attempts=3, + quota_full="wait", + global_limits=_NO_GLOBAL, + rng=lambda: 0.0, +): clock = clock or FakeClock() limiter = InMemoryLimiter( - scope="llm", sources={s.name: s for s in sources}, - global_limits=global_limits, lease_ttl_s=100.0, now=clock, + scope="llm", + sources={s.name: s for s in sources}, + global_limits=global_limits, + lease_ttl_s=100.0, + now=clock, ) gate = InMemoryGate(config=_BREAKER, now=clock) transport = FakeTransport(script) sleep = FakeSleep() mw = RetryMW( - scope="llm", sources=sources, selector=RoundRobinSelector(), - limiter=limiter, gate=gate, transport=transport, + scope="llm", + sources=sources, + selector=RoundRobinSelector(), + limiter=limiter, + gate=gate, + transport=transport, retry=RetryPolicy(max_attempts=max_attempts, backoff_base_s=2.0, backoff_max_s=30.0), backpressure=BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.01), - quota_full=quota_full, cooldown_memo=SourceCooldownMemo(now=clock), - emitter=None, now=clock, sleep=sleep, rng=rng, + quota_full=quota_full, + cooldown_memo=SourceCooldownMemo(now=clock), + emitter=None, + now=clock, + sleep=sleep, + rng=rng, ) return mw, limiter, gate, transport, sleep, clock @@ -140,7 +168,8 @@ class TestRetryAndFailover: async def test_max_attempts_is_total_attempts(self): mw, _, _, transport, _, _ = _harness( - [_src("a")], [TransientError("1"), TransientError("2"), TransientError("3")], + [_src("a")], + [TransientError("1"), TransientError("2"), TransientError("3")], max_attempts=3, ) with pytest.raises(AllSourcesExhausted) as ei: @@ -196,9 +225,7 @@ class TestScopeUnavailable: async def test_all_sources_circuit_open(self): clock = FakeClock() script = [TransientError(str(i)) for i in range(9)] - mw, _, gate, _, _, _ = _harness( - [_src("a")], script, clock=clock, max_attempts=99 - ) + mw, _, gate, _, _, _ = _harness([_src("a")], script, clock=clock, max_attempts=99) # 3 次失败后 a 开路 → 第 4 次尝试选不到源且 gate_rejections==全部 → CircuitOpen with pytest.raises(CircuitOpenError) as ei: await mw(_REQ) @@ -225,8 +252,11 @@ class TestScopeUnavailable: src = _src("a", max_concurrency=1) clock = FakeClock() limiter = InMemoryLimiter( - scope="llm", sources={"a": src}, global_limits=_NO_GLOBAL, - lease_ttl_s=100.0, now=clock, + scope="llm", + sources={"a": src}, + global_limits=_NO_GLOBAL, + lease_ttl_s=100.0, + now=clock, ) held = await limiter.try_acquire("a", 0) released = {"done": False} @@ -239,12 +269,20 @@ class TestScopeUnavailable: gate = InMemoryGate(config=_BREAKER, now=clock) transport = FakeTransport([_ok()]) mw = RetryMW( - scope="llm", sources=[src], selector=RoundRobinSelector(), - limiter=limiter, gate=gate, transport=transport, + scope="llm", + sources=[src], + selector=RoundRobinSelector(), + limiter=limiter, + gate=gate, + transport=transport, retry=RetryPolicy(max_attempts=3, backoff_base_s=2.0, backoff_max_s=30.0), backpressure=BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.01), - quota_full="wait", cooldown_memo=SourceCooldownMemo(now=clock), - emitter=None, now=clock, sleep=sleep_and_release, rng=lambda: 0.0, + quota_full="wait", + cooldown_memo=SourceCooldownMemo(now=clock), + emitter=None, + now=clock, + sleep=sleep_and_release, + rng=lambda: 0.0, ) resp = await mw(_REQ) assert resp.content == "ok" and released["done"] @@ -265,7 +303,8 @@ class TestCancellation: mw, _, gate, _, _, _ = _harness( [_src("a")], [TransientError("1"), TransientError("2"), TransientError("3"), "hang"], - clock=clock, max_attempts=99, + clock=clock, + max_attempts=99, ) # 三连失败开路 with pytest.raises(CircuitOpenError): diff --git a/tests/unit/test_sources.py b/tests/unit/test_sources.py index 30eef38..3982984 100644 --- a/tests/unit/test_sources.py +++ b/tests/unit/test_sources.py @@ -6,8 +6,12 @@ from polygateway.types import SourceConfig, SourceStats def _src(name): return SourceConfig( - name=name, provider="openai", base_url="https://gw.example/v1", - api_key="sk", model="m", timeout_s=10.0, + name=name, + provider="openai", + base_url="https://gw.example/v1", + api_key="sk", + model="m", + timeout_s=10.0, ) diff --git a/tests/unit/test_streaming.py b/tests/unit/test_streaming.py index f20fbab..6a98d04 100644 --- a/tests/unit/test_streaming.py +++ b/tests/unit/test_streaming.py @@ -47,7 +47,9 @@ class TestThreeLayers: async def test_inter_token_timeout(self): wrapped = stream_with_liveness_timeouts( _emit(["a", "b"], delay_s=0.2, first_delay_s=0.0), - ttft_s=1.0, inter_token_s=0.05, total_s=5.0, + ttft_s=1.0, + inter_token_s=0.05, + total_s=5.0, ) with pytest.raises(StreamLivenessTimeout) as ei: await _collect(wrapped) diff --git a/tests/unit/test_structured.py b/tests/unit/test_structured.py index dbcc0bf..889da54 100644 --- a/tests/unit/test_structured.py +++ b/tests/unit/test_structured.py @@ -21,9 +21,19 @@ class Verdict(BaseModel): def _resp(content): return LLMResponse( - content=content, thinking="", model="m", provider="p", prompt_tokens=1, - completion_tokens=2, latency_ms=10, ttft_ms=None, max_inter_token_ms=None, - cache_hit=False, call_id="cid", source_name="s1", usage_source="measured", + content=content, + thinking="", + model="m", + provider="p", + prompt_tokens=1, + completion_tokens=2, + latency_ms=10, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="cid", + source_name="s1", + usage_source="measured", ) @@ -43,10 +53,10 @@ class TestJsonRepairStrategy: @pytest.mark.parametrize( "dirty", [ - '```json\n{"answer": 1, "reason": "ok"}\n```', # 围栏 - '{"answer": 1, "reason": "ok",}', # 尾逗号 - "{'answer': 1, 'reason': 'ok'}", # 单引号 - '{"answer": 1, "reason": "ok"', # 缺右括号 + '```json\n{"answer": 1, "reason": "ok"}\n```', # 围栏 + '{"answer": 1, "reason": "ok",}', # 尾逗号 + "{'answer': 1, 'reason': 'ok'}", # 单引号 + '{"answer": 1, "reason": "ok"', # 缺右括号 ], ) def test_repairs_real_world_dirt(self, dirty): @@ -79,7 +89,7 @@ class TestNativeSchemaStrategy: def _mw(**kwargs): - defaults = dict(strategy=JsonRepairStrategy(), max_retries=1, escalation=None) + defaults = {"strategy": JsonRepairStrategy(), "max_retries": 1, "escalation": None} defaults.update(kwargs) return StructuredMW(**defaults) diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 196394d..be0245f 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -15,37 +15,80 @@ from polygateway.types import ChatRequest, LLMResponse, SourceConfig _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1") _EXPECTED_COLUMNS = [ - "call_id", "parent_call_id", "session_id", "model", "provider", "source_name", - "messages", "response", "thinking", "prompt_tokens", "completion_tokens", - "usage_source", "latency_ms", "ttft_ms", "max_inter_token_ms", "cache_hit", - "error", "cost", "created_at", + "call_id", + "parent_call_id", + "session_id", + "model", + "provider", + "source_name", + "messages", + "response", + "thinking", + "prompt_tokens", + "completion_tokens", + "usage_source", + "latency_ms", + "ttft_ms", + "max_inter_token_ms", + "cache_hit", + "error", + "cost", + "created_at", ] def _resp(**overrides): - base = dict( - content="ok", thinking="", model="m", provider="p", prompt_tokens=1, - completion_tokens=2, latency_ms=30, ttft_ms=None, max_inter_token_ms=None, - cache_hit=False, call_id="cid-1", source_name="s1", usage_source="measured", - ) + base = { + "content": "ok", + "thinking": "", + "model": "m", + "provider": "p", + "prompt_tokens": 1, + "completion_tokens": 2, + "latency_ms": 30, + "ttft_ms": None, + "max_inter_token_ms": None, + "cache_hit": False, + "call_id": "cid-1", + "source_name": "s1", + "usage_source": "measured", + } base.update(overrides) return LLMResponse(**base) def _source(): return SourceConfig( - name="s1", provider="p", base_url="https://gw.example/v1", - api_key="sk", model="m", timeout_s=10.0, + name="s1", + provider="p", + base_url="https://gw.example/v1", + api_key="sk", + model="m", + timeout_s=10.0, ) async def _record_minimal(recorder, call_id="c1", **overrides): - fields = dict( - call_id=call_id, parent_call_id=None, session_id="sess-1", model="m", - provider="p", source_name="s1", messages="[]", response="ok", thinking="", - prompt_tokens=1, completion_tokens=2, usage_source="measured", latency_ms=10, - ttft_ms=None, max_inter_token_ms=None, cache_hit=False, error=None, cost=None, - ) + fields = { + "call_id": call_id, + "parent_call_id": None, + "session_id": "sess-1", + "model": "m", + "provider": "p", + "source_name": "s1", + "messages": "[]", + "response": "ok", + "thinking": "", + "prompt_tokens": 1, + "completion_tokens": 2, + "usage_source": "measured", + "latency_ms": 10, + "ttft_ms": None, + "max_inter_token_ms": None, + "cache_hit": False, + "error": None, + "cost": None, + } fields.update(overrides) await recorder.record_llm_call(**fields) @@ -55,9 +98,9 @@ class TestSQLiteRecorder: recorder = SQLiteRecorder(tmp_path / "t.db") await _record_minimal(recorder) recorder.close() - cols = [r[1] for r in sqlite3.connect(tmp_path / "t.db").execute( - "PRAGMA table_info(llm_calls)" - )] + cols = [ + r[1] for r in sqlite3.connect(tmp_path / "t.db").execute("PRAGMA table_info(llm_calls)") + ] assert cols == _EXPECTED_COLUMNS async def test_call_id_idempotent(self, tmp_path): @@ -65,18 +108,20 @@ class TestSQLiteRecorder: await _record_minimal(recorder, call_id="dup") await _record_minimal(recorder, call_id="dup", response="second") recorder.close() - rows = sqlite3.connect(tmp_path / "t.db").execute( - "SELECT response FROM llm_calls WHERE call_id='dup'" - ).fetchall() + rows = ( + sqlite3.connect(tmp_path / "t.db") + .execute("SELECT response FROM llm_calls WHERE call_id='dup'") + .fetchall() + ) assert rows == [("ok",)] # INSERT OR IGNORE: 第二次静默忽略 async def test_concurrent_writes_all_land(self, tmp_path): recorder = SQLiteRecorder(tmp_path / "t.db") await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50))) recorder.close() - (count,) = sqlite3.connect(tmp_path / "t.db").execute( - "SELECT COUNT(*) FROM llm_calls" - ).fetchone() + (count,) = ( + sqlite3.connect(tmp_path / "t.db").execute("SELECT COUNT(*) FROM llm_calls").fetchone() + ) assert count == 50 async def test_unwritable_path_degrades_silently(self): @@ -98,8 +143,12 @@ class TestEmitter: rec = _MemoryRecorder() emitter = TelemetryEmitter(rec) await emitter.emit_attempt( - request=_REQ, source=_source(), call_id="cid-1", latency_ms=42, - response=_resp(), error=None, + request=_REQ, + source=_source(), + call_id="cid-1", + latency_ms=42, + response=_resp(), + error=None, ) row = rec.rows[0] assert row["call_id"] == "cid-1" and row["error"] is None @@ -110,8 +159,12 @@ class TestEmitter: rec = _MemoryRecorder() emitter = TelemetryEmitter(rec) await emitter.emit_attempt( - request=_REQ, source=_source(), call_id="cid-2", latency_ms=7, - response=None, error="TransientError: boom", + request=_REQ, + source=_source(), + call_id="cid-2", + latency_ms=7, + response=None, + error="TransientError: boom", ) row = rec.rows[0] assert row["error"].startswith("TransientError") @@ -121,12 +174,23 @@ class TestEmitter: rec = _MemoryRecorder() emitter = TelemetryEmitter(rec) big = "data:image/png;base64," + "A" * 100_000 - req = ChatRequest(messages=[{"role": "user", "content": [ - {"type": "image_url", "image_url": {"url": big}}, - ]}]) + req = ChatRequest( + messages=[ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": big}}, + ], + } + ] + ) await emitter.emit_attempt( - request=req, source=_source(), call_id="c", latency_ms=1, - response=None, error="x", + request=req, + source=_source(), + call_id="c", + latency_ms=1, + response=None, + error="x", ) assert len(rec.rows[0]["messages"]) < 500 # base64 不整段进库(VT R12) @@ -137,8 +201,12 @@ class TestEmitter: emitter = TelemetryEmitter(Broken()) await emitter.emit_attempt( - request=_REQ, source=_source(), call_id="c", latency_ms=1, - response=_resp(), error=None, + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=_resp(), + error=None, ) # 不抛(降级不冒泡) @@ -194,10 +262,9 @@ def test_single_emitter_discipline(): """铁律执法: record_llm_call 在 src/ 的调用点只允许出现在 telemetry emitter。""" out = subprocess.run( ["grep", "-rln", "record_llm_call(", "src/polygateway"], - capture_output=True, text=True, cwd=Path(__file__).resolve().parents[2], + capture_output=True, + text=True, + cwd=Path(__file__).resolve().parents[2], ).stdout.splitlines() - callers = [ - p for p in out - if not p.endswith(("ports.py", "telemetry/sqlite.py")) - ] + callers = [p for p in out if not p.endswith(("ports.py", "telemetry/sqlite.py"))] assert callers == ["src/polygateway/middleware/telemetry.py"] diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index c258255..de44e03 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -19,14 +19,14 @@ from polygateway.types import ( 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 = { + "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) @@ -144,7 +144,13 @@ class TestAuxTypes: 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"}, + 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"