69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
"""核心数据模型（纯标准库，无外部依赖）"""
|
|
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
|
|
|