Author SHA1 Message Date
tzt 06700a5aa3 feat(v1): T-R3 采纳 llmrouter 词边界与失败安全——难度英文标记整词命中 + HF 分类器两级回落
- difficulty:标记表编译拆分——英文标记改词边界正则(修真实误判:'int' 子串
  命中 'print'/'point'、'log' 命中 'logic'、'list' 命中 'listen'),中文标记
  保持子串语义;命中行为对合法用例不变(整词出现照常计数)
- classifier:HuggingFaceClassifier 推理期异常回落内置 RuleClassifier(单次
  推理异常不打垮路由);build_classifier 的 hf 分支构造失败(ML 依赖缺失/
  模型加载失败)打印提示并回落规则分类器(外置规则照常合并)
- 新增 tests/test_failsafe.py 5 项;全量 43 passed(38+5)
2026-09-19 09:27:15 +08:00
tzt 0dad895aac feat(v1): T-R2 采纳 llmrouter「规则文档即配置」——分类规则外置 config/routes.json
- classifier.load_domain_rules():支持 .json(必有)/.yaml(pyyaml 可选,与
  config.py 同一可选依赖纪律);文件按 domain 整域覆盖内置 DOMAIN_RULES
  (与 v2 知识库'文件按 id 覆盖'同一惯例);缺失/格式非法/条目非法整体
  安全回退内置(llmrouter 失败安全思想,规则文档编辑错误不打垮路由)
- build_classifier:cfg.rules_file 显式指定,未指定时约定路径 config/routes.json
  存在即自动加载(约定优于配置,ROUTES.md 精髓:改文档即改行为,可 review 可版本化)
- 新增 tests/test_rules_external.py 5 项;全量 38 passed(33+5)
2026-09-19 09:24:56 +08:00
tzt a9f60159e7 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)
2026-09-19 09:23:04 +08:00
tzt 5b28c2a583 chore(v1): 外部优点融入前置锚点——本轮五项目吸收的回滚基线 2026-09-19 09:16:52 +08:00
tzt c072a1a237 feat(v1): 架构与算法优化二轮——共享 LLM 客户端去重、语义缓存 O(1) 提升/淘汰、分类器热路径清理
- 新增 router_system/llm_client.py:OpenAICompatClient 统一 experts/judge/fallback
  三处复制的懒建 AsyncClient + /chat/completions + choices/usage 解析(~60 行去重);
  密钥解析统一走 config.get_api_key(激活原死代码,顺带消除 experts 默认环境名不一致)
- 语义缓存 L2:条目容器 list→OrderedDict(提升/淘汰 O(n)→O(1)),按 query 天然去重;
  n-gram 向量 lru_cache 复用(同一次 miss 的 get/put 免重复分词);
  A/B:淘汰路径 0.040→0.034s,miss→put 往返 9.41→8.57s(-9%)
- RuleJudge 覆盖度:response.lower() 提出逐词循环(原 O(terms×len) 重复复制)
- extract_content_terms 纯函数 lru_cache 化(专家与 Judge 对同一查询免重复分词),返回 tuple
- RuleClassifier:_score 去掉败者领域白建的命中词 list(胜出后单独收集);
  修复 code 规则 ("api",0.7) 重复登记(原命中计 1.4 分)
- difficulty:正则模块级预编译
- tests:恢复上一轮引入的乱码中文 docstring;网关测试输入串恢复为可判 code 的中文查询
2026-09-18 23:27:26 +08:00
tzt 81b0bc9b9b chore(v1): 优化前置锚点——本轮架构/算法优化的回滚基线 2026-09-18 23:03:04 +08:00
tzt 9fdf91c2fb feat(v1): 架构与算法优化——语义缓存 2.37x、Router 去重、分类器确定性决胜
架构:
- Router:低置信度直连/专家异常两条兜底路径的收尾逻辑(回退→finalize→record)提取为
  _fallback_result 公共方法,消除三处重复收尾块

算法:
- RouterCache:语义条目写入时预计算向量范数(原每次两两比较重算)、
  语义查找单遍完成(原命中后二次 O(N) 查找)、相似度=1.0 提前终止扫描;
  微基准(3000 条目×200 查询):3986ms -> 1685ms,2.37x
- RuleClassifier:同分决胜改为按领域名字典序(与规则表排列顺序无关的确定性)、
  次高分由全排序改 O(n) 扫描

