Compare commits
6
Commits
9fdf91c2fb
...
06700a5aa3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
06700a5aa3 | ||
|
|
0dad895aac | ||
|
|
a9f60159e7 | ||
|
|
5b28c2a583 | ||
|
|
c072a1a237 | ||
|
|
81b0bc9b9b |
@@ -28,3 +28,6 @@ Thumbs.db
|
||||
|
||||
# 安全扫描器工作目录(不入库)
|
||||
.mimosa/
|
||||
|
||||
# 参考项目目录(不入库,仅作设计吸收来源)
|
||||
extra/
|
||||
|
||||
+95
-94
@@ -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)
|
||||
|
||||
+36
-19
@@ -12,11 +12,17 @@
|
||||
- 每条语义缓存条目在写入时预计算并缓存向量范数,查询时免重复计算(原来每对比较都重算)
|
||||
- 语义查找单遍完成:扫描即跟踪最优条目与命中计数,命中后不再二次线性查找
|
||||
- 相似度达到 1.0(完全相同查询)时提前终止扫描(余弦相似度上界,不可能更优)
|
||||
- 语义条目容器为 OrderedDict:提升/淘汰均为 O(1)(原 list pop 是 O(n) 搬移),
|
||||
且按 query 天然去重(原实现同一 query 可重复追加,重复条目白白参与扫描)
|
||||
- n-gram 向量经 lru_cache 复用:同一次未命中请求里 get 与紧随的 put
|
||||
不再重复做两遍分词+计数
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections import OrderedDict
|
||||
from dataclasses import dataclass
|
||||
from functools import lru_cache
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
@@ -52,6 +58,17 @@ def _dot(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float:
|
||||
return sum(v * vec_b.get(k, 0.0) for k, v in vec_a.items())
|
||||
|
||||
|
||||
@lru_cache(maxsize=1024)
|
||||
def _embed(text: str) -> Tuple[Dict[str, float], float]:
|
||||
"""n-gram 向量 + 范数(带缓存)。
|
||||
|
||||
纯函数且同一次未命中请求会在 get/put 各用一次,缓存可省一遍分词。
|
||||
契约:返回的 dict 仅供只读(内部无任何修改点),调用方不得改写。
|
||||
"""
|
||||
vec = _tf_vector(_ngrams(text))
|
||||
return vec, _norm(vec)
|
||||
|
||||
|
||||
class RouterCache:
|
||||
"""L1 精确缓存 + L2 语义缓存。"""
|
||||
|
||||
@@ -63,7 +80,8 @@ class RouterCache:
|
||||
self.max_exact = max_exact
|
||||
self.max_semantic = max_semantic
|
||||
self._exact: Dict[str, CacheEntry] = {}
|
||||
self._semantic: List[Tuple[str, CacheEntry]] = [] # (query, entry)
|
||||
# (query -> entry),插入序即淘汰序;len 兼容原 list 断言
|
||||
self._semantic: "OrderedDict[str, CacheEntry]" = OrderedDict()
|
||||
self._sem_vecs: Dict[str, Dict[str, float]] = {}
|
||||
self._sem_norms: Dict[str, float] = {} # 预计算范数,避免查询期重算
|
||||
self.hits = {"exact": 0, "semantic": 0}
|
||||
@@ -78,57 +96,56 @@ class RouterCache:
|
||||
return ("exact", entry.result)
|
||||
|
||||
if self.semantic_enabled:
|
||||
q_vec = _tf_vector(_ngrams(query))
|
||||
q_norm = _norm(q_vec)
|
||||
q_vec, q_norm = _embed(query)
|
||||
best_sim = 0.0
|
||||
best_idx = -1
|
||||
best_q = ""
|
||||
if q_norm > 0.0:
|
||||
# 单遍扫描:同时跟踪最优相似度与条目位置
|
||||
for i, (q, _e) in enumerate(self._semantic):
|
||||
# 单遍扫描(插入序):同时跟踪最优条目,sim=1.0 提前终止
|
||||
for q, _e in self._semantic.items():
|
||||
n_q = self._sem_norms.get(q, 0.0)
|
||||
if n_q <= 0.0:
|
||||
continue
|
||||
sim = _dot(q_vec, self._sem_vecs.get(q, {})) / (q_norm * n_q)
|
||||
if sim > best_sim:
|
||||
best_sim = sim
|
||||
best_idx = i
|
||||
best_q = q
|
||||
if sim >= 1.0:
|
||||
break # 余弦相似度上界:完全相同查询,提前终止
|
||||
if best_idx >= 0 and best_sim >= self.similarity_threshold:
|
||||
best_q, best_entry = self._semantic[best_idx]
|
||||
if best_q and best_sim >= self.similarity_threshold:
|
||||
best_entry = self._semantic[best_q]
|
||||
# 完全相同查询(相似度=1.0)计为 exact 命中
|
||||
is_exact = best_sim >= 0.999
|
||||
level = "exact" if is_exact else "semantic"
|
||||
self.hits[level] += 1
|
||||
self._bump_semantic(best_idx, best_q, best_entry)
|
||||
self._bump_semantic(best_q, best_entry)
|
||||
return (level, best_entry.result)
|
||||
|
||||
self.misses += 1
|
||||
return None
|
||||
|
||||
def _bump_semantic(self, idx: int, query: str, entry: CacheEntry):
|
||||
"""语义命中:累计命中次数,达到阈值提升为精确缓存(O(1),无需二次查找)。"""
|
||||
def _bump_semantic(self, query: str, entry: CacheEntry):
|
||||
"""语义命中:累计命中次数,达到阈值提升为精确缓存(OrderedDict 删除 O(1))。"""
|
||||
entry.hits += 1
|
||||
if entry.hits >= self.promote_frequency:
|
||||
self._exact[query] = entry
|
||||
self._semantic.pop(idx)
|
||||
del self._semantic[query]
|
||||
self._sem_vecs.pop(query, None)
|
||||
self._sem_norms.pop(query, None)
|
||||
|
||||
# ---- 写入 ----
|
||||
def put(self, query: str, result: Dict[str, Any]):
|
||||
if query in self._exact:
|
||||
return
|
||||
if query in self._exact or query in self._semantic:
|
||||
return # 已缓存(语义区按 query 天然去重,避免重复条目参与扫描)
|
||||
entry = CacheEntry(result=result)
|
||||
if self.semantic_enabled:
|
||||
if len(self._semantic) >= self.max_semantic:
|
||||
old_q, _ = self._semantic.pop(0)
|
||||
old_q, _old = self._semantic.popitem(last=False) # O(1) 淘汰最旧
|
||||
self._sem_vecs.pop(old_q, None)
|
||||
self._sem_norms.pop(old_q, None)
|
||||
self._semantic.append((query, entry))
|
||||
vec = _tf_vector(_ngrams(query))
|
||||
self._semantic[query] = entry
|
||||
vec, norm = _embed(query) # 与 get miss 共享同一份缓存向量
|
||||
self._sem_vecs[query] = vec
|
||||
self._sem_norms[query] = _norm(vec)
|
||||
self._sem_norms[query] = norm
|
||||
else:
|
||||
self._exact[query] = entry
|
||||
if len(self._exact) > self.max_exact:
|
||||
|
||||
+112
-28
@@ -6,11 +6,18 @@
|
||||
置信度设计:每个领域有一组 (关键词, 权重)。命中权重求和得原始分 s,
|
||||
confidence = 1 - exp(-s),保证 s=1 -> 0.63,s=2 -> 0.86,s=3 -> 0.95。
|
||||
无领域命中(或最高分领域为 general)时置信度低,触发 should_fallback。
|
||||
|
||||
规则外置(T-R2,采纳 llmrouter「规则文档即配置」设计):领域规则支持从
|
||||
config/routes.json(或 .yaml,pyyaml 可选)加载,文件按 domain 覆盖内置
|
||||
DOMAIN_RULES(与 v2 知识库"文件按 id 覆盖"同一惯例);文件缺失/格式非法
|
||||
整体安全回退内置(llmrouter 失败安全思想)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
from typing import Dict, List, Tuple
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from .difficulty import estimate_difficulty
|
||||
from .models import Classification
|
||||
@@ -33,7 +40,7 @@ DOMAIN_RULES: Dict[str, List[Tuple[str, float]]] = {
|
||||
("algorithm", 0.8), ("sort", 0.7), ("array", 0.7), ("regex", 0.8),
|
||||
("import", 0.7), ("loop", 0.7), ("recursion", 0.8), ("refactor", 0.8),
|
||||
("deploy", 0.8), ("docker", 0.8), ("kubernetes", 0.8),
|
||||
("async", 0.7), ("flask", 0.7), ("django", 0.7), ("api", 0.7),
|
||||
("async", 0.7), ("flask", 0.7), ("django", 0.7),
|
||||
("索引", 0.8), ("优化", 0.7), ("查询", 0.6),
|
||||
],
|
||||
"math": [
|
||||
@@ -82,6 +89,50 @@ _STOPWORDS = {
|
||||
"or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
|
||||
}
|
||||
|
||||
# 外置规则默认路径(约定优于配置:文件存在即自动加载,T-R2)
|
||||
DEFAULT_RULES_FILE = Path(__file__).resolve().parent.parent / "config" / "routes.json"
|
||||
|
||||
|
||||
def load_domain_rules(path: Optional[str | Path] = None
|
||||
) -> Optional[Dict[str, List[Tuple[str, float]]]]:
|
||||
"""加载外置分类规则文件,返回 {domain: [(关键词, 权重), ...]}。
|
||||
|
||||
文件缺失 / 格式非法 / 条目非法时返回 None(调用方安全回退内置规则,
|
||||
不抛异常——规则文档可被人工编辑,编辑错误不应打垮路由)。
|
||||
支持 .json(必有)与 .yaml(pyyaml 可选依赖)。
|
||||
"""
|
||||
p = Path(path) if path else DEFAULT_RULES_FILE
|
||||
if not p.exists():
|
||||
return None
|
||||
try:
|
||||
if p.suffix.lower() in (".yaml", ".yml"):
|
||||
try:
|
||||
import yaml # type: ignore
|
||||
except ImportError:
|
||||
return None
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
else:
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except (OSError, ValueError, Exception): # noqa: BLE001 解析失败一律回退
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
rules: Dict[str, List[Tuple[str, float]]] = {}
|
||||
for domain, entries in data.items():
|
||||
if not isinstance(domain, str) or not isinstance(entries, list):
|
||||
return None
|
||||
pairs: List[Tuple[str, float]] = []
|
||||
for entry in entries:
|
||||
if (not isinstance(entry, (list, tuple)) or len(entry) != 2
|
||||
or not isinstance(entry[0], str)
|
||||
or not isinstance(entry[1], (int, float))):
|
||||
return None
|
||||
pairs.append((entry[0], float(entry[1])))
|
||||
rules[domain] = pairs
|
||||
return rules
|
||||
|
||||
|
||||
class BaseClassifier:
|
||||
def classify(self, query: str) -> Classification:
|
||||
@@ -98,24 +149,29 @@ class RuleClassifier(BaseClassifier):
|
||||
self.confidence_floor = confidence_floor
|
||||
self.rules = DOMAIN_RULES
|
||||
|
||||
def _score(self, query: str) -> Tuple[Dict[str, float], Dict[str, List[str]]]:
|
||||
q = query.lower()
|
||||
def _score(self, q: str) -> Dict[str, float]:
|
||||
"""对已 lowercase 的查询按领域规则打分,返回有命中的领域分数。
|
||||
|
||||
命中词列表只对最终胜出领域有意义(classify 的唯一消费点),
|
||||
故不在各领域上白建 list——胜出后由 _matched_rules 单独收集。
|
||||
"""
|
||||
scores: Dict[str, float] = {}
|
||||
matched: Dict[str, List[str]] = {}
|
||||
for domain, rules in self.rules.items():
|
||||
s = 0.0
|
||||
hits = []
|
||||
for kw, w in rules:
|
||||
if kw in q:
|
||||
s += w
|
||||
hits.append(kw)
|
||||
if s > 0:
|
||||
scores[domain] = s
|
||||
matched[domain] = hits
|
||||
return scores, matched
|
||||
return scores
|
||||
|
||||
def _matched_rules(self, q: str, domain: str) -> List[str]:
|
||||
"""收集指定领域命中的关键词(保持登记顺序)。"""
|
||||
return [kw for kw, _w in self.rules.get(domain, ()) if kw in q]
|
||||
|
||||
def classify(self, query: str) -> Classification:
|
||||
raw, matched = self._score(query)
|
||||
q = query.lower()
|
||||
raw = self._score(q)
|
||||
if not raw:
|
||||
# 完全无命中 -> general,低置信度
|
||||
diff, ds = estimate_difficulty(query)
|
||||
@@ -150,7 +206,7 @@ class RuleClassifier(BaseClassifier):
|
||||
difficulty=diff,
|
||||
difficulty_score=ds,
|
||||
raw_scores={k: round(v, 3) for k, v in raw.items()},
|
||||
matched_rules=matched.get(best_domain, []),
|
||||
matched_rules=self._matched_rules(q, best_domain),
|
||||
)
|
||||
|
||||
|
||||
@@ -158,6 +214,8 @@ class HuggingFaceClassifier(BaseClassifier):
|
||||
"""可选:基于 transformers 的序列分类模型。
|
||||
|
||||
仅当安装 torch+transformers 且模型可加载时可用;否则抛错提示。
|
||||
失败安全(T-R3,采纳 llmrouter 分类失败静默降级思想):模型加载成功但
|
||||
推理期异常时,自动回落内置规则分类器,不让单次推理异常打垮路由。
|
||||
"""
|
||||
|
||||
def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55):
|
||||
@@ -173,32 +231,58 @@ class HuggingFaceClassifier(BaseClassifier):
|
||||
)
|
||||
self.labels = ["code", "math", "legal", "medical", "general"]
|
||||
self.confidence_floor = confidence_floor
|
||||
self._rule_fallback = RuleClassifier(confidence_floor=confidence_floor)
|
||||
|
||||
def classify(self, query: str) -> Classification:
|
||||
import torch # type: ignore
|
||||
try:
|
||||
import torch # type: ignore
|
||||
|
||||
inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
|
||||
with torch.no_grad():
|
||||
logits = self.model(**inputs).logits
|
||||
probs = torch.softmax(logits, dim=-1)[0]
|
||||
idx = int(probs.argmax())
|
||||
diff, ds = estimate_difficulty(query)
|
||||
return Classification(
|
||||
domain=self.labels[idx],
|
||||
confidence=round(float(probs[idx]), 4),
|
||||
difficulty=diff,
|
||||
difficulty_score=ds,
|
||||
raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))},
|
||||
)
|
||||
inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
|
||||
with torch.no_grad():
|
||||
logits = self.model(**inputs).logits
|
||||
probs = torch.softmax(logits, dim=-1)[0]
|
||||
idx = int(probs.argmax())
|
||||
diff, ds = estimate_difficulty(query)
|
||||
return Classification(
|
||||
domain=self.labels[idx],
|
||||
confidence=round(float(probs[idx]), 4),
|
||||
difficulty=diff,
|
||||
difficulty_score=ds,
|
||||
raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))},
|
||||
)
|
||||
except Exception: # noqa: BLE001 推理失败回落规则分类器(失败安全)
|
||||
return self._rule_fallback.classify(query)
|
||||
|
||||
|
||||
def build_classifier(cfg: Dict) -> BaseClassifier:
|
||||
"""根据配置构建分类器。cfg 为 classifier 段配置。"""
|
||||
"""根据配置构建分类器。cfg 为 classifier 段配置。
|
||||
|
||||
T-R2:cfg.rules_file 指定外置规则文件;未指定时若约定路径
|
||||
config/routes.json 存在则自动加载。文件按 domain 覆盖内置规则。
|
||||
"""
|
||||
ctype = cfg.get("type", "rule")
|
||||
floor = cfg.get("confidence_floor", 0.55)
|
||||
if ctype == "rule":
|
||||
return RuleClassifier(confidence_floor=floor)
|
||||
clf = RuleClassifier(confidence_floor=floor)
|
||||
rules_file = cfg.get("rules_file")
|
||||
external = load_domain_rules(rules_file) if rules_file \
|
||||
else (load_domain_rules() if DEFAULT_RULES_FILE.exists() else None)
|
||||
if external:
|
||||
clf.rules = {**DOMAIN_RULES, **external}
|
||||
return clf
|
||||
if ctype == "hf":
|
||||
return HuggingFaceClassifier(cfg.get("model", "Qwen/Qwen3-0.6B"), confidence_floor=floor)
|
||||
# 失败安全(T-R3):ML 依赖缺失 / 模型加载失败时回落规则分类器
|
||||
try:
|
||||
return HuggingFaceClassifier(
|
||||
cfg.get("model", "Qwen/Qwen3-0.6B"), confidence_floor=floor)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[classifier] HF 分类器不可用({type(e).__name__}),回落规则分类器")
|
||||
clf = RuleClassifier(confidence_floor=floor)
|
||||
rules_file = cfg.get("rules_file")
|
||||
external = load_domain_rules(rules_file) if rules_file \
|
||||
else (load_domain_rules() if DEFAULT_RULES_FILE.exists() else None)
|
||||
if external:
|
||||
clf.rules = {**DOMAIN_RULES, **external}
|
||||
return clf
|
||||
raise ValueError(f"未知分类器类型: {ctype}(支持 rule | hf)")
|
||||
|
||||
|
||||
+83
-52
@@ -1,52 +1,83 @@
|
||||
"""查询难度估计器(启发式,纯标准库)。
|
||||
|
||||
依据:RouterArena 用 Bloom 分类法把问题分为 easy/medium/hard。
|
||||
这里用查询长度、指令动词、数学/推理标记做轻量估计。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Tuple
|
||||
|
||||
# 触发 hard 的指令动词 / 推理标记
|
||||
_HARD_MARKERS = [
|
||||
"证明", "推导", "为什么", "如何", "对比", "比较", "分析", "评估", "设计", "优化",
|
||||
"复杂度", "时间复杂度", "空间复杂度", "原理", "机制", "优缺点", "区别", "推论", "定理",
|
||||
"proof", "prove", "derive", "explain why", "why", "how", "compare", "contrast",
|
||||
"analy", "evaluate", "design", "optimize", "refactor", "architect",
|
||||
"implement", "debug", "review", "plan", "synthesize",
|
||||
"int", "sum", "sqrt", "lim", "log", "derivative", "integral",
|
||||
]
|
||||
# 触发 medium 的标记
|
||||
_MEDIUM_MARKERS = [
|
||||
"用", "写", "计算", "求解", "生成", "翻译", "总结", "解释",
|
||||
"注意", "建议", "是否", "实现", "步骤",
|
||||
"write", "code", "function", "script", "calculate", "solve", "summarize",
|
||||
"translate", "fix", "explain", "describe", "list",
|
||||
]
|
||||
|
||||
|
||||
def estimate_difficulty(query: str) -> Tuple[str, float]:
|
||||
"""返回 (difficulty, score),score 属于 [0,1]。"""
|
||||
q = query.lower()
|
||||
hard_hits = sum(1 for m in _HARD_MARKERS if m in q)
|
||||
medium_hits = sum(1 for m in _MEDIUM_MARKERS if m in q)
|
||||
length = len(query)
|
||||
|
||||
score = 0.0
|
||||
score += min(0.30, length / 600.0) # 长度贡献
|
||||
score += min(0.55, hard_hits * 0.25) # 推理标记贡献
|
||||
score += min(0.30, medium_hits * 0.08) # 一般指令贡献
|
||||
|
||||
# 额外:代码 / 数学表达式(多步骤信号)
|
||||
if "```" in query or re.search(r"\b(def|class|function|import)\b", q):
|
||||
score += 0.15
|
||||
if re.search(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]", q):
|
||||
score += 0.15
|
||||
|
||||
score = max(0.0, min(1.0, score))
|
||||
if score >= 0.55:
|
||||
return "hard", score
|
||||
if score >= 0.25:
|
||||
return "medium", score
|
||||
return "easy", score␍
|
||||
"""查询难度估计器(启发式,纯标准库)。
|
||||
|
||||
依据:RouterArena 用 Bloom 分类法把问题分为 easy/medium/hard。
|
||||
这里用查询长度、指令动词、数学/推理标记做轻量估计。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Tuple
|
||||
|
||||
# 触发 hard 的指令动词 / 推理标记
|
||||
_HARD_MARKERS = [
|
||||
"证明", "推导", "为什么", "如何", "对比", "比较", "分析", "评估", "设计", "优化",
|
||||
"复杂度", "时间复杂度", "空间复杂度", "原理", "机制", "优缺点", "区别", "推论", "定理",
|
||||
"proof", "prove", "derive", "explain why", "why", "how", "compare", "contrast",
|
||||
"analy", "evaluate", "design", "optimize", "refactor", "architect",
|
||||
"implement", "debug", "review", "plan", "synthesize",
|
||||
"int", "sum", "sqrt", "lim", "log", "derivative", "integral",
|
||||
]
|
||||
# 触发 medium 的标记
|
||||
_MEDIUM_MARKERS = [
|
||||
"用", "写", "计算", "求解", "生成", "翻译", "总结", "解释",
|
||||
"注意", "建议", "是否", "实现", "步骤",
|
||||
"write", "code", "function", "script", "calculate", "solve", "summarize",
|
||||
"translate", "fix", "explain", "describe", "list",
|
||||
]
|
||||
|
||||
|
||||
def _compile_markers(markers):
|
||||
"""标记表编译:英文标记用词边界正则,中文标记保持子串匹配。
|
||||
|
||||
T-R3(采纳 llmrouter 词边界思想,super_hard 先于 hard 的同理):
|
||||
纯子串匹配会让 "int" 误命中 "print"/"point"、"log" 误命中 "logic"、
|
||||
"list" 误命中 "listen"——英文必须整词命中才计数。
|
||||
"""
|
||||
substr, words = [], []
|
||||
for m in markers:
|
||||
(words if m.isascii() else substr).append(m)
|
||||
pattern = re.compile(r"\b(?:%s)\b" % "|".join(re.escape(w) for w in words)) \
|
||||
if words else None
|
||||
return substr, pattern
|
||||
|
||||
|
||||
_HARD_SUBSTR, _HARD_RE = _compile_markers(_HARD_MARKERS)
|
||||
_MEDIUM_SUBSTR, _MEDIUM_RE = _compile_markers(_MEDIUM_MARKERS)
|
||||
|
||||
# 预编译正则(模块级一次,避免每次调用走 re 内部缓存查找)
|
||||
_RE_CODE_EXPR = re.compile(r"\b(def|class|function|import)\b")
|
||||
_RE_ARITH_EXPR = re.compile(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]")
|
||||
|
||||
|
||||
def _hit_count(substr, pattern, q: str) -> int:
|
||||
"""中文子串命中数 + 英文整词命中数。"""
|
||||
n = sum(1 for m in substr if m in q)
|
||||
if pattern is not None:
|
||||
n += len(pattern.findall(q))
|
||||
return n
|
||||
|
||||
|
||||
def estimate_difficulty(query: str) -> Tuple[str, float]:
|
||||
"""返回 (difficulty, score),score 属于 [0,1]。"""
|
||||
q = query.lower()
|
||||
hard_hits = _hit_count(_HARD_SUBSTR, _HARD_RE, q)
|
||||
medium_hits = _hit_count(_MEDIUM_SUBSTR, _MEDIUM_RE, q)
|
||||
length = len(query)
|
||||
|
||||
score = 0.0
|
||||
score += min(0.30, length / 600.0) # 长度贡献
|
||||
score += min(0.55, hard_hits * 0.25) # 推理标记贡献
|
||||
score += min(0.30, medium_hits * 0.08) # 一般指令贡献
|
||||
|
||||
# 额外:代码 / 数学表达式(多步骤信号)
|
||||
if "```" in query or _RE_CODE_EXPR.search(q):
|
||||
score += 0.15
|
||||
if _RE_ARITH_EXPR.search(q):
|
||||
score += 0.15
|
||||
|
||||
score = max(0.0, min(1.0, score))
|
||||
if score >= 0.55:
|
||||
return "hard", score
|
||||
if score >= 0.25:
|
||||
return "medium", score
|
||||
return "easy", score
|
||||
|
||||
+268
-270
@@ -1,270 +1,268 @@
|
||||
"""专家模型池:统一 Expert 接口,支持三种后端。
|
||||
|
||||
- MockExpert :确定性模板输出(零依赖,离线可跑,便于测试与演示)
|
||||
- HFExpert :HuggingFace transformers 真实小模型(可选,需 ML 依赖)
|
||||
- APIExpert :OpenAI 兼容 API(可选,需 API Key,如 DeepSeek)
|
||||
|
||||
成本估计:cost_est 按参数量粗估(美元/百万 token 的近似比例)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from .models import ExpertResponse
|
||||
|
||||
# 按模型规模粗估的相对成本($ / 1M output tokens,近似)
|
||||
MODEL_COST_EST = {
|
||||
"mock": 0.0,
|
||||
"0.5b": 0.02,
|
||||
"1b": 0.05,
|
||||
"1.7b": 0.08,
|
||||
"3b": 0.12,
|
||||
"4b": 0.15,
|
||||
"7b": 0.25,
|
||||
"70b": 2.50,
|
||||
"api": 1.00,
|
||||
}
|
||||
|
||||
|
||||
def _cost_for(model_name: str, default: str = "1b") -> float:
|
||||
mn = model_name.lower()
|
||||
for key in ("0.5b", "1.7b", "3b", "4b", "7b", "70b"):
|
||||
if key in mn:
|
||||
return MODEL_COST_EST[key]
|
||||
if "api" in mn or mn in ("deepseek-chat", "gpt-4o-mini", "claude"):
|
||||
return MODEL_COST_EST["api"]
|
||||
return MODEL_COST_EST.get(default, 0.1)
|
||||
|
||||
|
||||
def extract_content_terms(query: str) -> List[str]:
|
||||
"""抽取查询中的"内容词"(中文词/英文单词),用于 Judge 覆盖度与 Mock 回显。"""
|
||||
q = query.lower()
|
||||
terms: List[str] = []
|
||||
# 英文单词(>=2 字符)
|
||||
for w in re.findall(r"[a-z][a-z0-9_]{1,}", q):
|
||||
if w not in _STOPWORDS_EN and w not in terms:
|
||||
terms.append(w)
|
||||
# 中文:按 2-4 字窗口切分,保留含中文字符的片段
|
||||
cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q)
|
||||
for c in cn:
|
||||
terms.append(c)
|
||||
return terms
|
||||
|
||||
|
||||
_STOPWORDS_EN = {
|
||||
"the", "a", "an", "is", "are", "to", "of", "in", "on", "for", "with", "and",
|
||||
"or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
|
||||
"this", "that", "it", "be", "was", "were", "have", "has", "had", "will",
|
||||
"not", "no", "yes", "i", "you", "he", "she", "we", "they",
|
||||
}
|
||||
|
||||
|
||||
class Expert:
|
||||
name: str = "expert"
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class MockExpert(Expert):
|
||||
"""确定性模板专家:零依赖,离线可跑。
|
||||
|
||||
输出会回显查询中的内容词以提高 Judge 覆盖度,并带领域结构,
|
||||
使端到端管线(分类 -> 专家 -> Judge -> 缓存)可被稳定测试与演示。
|
||||
"""
|
||||
|
||||
def __init__(self, name: str, domain: str, model: str = "mock"):
|
||||
self.name = name
|
||||
self.domain = domain
|
||||
self.model = model
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
await asyncio.sleep(0.001) # 模拟极短推理延迟
|
||||
terms = extract_content_terms(query)
|
||||
body = self._template(query, terms, difficulty)
|
||||
# 预估 token 数:中文约 1.5 字符/token,英文约 4 字符/token
|
||||
tokens = max(8, int(len(body) / 2.2))
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=1.0,
|
||||
tokens=tokens,
|
||||
cost_est=_cost_for(self.model) * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
def _template(self, query: str, terms: List[str], difficulty: str) -> str:
|
||||
kw = "、".join(terms[:6]) if terms else "该主题"
|
||||
if self.domain == "code":
|
||||
return (
|
||||
f"(mock 代码专家)针对「{query}」的实现思路如下:\n\n"
|
||||
f"```python\n"
|
||||
f"def solve() -> None:\n"
|
||||
f" # 关键点:{kw}\n"
|
||||
f" # 1. 明确输入输出约束\n"
|
||||
f" # 2. 选择合适数据结构\n"
|
||||
f" # 3. 处理边界条件(空输入、极端值)\n"
|
||||
f" # 4. 补充单元测试\n"
|
||||
f" pass\n"
|
||||
f"```\n\n"
|
||||
f"复杂度:平均 O(n)。请按上述步骤补充具体实现。"
|
||||
)
|
||||
if self.domain == "math":
|
||||
return (
|
||||
f"(mock 数学专家)求解「{query}」的步骤:\n\n"
|
||||
f"1. 明确已知条件与目标:{kw}\n"
|
||||
f"2. 选择合适的方法(代数变形 / 积分 / 归纳等)\n"
|
||||
f"3. 逐步推导并验证中间结果\n"
|
||||
f"4. 检查边界与特殊情况\n\n"
|
||||
f"结论:在标准假设下,结果可化简为闭合形式。完整推导见正式解答。"
|
||||
)
|
||||
if self.domain == "legal":
|
||||
return (
|
||||
f"(mock 法律专家)关于「{query}」的初步法律分析:\n\n"
|
||||
f"相关要点:{kw}\n"
|
||||
f"1. 适用法规:请以现行有效法条为准(建议核对最新修订版)\n"
|
||||
f"2. 合同/合规风险点识别\n"
|
||||
f"3. 责任划分与救济途径\n\n"
|
||||
f"⚠️ 提示:以上为一般性分析,不构成正式法律意见,个案请咨询执业律师。"
|
||||
)
|
||||
if self.domain == "medical":
|
||||
return (
|
||||
f"(mock 医学专家)关于「{query}」的科普性说明:\n\n"
|
||||
f"相关关键词:{kw}\n"
|
||||
f"1. 常见表现与可能原因\n"
|
||||
f"2. 一般处理建议与注意事项\n"
|
||||
f"3. 何时需要就医(警示信号)\n\n"
|
||||
f"⚠️ 提示:内容仅供健康科普,不能替代医生诊断;如有不适请及时就医。"
|
||||
)
|
||||
return (
|
||||
f"(mock 通用专家)关于「{query}」的回答:\n\n"
|
||||
f"核心要点:{kw}\n"
|
||||
f"1. 背景与定义\n"
|
||||
f"2. 主要分类/维度\n"
|
||||
f"3. 实际应用与注意事项\n\n"
|
||||
f"如需更深入的分析,可以补充更多上下文。"
|
||||
)
|
||||
|
||||
|
||||
class HFExpert(Expert):
|
||||
"""可选:HuggingFace 真实小模型(需 requirements-ml.txt)。"""
|
||||
|
||||
def __init__(self, name: str, domain: str, model: str):
|
||||
self.name = name
|
||||
self.domain = domain
|
||||
self.model = model
|
||||
self._loaded = False
|
||||
self._model = None
|
||||
self._tokenizer = None
|
||||
|
||||
def _ensure_loaded(self):
|
||||
if self._loaded:
|
||||
return
|
||||
try:
|
||||
from transformers import AutoModelForCausalLM, AutoTokenizer
|
||||
except ImportError as e:
|
||||
raise RuntimeError("HFExpert 需要安装 ML 依赖:pip install -r requirements-ml.txt") from e
|
||||
self._tokenizer = AutoTokenizer.from_pretrained(self.model)
|
||||
self._model = AutoModelForCausalLM.from_pretrained(
|
||||
self.model, device_map="auto", torch_dtype="auto"
|
||||
)
|
||||
self._loaded = True
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
self._ensure_loaded()
|
||||
return await asyncio.to_thread(self._generate_sync, query)
|
||||
|
||||
def _generate_sync(self, query: str) -> ExpertResponse:
|
||||
messages = [{"role": "user", "content": query}]
|
||||
text = self._tokenizer.apply_chat_template(messages, tokenize=False)
|
||||
inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device)
|
||||
outputs = self._model.generate(
|
||||
**inputs,
|
||||
max_new_tokens=512,
|
||||
temperature=0.2,
|
||||
do_sample=True,
|
||||
)
|
||||
body = self._tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=512,
|
||||
cost_est=_cost_for(self.model) * 512 / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
class APIExpert(Expert):
|
||||
"""可选:OpenAI 兼容 Chat Completions(DeepSeek / OpenAI / 本地 vLLM)。"""
|
||||
|
||||
def __init__(self, name: str, domain: str, model: str, base_url: str, api_key: str):
|
||||
self.name = name
|
||||
self.domain = domain
|
||||
self.model = model
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client is None:
|
||||
import httpx
|
||||
self._client = httpx.AsyncClient(timeout=60.0)
|
||||
return self._client
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
client = self._get_client()
|
||||
resp = await client.post(
|
||||
f"{self.base_url}/chat/completions",
|
||||
headers={"Authorization": f"Bearer {self.api_key}"},
|
||||
json={
|
||||
"model": self.model,
|
||||
"messages": [{"role": "user", "content": query}],
|
||||
"temperature": 0.2,
|
||||
"max_tokens": 1024,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
body = data["choices"][0]["message"]["content"]
|
||||
usage = data.get("usage", {})
|
||||
tokens = usage.get("completion_tokens", int(len(body) / 2.2))
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=tokens,
|
||||
cost_est=_cost_for(self.model) * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
def build_expert(domain: str, cfg: Dict) -> Expert:
|
||||
"""根据配置构建领域专家。cfg 为 experts.<domain> 段配置。"""
|
||||
etype = cfg.get("type", "mock")
|
||||
model = cfg.get("model", "mock")
|
||||
name = f"expert-{domain}"
|
||||
if etype == "mock":
|
||||
return MockExpert(name, domain, model)
|
||||
if etype == "hf":
|
||||
return HFExpert(name, domain, model)
|
||||
if etype == "api":
|
||||
base_url = cfg.get("base_url", "https://api.deepseek.com/v1")
|
||||
api_key = cfg.get("api_key") or _env(cfg.get("api_key_env", ""))
|
||||
if not api_key:
|
||||
raise RuntimeError(f"APIExpert({domain}) 缺少 API Key(env: {cfg.get('api_key_env')})")
|
||||
return APIExpert(name, domain, model, base_url, api_key)
|
||||
raise ValueError(f"未知专家后端类型: {etype}(支持 mock | hf | api)")
|
||||
|
||||
|
||||
def _env(name: str) -> Optional[str]:
|
||||
import os
|
||||
return os.environ.get(name) if name else None
|
||||
|
||||
|
||||
def build_expert_pool(experts_cfg: Dict[str, Dict], domains: List[str]) -> Dict[str, Expert]:
|
||||
"""构建完整专家池。"""
|
||||
pool: Dict[str, Expert] = {}
|
||||
for domain in domains:
|
||||
cfg = experts_cfg.get(domain, {"type": "mock", "model": "mock"})
|
||||
pool[domain] = build_expert(domain, cfg)
|
||||
return pool␍
|
||||
"""专家模型池:统一 Expert 接口,支持三种后端。
|
||||
|
||||
- MockExpert :确定性模板输出(零依赖,离线可跑,便于测试与演示)
|
||||
- HFExpert :HuggingFace transformers 真实小模型(可选,需 ML 依赖)
|
||||
- APIExpert :OpenAI 兼容 API(可选,需 API Key,如 DeepSeek)
|
||||
|
||||
成本估计:cost_est 按参数量粗估(美元/百万 token 的近似比例)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from functools import lru_cache
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from .config import get_api_key
|
||||
from .llm_client import OpenAICompatClient
|
||||
from .models import ExpertResponse
|
||||
|
||||
# 按模型规模粗估的相对成本($ / 1M output tokens,近似)
|
||||
MODEL_COST_EST = {
|
||||
"mock": 0.0,
|
||||
"0.5b": 0.02,
|
||||
"1b": 0.05,
|
||||
"1.7b": 0.08,
|
||||
"3b": 0.12,
|
||||
"4b": 0.15,
|
||||
"7b": 0.25,
|
||||
"70b": 2.50,
|
||||
"api": 1.00,
|
||||
}
|
||||
|
||||
|
||||
def _cost_for(model_name: str, default: str = "1b") -> float:
|
||||
mn = model_name.lower()
|
||||
for key in ("0.5b", "1.7b", "3b", "4b", "7b", "70b"):
|
||||
if key in mn:
|
||||
return MODEL_COST_EST[key]
|
||||
if "api" in mn or mn in ("deepseek-chat", "gpt-4o-mini", "claude"):
|
||||
return MODEL_COST_EST["api"]
|
||||
return MODEL_COST_EST.get(default, 0.1)
|
||||
|
||||
|
||||
@lru_cache(maxsize=2048)
|
||||
def _extract_content_terms_cached(query: str) -> Tuple[str, ...]:
|
||||
q = query.lower()
|
||||
terms: List[str] = []
|
||||
# 英文单词(>=2 字符)
|
||||
for w in re.findall(r"[a-z][a-z0-9_]{1,}", q):
|
||||
if w not in _STOPWORDS_EN and w not in terms:
|
||||
terms.append(w)
|
||||
# 中文:按 2-4 字窗口切分,保留含中文字符的片段
|
||||
cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q)
|
||||
for c in cn:
|
||||
terms.append(c)
|
||||
return tuple(terms)
|
||||
|
||||
|
||||
def extract_content_terms(query: str) -> Tuple[str, ...]:
|
||||
"""抽取查询中的"内容词"(中文词/英文单词),用于 Judge 覆盖度与 Mock 回显。
|
||||
|
||||
纯函数,但 Router 主链路中专家生成与 Judge 评估会对同一查询各调一次,
|
||||
故以 lru_cache 去重;返回不可变 tuple(调用方只读)。
|
||||
"""
|
||||
return _extract_content_terms_cached(query)
|
||||
|
||||
|
||||
_STOPWORDS_EN = {
|
||||
"the", "a", "an", "is", "are", "to", "of", "in", "on", "for", "with", "and",
|
||||
"or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
|
||||
"this", "that", "it", "be", "was", "were", "have", "has", "had", "will",
|
||||
"not", "no", "yes", "i", "you", "he", "she", "we", "they",
|
||||
}
|
||||
|
||||
|
||||
class Expert:
|
||||
name: str = "expert"
|
||||
nominal_latency_ms: float = 100.0 # 标称延迟元数据(可解释评分 speed 维输入)
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class MockExpert(Expert):
|
||||
"""确定性模板专家:零依赖,离线可跑。
|
||||
|
||||
输出会回显查询中的内容词以提高 Judge 覆盖度,并带领域结构,
|
||||
使端到端管线(分类 -> 专家 -> Judge -> 缓存)可被稳定测试与演示。
|
||||
"""
|
||||
|
||||
nominal_latency_ms = 1.0
|
||||
|
||||
def __init__(self, name: str, domain: str, model: str = "mock"):
|
||||
self.name = name
|
||||
self.domain = domain
|
||||
self.model = model
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
await asyncio.sleep(0.001) # 模拟极短推理延迟
|
||||
terms = extract_content_terms(query)
|
||||
body = self._template(query, terms, difficulty)
|
||||
# 预估 token 数:中文约 1.5 字符/token,英文约 4 字符/token
|
||||
tokens = max(8, int(len(body) / 2.2))
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=1.0,
|
||||
tokens=tokens,
|
||||
cost_est=_cost_for(self.model) * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
def _template(self, query: str, terms: List[str], difficulty: str) -> str:
|
||||
kw = "、".join(terms[:6]) if terms else "该主题"
|
||||
if self.domain == "code":
|
||||
return (
|
||||
f"(mock 代码专家)针对「{query}」的实现思路如下:\n\n"
|
||||
f"```python\n"
|
||||
f"def solve() -> None:\n"
|
||||
f" # 关键点:{kw}\n"
|
||||
f" # 1. 明确输入输出约束\n"
|
||||
f" # 2. 选择合适数据结构\n"
|
||||
f" # 3. 处理边界条件(空输入、极端值)\n"
|
||||
f" # 4. 补充单元测试\n"
|
||||
f" pass\n"
|
||||
f"```\n\n"
|
||||
f"复杂度:平均 O(n)。请按上述步骤补充具体实现。"
|
||||
)
|
||||
if self.domain == "math":
|
||||
return (
|
||||
f"(mock 数学专家)求解「{query}」的步骤:\n\n"
|
||||
f"1. 明确已知条件与目标:{kw}\n"
|
||||
f"2. 选择合适的方法(代数变形 / 积分 / 归纳等)\n"
|
||||
f"3. 逐步推导并验证中间结果\n"
|
||||
f"4. 检查边界与特殊情况\n\n"
|
||||
f"结论:在标准假设下,结果可化简为闭合形式。完整推导见正式解答。"
|
||||
)
|
||||
if self.domain == "legal":
|
||||
return (
|
||||
f"(mock 法律专家)关于「{query}」的初步法律分析:\n\n"
|
||||
f"相关要点:{kw}\n"
|
||||
f"1. 适用法规:请以现行有效法条为准(建议核对最新修订版)\n"
|
||||
f"2. 合同/合规风险点识别\n"
|
||||
f"3. 责任划分与救济途径\n\n"
|
||||
f"⚠️ 提示:以上为一般性分析,不构成正式法律意见,个案请咨询执业律师。"
|
||||
)
|
||||
if self.domain == "medical":
|
||||
return (
|
||||
f"(mock 医学专家)关于「{query}」的科普性说明:\n\n"
|
||||
f"相关关键词:{kw}\n"
|
||||
f"1. 常见表现与可能原因\n"
|
||||
f"2. 一般处理建议与注意事项\n"
|
||||
f"3. 何时需要就医(警示信号)\n\n"
|
||||
f"⚠️ 提示:内容仅供健康科普,不能替代医生诊断;如有不适请及时就医。"
|
||||
)
|
||||
return (
|
||||
f"(mock 通用专家)关于「{query}」的回答:\n\n"
|
||||
f"核心要点:{kw}\n"
|
||||
f"1. 背景与定义\n"
|
||||
f"2. 主要分类/维度\n"
|
||||
f"3. 实际应用与注意事项\n\n"
|
||||
f"如需更深入的分析,可以补充更多上下文。"
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
self.model = model
|
||||
self._loaded = False
|
||||
self._model = None
|
||||
self._tokenizer = None
|
||||
|
||||
def _ensure_loaded(self):
|
||||
if self._loaded:
|
||||
return
|
||||
try:
|
||||
from transformers import AutoModelForCausalLM, AutoTokenizer
|
||||
except ImportError as e:
|
||||
raise RuntimeError("HFExpert 需要安装 ML 依赖:pip install -r requirements-ml.txt") from e
|
||||
self._tokenizer = AutoTokenizer.from_pretrained(self.model)
|
||||
self._model = AutoModelForCausalLM.from_pretrained(
|
||||
self.model, device_map="auto", torch_dtype="auto"
|
||||
)
|
||||
self._loaded = True
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
self._ensure_loaded()
|
||||
return await asyncio.to_thread(self._generate_sync, query)
|
||||
|
||||
def _generate_sync(self, query: str) -> ExpertResponse:
|
||||
messages = [{"role": "user", "content": query}]
|
||||
text = self._tokenizer.apply_chat_template(messages, tokenize=False)
|
||||
inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device)
|
||||
outputs = self._model.generate(
|
||||
**inputs,
|
||||
max_new_tokens=512,
|
||||
temperature=0.2,
|
||||
do_sample=True,
|
||||
)
|
||||
body = self._tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=512,
|
||||
cost_est=_cost_for(self.model) * 512 / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
self.model = model
|
||||
self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
|
||||
|
||||
async def generate(self, query: str, difficulty: str) -> ExpertResponse:
|
||||
data = await self._client.chat(
|
||||
self.model,
|
||||
[{"role": "user", "content": query}],
|
||||
temperature=0.2,
|
||||
max_tokens=1024,
|
||||
)
|
||||
body = self._client.completion_text(data)
|
||||
tokens = self._client.completion_tokens(data, body)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=tokens,
|
||||
cost_est=_cost_for(self.model) * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
def build_expert(domain: str, cfg: Dict) -> Expert:
|
||||
"""根据配置构建领域专家。cfg 为 experts.<domain> 段配置。"""
|
||||
etype = cfg.get("type", "mock")
|
||||
model = cfg.get("model", "mock")
|
||||
name = f"expert-{domain}"
|
||||
if etype == "mock":
|
||||
return MockExpert(name, domain, model)
|
||||
if etype == "hf":
|
||||
return HFExpert(name, domain, model)
|
||||
if etype == "api":
|
||||
base_url = cfg.get("base_url", "https://api.deepseek.com/v1")
|
||||
api_key = cfg.get("api_key") or get_api_key(cfg)
|
||||
if not api_key:
|
||||
raise RuntimeError(f"APIExpert({domain}) 缺少 API Key(env: {cfg.get('api_key_env')})")
|
||||
return APIExpert(name, domain, model, base_url, api_key)
|
||||
raise ValueError(f"未知专家后端类型: {etype}(支持 mock | hf | api)")
|
||||
|
||||
|
||||
def build_expert_pool(experts_cfg: Dict[str, Dict], domains: List[str]) -> Dict[str, Expert]:
|
||||
"""构建完整专家池。"""
|
||||
pool: Dict[str, Expert] = {}
|
||||
for domain in domains:
|
||||
cfg = experts_cfg.get(domain, {"type": "mock", "model": "mock"})
|
||||
pool[domain] = build_expert(domain, cfg)
|
||||
return pool
|
||||
|
||||
@@ -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)
|
||||
+84
-99
@@ -1,99 +1,84 @@
|
||||
"""大模型回退层:Mock 与 OpenAI 兼容 API 两种后端。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Dict, Optional
|
||||
|
||||
from .models import ExpertResponse
|
||||
|
||||
|
||||
class FallbackProvider:
|
||||
name: str = "fallback"
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class MockFallback(FallbackProvider):
|
||||
"""确定性 mock 大模型:标识为 fallback,便于测试升级路径。"""
|
||||
|
||||
def __init__(self, model: str = "mock-large"):
|
||||
self.model = model
|
||||
self.name = f"fallback-{model}"
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
await asyncio.sleep(0.002)
|
||||
body = (
|
||||
f"(大模型回退)「{query}」\n\n"
|
||||
"这是一条来自大模型回退路径的完整回答。\n"
|
||||
"要点:\n"
|
||||
"1. 对复杂/跨域任务给出综合推理\n"
|
||||
"2. 补充领域专家未覆盖的上下文\n"
|
||||
"3. 给出可执行的后续建议\n"
|
||||
)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=2.0,
|
||||
tokens=120,
|
||||
cost_est=2.0 * 120 / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
class APIFallback(FallbackProvider):
|
||||
"""OpenAI 兼容大模型 API(如 DeepSeek / OpenAI / 本地 vLLM)。"""
|
||||
|
||||
def __init__(self, model: str, base_url: str, api_key: str):
|
||||
self.model = model
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
self.name = f"fallback-{model}"
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client is None:
|
||||
import httpx
|
||||
self._client = httpx.AsyncClient(timeout=90.0)
|
||||
return self._client
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
client = self._get_client()
|
||||
resp = await client.post(
|
||||
f"{self.base_url}/chat/completions",
|
||||
headers={"Authorization": f"Bearer {self.api_key}"},
|
||||
json={
|
||||
"model": self.model,
|
||||
"messages": [{"role": "user", "content": query}],
|
||||
"temperature": 0.3,
|
||||
"max_tokens": 2048,
|
||||
},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
body = data["choices"][0]["message"]["content"]
|
||||
usage = data.get("usage", {})
|
||||
tokens = usage.get("completion_tokens", int(len(body) / 2.2))
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=tokens,
|
||||
cost_est=2.0 * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
def build_fallback(cfg: Dict) -> FallbackProvider:
|
||||
"""cfg 为 fallback 段配置。"""
|
||||
ftype = cfg.get("type", "mock")
|
||||
model = cfg.get("model", "deepseek-chat")
|
||||
if ftype == "mock":
|
||||
return MockFallback(model=model)
|
||||
if ftype == "api":
|
||||
import os
|
||||
api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY"))
|
||||
if not api_key:
|
||||
raise RuntimeError(
|
||||
f"APIFallback 缺少 API Key:请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key"
|
||||
)
|
||||
return APIFallback(model, cfg.get("base_url", "https://api.deepseek.com/v1"), api_key)
|
||||
raise ValueError(f"未知 fallback 类型: {ftype}(支持 mock | api)")␍
|
||||
"""大模型回退层:Mock 与 OpenAI 兼容 API 两种后端。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import Dict
|
||||
|
||||
from .config import get_api_key
|
||||
from .llm_client import OpenAICompatClient
|
||||
from .models import ExpertResponse
|
||||
|
||||
|
||||
class FallbackProvider:
|
||||
name: str = "fallback"
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class MockFallback(FallbackProvider):
|
||||
"""确定性 mock 大模型:标识为 fallback,便于测试升级路径。"""
|
||||
|
||||
def __init__(self, model: str = "mock-large"):
|
||||
self.model = model
|
||||
self.name = f"fallback-{model}"
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
await asyncio.sleep(0.002)
|
||||
body = (
|
||||
f"(大模型回退)「{query}」\n\n"
|
||||
"这是一条来自大模型回退路径的完整回答。\n"
|
||||
"要点:\n"
|
||||
"1. 对复杂/跨域任务给出综合推理\n"
|
||||
"2. 补充领域专家未覆盖的上下文\n"
|
||||
"3. 给出可执行的后续建议\n"
|
||||
)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=2.0,
|
||||
tokens=120,
|
||||
cost_est=2.0 * 120 / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
class APIFallback(FallbackProvider):
|
||||
"""OpenAI 兼容大模型 API(如 DeepSeek / OpenAI / 本地 vLLM)。"""
|
||||
|
||||
def __init__(self, model: str, base_url: str, api_key: str):
|
||||
self.model = model
|
||||
self.name = f"fallback-{model}"
|
||||
self._client = OpenAICompatClient(base_url, api_key, timeout=90.0)
|
||||
|
||||
async def generate(self, query: str) -> ExpertResponse:
|
||||
data = await self._client.chat(
|
||||
self.model,
|
||||
[{"role": "user", "content": query}],
|
||||
temperature=0.3,
|
||||
max_tokens=2048,
|
||||
)
|
||||
body = self._client.completion_text(data)
|
||||
tokens = self._client.completion_tokens(data, body)
|
||||
return ExpertResponse(
|
||||
text=body,
|
||||
model_used=self.model,
|
||||
latency_ms=0.0,
|
||||
tokens=tokens,
|
||||
cost_est=2.0 * tokens / 1_000_000,
|
||||
)
|
||||
|
||||
|
||||
def build_fallback(cfg: Dict) -> FallbackProvider:
|
||||
"""cfg 为 fallback 段配置。"""
|
||||
ftype = cfg.get("type", "mock")
|
||||
model = cfg.get("model", "deepseek-chat")
|
||||
if ftype == "mock":
|
||||
return MockFallback(model=model)
|
||||
if ftype == "api":
|
||||
api_key = cfg.get("api_key") or get_api_key(cfg)
|
||||
if not api_key:
|
||||
raise RuntimeError(
|
||||
f"APIFallback 缺少 API Key:请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key"
|
||||
)
|
||||
return APIFallback(model, cfg.get("base_url", "https://api.deepseek.com/v1"), api_key)
|
||||
raise ValueError(f"未知 fallback 类型: {ftype}(支持 mock | api)")
|
||||
|
||||
+168
-174
@@ -1,174 +1,168 @@
|
||||
"""质量控制器(Judge):评估专家输出,决定是否升级大模型。
|
||||
|
||||
- RuleJudge:零依赖启发式(内容覆盖度 / 长度充分性 / 领域格式 / 安全提示),
|
||||
稳定可测,适合 MVP 与离线演示。
|
||||
- LLMJudge:可选,基于 transformers 小模型或 API 的 LLM-as-Judge。
|
||||
|
||||
设计对齐实现方案:overall_score < judge_fallback_threshold -> 升级大模型。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List
|
||||
|
||||
from .experts import extract_content_terms
|
||||
|
||||
# 各领域期望的响应长度范围(字符数)
|
||||
_EXPECTED_LEN = {
|
||||
"code": (60, 2000),
|
||||
"math": (60, 2000),
|
||||
"legal": (80, 3000),
|
||||
"medical": (80, 3000),
|
||||
"general": (40, 2000),
|
||||
}
|
||||
|
||||
# 领域格式检查:响应应包含的标记
|
||||
_DOMAIN_FORMAT_HINTS = {
|
||||
"code": ["```", "def ", "function", "class "],
|
||||
"math": ["步骤", "推导", "=", "解", "step"],
|
||||
"legal": ["⚠", "法律", "意见", "合规", "contract", "law"],
|
||||
"medical": ["⚠", "就医", "医生", "症状", "诊断", "symptom"],
|
||||
"general": [],
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class QualityEvaluation:
|
||||
overall_score: float
|
||||
scores: Dict[str, float] = field(default_factory=dict)
|
||||
needs_fallback: bool = False
|
||||
reasons: List[str] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"overall_score": round(self.overall_score, 4),
|
||||
"scores": {k: round(v, 4) for k, v in self.scores.items()},
|
||||
"needs_fallback": self.needs_fallback,
|
||||
"reasons": self.reasons,
|
||||
}
|
||||
|
||||
|
||||
class BaseJudge:
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class RuleJudge(BaseJudge):
|
||||
"""启发式质量评估(零依赖)。"""
|
||||
|
||||
def __init__(self, fallback_threshold: float = 0.70):
|
||||
self.fallback_threshold = fallback_threshold
|
||||
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
scores: Dict[str, float] = {}
|
||||
reasons: List[str] = []
|
||||
|
||||
# 1) 内容覆盖度:查询中的内容词有多少出现在响应里
|
||||
terms = extract_content_terms(query)
|
||||
if terms:
|
||||
hit = sum(1 for t in terms if t in response.lower())
|
||||
coverage = hit / len(terms)
|
||||
scores["coverage"] = coverage
|
||||
if coverage < 0.4:
|
||||
reasons.append(f"内容覆盖度低 ({coverage:.0%})")
|
||||
else:
|
||||
scores["coverage"] = 1.0
|
||||
|
||||
# 2) 长度充分性
|
||||
lo, hi = _EXPECTED_LEN.get(domain, (40, 2000))
|
||||
n = len(response)
|
||||
if n < lo:
|
||||
scores["length"] = max(0.0, n / lo)
|
||||
reasons.append(f"响应过短 ({n} 字符)")
|
||||
elif n > hi:
|
||||
scores["length"] = 0.8
|
||||
reasons.append(f"响应过长 ({n} 字符)")
|
||||
else:
|
||||
scores["length"] = 1.0
|
||||
|
||||
# 3) 领域格式检查
|
||||
hints = _DOMAIN_FORMAT_HINTS.get(domain, [])
|
||||
if hints:
|
||||
hit_hints = sum(1 for h in hints if h in response)
|
||||
scores["format"] = min(1.0, 0.4 + 0.2 * hit_hints)
|
||||
if hit_hints == 0:
|
||||
reasons.append("缺少领域格式特征")
|
||||
else:
|
||||
scores["format"] = 1.0
|
||||
|
||||
# 4) 安全/免责提示(法律、医疗领域应有警示语)
|
||||
if domain in ("legal", "medical") and ("⚠" not in response and "提示" not in response):
|
||||
scores["safety"] = 0.6
|
||||
reasons.append("缺少免责提示")
|
||||
else:
|
||||
scores["safety"] = 1.0
|
||||
|
||||
weights = {"coverage": 0.4, "length": 0.2, "format": 0.2, "safety": 0.2}
|
||||
overall = sum(scores.get(k, 0.0) * w for k, w in weights.items())
|
||||
needs = overall < self.fallback_threshold
|
||||
if needs:
|
||||
reasons.append("质量分低于阈值,建议升级大模型")
|
||||
return QualityEvaluation(
|
||||
overall_score=round(overall, 4),
|
||||
scores=scores,
|
||||
needs_fallback=needs,
|
||||
reasons=reasons,
|
||||
)
|
||||
|
||||
|
||||
class LLMJudge(BaseJudge):
|
||||
"""可选:LLM-as-Judge(API 后端)。"""
|
||||
|
||||
def __init__(self, model: str, base_url: str, api_key: str, fallback_threshold: float = 0.70):
|
||||
self.model = model
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
self.fallback_threshold = fallback_threshold
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client is None:
|
||||
import httpx
|
||||
self._client = httpx.AsyncClient(timeout=60.0)
|
||||
return self._client
|
||||
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
client = self._get_client()
|
||||
prompt = (
|
||||
f"你是质量评审员。评估以下回答对查询的满足程度,输出 0-1 分(相关性/正确性/完整性)。\n"
|
||||
f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n"
|
||||
f"只输出一个 0 到 1 之间的数字。"
|
||||
)
|
||||
resp = await client.post(
|
||||
f"{self.base_url}/chat/completions",
|
||||
headers={"Authorization": f"Bearer {self.api_key}"},
|
||||
json={"model": self.model, "messages": [{"role": "user", "content": prompt}]},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
try:
|
||||
score = float(resp.json()["choices"][0]["message"]["content"].strip())
|
||||
score = max(0.0, min(1.0, score))
|
||||
except Exception:
|
||||
score = 0.5
|
||||
return QualityEvaluation(
|
||||
overall_score=score,
|
||||
scores={"llm_judge": score},
|
||||
needs_fallback=score < self.fallback_threshold,
|
||||
)
|
||||
|
||||
|
||||
def build_judge(cfg: Dict, fallback_threshold: float = 0.70) -> BaseJudge:
|
||||
"""cfg 为 judge 段配置。"""
|
||||
jtype = cfg.get("type", "rule")
|
||||
if jtype == "rule":
|
||||
return RuleJudge(fallback_threshold=fallback_threshold)
|
||||
if jtype == "llm":
|
||||
import os
|
||||
api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY"))
|
||||
return LLMJudge(
|
||||
cfg.get("model", "deepseek-chat"),
|
||||
cfg.get("base_url", "https://api.deepseek.com/v1"),
|
||||
api_key or "",
|
||||
fallback_threshold=fallback_threshold,
|
||||
)
|
||||
raise ValueError(f"未知 judge 类型: {jtype}(支持 rule | llm)")␍
|
||||
"""质量控制器(Judge):评估专家输出,决定是否升级大模型。
|
||||
|
||||
- RuleJudge:零依赖启发式(内容覆盖度 / 长度充分性 / 领域格式 / 安全提示),
|
||||
稳定可测,适合 MVP 与离线演示。
|
||||
- LLMJudge:可选,基于 transformers 小模型或 API 的 LLM-as-Judge。
|
||||
|
||||
设计对齐实现方案:overall_score < judge_fallback_threshold -> 升级大模型。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List
|
||||
|
||||
from .experts import extract_content_terms
|
||||
|
||||
# 各领域期望的响应长度范围(字符数)
|
||||
_EXPECTED_LEN = {
|
||||
"code": (60, 2000),
|
||||
"math": (60, 2000),
|
||||
"legal": (80, 3000),
|
||||
"medical": (80, 3000),
|
||||
"general": (40, 2000),
|
||||
}
|
||||
|
||||
# 领域格式检查:响应应包含的标记
|
||||
_DOMAIN_FORMAT_HINTS = {
|
||||
"code": ["```", "def ", "function", "class "],
|
||||
"math": ["步骤", "推导", "=", "解", "step"],
|
||||
"legal": ["⚠", "法律", "意见", "合规", "contract", "law"],
|
||||
"medical": ["⚠", "就医", "医生", "症状", "诊断", "symptom"],
|
||||
"general": [],
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class QualityEvaluation:
|
||||
overall_score: float
|
||||
scores: Dict[str, float] = field(default_factory=dict)
|
||||
needs_fallback: bool = False
|
||||
reasons: List[str] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"overall_score": round(self.overall_score, 4),
|
||||
"scores": {k: round(v, 4) for k, v in self.scores.items()},
|
||||
"needs_fallback": self.needs_fallback,
|
||||
"reasons": self.reasons,
|
||||
}
|
||||
|
||||
|
||||
class BaseJudge:
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class RuleJudge(BaseJudge):
|
||||
"""启发式质量评估(零依赖)。"""
|
||||
|
||||
def __init__(self, fallback_threshold: float = 0.70):
|
||||
self.fallback_threshold = fallback_threshold
|
||||
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
scores: Dict[str, float] = {}
|
||||
reasons: List[str] = []
|
||||
|
||||
# 1) 内容覆盖度:查询中的内容词有多少出现在响应里
|
||||
terms = extract_content_terms(query)
|
||||
if terms:
|
||||
lowered = response.lower() # lowercase 一次,避免逐词重复复制长响应
|
||||
hit = sum(1 for t in terms if t in lowered)
|
||||
coverage = hit / len(terms)
|
||||
scores["coverage"] = coverage
|
||||
if coverage < 0.4:
|
||||
reasons.append(f"内容覆盖度低 ({coverage:.0%})")
|
||||
else:
|
||||
scores["coverage"] = 1.0
|
||||
|
||||
# 2) 长度充分性
|
||||
lo, hi = _EXPECTED_LEN.get(domain, (40, 2000))
|
||||
n = len(response)
|
||||
if n < lo:
|
||||
scores["length"] = max(0.0, n / lo)
|
||||
reasons.append(f"响应过短 ({n} 字符)")
|
||||
elif n > hi:
|
||||
scores["length"] = 0.8
|
||||
reasons.append(f"响应过长 ({n} 字符)")
|
||||
else:
|
||||
scores["length"] = 1.0
|
||||
|
||||
# 3) 领域格式检查
|
||||
hints = _DOMAIN_FORMAT_HINTS.get(domain, [])
|
||||
if hints:
|
||||
hit_hints = sum(1 for h in hints if h in response)
|
||||
scores["format"] = min(1.0, 0.4 + 0.2 * hit_hints)
|
||||
if hit_hints == 0:
|
||||
reasons.append("缺少领域格式特征")
|
||||
else:
|
||||
scores["format"] = 1.0
|
||||
|
||||
# 4) 安全/免责提示(法律、医疗领域应有警示语)
|
||||
if domain in ("legal", "medical") and ("⚠" not in response and "提示" not in response):
|
||||
scores["safety"] = 0.6
|
||||
reasons.append("缺少免责提示")
|
||||
else:
|
||||
scores["safety"] = 1.0
|
||||
|
||||
weights = {"coverage": 0.4, "length": 0.2, "format": 0.2, "safety": 0.2}
|
||||
overall = sum(scores.get(k, 0.0) * w for k, w in weights.items())
|
||||
needs = overall < self.fallback_threshold
|
||||
if needs:
|
||||
reasons.append("质量分低于阈值,建议升级大模型")
|
||||
return QualityEvaluation(
|
||||
overall_score=round(overall, 4),
|
||||
scores=scores,
|
||||
needs_fallback=needs,
|
||||
reasons=reasons,
|
||||
)
|
||||
|
||||
|
||||
class LLMJudge(BaseJudge):
|
||||
"""可选:LLM-as-Judge(API 后端)。"""
|
||||
|
||||
def __init__(self, model: str, base_url: str, api_key: str, fallback_threshold: float = 0.70):
|
||||
from .llm_client import OpenAICompatClient
|
||||
self.model = model
|
||||
self.fallback_threshold = fallback_threshold
|
||||
# 请求体与旧实现一致:不下发 temperature / max_tokens
|
||||
self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
|
||||
|
||||
async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
|
||||
prompt = (
|
||||
f"你是质量评审员。评估以下回答对查询的满足程度,输出 0-1 分(相关性/正确性/完整性)。\n"
|
||||
f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n"
|
||||
f"只输出一个 0 到 1 之间的数字。"
|
||||
)
|
||||
data = await self._client.chat(
|
||||
self.model,
|
||||
[{"role": "user", "content": prompt}],
|
||||
temperature=None,
|
||||
max_tokens=None,
|
||||
)
|
||||
try:
|
||||
score = float(self._client.completion_text(data).strip())
|
||||
score = max(0.0, min(1.0, score))
|
||||
except Exception:
|
||||
score = 0.5
|
||||
return QualityEvaluation(
|
||||
overall_score=score,
|
||||
scores={"llm_judge": score},
|
||||
needs_fallback=score < self.fallback_threshold,
|
||||
)
|
||||
|
||||
|
||||
def build_judge(cfg: Dict, fallback_threshold: float = 0.70) -> BaseJudge:
|
||||
"""cfg 为 judge 段配置。"""
|
||||
jtype = cfg.get("type", "rule")
|
||||
if jtype == "rule":
|
||||
return RuleJudge(fallback_threshold=fallback_threshold)
|
||||
if jtype == "llm":
|
||||
from .config import get_api_key
|
||||
api_key = cfg.get("api_key") or get_api_key(cfg)
|
||||
return LLMJudge(
|
||||
cfg.get("model", "deepseek-chat"),
|
||||
cfg.get("base_url", "https://api.deepseek.com/v1"),
|
||||
api_key or "",
|
||||
fallback_threshold=fallback_threshold,
|
||||
)
|
||||
raise ValueError(f"未知 judge 类型: {jtype}(支持 rule | llm)")
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""OpenAI 兼容 Chat Completions 共享客户端(架构去重)。
|
||||
|
||||
此前 experts.APIExpert / judge.LLMJudge / fallback.APIFallback 各自复制一份
|
||||
"懒建 AsyncClient + POST /chat/completions + choices/usage 解析"的等价逻辑
|
||||
(超时、温度、token 上限各自维护)。本模块将其统一为单一组件:
|
||||
|
||||
- 懒建 httpx.AsyncClient 并长期复用(与仓库内其他客户端一致)
|
||||
- temperature / max_tokens 传 None 时不出现在请求体(保持各后端原请求形状)
|
||||
- 密钥解析统一走 config.get_api_key(此前 experts 的默认环境名与另两处不一致)
|
||||
|
||||
依赖说明:httpx 仅 api 类型后端需要,import 延迟到连接创建时,
|
||||
router_system 核心保持零第三方依赖约束不变。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
class OpenAICompatClient:
|
||||
"""OpenAI 兼容 /chat/completions 客户端(懒建连接、跨调用复用)。"""
|
||||
|
||||
def __init__(self, base_url: str, api_key: str, timeout: float = 60.0):
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
self._timeout = timeout
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client is None:
|
||||
import httpx
|
||||
self._client = httpx.AsyncClient(timeout=self._timeout)
|
||||
return self._client
|
||||
|
||||
async def chat(
|
||||
self,
|
||||
model: str,
|
||||
messages: List[Dict[str, str]],
|
||||
temperature: Optional[float] = 0.2,
|
||||
max_tokens: Optional[int] = 1024,
|
||||
) -> Dict[str, Any]:
|
||||
"""调用 /chat/completions,返回完整响应 JSON(含 usage)。
|
||||
|
||||
HTTP 非 2xx 时抛 httpx.HTTPStatusError;temperature/max_tokens
|
||||
为 None 时对应字段不下发(与旧实现中 LLMJudge 的请求体一致)。
|
||||
"""
|
||||
payload: Dict[str, Any] = {"model": model, "messages": messages}
|
||||
if temperature is not None:
|
||||
payload["temperature"] = temperature
|
||||
if max_tokens is not None:
|
||||
payload["max_tokens"] = max_tokens
|
||||
client = self._get_client()
|
||||
resp = await client.post(
|
||||
f"{self.base_url}/chat/completions",
|
||||
headers={"Authorization": f"Bearer {self.api_key}"},
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
@staticmethod
|
||||
def completion_text(data: Dict[str, Any]) -> str:
|
||||
"""提取首条回复文本。"""
|
||||
return data["choices"][0]["message"]["content"]
|
||||
|
||||
@staticmethod
|
||||
def completion_tokens(data: Dict[str, Any], text: str) -> int:
|
||||
"""提取 completion token 数;usage 缺失时按 ~2.2 字符/token 估算。"""
|
||||
usage = data.get("usage", {})
|
||||
return usage.get("completion_tokens", int(len(text) / 2.2))
|
||||
+70
-68
@@ -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
|
||||
|
||||
+43
-7
@@ -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):
|
||||
|
||||
+74
-63
@@ -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,
|
||||
}
|
||||
|
||||
+1
-1
@@ -10,5 +10,5 @@ from router_system.router import build_router
|
||||
|
||||
@pytest.fixture()
|
||||
def router():
|
||||
"""????????? mock ?????????????"""
|
||||
"""构建全 mock 链路的 Router 实例(离线可跑,供各测试复用)。"""
|
||||
return build_router()
|
||||
|
||||
+4
-4
@@ -49,9 +49,9 @@ def test_exact_hit():
|
||||
|
||||
def test_semantic_hit():
|
||||
c = RouterCache(semantic_enabled=True, similarity_threshold=0.5)
|
||||
c.put("?python?????", {"response": "code", "domain": "code"})
|
||||
# ?????????? L2
|
||||
hit = c.get("?python????????")
|
||||
c.put("用 python 实现快速排序", {"response": "code", "domain": "code"})
|
||||
# 相似但不完全相同的查询应命中 L2 语义缓存
|
||||
hit = c.get("用 python 实现快速排序的算法")
|
||||
assert hit is not None
|
||||
assert hit[0] == "semantic"
|
||||
|
||||
@@ -60,7 +60,7 @@ def test_promote_to_exact():
|
||||
c = RouterCache(promote_frequency=3)
|
||||
result = {"response": "x", "domain": "general"}
|
||||
c.put("query", result)
|
||||
# ?????? 3 ? ? ???????
|
||||
# 语义命中累计 3 次后提升为精确缓存
|
||||
for _ in range(3):
|
||||
hit = c.get("query")
|
||||
assert hit is not None
|
||||
|
||||
@@ -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"
|
||||
@@ -0,0 +1,48 @@
|
||||
"""失败安全与词边界单元测试(T-R3,采纳 llmrouter 设计)。"""
|
||||
from router_system.classifier import RuleClassifier, build_classifier
|
||||
from router_system.difficulty import estimate_difficulty
|
||||
|
||||
|
||||
def test_word_boundary_stops_substring_false_positive():
|
||||
"""英文标记整词命中:'int' 不再被 'print'/'point' 误触发 hard。"""
|
||||
diff, score = estimate_difficulty("如何用 print 函数打印结果")
|
||||
# 旧实现 'int' 误命中 print 叠加 如何 -> hard;现在应为 medium 以下
|
||||
assert diff in ("easy", "medium")
|
||||
|
||||
diff2, _ = estimate_difficulty("请总结这段逻辑代码的思路")
|
||||
# 'sum' 不被 'summary/总结' 类场景误判:'逻辑' 含 'log' 也不再整词误命中
|
||||
assert diff2 in ("easy", "medium")
|
||||
|
||||
|
||||
def test_word_boundary_still_counts_whole_words():
|
||||
"""整词出现照常计数:int/log 作为真词仍触发 hard 信号。"""
|
||||
diff, _ = estimate_difficulty("explain how to implement int overflow and log parsing in depth")
|
||||
assert diff in ("medium", "hard")
|
||||
|
||||
|
||||
def test_chinese_markers_keep_substring_semantics():
|
||||
"""中文标记保持子串语义:'证明' 命中 '证明费马大定理'。"""
|
||||
diff, _ = estimate_difficulty("证明费马大定理并推导其推论,给出详细步骤")
|
||||
assert diff in ("medium", "hard")
|
||||
|
||||
|
||||
def test_build_hf_falls_back_when_unavailable(monkeypatch):
|
||||
"""HF 分类器构造失败(依赖缺失/模型加载失败)安全回落规则分类器。"""
|
||||
def _boom(self, *a, **k):
|
||||
raise RuntimeError("HuggingFaceClassifier 需要安装 ML 依赖")
|
||||
import router_system.classifier as C
|
||||
monkeypatch.setattr(C.HuggingFaceClassifier, "__init__", _boom)
|
||||
clf = build_classifier({"type": "hf", "model": "whatever"})
|
||||
assert isinstance(clf, RuleClassifier)
|
||||
assert clf.classify("用 Python 写一个快速排序函数").domain == "code"
|
||||
|
||||
|
||||
def test_hf_runtime_failure_falls_back_to_rules():
|
||||
"""推理期异常(torch 缺失/tokenizer 损坏)回落内置规则分类器。"""
|
||||
from router_system.classifier import HuggingFaceClassifier
|
||||
hf = HuggingFaceClassifier.__new__(HuggingFaceClassifier)
|
||||
hf.labels = ["code", "math", "legal", "medical", "general"]
|
||||
hf.confidence_floor = 0.55
|
||||
hf._rule_fallback = RuleClassifier(confidence_floor=0.55)
|
||||
r = hf.classify("用 Python 写一个快速排序函数") # 本环境无 torch/tokenizer -> 必走回落
|
||||
assert r.domain == "code"
|
||||
@@ -1,4 +1,4 @@
|
||||
"""???????? fastapi + httpx??"""
|
||||
"""网关集成测试(需 fastapi + httpx,链路全 mock)。"""
|
||||
import pytest
|
||||
|
||||
pytest.importorskip("fastapi")
|
||||
@@ -23,7 +23,7 @@ def test_health(client):
|
||||
|
||||
|
||||
def test_chat(client):
|
||||
resp = client.post("/chat", json={"query": "? Python ???????"})
|
||||
resp = client.post("/chat", json={"query": "用 Python 写一个快速排序函数"})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["response"]
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""分类规则外置单元测试(T-R2,采纳 llmrouter「规则文档即配置」设计)。"""
|
||||
import json
|
||||
|
||||
from router_system.classifier import (DOMAIN_RULES, RuleClassifier, build_classifier,
|
||||
load_domain_rules)
|
||||
|
||||
|
||||
def test_load_missing_file_returns_none(tmp_path):
|
||||
"""文件缺失:返回 None(调用方回退内置),不抛异常。"""
|
||||
assert load_domain_rules(tmp_path / "nope.json") is None
|
||||
|
||||
|
||||
def test_external_rules_override_domain(tmp_path):
|
||||
"""外置文件按 domain 覆盖内置:改写 code 规则即改变路由行为。"""
|
||||
f = tmp_path / "routes.json"
|
||||
f.write_text(json.dumps({"code": [["速排", 2.0]]}, ensure_ascii=False),
|
||||
encoding="utf-8")
|
||||
rules = load_domain_rules(f)
|
||||
assert rules is not None and rules["code"] == [("速排", 2.0)]
|
||||
|
||||
clf = RuleClassifier()
|
||||
clf.rules = {**DOMAIN_RULES, **rules}
|
||||
r = clf.classify("讲讲速排的思路")
|
||||
assert r.domain == "code"
|
||||
|
||||
|
||||
def test_build_classifier_merges_over_builtin(tmp_path):
|
||||
"""build_classifier 合并语义:文件只写 legal,其余领域保持内置。"""
|
||||
f = tmp_path / "routes.json"
|
||||
f.write_text(json.dumps({"legal": [["劳动合同", 3.0]]}, ensure_ascii=False),
|
||||
encoding="utf-8")
|
||||
clf = build_classifier({"type": "rule", "rules_file": str(f)})
|
||||
assert clf.rules["legal"] == [("劳动合同", 3.0)]
|
||||
assert clf.rules["code"] == DOMAIN_RULES["code"] # 未覆盖领域保持内置
|
||||
r = clf.classify("劳动合同到期不续签需要支付经济补偿吗")
|
||||
assert r.domain == "legal"
|
||||
|
||||
|
||||
def test_malformed_file_falls_back_to_builtin(tmp_path):
|
||||
"""格式非法:整体回退内置(失败安全,编辑错误不打垮路由)。"""
|
||||
f = tmp_path / "routes.json"
|
||||
f.write_text('{"code": [["坏数据", "不是数字"]]}', encoding="utf-8")
|
||||
assert load_domain_rules(f) is None
|
||||
clf = build_classifier({"type": "rule", "rules_file": str(f)})
|
||||
assert clf.rules == DOMAIN_RULES
|
||||
r = clf.classify("用 Python 写一个快速排序函数")
|
||||
assert r.domain == "code"
|
||||
|
||||
|
||||
def test_explicit_rules_file_missing_uses_builtin(tmp_path):
|
||||
"""显式路径不存在:静默回退内置,分类行为不变。"""
|
||||
clf = build_classifier({"type": "rule", "rules_file": str(tmp_path / "no.json")})
|
||||
assert clf.rules == DOMAIN_RULES
|
||||
Reference in New Issue
Block a user