feat(v1): T-R1 采纳 ai-model-router 可解释路由评分——五维加权+硬过滤带拒绝理由+置信度分差推导

- 新增 router_system/explain.py:CandidateProfile(capability/cost/latency/
  reliability/difficulty 画像)+ explain_routing(硬过滤先行、逐条人话拒绝理由、
  五维归一评分、同分按领域字典序、confidence=0.8+分差推导封顶 0.99)
- Router._explain 纯解释层装配(解释层任何异常不影响主链路,D-G4 同款纪律);
  路由胜负与既有决策完全等价,/chat 响应新增 route_explanation 字段(缓存命中为 None)
- experts:Expert.nominal_latency_ms 标称延迟元数据(mock 1/hf 300/api 800)
- stats:upgraded_by_domain 计数 + domain_reliability()(无数据给 0.9 中性先验)
- .gitignore 登记 extra/(参考项目目录不入库)
- 新增 tests/test_explain.py 8 项;全量 33 passed(基线 25)
This commit is contained in:
tzt
2026-09-19 09:23:04 +08:00
parent 5b28c2a583
commit a9f60159e7
8 changed files with 488 additions and 232 deletions
+70 -68
View File
@@ -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