测试:新增 5 项(缓存范数一致性/提升后无残留/淘汰同步清理、决胜确定性、区分度惩罚)
pytest 25 passed(原 20 全绿 + 新增 5)
基线检查点:66b6fd8(操作前已提交,20 passed)
2026-09-18 08:10:02 +08:00
tzt 66b6fd8c3e docs(branch): 第一代快照说明——本地多智能体协作模型路由 2026-09-15 10:02:06 +08:00
21 changed files with 1893 additions and 1259 deletions
+6
View File
@@ -25,3 +25,9 @@ cached_results/
Thumbs.db Thumbs.db
.idea/ .idea/
.vscode/ .vscode/
# 安全扫描器工作目录(不入库)
.mimosa/
# 参考项目目录(不入库,仅作设计吸收来源)
extra/
+29
View File
@@ -0,0 +1,29 @@
# 分支:v1-model-routing — 本地多智能体协作模型路由(第一代)
> **快照点**:`1e51167`(v1 MVP 完成时点,仓库首个提交)。
> 此分支为**历史路标**,冻结不再演进;集成主线见 `master`。
## 这一代是什么
- **核心命题**:用"规则知识库 + 任务拆解 Planner + 专业执行器池 + 质量控制器(Judge) + 最后处理者"
在限定条件下替代单一通用大模型——**本地多智能体协作路由**
- 架构:8 领域规则知识库 → 规则分类器(置信度 1−e^−s)→ Planner 任务拆解(DAG)
→ 黑板/前向链 → 规则执行器(L0 零参数、确定性模板)→ RuleJudge 五维评分 → mock 兜底
- 两级路由:`domain_groups`(tech/professional/lifestyle/general)→ 组内 RuleClassifier;三级子领域识别
- 接口:FastAPI `/chat` `/health` `/metrics`;核心包 `router_system/` 零第三方依赖
## 如何运行
```powershell
.venv\Scripts\python.exe scripts/demo.py --trace # L0 演示(离线可跑)
.venv\Scripts\python.exe scripts/eval.py # 迷你评估
.venv\Scripts\python.exe -m pytest tests -q # 测试(初版 20 项)
.venv\Scripts\python.exe scripts/serve.py --port 8000 # 网关
```
## 定位与结论
- **方向结论**:规则路由在 9 域平衡集实测 74.4% 准确率、68.9% 升级率(`research/routerarena/`)——
假阳性过高,**被 v2 端云协作风取代**;本代代码在主线保留为 legacy(`POST /chat/legacy`)
- 设计文档:`实现方案_多专业小模型+路由模型.md`、`可行性调研与落地实现路线报告.md`、
`research/2026_papers_survey.md`、`research/routerarena/01_results_and_gap_analysis.md`
+2 -1
View File
@@ -9,7 +9,7 @@
""" """
from __future__ import annotations from __future__ import annotations
from typing import List, Optional from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@@ -46,6 +46,7 @@ class QueryResponse(BaseModel):
cache_level: Optional[str] cache_level: Optional[str]
cost_est: float cost_est: float
error: Optional[str] = None error: Optional[str] = None
route_explanation: Optional[Dict[str, Any]] = None # T-R1 可解释评分(缓存命中为 None)
class HealthResponse(BaseModel): class HealthResponse(BaseModel):
+71 -40
View File
@@ -7,11 +7,22 @@
高频语义命中会提升为 O(1) 的精确缓存条目。 高频语义命中会提升为 O(1) 的精确缓存条目。
只缓存"未升级"的结果(升级路径每次都走大模型,不缓存,避免陈旧)。 只缓存"未升级"的结果(升级路径每次都走大模型,不缓存,避免陈旧)。
性能设计(2026-09 优化):
- 每条语义缓存条目在写入时预计算并缓存向量范数,查询时免重复计算(原来每对比较都重算)
- 语义查找单遍完成:扫描即跟踪最优条目与命中计数,命中后不再二次线性查找
- 相似度达到 1.0(完全相同查询)时提前终止扫描(余弦相似度上界,不可能更优)
- 语义条目容器为 OrderedDict:提升/淘汰均为 O(1)(原 list pop 是 O(n) 搬移),
且按 query 天然去重(原实现同一 query 可重复追加,重复条目白白参与扫描)
- n-gram 向量经 lru_cache 复用:同一次未命中请求里 get 与紧随的 put
不再重复做两遍分词+计数
""" """
from __future__ import annotations from __future__ import annotations
import re import re
from collections import OrderedDict
from dataclasses import dataclass from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Dict, List, Optional, Tuple from typing import Any, Dict, List, Optional, Tuple
@@ -29,18 +40,6 @@ def _ngrams(text: str, n: int = 3) -> List[str]:
return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)] return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)]
def _cosine(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float:
if not vec_a or not vec_b:
return 0.0
common = set(vec_a) & set(vec_b)
dot = sum(vec_a[k] * vec_b[k] for k in common)
na = sum(v * v for v in vec_a.values()) ** 0.5
nb = sum(v * v for v in vec_b.values()) ** 0.5
if na == 0 or nb == 0:
return 0.0
return dot / (na * nb)
def _tf_vector(grams: List[str]) -> Dict[str, float]: def _tf_vector(grams: List[str]) -> Dict[str, float]:
vec: Dict[str, float] = {} vec: Dict[str, float] = {}
for g in grams: for g in grams:
@@ -48,6 +47,28 @@ def _tf_vector(grams: List[str]) -> Dict[str, float]:
return vec return vec
def _norm(vec: Dict[str, float]) -> float:
return sum(v * v for v in vec.values()) ** 0.5
def _dot(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float:
"""点积:遍历较小的一方,另一侧用 get 兜底。"""
if len(vec_a) > len(vec_b):
vec_a, vec_b = vec_b, vec_a
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: class RouterCache:
"""L1 精确缓存 + L2 语义缓存。""" """L1 精确缓存 + L2 语义缓存。"""
@@ -59,8 +80,10 @@ class RouterCache:
self.max_exact = max_exact self.max_exact = max_exact
self.max_semantic = max_semantic self.max_semantic = max_semantic
self._exact: Dict[str, CacheEntry] = {} 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_vecs: Dict[str, Dict[str, float]] = {}
self._sem_norms: Dict[str, float] = {} # 预计算范数,避免查询期重算
self.hits = {"exact": 0, "semantic": 0} self.hits = {"exact": 0, "semantic": 0}
self.misses = 0 self.misses = 0
@@ -73,49 +96,56 @@ class RouterCache:
return ("exact", entry.result) return ("exact", entry.result)
if self.semantic_enabled: if self.semantic_enabled:
q_vec = _tf_vector(_ngrams(query)) q_vec, q_norm = _embed(query)
best_sim = 0.0 best_sim = 0.0
best_query: Optional[str] = None best_q = ""
best_result: Optional[Dict[str, Any]] = None if q_norm > 0.0:
for q, e in self._semantic: # 单遍扫描(插入序):同时跟踪最优条目,sim=1.0 提前终止
sim = _cosine(q_vec, self._sem_vecs.get(q, {})) for q, _e in self._semantic.items():
if sim > best_sim: n_q = self._sem_norms.get(q, 0.0)
best_sim = sim if n_q <= 0.0:
best_query = q continue
best_result = e.result sim = _dot(q_vec, self._sem_vecs.get(q, {})) / (q_norm * n_q)
if best_query is not None and best_sim >= self.similarity_threshold: if sim > best_sim:
best_sim = sim
best_q = q
if sim >= 1.0:
break # 余弦相似度上界:完全相同查询,提前终止
if best_q and best_sim >= self.similarity_threshold:
best_entry = self._semantic[best_q]
# 完全相同查询(相似度=1.0)计为 exact 命中 # 完全相同查询(相似度=1.0)计为 exact 命中
is_exact = best_sim >= 0.999 is_exact = best_sim >= 0.999
level = "exact" if is_exact else "semantic" level = "exact" if is_exact else "semantic"
self.hits[level] += 1 self.hits[level] += 1
self._semantic_hit(best_query) self._bump_semantic(best_q, best_entry)
return (level, best_result) return (level, best_entry.result)
self.misses += 1 self.misses += 1
return None return None
def _semantic_hit(self, query: str): def _bump_semantic(self, query: str, entry: CacheEntry):
"""语义命中:累计命中次数,达到阈值提升为精确缓存。""" """语义命中:累计命中次数,达到阈值提升为精确缓存(OrderedDict 删除 O(1))。"""
for i, (q, e) in enumerate(self._semantic): entry.hits += 1
if q == query: if entry.hits >= self.promote_frequency:
e.hits += 1 self._exact[query] = entry
if e.hits >= self.promote_frequency: del self._semantic[query]
self._exact[query] = e self._sem_vecs.pop(query, None)
self._semantic.pop(i) self._sem_norms.pop(query, None)
self._sem_vecs.pop(query, None)
break
# ---- 写入 ---- # ---- 写入 ----
def put(self, query: str, result: Dict[str, Any]): def put(self, query: str, result: Dict[str, Any]):
if query in self._exact: if query in self._exact or query in self._semantic:
return return # 已缓存(语义区按 query 天然去重,避免重复条目参与扫描)
entry = CacheEntry(result=result) entry = CacheEntry(result=result)
if self.semantic_enabled: if self.semantic_enabled:
if len(self._semantic) >= self.max_semantic: 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_vecs.pop(old_q, None)
self._semantic.append((query, entry)) self._sem_norms.pop(old_q, None)
self._sem_vecs[query] = _tf_vector(_ngrams(query)) self._semantic[query] = entry
vec, norm = _embed(query) # 与 get miss 共享同一份缓存向量
self._sem_vecs[query] = vec
self._sem_norms[query] = norm
else: else:
self._exact[query] = entry self._exact[query] = entry
if len(self._exact) > self.max_exact: if len(self._exact) > self.max_exact:
@@ -137,5 +167,6 @@ class RouterCache:
self._exact.clear() self._exact.clear()
self._semantic.clear() self._semantic.clear()
self._sem_vecs.clear() self._sem_vecs.clear()
self._sem_norms.clear()
self.hits = {"exact": 0, "semantic": 0} self.hits = {"exact": 0, "semantic": 0}
self.misses = 0 self.misses = 0
+115 -30
View File
@@ -6,11 +6,18 @@
置信度设计:每个领域有一组 (关键词, 权重)。命中权重求和得原始分 s, 置信度设计:每个领域有一组 (关键词, 权重)。命中权重求和得原始分 s,
confidence = 1 - exp(-s),保证 s=1 -> 0.63,s=2 -> 0.86,s=3 -> 0.95。 confidence = 1 - exp(-s),保证 s=1 -> 0.63,s=2 -> 0.86,s=3 -> 0.95。
无领域命中(或最高分领域为 general)时置信度低,触发 should_fallback。 无领域命中(或最高分领域为 general)时置信度低,触发 should_fallback。
规则外置(T-R2,采纳 llmrouter「规则文档即配置」设计):领域规则支持从
config/routes.json(或 .yaml,pyyaml 可选)加载,文件按 domain 覆盖内置
DOMAIN_RULES(与 v2 知识库"文件按 id 覆盖"同一惯例);文件缺失/格式非法
整体安全回退内置(llmrouter 失败安全思想)。
""" """
from __future__ import annotations from __future__ import annotations
import json
import math 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 .difficulty import estimate_difficulty
from .models import Classification 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), ("algorithm", 0.8), ("sort", 0.7), ("array", 0.7), ("regex", 0.8),
("import", 0.7), ("loop", 0.7), ("recursion", 0.8), ("refactor", 0.8), ("import", 0.7), ("loop", 0.7), ("recursion", 0.8), ("refactor", 0.8),
("deploy", 0.8), ("docker", 0.8), ("kubernetes", 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), ("索引", 0.8), ("优化", 0.7), ("查询", 0.6),
], ],
"math": [ "math": [
@@ -82,6 +89,50 @@ _STOPWORDS = {
"or", "do", "does", "can", "could", "would", "should", "please", "me", "my", "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: class BaseClassifier:
def classify(self, query: str) -> Classification: def classify(self, query: str) -> Classification:
@@ -98,24 +149,29 @@ class RuleClassifier(BaseClassifier):
self.confidence_floor = confidence_floor self.confidence_floor = confidence_floor
self.rules = DOMAIN_RULES self.rules = DOMAIN_RULES
def _score(self, query: str) -> Tuple[Dict[str, float], Dict[str, List[str]]]: def _score(self, q: str) -> Dict[str, float]:
q = query.lower() """对已 lowercase 的查询按领域规则打分,返回有命中的领域分数。
命中词列表只对最终胜出领域有意义(classify 的唯一消费点),
故不在各领域上白建 list——胜出后由 _matched_rules 单独收集。
"""
scores: Dict[str, float] = {} scores: Dict[str, float] = {}
matched: Dict[str, List[str]] = {}
for domain, rules in self.rules.items(): for domain, rules in self.rules.items():
s = 0.0 s = 0.0
hits = []
for kw, w in rules: for kw, w in rules:
if kw in q: if kw in q:
s += w s += w
hits.append(kw)
if s > 0: if s > 0:
scores[domain] = s scores[domain] = s
matched[domain] = hits return scores
return scores, matched
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: def classify(self, query: str) -> Classification:
raw, matched = self._score(query) q = query.lower()
raw = self._score(q)
if not raw: if not raw:
# 完全无命中 -> general,低置信度 # 完全无命中 -> general,低置信度
diff, ds = estimate_difficulty(query) diff, ds = estimate_difficulty(query)
@@ -128,7 +184,8 @@ class RuleClassifier(BaseClassifier):
matched_rules=[], matched_rules=[],
) )
best_domain = max(raw, key=raw.get) # 同分决胜:按领域名字典序,保证与规则表排列顺序无关的确定性
best_domain = max(sorted(raw), key=lambda d: raw[d])
best_score = raw[best_domain] best_score = raw[best_domain]
confidence = 1.0 - math.exp(-best_score) confidence = 1.0 - math.exp(-best_score)
@@ -138,7 +195,7 @@ class RuleClassifier(BaseClassifier):
# 与次高分的差距影响置信度(区分度) # 与次高分的差距影响置信度(区分度)
if len(raw) > 1: if len(raw) > 1:
second = sorted(raw.values(), reverse=True)[1] second = max(v for d, v in raw.items() if d != best_domain)
if second > 0.7 * best_score: if second > 0.7 * best_score:
confidence *= 0.85 confidence *= 0.85
@@ -149,7 +206,7 @@ class RuleClassifier(BaseClassifier):
difficulty=diff, difficulty=diff,
difficulty_score=ds, difficulty_score=ds,
raw_scores={k: round(v, 3) for k, v in raw.items()}, 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),
) )
@@ -157,6 +214,8 @@ class HuggingFaceClassifier(BaseClassifier):
"""可选:基于 transformers 的序列分类模型。 """可选:基于 transformers 的序列分类模型。
仅当安装 torch+transformers 且模型可加载时可用;否则抛错提示。 仅当安装 torch+transformers 且模型可加载时可用;否则抛错提示。
失败安全(T-R3,采纳 llmrouter 分类失败静默降级思想):模型加载成功但
推理期异常时,自动回落内置规则分类器,不让单次推理异常打垮路由。
""" """
def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55): def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55):
@@ -172,32 +231,58 @@ class HuggingFaceClassifier(BaseClassifier):
) )
self.labels = ["code", "math", "legal", "medical", "general"] self.labels = ["code", "math", "legal", "medical", "general"]
self.confidence_floor = confidence_floor self.confidence_floor = confidence_floor
self._rule_fallback = RuleClassifier(confidence_floor=confidence_floor)
def classify(self, query: str) -> Classification: 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) inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
with torch.no_grad(): with torch.no_grad():
logits = self.model(**inputs).logits logits = self.model(**inputs).logits
probs = torch.softmax(logits, dim=-1)[0] probs = torch.softmax(logits, dim=-1)[0]
idx = int(probs.argmax()) idx = int(probs.argmax())
diff, ds = estimate_difficulty(query) diff, ds = estimate_difficulty(query)
return Classification( return Classification(
domain=self.labels[idx], domain=self.labels[idx],
confidence=round(float(probs[idx]), 4), confidence=round(float(probs[idx]), 4),
difficulty=diff, difficulty=diff,
difficulty_score=ds, difficulty_score=ds,
raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))}, 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: def build_classifier(cfg: Dict) -> BaseClassifier:
"""根据配置构建分类器。cfg 为 classifier 段配置。""" """根据配置构建分类器。cfg 为 classifier 段配置。
T-R2:cfg.rules_file 指定外置规则文件;未指定时若约定路径
config/routes.json 存在则自动加载。文件按 domain 覆盖内置规则。
"""
ctype = cfg.get("type", "rule") ctype = cfg.get("type", "rule")
floor = cfg.get("confidence_floor", 0.55) floor = cfg.get("confidence_floor", 0.55)
if ctype == "rule": 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": 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)") raise ValueError(f"未知分类器类型: {ctype}(支持 rule | hf)")
+35 -4
View File
@@ -26,11 +26,42 @@ _MEDIUM_MARKERS = [
] ]
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]: def estimate_difficulty(query: str) -> Tuple[str, float]:
"""返回 (difficulty, score),score 属于 [0,1]。""" """返回 (difficulty, score),score 属于 [0,1]。"""
q = query.lower() q = query.lower()
hard_hits = sum(1 for m in _HARD_MARKERS if m in q) hard_hits = _hit_count(_HARD_SUBSTR, _HARD_RE, q)
medium_hits = sum(1 for m in _MEDIUM_MARKERS if m in q) medium_hits = _hit_count(_MEDIUM_SUBSTR, _MEDIUM_RE, q)
length = len(query) length = len(query)
score = 0.0 score = 0.0
@@ -39,9 +70,9 @@ def estimate_difficulty(query: str) -> Tuple[str, float]:
score += min(0.30, medium_hits * 0.08) # 一般指令贡献 score += min(0.30, medium_hits * 0.08) # 一般指令贡献
# 额外:代码 / 数学表达式(多步骤信号) # 额外:代码 / 数学表达式(多步骤信号)
if "```" in query or re.search(r"\b(def|class|function|import)\b", q): if "```" in query or _RE_CODE_EXPR.search(q):
score += 0.15 score += 0.15
if re.search(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]", q): if _RE_ARITH_EXPR.search(q):
score += 0.15 score += 0.15
score = max(0.0, min(1.0, score)) score = max(0.0, min(1.0, score))
+32 -34
View File
@@ -10,8 +10,11 @@ from __future__ import annotations
import asyncio import asyncio
import re import re
from typing import Dict, List, Optional 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 from .models import ExpertResponse
# 按模型规模粗估的相对成本($ / 1M output tokens,近似) # 按模型规模粗估的相对成本($ / 1M output tokens,近似)
@@ -38,8 +41,8 @@ def _cost_for(model_name: str, default: str = "1b") -> float:
return MODEL_COST_EST.get(default, 0.1) return MODEL_COST_EST.get(default, 0.1)
def extract_content_terms(query: str) -> List[str]: @lru_cache(maxsize=2048)
"""抽取查询中的"内容词"(中文词/英文单词),用于 Judge 覆盖度与 Mock 回显。""" def _extract_content_terms_cached(query: str) -> Tuple[str, ...]:
q = query.lower() q = query.lower()
terms: List[str] = [] terms: List[str] = []
# 英文单词(>=2 字符) # 英文单词(>=2 字符)
@@ -50,7 +53,16 @@ def extract_content_terms(query: str) -> List[str]:
cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q) cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q)
for c in cn: for c in cn:
terms.append(c) terms.append(c)
return terms 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 = { _STOPWORDS_EN = {
@@ -63,6 +75,7 @@ _STOPWORDS_EN = {
class Expert: class Expert:
name: str = "expert" name: str = "expert"
nominal_latency_ms: float = 100.0 # 标称延迟元数据(可解释评分 speed 维输入)
async def generate(self, query: str, difficulty: str) -> ExpertResponse: async def generate(self, query: str, difficulty: str) -> ExpertResponse:
raise NotImplementedError raise NotImplementedError
@@ -75,6 +88,8 @@ class MockExpert(Expert):
使端到端管线(分类 -> 专家 -> Judge -> 缓存)可被稳定测试与演示。 使端到端管线(分类 -> 专家 -> Judge -> 缓存)可被稳定测试与演示。
""" """
nominal_latency_ms = 1.0
def __init__(self, name: str, domain: str, model: str = "mock"): def __init__(self, name: str, domain: str, model: str = "mock"):
self.name = name self.name = name
self.domain = domain self.domain = domain
@@ -150,6 +165,8 @@ class MockExpert(Expert):
class HFExpert(Expert): class HFExpert(Expert):
"""可选:HuggingFace 真实小模型(需 requirements-ml.txt)。""" """可选:HuggingFace 真实小模型(需 requirements-ml.txt)。"""
nominal_latency_ms = 300.0
def __init__(self, name: str, domain: str, model: str): def __init__(self, name: str, domain: str, model: str):
self.name = name self.name = name
self.domain = domain self.domain = domain
@@ -198,37 +215,23 @@ class HFExpert(Expert):
class APIExpert(Expert): class APIExpert(Expert):
"""可选:OpenAI 兼容 Chat Completions(DeepSeek / OpenAI / 本地 vLLM)。""" """可选: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): def __init__(self, name: str, domain: str, model: str, base_url: str, api_key: str):
self.name = name self.name = name
self.domain = domain self.domain = domain
self.model = model self.model = model
self.base_url = base_url.rstrip("/") self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
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: async def generate(self, query: str, difficulty: str) -> ExpertResponse:
client = self._get_client() data = await self._client.chat(
resp = await client.post( self.model,
f"{self.base_url}/chat/completions", [{"role": "user", "content": query}],
headers={"Authorization": f"Bearer {self.api_key}"}, temperature=0.2,
json={ max_tokens=1024,
"model": self.model,
"messages": [{"role": "user", "content": query}],
"temperature": 0.2,
"max_tokens": 1024,
},
) )
resp.raise_for_status() body = self._client.completion_text(data)
data = resp.json() tokens = self._client.completion_tokens(data, body)
body = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
tokens = usage.get("completion_tokens", int(len(body) / 2.2))
return ExpertResponse( return ExpertResponse(
text=body, text=body,
model_used=self.model, model_used=self.model,
@@ -249,18 +252,13 @@ def build_expert(domain: str, cfg: Dict) -> Expert:
return HFExpert(name, domain, model) return HFExpert(name, domain, model)
if etype == "api": if etype == "api":
base_url = cfg.get("base_url", "https://api.deepseek.com/v1") base_url = cfg.get("base_url", "https://api.deepseek.com/v1")
api_key = cfg.get("api_key") or _env(cfg.get("api_key_env", "")) api_key = cfg.get("api_key") or get_api_key(cfg)
if not api_key: if not api_key:
raise RuntimeError(f"APIExpert({domain}) 缺少 API Key(env: {cfg.get('api_key_env')})") raise RuntimeError(f"APIExpert({domain}) 缺少 API Key(env: {cfg.get('api_key_env')})")
return APIExpert(name, domain, model, base_url, api_key) return APIExpert(name, domain, model, base_url, api_key)
raise ValueError(f"未知专家后端类型: {etype}(支持 mock | hf | api)") 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]: def build_expert_pool(experts_cfg: Dict[str, Dict], domains: List[str]) -> Dict[str, Expert]:
"""构建完整专家池。""" """构建完整专家池。"""
pool: Dict[str, Expert] = {} pool: Dict[str, Expert] = {}
+106
View File
@@ -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)
+12 -27
View File
@@ -2,8 +2,10 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from typing import Dict, Optional from typing import Dict
from .config import get_api_key
from .llm_client import OpenAICompatClient
from .models import ExpertResponse from .models import ExpertResponse
@@ -45,34 +47,18 @@ class APIFallback(FallbackProvider):
def __init__(self, model: str, base_url: str, api_key: str): def __init__(self, model: str, base_url: str, api_key: str):
self.model = model self.model = model
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.name = f"fallback-{model}" self.name = f"fallback-{model}"
self._client = None self._client = OpenAICompatClient(base_url, api_key, timeout=90.0)
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: async def generate(self, query: str) -> ExpertResponse:
client = self._get_client() data = await self._client.chat(
resp = await client.post( self.model,
f"{self.base_url}/chat/completions", [{"role": "user", "content": query}],
headers={"Authorization": f"Bearer {self.api_key}"}, temperature=0.3,
json={ max_tokens=2048,
"model": self.model,
"messages": [{"role": "user", "content": query}],
"temperature": 0.3,
"max_tokens": 2048,
},
) )
resp.raise_for_status() body = self._client.completion_text(data)
data = resp.json() tokens = self._client.completion_tokens(data, body)
body = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
tokens = usage.get("completion_tokens", int(len(body) / 2.2))
return ExpertResponse( return ExpertResponse(
text=body, text=body,
model_used=self.model, model_used=self.model,
@@ -89,8 +75,7 @@ def build_fallback(cfg: Dict) -> FallbackProvider:
if ftype == "mock": if ftype == "mock":
return MockFallback(model=model) return MockFallback(model=model)
if ftype == "api": if ftype == "api":
import os api_key = cfg.get("api_key") or get_api_key(cfg)
api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY"))
if not api_key: if not api_key:
raise RuntimeError( raise RuntimeError(
f"APIFallback 缺少 API Key:请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key" f"APIFallback 缺少 API Key:请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key"
+13 -19
View File
@@ -66,7 +66,8 @@ class RuleJudge(BaseJudge):
# 1) 内容覆盖度:查询中的内容词有多少出现在响应里 # 1) 内容覆盖度:查询中的内容词有多少出现在响应里
terms = extract_content_terms(query) terms = extract_content_terms(query)
if terms: if terms:
hit = sum(1 for t in terms if t in response.lower()) lowered = response.lower() # lowercase 一次,避免逐词重复复制长响应
hit = sum(1 for t in terms if t in lowered)
coverage = hit / len(terms) coverage = hit / len(terms)
scores["coverage"] = coverage scores["coverage"] = coverage
if coverage < 0.4: if coverage < 0.4:
@@ -120,33 +121,26 @@ class LLMJudge(BaseJudge):
"""可选:LLM-as-Judge(API 后端)。""" """可选:LLM-as-Judge(API 后端)。"""
def __init__(self, model: str, base_url: str, api_key: str, fallback_threshold: float = 0.70): 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.model = model
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.fallback_threshold = fallback_threshold self.fallback_threshold = fallback_threshold
self._client = None # 请求体与旧实现一致:不下发 temperature / max_tokens
self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
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: async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
client = self._get_client()
prompt = ( prompt = (
f"你是质量评审员。评估以下回答对查询的满足程度,输出 0-1 分(相关性/正确性/完整性)。\n" f"你是质量评审员。评估以下回答对查询的满足程度,输出 0-1 分(相关性/正确性/完整性)。\n"
f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n" f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n"
f"只输出一个 0 到 1 之间的数字。" f"只输出一个 0 到 1 之间的数字。"
) )
resp = await client.post( data = await self._client.chat(
f"{self.base_url}/chat/completions", self.model,
headers={"Authorization": f"Bearer {self.api_key}"}, [{"role": "user", "content": prompt}],
json={"model": self.model, "messages": [{"role": "user", "content": prompt}]}, temperature=None,
max_tokens=None,
) )
resp.raise_for_status()
try: try:
score = float(resp.json()["choices"][0]["message"]["content"].strip()) score = float(self._client.completion_text(data).strip())
score = max(0.0, min(1.0, score)) score = max(0.0, min(1.0, score))
except Exception: except Exception:
score = 0.5 score = 0.5
@@ -163,8 +157,8 @@ def build_judge(cfg: Dict, fallback_threshold: float = 0.70) -> BaseJudge:
if jtype == "rule": if jtype == "rule":
return RuleJudge(fallback_threshold=fallback_threshold) return RuleJudge(fallback_threshold=fallback_threshold)
if jtype == "llm": if jtype == "llm":
import os from .config import get_api_key
api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY")) api_key = cfg.get("api_key") or get_api_key(cfg)
return LLMJudge( return LLMJudge(
cfg.get("model", "deepseek-chat"), cfg.get("model", "deepseek-chat"),
cfg.get("base_url", "https://api.deepseek.com/v1"), cfg.get("base_url", "https://api.deepseek.com/v1"),
+69
View File
@@ -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))
+2
View File
@@ -44,6 +44,7 @@ class RouterResult:
cache_level: Optional[str] = None # exact | semantic cache_level: Optional[str] = None # exact | semantic
cost_est: float = 0.0 cost_est: float = 0.0
error: Optional[str] = None error: Optional[str] = None
route_explanation: Optional[Dict[str, Any]] = None # 可解释评分(T-R1,缓存命中为 None)
def to_dict(self) -> Dict[str, Any]: def to_dict(self) -> Dict[str, Any]:
return { return {
@@ -61,6 +62,7 @@ class RouterResult:
"cache_level": self.cache_level, "cache_level": self.cache_level,
"cost_est": round(self.cost_est, 6), "cost_est": round(self.cost_est, 6),
"error": self.error, "error": self.error,
"route_explanation": self.route_explanation,
} }
+58 -18
View File
@@ -45,6 +45,8 @@ class Router:
self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60) self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60)
self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70) self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70)
self.cache_enabled = cfg.get("cache", {}).get("enabled", True) 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: async def route(self, query: str) -> RouterResult:
@@ -80,16 +82,16 @@ class Router:
classification = self.classifier.classify(query) classification = self.classifier.classify(query)
route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}") 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): if self.classifier.should_fallback(classification, self.low_confidence_threshold):
route.append("direct_fallback") route.append("direct_fallback")
fb = await self._call_fallback(query) return await self._fallback_result(query, classification, route, start,
latency = now_ms() - start explanation=explanation)
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)
self._record(result, latency)
return result
# ---- Step 3: 选择专家 ---- # ---- Step 3: 选择专家 ----
domain = classification.domain domain = classification.domain
@@ -106,14 +108,8 @@ class Router:
except Exception as e: except Exception as e:
self.stats.record_error() self.stats.record_error()
route.append(f"expert_error:{type(e).__name__}") route.append(f"expert_error:{type(e).__name__}")
fb = await self._call_fallback(query) return await self._fallback_result(query, classification, route, start,
latency = now_ms() - start error=str(e), explanation=explanation)
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=str(e))
self._record(result, latency)
return result
# ---- Step 5: Judge 评估 ---- # ---- Step 5: Judge 评估 ----
try: try:
@@ -135,16 +131,59 @@ class Router:
latency = now_ms() - start latency = now_ms() - start
result = self._finalize(query, classification, final_resp, quality_score=quality_score, result = self._finalize(query, classification, final_resp, quality_score=quality_score,
upgraded=upgraded, route=route, latency_ms=latency, 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) self._record(result, latency)
# 未升级的结果写缓存 # 未升级的结果写缓存(解释不随缓存复用——缓存命中路径 explanation=None)
if self.cache_enabled and not upgraded and result.response: if self.cache_enabled and not upgraded and result.response:
self.cache.put(query, result.to_dict()) self.cache.put(query, result.to_dict())
return result 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,
explanation=None) -> RouterResult:
"""兜底路径的公共收尾:调用大模型回退 -> finalize -> 记录指标。
低置信度直连、专家异常两条路径共用,避免收尾逻辑三处重复。
"""
fb = await self._call_fallback(query)
latency = now_ms() - start
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, explanation=explanation)
self._record(result, latency)
return result
async def _call_fallback(self, query: str) -> ExpertResponse: async def _call_fallback(self, query: str) -> ExpertResponse:
try: try:
return await self.fallback.generate(query) return await self.fallback.generate(query)
@@ -160,7 +199,7 @@ class Router:
def _finalize(query: str, classification: Classification, resp: ExpertResponse, def _finalize(query: str, classification: Classification, resp: ExpertResponse,
quality_score: float, upgraded: bool, route: list, quality_score: float, upgraded: bool, route: list,
latency_ms: float, model_used: str, cost_est: float, latency_ms: float, model_used: str, cost_est: float,
error: Optional[str] = None) -> RouterResult: error: Optional[str] = None, explanation=None) -> RouterResult:
return RouterResult( return RouterResult(
query=query, query=query,
response=resp.text, response=resp.text,
@@ -175,6 +214,7 @@ class Router:
cache_hit=False, cache_hit=False,
cost_est=cost_est, cost_est=cost_est,
error=error, error=error,
route_explanation=explanation.to_dict() if explanation is not None else None,
) )
def _record(self, result: RouterResult, latency_ms: float): def _record(self, result: RouterResult, latency_ms: float):
+11
View File
@@ -13,6 +13,7 @@ class Stats:
self.domain_counter: Counter = Counter() self.domain_counter: Counter = Counter()
self.difficulty_counter: Counter = Counter() self.difficulty_counter: Counter = Counter()
self.upgraded = 0 self.upgraded = 0
self.upgraded_by_domain: Counter = Counter() # 可解释评分 reliability 维输入
self.cache_hits = 0 self.cache_hits = 0
self.cache_levels: Counter = Counter() self.cache_levels: Counter = Counter()
self.errors = 0 self.errors = 0
@@ -29,6 +30,7 @@ class Stats:
self.difficulty_counter[difficulty] += 1 self.difficulty_counter[difficulty] += 1
if upgraded: if upgraded:
self.upgraded += 1 self.upgraded += 1
self.upgraded_by_domain[domain] += 1
if cache_hit: if cache_hit:
self.cache_hits += 1 self.cache_hits += 1
if cache_level: if cache_level:
@@ -41,6 +43,15 @@ class Stats:
with self._lock: with self._lock:
self.errors += 1 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]: def summary(self) -> Dict[str, Any]:
with self._lock: with self._lock:
n = self.requests n = self.requests
+1 -1
View File
@@ -10,5 +10,5 @@ from router_system.router import build_router
@pytest.fixture() @pytest.fixture()
def router(): def router():
"""????????? mock ?????????????""" """构建全 mock 链路的 Router 实例(离线可跑,供各测试复用)。"""
return build_router() return build_router()
+40 -4
View File
@@ -1,6 +1,42 @@
from router_system.cache import RouterCache from router_system.cache import RouterCache
def test_semantic_lookup_after_many_entries():
"""多条目下语义命中正确(范数预计算 + 单遍扫描的回归)。"""
c = RouterCache(similarity_threshold=0.5)
for i in range(50):
c.put(f"完全不相关的查询主题编号{i}关于烹饪的意见", {"response": f"r{i}"})
c.put("用 Python 实现快速排序函数", {"response": "code-answer"})
level, got = c.get("用 Python 实现快速排序的函数写法") # 相似但不完全相同
assert level in ("semantic", "exact")
assert got["response"] == "code-answer"
def test_promotion_clears_semantic_state():
"""提升为精确缓存后,语义列表与范数索引无残留。"""
c = RouterCache(promote_frequency=2)
c.put("查询甲", {"response": "a"})
first = c.get("查询甲") # 相似度=1.0 计 exact,hits 达阈值即提升
assert first is not None and first[0] == "exact"
second = c.get("查询甲")
assert second is not None and second[0] == "exact"
assert c.stats()["exact_size"] == 1
assert c.stats()["semantic_size"] == 0
assert len(c._sem_norms) == 0
def test_semantic_eviction_clears_norms():
"""语义缓存满员淘汰最旧条目时,向量与范数索引同步清理。"""
c = RouterCache(max_semantic=2)
c.put("查询一", {"response": "1"})
c.put("查询二", {"response": "2"})
c.put("查询三", {"response": "3"}) # 淘汰查询一
assert len(c._semantic) == 2
assert len(c._sem_vecs) == 2
assert len(c._sem_norms) == 2
assert c.get("查询一") is None
def test_exact_hit(): def test_exact_hit():
c = RouterCache() c = RouterCache()
result = {"response": "hello", "domain": "general"} result = {"response": "hello", "domain": "general"}
@@ -13,9 +49,9 @@ def test_exact_hit():
def test_semantic_hit(): def test_semantic_hit():
c = RouterCache(semantic_enabled=True, similarity_threshold=0.5) c = RouterCache(semantic_enabled=True, similarity_threshold=0.5)
c.put("?python?????", {"response": "code", "domain": "code"}) c.put("用 python 实现快速排序", {"response": "code", "domain": "code"})
# ?????????? L2 # 相似但不完全相同的查询应命中 L2 语义缓存
hit = c.get("?python????????") hit = c.get("用 python 实现快速排序的算法")
assert hit is not None assert hit is not None
assert hit[0] == "semantic" assert hit[0] == "semantic"
@@ -24,7 +60,7 @@ def test_promote_to_exact():
c = RouterCache(promote_frequency=3) c = RouterCache(promote_frequency=3)
result = {"response": "x", "domain": "general"} result = {"response": "x", "domain": "general"}
c.put("query", result) c.put("query", result)
# ?????? 3 ? ? ??????? # 语义命中累计 3 次后提升为精确缓存
for _ in range(3): for _ in range(3):
hit = c.get("query") hit = c.get("query")
assert hit is not None assert hit is not None
+19
View File
@@ -2,6 +2,25 @@
from router_system.classifier import RuleClassifier from router_system.classifier import RuleClassifier
def test_tie_break_is_deterministic():
"""同分决胜:按领域名字典序,与规则表排列顺序无关。"""
clf = RuleClassifier()
clf.rules = {"zeta": [("x", 1.0)], "alpha": [("x", 1.0)]}
r = clf.classify("x")
assert r.domain == "alpha"
def test_distinctiveness_penalty():
"""次高分占比高(语义含混)时置信度被压低;单一领域命中不受影响。"""
clf = RuleClassifier()
clf.rules = {"a": [("kw", 1.0)], "b": [("kw", 0.9)]}
r_ambiguous = clf.classify("kw")
clf_clear = RuleClassifier()
clf_clear.rules = {"a": [("kw", 1.0)], "b": [("other", 0.1)]}
r_clear = clf_clear.classify("kw")
assert r_clear.confidence > r_ambiguous.confidence
def test_code_classification(): def test_code_classification():
clf = RuleClassifier() clf = RuleClassifier()
r = clf.classify("用 Python 写一个快速排序函数") r = clf.classify("用 Python 写一个快速排序函数")
+90
View File
@@ -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"
+48
View File
@@ -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"
+2 -2
View File
@@ -1,4 +1,4 @@
"""???????? fastapi + httpx??""" """网关集成测试(需 fastapi + httpx,链路全 mock)。"""
import pytest import pytest
pytest.importorskip("fastapi") pytest.importorskip("fastapi")
@@ -23,7 +23,7 @@ def test_health(client):
def test_chat(client): def test_chat(client):
resp = client.post("/chat", json={"query": "? Python ???????"}) resp = client.post("/chat", json={"query": "用 Python 写一个快速排序函数"})
assert resp.status_code == 200 assert resp.status_code == 200
data = resp.json() data = resp.json()
assert data["response"] assert data["response"]
+53
View File
@@ -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