"""Scene-based routing table (架构文档 §8.1 场景化路由). ``scene → (provider, model, temperature, max_tokens, timeout)`` is configuration, so switching models never requires a code change (NFR-MAINT-05). Defaults live here; ``LLM_SCENE_OVERRIDES_JSON`` overrides any field per scene at deploy time. Token budgets follow 架构文档 §7.2: 输入合计 ≤ 3700, 输出 ≤ 800. """ from __future__ import annotations from typing import Any from packages.llm.errors import SceneNotRoutedError from packages.llm.types import Scene, SceneConfig _DEFAULTS: dict[Scene, dict[str, Any]] = { Scene.AGENT_CHAT: { "temperature": 0.7, "max_tokens": 800, "timeout_seconds": 30.0, "prompt_id": "agent_chat", "prompt_version": "v1", "critical": False, }, Scene.INTENT_CLASSIFY: { "temperature": 0.0, "max_tokens": 128, "timeout_seconds": 10.0, "prompt_id": "intent_classify", "prompt_version": "v1", # Intent classification gates 越界拒答 (FR-AGT-02); degrading it silently # would let out-of-scope questions through, so it is never budget-gated. "critical": True, }, Scene.PREFERENCE_EXTRACT: { "temperature": 0.0, "max_tokens": 256, "timeout_seconds": 15.0, "prompt_id": "preference_extract", "prompt_version": "v1", "critical": False, }, Scene.REPORT_COPY: { "temperature": 0.6, "max_tokens": 3000, "timeout_seconds": 90.0, "prompt_id": "report_copy", "prompt_version": "v1", "critical": False, }, } def build_scene_configs( *, primary_provider: str, backup_provider: str | None = None, structured_max_retries: int = 2, models: dict[str, str] | None = None, overrides: dict[str, dict[str, Any]] | None = None, ) -> dict[Scene, SceneConfig]: """Materialise the routing table from settings.""" providers = tuple(p for p in (primary_provider, backup_provider) if p) configs: dict[Scene, SceneConfig] = {} for scene, defaults in _DEFAULTS.items(): merged: dict[str, Any] = { **defaults, "structured_max_retries": structured_max_retries, **(overrides or {}).get(scene.value, {}), } configs[scene] = SceneConfig( scene=scene, providers=tuple(merged.pop("providers", providers)), models=dict(merged.pop("models", models or {})), **merged, ) return configs class SceneRouter: def __init__(self, configs: dict[Scene, SceneConfig]) -> None: self._configs = dict(configs) def resolve(self, scene: Scene) -> SceneConfig: config = self._configs.get(scene) if config is None: msg = f"no routing configured for scene {scene.value}" raise SceneNotRoutedError(msg) return config def scenes(self) -> tuple[Scene, ...]: return tuple(self._configs)