diff --git a/.gitignore b/.gitignore index 5f9b831..9f66ebd 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,6 @@ Thumbs.db # 安全扫描器工作目录（不入库） .mimosa/ + +# 参考项目目录（不入库，仅作设计吸收来源） +extra/ diff --git a/gateway/api.py b/gateway/api.py index 38cd2aa..faf72ed 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -1,94 +1,95 @@ -"""FastAPI 网关：对外提供 /chat /health /metrics 接口。 - -启动： - uvicorn gateway.api:app --host 0.0.0.0 --port 8000 -或： - python -m gateway.api - -依赖：fastapi, uvicorn, pydantic（见 requirements.txt） -""" -from __future__ import annotations - -from typing import List, Optional - -from pydantic import BaseModel, Field - -from router_system.config import load_config -from router_system.router import Router, build_router - -# ---- 全局单例 ---- -_router: Optional[Router] = None - - -def get_router() -> Router: - global _router - if _router is None: - _router = build_router() - return _router - - -# ---- 请求/响应模型 ---- -class QueryRequest(BaseModel): - query: str = Field(..., min_length=1, max_length=8000, description="用户查询") - - -class QueryResponse(BaseModel): - response: str - domain: str - difficulty: str - confidence: float - upgraded: bool - quality_score: float - model_used: str - route: List[str] - latency_ms: float - cache_hit: bool - cache_level: Optional[str] - cost_est: float - error: Optional[str] = None - - -class HealthResponse(BaseModel): - status: str - domains: List[str] - classifier: str - judge: str - fallback: str - - -# ---- FastAPI 应用 ---- -try: - from fastapi import FastAPI - - app = FastAPI( - title="Multi-Expert Router API", - description="多专业小模型 + 路由模型系统（MVP）", - version="0.1.0", - ) - - @app.get("/health", response_model=HealthResponse, tags=["system"]) - async def health(): - return get_router().health() - - @app.post("/chat", response_model=QueryResponse, tags=["chat"]) - async def chat(req: QueryRequest): - result = await get_router().route(req.query) - return QueryResponse(**result.to_dict()) - - @app.get("/metrics", tags=["system"]) - async def metrics(): - r = get_router() - return { - "router": r.stats.summary(), - "cache": r.cache.stats(), - } - -except ImportError: - # fastapi 未安装时，提供 CLI 入口提示 - app = None - print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt") - - -if __name__ == "__main__": - import uvicorn - uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False) +"""FastAPI 网关：对外提供 /chat /health /metrics 接口。 + +启动： + uvicorn gateway.api:app --host 0.0.0.0 --port 8000 +或： + python -m gateway.api + +依赖：fastapi, uvicorn, pydantic（见 requirements.txt） +""" +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + +from router_system.config import load_config +from router_system.router import Router, build_router + +# ---- 全局单例 ---- +_router: Optional[Router] = None + + +def get_router() -> Router: + global _router + if _router is None: + _router = build_router() + return _router + + +# ---- 请求/响应模型 ---- +class QueryRequest(BaseModel): + query: str = Field(..., min_length=1, max_length=8000, description="用户查询") + + +class QueryResponse(BaseModel): + response: str + domain: str + difficulty: str + confidence: float + upgraded: bool + quality_score: float + model_used: str + route: List[str] + latency_ms: float + cache_hit: bool + cache_level: Optional[str] + cost_est: float + error: Optional[str] = None + route_explanation: Optional[Dict[str, Any]] = None # T-R1 可解释评分（缓存命中为 None） + + +class HealthResponse(BaseModel): + status: str + domains: List[str] + classifier: str + judge: str + fallback: str + + +# ---- FastAPI 应用 ---- +try: + from fastapi import FastAPI + + app = FastAPI( + title="Multi-Expert Router API", + description="多专业小模型 + 路由模型系统（MVP）", + version="0.1.0", + ) + + @app.get("/health", response_model=HealthResponse, tags=["system"]) + async def health(): + return get_router().health() + + @app.post("/chat", response_model=QueryResponse, tags=["chat"]) + async def chat(req: QueryRequest): + result = await get_router().route(req.query) + return QueryResponse(**result.to_dict()) + + @app.get("/metrics", tags=["system"]) + async def metrics(): + r = get_router() + return { + "router": r.stats.summary(), + "cache": r.cache.stats(), + } + +except ImportError: + # fastapi 未安装时，提供 CLI 入口提示 + app = None + print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt") + + +if __name__ == "__main__": + import uvicorn + uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False) diff --git a/router_system/experts.py b/router_system/experts.py index 679287a..405a79d 100644 --- a/router_system/experts.py +++ b/router_system/experts.py @@ -75,6 +75,7 @@ _STOPWORDS_EN = { class Expert: name: str = "expert" + nominal_latency_ms: float = 100.0 # 标称延迟元数据（可解释评分 speed 维输入） async def generate(self, query: str, difficulty: str) -> ExpertResponse: raise NotImplementedError @@ -87,6 +88,8 @@ class MockExpert(Expert): 使端到端管线（分类 -> 专家 -> Judge -> 缓存）可被稳定测试与演示。 """ + nominal_latency_ms = 1.0 + def __init__(self, name: str, domain: str, model: str = "mock"): self.name = name self.domain = domain @@ -162,6 +165,8 @@ class MockExpert(Expert): class HFExpert(Expert): """可选：HuggingFace 真实小模型（需 requirements-ml.txt）。""" + nominal_latency_ms = 300.0 + def __init__(self, name: str, domain: str, model: str): self.name = name self.domain = domain @@ -210,6 +215,8 @@ class HFExpert(Expert): class APIExpert(Expert): """可选：OpenAI 兼容 Chat Completions（DeepSeek / OpenAI / 本地 vLLM）。""" + nominal_latency_ms = 800.0 + def __init__(self, name: str, domain: str, model: str, base_url: str, api_key: str): self.name = name self.domain = domain diff --git a/router_system/explain.py b/router_system/explain.py new file mode 100644 index 0000000..d759f5a --- /dev/null +++ b/router_system/explain.py @@ -0,0 +1,106 @@ +"""可解释路由评分层（T-R1，采纳 ai-model-router 五维评分 + 硬过滤带拒绝理由设计）。 + +对分类产生的候选领域专家做透明评分与解释。路由胜负保持既有逻辑不变 +（本模块是纯解释层，不参与决策）： + +- 硬过滤先行：无规则命中/低置信度/缺专家的候选先淘汰，逐条记录人话拒绝理由 +- 五维加权评分（各维归一到 0-1，权重可配）： + capability 分类器该领域原始分归一（语义匹配强度） + cost_efficiency 1/(1+cost*100) 平滑映射，零成本本地模型恒 1.0 + speed 1/(1+标称延迟/200)（200ms 参考延迟） + reliability Judge 通过率（无数据时 0.9 中性先验） + quality 难度 × 领域纵深启发式 +- 置信度可推导：confidence = 0.8 + (最高分 - 次高分)，分差越大越自信（封顶 0.99） +- 同分决胜按领域名字典序（与仓库既有确定性约定一致） + +输出 RoutingExplanation.to_dict() 直接进 /chat 响应的 route_explanation 字段。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +# 领域纵深启发式（该领域知识可承载的任务深度，0-1） +_DOMAIN_DEPTH = { + "code": 0.90, "math": 0.90, "legal": 0.85, "medical": 0.85, "general": 0.60, +} + +DEFAULT_WEIGHTS: Dict[str, float] = { + "capability": 0.35, + "cost_efficiency": 0.20, + "speed": 0.15, + "reliability": 0.20, + "quality": 0.10, +} + + +@dataclass +class CandidateProfile: + """一个候选领域专家的评分输入（由 Router 从分类/专家池/统计装配）。""" + domain: str + model: str # 专家模型名 + capability: float # 分类器该领域原始分归一 0-1 + cost_per_mtok: float # $ / 1M output tokens 估计 + nominal_latency_ms: float # 标称延迟（专家后端元数据） + reliability: float # Judge 通过率（无数据给中性先验） + difficulty_score: float # 查询难度分 0-1（质量维输入） + hard_fail_reason: Optional[str] = None # 非 None 即硬过滤淘汰（人话理由） + + +@dataclass +class RoutingExplanation: + """一次路由的完整解释：胜者 + 排名 + 拒绝名单 + 权重快照。""" + winner: str # 胜出领域；无存活候选时为 "fallback" + confidence: float # 0.8 + 分差推导（封顶 0.99） + ranked: List[Dict[str, Any]] = field(default_factory=list) + rejected: List[Dict[str, Any]] = field(default_factory=list) + weights: Dict[str, float] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "winner": self.winner, + "confidence": round(self.confidence, 4), + "ranked": self.ranked, + "rejected": self.rejected, + "weights": dict(self.weights), + } + + +def score_candidate(c: CandidateProfile, weights: Dict[str, float]) -> Dict[str, Any]: + """单候选五维评分（全部绝对映射 0-1，跨候选可比、可单测）。""" + dims = { + "capability": max(0.0, min(1.0, c.capability)), + "cost_efficiency": 1.0 / (1.0 + max(0.0, c.cost_per_mtok) * 100.0), + "speed": 1.0 / (1.0 + max(0.0, c.nominal_latency_ms) / 200.0), + "reliability": max(0.0, min(1.0, c.reliability)), + "quality": max(0.0, min(1.0, c.difficulty_score)) * _DOMAIN_DEPTH.get(c.domain, 0.60), + } + total = sum(dims[k] * weights.get(k, 0.0) for k in dims) + return {"domain": c.domain, "model": c.model, + "total": round(total, 4), "dims": {k: round(v, 4) for k, v in dims.items()}} + + +def explain_routing(candidates: List[CandidateProfile], + weights: Optional[Dict[str, float]] = None) -> RoutingExplanation: + """硬过滤 -> 五维加权 -> 排名与置信度推导（确定性：同分按领域名字典序）。""" + w = dict(DEFAULT_WEIGHTS if weights is None else weights) + rejected: List[Dict[str, Any]] = [] + alive: List[CandidateProfile] = [] + for c in candidates: + if c.hard_fail_reason is not None: + rejected.append({"domain": c.domain, "reason": c.hard_fail_reason}) + else: + alive.append(c) + + scored = [score_candidate(c, w) for c in alive] + scored.sort(key=lambda s: (-s["total"], s["domain"])) + + if not scored: + # 全部被硬过滤（含低置信度直连回退路径）：决策即走大模型回退 + return RoutingExplanation(winner="fallback", confidence=0.80, + ranked=[], rejected=rejected, weights=w) + + gap = scored[0]["total"] - (scored[1]["total"] if len(scored) > 1 else 0.0) + confidence = min(0.99, 0.80 + max(0.0, gap)) + return RoutingExplanation(winner=scored[0]["domain"], confidence=confidence, + ranked=scored, rejected=rejected, weights=w) diff --git a/router_system/models.py b/router_system/models.py index e925f5e..4446b62 100644 --- a/router_system/models.py +++ b/router_system/models.py @@ -1,68 +1,70 @@ -"""核心数据模型（纯标准库，无外部依赖）""" -from __future__ import annotations - -import time -from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional - - -@dataclass -class Classification: - """分类器输出：领域 + 置信度 + 难度""" - domain: str - confidence: float - difficulty: str # easy | medium | hard - difficulty_score: float = 0.5 - raw_scores: Dict[str, float] = field(default_factory=dict) - matched_rules: List[str] = field(default_factory=list) - - -@dataclass -class ExpertResponse: - """专家模型输出""" - text: str - model_used: str - latency_ms: float = 0.0 - tokens: int = 0 - cost_est: float = 0.0 # 相对成本估计（美元，近似） - - -@dataclass -class RouterResult: - """一次路由的完整结果""" - query: str - response: str - domain: str - difficulty: str - confidence: float - upgraded: bool # 是否升级到大模型 - quality_score: float - model_used: str - route: List[str] = field(default_factory=list) # 路由决策轨迹 - latency_ms: float = 0.0 - cache_hit: bool = False - cache_level: Optional[str] = None # exact | semantic - cost_est: float = 0.0 - error: Optional[str] = None - - def to_dict(self) -> Dict[str, Any]: - return { - "query": self.query, - "response": self.response, - "domain": self.domain, - "difficulty": self.difficulty, - "confidence": round(self.confidence, 4), - "upgraded": self.upgraded, - "quality_score": round(self.quality_score, 4), - "model_used": self.model_used, - "route": self.route, - "latency_ms": round(self.latency_ms, 2), - "cache_hit": self.cache_hit, - "cache_level": self.cache_level, - "cost_est": round(self.cost_est, 6), - "error": self.error, - } - - -def now_ms() -> float: - return time.perf_counter() * 1000.0 +"""核心数据模型（纯标准库，无外部依赖）""" +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class Classification: + """分类器输出：领域 + 置信度 + 难度""" + domain: str + confidence: float + difficulty: str # easy | medium | hard + difficulty_score: float = 0.5 + raw_scores: Dict[str, float] = field(default_factory=dict) + matched_rules: List[str] = field(default_factory=list) + + +@dataclass +class ExpertResponse: + """专家模型输出""" + text: str + model_used: str + latency_ms: float = 0.0 + tokens: int = 0 + cost_est: float = 0.0 # 相对成本估计（美元，近似） + + +@dataclass +class RouterResult: + """一次路由的完整结果""" + query: str + response: str + domain: str + difficulty: str + confidence: float + upgraded: bool # 是否升级到大模型 + quality_score: float + model_used: str + route: List[str] = field(default_factory=list) # 路由决策轨迹 + latency_ms: float = 0.0 + cache_hit: bool = False + cache_level: Optional[str] = None # exact | semantic + cost_est: float = 0.0 + error: Optional[str] = None + route_explanation: Optional[Dict[str, Any]] = None # 可解释评分（T-R1，缓存命中为 None） + + def to_dict(self) -> Dict[str, Any]: + return { + "query": self.query, + "response": self.response, + "domain": self.domain, + "difficulty": self.difficulty, + "confidence": round(self.confidence, 4), + "upgraded": self.upgraded, + "quality_score": round(self.quality_score, 4), + "model_used": self.model_used, + "route": self.route, + "latency_ms": round(self.latency_ms, 2), + "cache_hit": self.cache_hit, + "cache_level": self.cache_level, + "cost_est": round(self.cost_est, 6), + "error": self.error, + "route_explanation": self.route_explanation, + } + + +def now_ms() -> float: + return time.perf_counter() * 1000.0 diff --git a/router_system/router.py b/router_system/router.py index 79410ab..e40147f 100644 --- a/router_system/router.py +++ b/router_system/router.py @@ -45,6 +45,8 @@ class Router: self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60) self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70) self.cache_enabled = cfg.get("cache", {}).get("enabled", True) + # T-R1：可解释路由评分层（纯解释，不影响决策；默认开启） + self.explain_enabled = bool(rcfg.get("explain_enabled", True)) # --------------------------------------------------------------- async def route(self, query: str) -> RouterResult: @@ -80,10 +82,16 @@ class Router: classification = self.classifier.classify(query) route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}") + # T-R1：候选装配与可解释评分（纯解释层，不改变后续任何决策分支） + explanation = self._explain(classification) if self.explain_enabled else None + if explanation is not None: + route.append(f"explain:{explanation.winner}@{explanation.confidence:.2f}") + # 低置信度 -> 直接走大模型 if self.classifier.should_fallback(classification, self.low_confidence_threshold): route.append("direct_fallback") - return await self._fallback_result(query, classification, route, start) + return await self._fallback_result(query, classification, route, start, + explanation=explanation) # ---- Step 3: 选择专家 ---- domain = classification.domain @@ -101,7 +109,7 @@ class Router: self.stats.record_error() route.append(f"expert_error:{type(e).__name__}") return await self._fallback_result(query, classification, route, start, - error=str(e)) + error=str(e), explanation=explanation) # ---- Step 5: Judge 评估 ---- try: @@ -123,19 +131,46 @@ class Router: latency = now_ms() - start result = self._finalize(query, classification, final_resp, quality_score=quality_score, upgraded=upgraded, route=route, latency_ms=latency, - model_used=final_resp.model_used, cost_est=final_resp.cost_est) + model_used=final_resp.model_used, cost_est=final_resp.cost_est, + explanation=explanation) self._record(result, latency) - # 未升级的结果写缓存 + # 未升级的结果写缓存（解释不随缓存复用——缓存命中路径 explanation=None） if self.cache_enabled and not upgraded and result.response: self.cache.put(query, result.to_dict()) return result + # --------------------------------------------------------------- + def _explain(self, classification: Classification): + """装配候选画像并生成可解释评分（T-R1；任何异常不影响主链路）。""" + try: + from .explain import CandidateProfile, explain_routing + from .experts import _cost_for + + raw = classification.raw_scores or {} + raw_max = max(raw.values()) if raw else 0.0 + candidates = [] + for dom, expert in self.experts.items(): + cap = (raw.get(dom, 0.0) / raw_max) if raw_max > 0 else 0.0 + reason = "无规则命中（capability=0）" if cap <= 0.0 else None + candidates.append(CandidateProfile( + domain=dom, model=getattr(expert, "model", expert.name), + capability=cap, + cost_per_mtok=_cost_for(getattr(expert, "model", "mock")), + nominal_latency_ms=getattr(expert, "nominal_latency_ms", 100.0), + reliability=self.stats.domain_reliability(dom), + difficulty_score=classification.difficulty_score, + hard_fail_reason=reason)) + return explain_routing(candidates) + except Exception: # noqa: BLE001 解释层故障不影响路由（D-G4 同款纪律） + return None + # --------------------------------------------------------------- async def _fallback_result(self, query: str, classification: Classification, route: list, start: float, - error: Optional[str] = None) -> RouterResult: + error: Optional[str] = None, + explanation=None) -> RouterResult: """兜底路径的公共收尾：调用大模型回退 -> finalize -> 记录指标。 低置信度直连、专家异常两条路径共用，避免收尾逻辑三处重复。 @@ -145,7 +180,7 @@ class Router: result = self._finalize(query, classification, fb, quality_score=0.0, upgraded=True, route=route, latency_ms=latency, model_used=fb.model_used, cost_est=fb.cost_est, - error=error) + error=error, explanation=explanation) self._record(result, latency) return result @@ -164,7 +199,7 @@ class Router: def _finalize(query: str, classification: Classification, resp: ExpertResponse, quality_score: float, upgraded: bool, route: list, latency_ms: float, model_used: str, cost_est: float, - error: Optional[str] = None) -> RouterResult: + error: Optional[str] = None, explanation=None) -> RouterResult: return RouterResult( query=query, response=resp.text, @@ -179,6 +214,7 @@ class Router: cache_hit=False, cost_est=cost_est, error=error, + route_explanation=explanation.to_dict() if explanation is not None else None, ) def _record(self, result: RouterResult, latency_ms: float): diff --git a/router_system/stats.py b/router_system/stats.py index 5ffd70c..7f54f85 100644 --- a/router_system/stats.py +++ b/router_system/stats.py @@ -1,63 +1,74 @@ -"""运行指标收集（线程安全，零依赖）。""" -from __future__ import annotations - -import threading -from collections import Counter, deque -from typing import Any, Deque, Dict - - -class Stats: - def __init__(self, window: int = 1000): - self._lock = threading.Lock() - self.requests = 0 - self.domain_counter: Counter = Counter() - self.difficulty_counter: Counter = Counter() - self.upgraded = 0 - self.cache_hits = 0 - self.cache_levels: Counter = Counter() - self.errors = 0 - self.latencies: Deque[float] = deque(maxlen=window) - self.cost_total = 0.0 - self.model_usage: Counter = Counter() - - def record(self, latency_ms: float, domain: str, difficulty: str, - upgraded: bool, cache_hit: bool, cache_level: str | None, - cost_est: float, model_used: str): - with self._lock: - self.requests += 1 - self.domain_counter[domain] += 1 - self.difficulty_counter[difficulty] += 1 - if upgraded: - self.upgraded += 1 - if cache_hit: - self.cache_hits += 1 - if cache_level: - self.cache_levels[cache_level] += 1 - self.latencies.append(latency_ms) - self.cost_total += cost_est - self.model_usage[model_used] += 1 - - def record_error(self): - with self._lock: - self.errors += 1 - - def summary(self) -> Dict[str, Any]: - with self._lock: - n = self.requests - lat = list(self.latencies) - avg_lat = sum(lat) / len(lat) if lat else 0.0 - p99 = sorted(lat)[int(len(lat) * 0.99) - 1] if len(lat) >= 100 else (max(lat) if lat else 0.0) - return { - "total_requests": n, - "domain_distribution": dict(self.domain_counter), - "difficulty_distribution": dict(self.difficulty_counter), - "fallback_rate": round(self.upgraded / n, 4) if n else 0.0, - "upgraded_requests": self.upgraded, - "cache_hit_rate": round(self.cache_hits / n, 4) if n else 0.0, - "cache_levels": dict(self.cache_levels), - "avg_latency_ms": round(avg_lat, 3), - "p99_latency_ms": round(p99, 3), - "total_cost_est_usd": round(self.cost_total, 6), - "model_usage": dict(self.model_usage), - "errors": self.errors, - } +"""运行指标收集（线程安全，零依赖）。""" +from __future__ import annotations + +import threading +from collections import Counter, deque +from typing import Any, Deque, Dict + + +class Stats: + def __init__(self, window: int = 1000): + self._lock = threading.Lock() + self.requests = 0 + self.domain_counter: Counter = Counter() + self.difficulty_counter: Counter = Counter() + self.upgraded = 0 + self.upgraded_by_domain: Counter = Counter() # 可解释评分 reliability 维输入 + self.cache_hits = 0 + self.cache_levels: Counter = Counter() + self.errors = 0 + self.latencies: Deque[float] = deque(maxlen=window) + self.cost_total = 0.0 + self.model_usage: Counter = Counter() + + def record(self, latency_ms: float, domain: str, difficulty: str, + upgraded: bool, cache_hit: bool, cache_level: str | None, + cost_est: float, model_used: str): + with self._lock: + self.requests += 1 + self.domain_counter[domain] += 1 + self.difficulty_counter[difficulty] += 1 + if upgraded: + self.upgraded += 1 + self.upgraded_by_domain[domain] += 1 + if cache_hit: + self.cache_hits += 1 + if cache_level: + self.cache_levels[cache_level] += 1 + self.latencies.append(latency_ms) + self.cost_total += cost_est + self.model_usage[model_used] += 1 + + def record_error(self): + with self._lock: + self.errors += 1 + + def domain_reliability(self, domain: str, prior: float = 0.9) -> float: + """领域可靠性 = 1 - 升级率（无数据时返回中性先验，可解释评分用）。""" + with self._lock: + n = self.domain_counter.get(domain, 0) + if n <= 0: + return prior + up = self.upgraded_by_domain.get(domain, 0) + return max(0.0, min(1.0, 1.0 - up / n)) + + def summary(self) -> Dict[str, Any]: + with self._lock: + n = self.requests + lat = list(self.latencies) + avg_lat = sum(lat) / len(lat) if lat else 0.0 + p99 = sorted(lat)[int(len(lat) * 0.99) - 1] if len(lat) >= 100 else (max(lat) if lat else 0.0) + return { + "total_requests": n, + "domain_distribution": dict(self.domain_counter), + "difficulty_distribution": dict(self.difficulty_counter), + "fallback_rate": round(self.upgraded / n, 4) if n else 0.0, + "upgraded_requests": self.upgraded, + "cache_hit_rate": round(self.cache_hits / n, 4) if n else 0.0, + "cache_levels": dict(self.cache_levels), + "avg_latency_ms": round(avg_lat, 3), + "p99_latency_ms": round(p99, 3), + "total_cost_est_usd": round(self.cost_total, 6), + "model_usage": dict(self.model_usage), + "errors": self.errors, + } diff --git a/tests/test_explain.py b/tests/test_explain.py new file mode 100644 index 0000000..6e11a94 --- /dev/null +++ b/tests/test_explain.py @@ -0,0 +1,90 @@ +"""可解释路由评分层单元测试（T-R1，采纳 ai-model-router 设计）。""" +import pytest + +from router_system.explain import CandidateProfile, explain_routing +from router_system.router import build_router + + +def _cand(domain, capability, cost=1.0, latency=100.0, reliability=0.9, fail=None): + return CandidateProfile(domain=domain, model=f"m-{domain}", capability=capability, + cost_per_mtok=cost, nominal_latency_ms=latency, + reliability=reliability, difficulty_score=0.5, + hard_fail_reason=fail) + + +def test_hard_filter_rejects_with_reason(): + """硬过滤淘汰带人话理由，被拒者不进排名。""" + exp = explain_routing([ + _cand("code", 1.0), + _cand("math", 0.0, fail="无规则命中（capability=0）"), + ]) + assert exp.winner == "code" + assert any(r["domain"] == "math" and "无规则命中" in r["reason"] for r in exp.rejected) + assert all(s["domain"] != "math" for s in exp.ranked) + + +def test_all_rejected_falls_back(): + """全部候选被硬过滤：决策者是大模型回退，置信度取基准 0.8。""" + exp = explain_routing([_cand("code", 0.0, fail="x"), _cand("math", 0.0, fail="y")]) + assert exp.winner == "fallback" + assert exp.confidence == 0.80 + assert len(exp.rejected) == 2 + + +def test_zero_cost_local_beats_costly(): + """成本效率维平滑映射：零成本恒 1.0，贵模型显著吃亏。""" + cheap = explain_routing([_cand("code", 1.0, cost=0.0, latency=1.0)]).ranked[0] + pricey = explain_routing([_cand("math", 1.0, cost=2.5, latency=800.0)]).ranked[0] + assert cheap["dims"]["cost_efficiency"] == 1.0 + assert pricey["dims"]["cost_efficiency"] < 0.5 + assert cheap["total"] > pricey["total"] + + +def test_confidence_gap_derivation(): + """置信度 = 0.8 + 分差：独苗封顶 0.99，双候选同分回落 0.8。""" + single = explain_routing([_cand("code", 1.0)]) + tied = explain_routing([_cand("code", 1.0), _cand("math", 1.0)]) + assert single.confidence == 0.99 + assert tied.confidence == 0.80 + + +def test_tie_break_is_deterministic(): + """同分决胜按领域名字典序（与仓库既有确定性约定一致）。""" + exp = explain_routing([_cand("zeta", 1.0), _cand("alpha", 1.0)]) + assert exp.winner == "alpha" + + +def test_weights_snapshot_and_rank_order(): + """排名按 total 降序，权重快照随解释输出。""" + exp = explain_routing([ + _cand("code", 1.0, cost=0.0, latency=1.0), + _cand("math", 0.5, cost=2.0, latency=900.0), + ]) + totals = [s["total"] for s in exp.ranked] + assert totals == sorted(totals, reverse=True) + assert exp.weights["capability"] == 0.35 + + +@pytest.mark.asyncio +async def test_route_attaches_explanation(): + """主链路返回结构化 route_explanation；缓存命中路径为 None。""" + router = build_router() + r1 = await router.route("用 Python 写一个快速排序函数") + assert r1.route_explanation is not None + assert r1.route_explanation["winner"] == "code" + assert any("explain:" in step for step in r1.route) + assert any("无规则命中" in r["reason"] + for r in r1.route_explanation["rejected"]) + r2 = await router.route("用 Python 写一个快速排序函数") + assert r2.cache_hit is True + assert r2.route_explanation is None + + +@pytest.mark.asyncio +async def test_low_confidence_explains_fallback(): + """低置信度直连回退：解释层给出 fallback 胜者与拒绝名单。""" + router = build_router() + r = await router.route("今天天气怎么样") + assert r.upgraded is True + assert r.route_explanation is not None + assert r.route_explanation["winner"] == "fallback"