"""配置聚合与装配期校验(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 json import os from dataclasses import dataclass from typing import TYPE_CHECKING from dotenv import dotenv_values from loguru import logger 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"), "EXTRA_BODY": ("extra_body", "json"), } _RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE"}) _SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"}) _QUOTA_FULL = frozenset({"wait", "fail_fast"}) # 后端合法域: env 解析与构造期校验共用一份定义,避免两处分叉 _LIMITER_BACKENDS = frozenset({"memory", "redis"}) _BREAKER_BACKENDS = frozenset({"memory", "redis"}) _CACHE_BACKENDS = frozenset({"redis", "memory", "none"}) _TELEMETRY_BACKENDS = frozenset({"sqlite", "postgres", "none"}) _REDIS_DEPENDENT_BACKENDS = ("limiter_backend", "breaker_backend", "cache_backend") # 背压默认(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 同源 _PROBE_GRACE_S = 5.0 # 半开探针租约相对最慢调用的清理宽限(CHS container.py:274-275) 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}") if kind == "json": # JSONDecodeError 是 ValueError 子类,复用下方的统一包装 parsed = json.loads(raw) if not isinstance(parsed, dict): raise ValueError(f"必须是 JSON 对象(而非数组/标量): {raw!r}") return parsed 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 的完整装配配置;**任何**构造路径都通过全部装配守卫(ARCH §7.3)。 守卫校验的是**跨字段**不变量: 单看一个字段都合法,组合起来才会在运行时 咬人(租约先于请求过期、正常慢首包被误判卡死、半开探针在途被接管)。 types.py 各子配置的 `__post_init__` 只看得见自己的字段,故由本类把关。 放在 `__post_init__` 而非某个工厂里: 这些约束是本类定义的一部分,不是 某个入口的输入检查。挂在构造期,直接构造、`dataclasses.replace` 与全部 装配工厂一并覆盖;挂在工厂里则每加一个工厂就多一处要同步。 """ 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 telemetry_pg_dsn: str | None redis_url: str | None pricing_path: str | None structured_max_retries: int lease_ttl_s: float def __post_init__(self) -> None: self._normalize() self._validate_identity() self._validate_backends() self._validate_cache() self._validate_telemetry() self._validate_lease() self._validate_stall() self._validate_probe() def _normalize(self) -> None: """把 `from_env` 一直在做的规范化补到构造路上,两条路必须产出同一个值。 `scope` 的 strip 才是要紧的那一半: 它进 Redis key(`pgw:limit:{scope}:…` /`pgw:gate:{scope}:…`),而两个 Redis 后端在构造函数里只 `.lower()` **不 strip** ——`"llm "` 会产出 `pgw:limit:llm :…`,与 `from_env` 路的进程分裂成两套命名空间。 大小写则不会: 后端自 v1.0.0 起各自 lower,`from_settings` 传 "LLM" 也落在同一 套 key 上(此处 lower 只为让 `GatewaySettings.scope` 属性两路取值一致)。 空串归 None 同理: 留着空串会骗过 `is None` 判断,把错误推迟到 redis 客户端 抛连接串解析异常。`telemetry_pg_dsn` 的驱动后缀因为要看 backend 且需告警, 规范化留在 `_validate_telemetry`。 """ normalized_scope = self.scope.strip().lower() if normalized_scope != self.scope: object.__setattr__(self, "scope", normalized_scope) for field in ("redis_url", "pricing_path"): if getattr(self, field) == "": object.__setattr__(self, field, None) def _validate_identity(self) -> None: """本类自身字段的基本域: 空 scope 会污染遥测与缓存命名空间;零源必然选源失败。""" if not self.scope.strip(): raise ValueError("GatewaySettings.scope 不能为空") if not self.sources: raise ValueError("GatewaySettings.sources 不能为空: 至少一个源") if self.structured_max_retries < 0: raise ValueError(f"structured_max_retries 不能为负: {self.structured_max_retries}") def _validate_backends(self) -> None: """后端选择必须落在合法域内,取 redis 的还必须有连接串。 域外取值此前只有 `from_env` 拦得住,直接构造会一路走到 `client.py` 的 `_build_*`,落进 else 分支静默不建后端,或撞上那里的断言。 """ for field, allowed in ( ("limiter_backend", _LIMITER_BACKENDS), ("breaker_backend", _BREAKER_BACKENDS), ("cache_backend", _CACHE_BACKENDS), ("telemetry_backend", _TELEMETRY_BACKENDS), ("selector", _SELECTORS), ("quota_full", _QUOTA_FULL), ): value = getattr(self, field) if value not in allowed: raise ValueError(f"{field} 非法值 {value!r};允许: {sorted(allowed)}") on_redis = [f for f in _REDIS_DEPENDENT_BACKENDS if getattr(self, f) == "redis"] if on_redis and self.redis_url is None: raise ValueError(f"{'、'.join(on_redis)} 取 redis 时必须提供 redis_url") def _validate_cache(self) -> None: """启用缓存必须有命名空间与正 TTL(缺命名空间即失去租户隔离,会毒化缓存)。""" if self.cache_backend == "none": return if not self.cache_namespace: raise ValueError("启用缓存时 cache_namespace 不能为空: 缓存 key 靠它做租户隔离") if self.cache_ttl_s is None or self.cache_ttl_s <= 0: raise ValueError(f"cache_ttl_s 必须 > 0(禁止永不过期): {self.cache_ttl_s}") def _validate_telemetry(self) -> None: """遥测后端各自的落点必填;顺带剥掉 asyncpg 不认的 SQLAlchemy 驱动后缀。 剥而不是拒: 两条装配路对同一 DSN 应产出同一结果。但不静默——`from_env` 那条路在 `_load_pg_dsn` 就剥干净了,能走到这里的只有手工构造的调用方, 他有权知道库动了他给的值。 """ if self.telemetry_backend == "sqlite" and not self.telemetry_sqlite_path: raise ValueError("telemetry_backend=sqlite 时必须提供 telemetry_sqlite_path") if self.telemetry_backend != "postgres": return if not self.telemetry_pg_dsn: raise ValueError("telemetry_backend=postgres 时必须提供 telemetry_pg_dsn") stripped = _strip_dsn_driver(self.telemetry_pg_dsn) if stripped != self.telemetry_pg_dsn: # 只报 scheme 段: DSN 带密码,整串不得进日志(P5 敏感信息只走 .env) logger.warning( "telemetry_pg_dsn 的 scheme 含 asyncpg 不认的驱动后缀,已由 {} 剥为 {}", self.telemetry_pg_dsn.partition("://")[0], stripped.partition("://")[0], ) object.__setattr__(self, "telemetry_pg_dsn", stripped) def _validate_lease(self) -> None: """调用超时须 ≤ permit 租约 TTL,防租约先于请求过期使并发超出配额。""" slowest = max(s.timeout_s for s in self.sources) if slowest > self.lease_ttl_s: raise ValueError( f"源最大 timeout_s({slowest})超过 permit 租约 lease_ttl_s" f"({self.lease_ttl_s});调大 lease_ttl_s 或调小源的 timeout_s" ) def _validate_stall(self) -> None: """stall 窗口须 ≥ 最慢源 TTFT 上限,防把正常慢首包误判为卡死。""" ttfts = [s.ttft_timeout_s for s in self.sources if s.ttft_timeout_s is not None] if ttfts and self.backpressure.stall_window_s < max(ttfts): raise ValueError( f"backpressure.stall_window_s({self.backpressure.stall_window_s})须 ≥ " f"最大源 ttft_timeout_s({max(ttfts)});调大 stall_window_s 或调小 ttft_timeout_s" ) def _validate_probe(self) -> None: """半开探针租约须撑过一次最慢调用,否则探针在途即被接管(M2 设计 §3)。""" floor = max(s.timeout_s for s in self.sources) + _PROBE_GRACE_S if self.breaker.probe_ttl_s < floor: raise ValueError( f"breaker.probe_ttl_s({self.breaker.probe_ttl_s})须 ≥ 最慢源 " f"timeout_s + {_PROBE_GRACE_S}({floor});调大 probe_ttl_s 或调小源的 timeout_s" ) @classmethod def from_env( cls, scope: str = "LLM", env: Mapping[str, str] | None = None, *, env_file: str = ".env", ) -> GatewaySettings: """聚合 env(缺省 .env + os.environ,后者优先);守卫由 `__post_init__` 执行。""" 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) return 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, "health_aware"), quota_full=_load_choice(env, f"{scope_u}__QUOTA_FULL", _QUOTA_FULL, "wait"), **_load_pgw(env), ) 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)。M2.5 修正: 只看源级并发——M2 曾用 # 全局并发抬升(SOAK 100→阈值 200)使熔断失灵(病灶 2,设计 2026-07-21-m25) concurrency = max((s.max_concurrency for s in sources), default=0) if concurrency > 0: threshold = max(threshold, concurrency * 2) slowest = max(s.timeout_s for s in sources) probe_floor = slowest + _PROBE_GRACE_S probe = _first(env, f"{scope}__BREAKER__PROBE_TTL_S") if probe is not None: # 配置值不在此校验: 探针租约下限是跨字段不变量,由 GatewaySettings._validate_probe # 统一把关(否则直接构造那条装配路会绕过) probe_ttl_s = float(_cast(probe[1], "float", probe[0])) else: # 派生规则: 探针租约须撑过一次最慢调用,且不短于冷却期(第三项保证守卫恒成立) probe_ttl_s = max(2 * slowest, cooldown_s, probe_floor) # M2.5 失败率通道参数(可选键,库缺省——韧性参数缺省先例同 backpressure) raw_min = env.get(f"{scope}__BREAKER__MIN_CALLS") min_calls = int(_cast(raw_min, "int", f"{scope}__BREAKER__MIN_CALLS")) if raw_min else 10 fail_rate = _opt_float(env, f"{scope}__BREAKER__FAIL_RATE", 0.6) window_s = _opt_float(env, f"{scope}__BREAKER__WINDOW_S", 60.0) max_cooldown_s = _opt_float(env, f"{scope}__BREAKER__MAX_COOLDOWN_S", max(300.0, cooldown_s)) return BreakerConfig( fail_threshold=threshold, cooldown_s=cooldown_s, probe_ttl_s=probe_ttl_s, min_calls=min_calls, fail_rate=fail_rate, window_s=window_s, max_cooldown_s=max_cooldown_s, ) def _opt_float(env: Mapping[str, str], key: str, default: float) -> float: """可选韧性参数: 缺省用库值,显式配置则解析(坏值 fail-loud)。""" raw = env.get(key) if raw is None or raw == "": return default return float(_cast(raw, "float", key)) 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]: # 合法域与构造期守卫共用常量;此处的检查保留是为了报错能点出 env 键名, # 构造期那道点的是字段名(两类调用方各看得懂自己那套) limiter_backend = _load_choice(env, "PGW_LIMITER_BACKEND", _LIMITER_BACKENDS, "memory") breaker_backend = _load_choice(env, "PGW_BREAKER_BACKEND", _BREAKER_BACKENDS, "memory") _, cache_backend = _require(env, "PGW_CACHE_BACKEND") _, telemetry_backend = _require(env, "PGW_TELEMETRY_BACKEND") if cache_backend not in _CACHE_BACKENDS: raise ValueError(f"PGW_CACHE_BACKEND 非法值 {cache_backend!r}") if telemetry_backend not in _TELEMETRY_BACKENDS: raise ValueError(f"PGW_TELEMETRY_BACKEND 非法值 {telemetry_backend!r}") redis_url = env.get("REDIS_URL") or None if "redis" in (limiter_backend, breaker_backend) and redis_url is None: raise ValueError("缺关键配置: 限流/熔断后端取 redis 需设置 REDIS_URL") 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, "telemetry_pg_dsn": _load_pg_dsn(env) if telemetry_backend == "postgres" else None, "redis_url": redis_url, "pricing_path": env.get("PGW_PRICING_PATH") or None, "structured_max_retries": _load_structured_retries(env), "lease_ttl_s": _load_lease_ttl(env), } def _strip_dsn_driver(dsn: str) -> str: """剥 SQLAlchemy 风格的 `+driver` 后缀(asyncpg 不认);已干净的原样返回。""" scheme, sep, rest = dsn.partition("://") return f"{scheme.partition('+')[0]}{sep}{rest}" def _load_pg_dsn(env: Mapping[str, str]) -> str: """读取 Postgres DSN 并剥驱动后缀。 env 路在此剥干净,构造期那道就无事可做——三项目 `.env` 里的 SQLAlchemy 写法不会每次装配都刷一条 warning。 """ _, dsn = _require(env, "PGW_TELEMETRY_PG_DSN") return _strip_dsn_driver(dsn) 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 2 # M2.5 迭代4: 1→2 if value < 0: raise ValueError("PGW_STRUCTURED_MAX_RETRIES 不能为负") return value 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 @dataclass(frozen=True) class EmbeddingSettings: """Embedding scope 装配配置(M2 §7): 复用 GatewaySettings + embedding 专用键。 专用键不进 GatewaySettings(LLM scope 不受影响): `{SCOPE}__BATCH_SIZE` 必填(分批是行为关键,不设默认)、`{SCOPE}__NORMALIZE`/`{SCOPE}__EXPECTED_DIM` 可选。cache/structured 键对 embedding 无意义,装配时忽略。 """ gateway: GatewaySettings batch_size: int normalize: bool = False expected_dim: int | None = None def __post_init__(self) -> None: """自身字段的域校验;内嵌的 gateway 由 `GatewaySettings.__post_init__` 自己把关。""" if self.batch_size < 1: raise ValueError(f"EmbeddingSettings.batch_size 必须 ≥ 1: {self.batch_size}") if self.expected_dim is not None and self.expected_dim < 1: raise ValueError(f"EmbeddingSettings.expected_dim 必须 ≥ 1: {self.expected_dim}") @classmethod def from_env( cls, scope: str = "EMBED", env: Mapping[str, str] | None = None, *, env_file: str = ".env", ) -> EmbeddingSettings: 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() gateway = GatewaySettings.from_env(scope_u, env=env) key, raw = _require(env, f"{scope_u}__BATCH_SIZE") batch_size = int(_cast(raw, "int", key)) if batch_size < 1: raise ValueError(f"{scope_u}__BATCH_SIZE 必须 ≥ 1") norm = _first(env, f"{scope_u}__NORMALIZE") dim = _first(env, f"{scope_u}__EXPECTED_DIM") expected_dim = int(_cast(dim[1], "int", dim[0])) if dim else None if expected_dim is not None and expected_dim < 1: raise ValueError(f"{scope_u}__EXPECTED_DIM 必须 ≥ 1") return cls( gateway=gateway, batch_size=batch_size, normalize=bool(_cast(norm[1], "bool", norm[0])) if norm else False, expected_dim=expected_dim, ) @dataclass(frozen=True) class OcrSettings: """OCR scope 装配配置(M3 设计 §3.4): 复用 GatewaySettings,无 OCR 专用键。 cache/structured/pricing 键对 OCR 无意义,装配时忽略;TPM 闸不启用 (tpm=0)、看门狗键不配(非流式)由源键缺省自然达成。api_key 惯例填 "none"(MonkeyOCR 无鉴权,SourceConfig 非空校验用占位)。 """ gateway: GatewaySettings @classmethod def from_env( cls, scope: str = "OCR", env: Mapping[str, str] | None = None, *, env_file: str = ".env", ) -> OcrSettings: if env is None: env = { k: v for k, v in {**dotenv_values(env_file), **os.environ}.items() if v is not None } return cls(gateway=GatewaySettings.from_env(scope.upper(), env=env))