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）

.gitignore

@@ -25,3 +25,6 @@ cached_results/
 Thumbs.db
 .idea/
 .vscode/
+
+# 安全扫描器工作目录（不入库）
+.mimosa/

BRANCH.md

@@ -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`

router_system/cache.py

@@ -7,6 +7,11 @@
 高频语义命中会提升为 O(1) 的精确缓存条目。
 
 只缓存"未升级"的结果（升级路径每次都走大模型，不缓存，避免陈旧）。
+
+性能设计（2026-09 优化）：
+- 每条语义缓存条目在写入时预计算并缓存向量范数，查询时免重复计算（原来每对比较都重算）
+- 语义查找单遍完成：扫描即跟踪最优条目与命中计数，命中后不再二次线性查找
+- 相似度达到 1.0（完全相同查询）时提前终止扫描（余弦相似度上界，不可能更优）
 """
 from __future__ import annotations
 
@@ -29,18 +34,6 @@ def _ngrams(text: str, n: int = 3) -> List[str]:
     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]:
     vec: Dict[str, float] = {}
     for g in grams:
@@ -48,6 +41,17 @@ def _tf_vector(grams: List[str]) -> Dict[str, float]:
     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())
+
+
 class RouterCache:
     """L1 精确缓存 + L2 语义缓存。"""
 
@@ -61,6 +65,7 @@ class RouterCache:
         self._exact: Dict[str, CacheEntry] = {}
         self._semantic: List[Tuple[str, CacheEntry]] = []  # (query, entry)
         self._sem_vecs: Dict[str, Dict[str, float]] = {}
+        self._sem_norms: Dict[str, float] = {}             # 预计算范数，避免查询期重算
         self.hits = {"exact": 0, "semantic": 0}
         self.misses = 0
 
@@ -74,36 +79,41 @@ class RouterCache:
 
         if self.semantic_enabled:
             q_vec = _tf_vector(_ngrams(query))
+            q_norm = _norm(q_vec)
             best_sim = 0.0
-            best_query: Optional[str] = None
-            best_result: Optional[Dict[str, Any]] = None
-            for q, e in self._semantic:
-                sim = _cosine(q_vec, self._sem_vecs.get(q, {}))
-                if sim > best_sim:
-                    best_sim = sim
-                    best_query = q
-                    best_result = e.result
-            if best_query is not None and best_sim >= self.similarity_threshold:
+            best_idx = -1
+            if q_norm > 0.0:
+                # 单遍扫描：同时跟踪最优相似度与条目位置
+                for i, (q, _e) in enumerate(self._semantic):
+                    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
+                        if sim >= 1.0:
+                            break  # 余弦相似度上界：完全相同查询，提前终止
+            if best_idx >= 0 and best_sim >= self.similarity_threshold:
+                best_q, best_entry = self._semantic[best_idx]
                 # 完全相同查询（相似度=1.0）计为 exact 命中
                 is_exact = best_sim >= 0.999
                 level = "exact" if is_exact else "semantic"
                 self.hits[level] += 1
-                self._semantic_hit(best_query)
-                return (level, best_result)
+                self._bump_semantic(best_idx, best_q, best_entry)
+                return (level, best_entry.result)
 
         self.misses += 1
         return None
 
-    def _semantic_hit(self, query: str):
-        """语义命中：累计命中次数，达到阈值提升为精确缓存。"""
-        for i, (q, e) in enumerate(self._semantic):
-            if q == query:
-                e.hits += 1
-                if e.hits >= self.promote_frequency:
-                    self._exact[query] = e
-                    self._semantic.pop(i)
-                    self._sem_vecs.pop(query, None)
-                break
+    def _bump_semantic(self, idx: int, query: str, entry: CacheEntry):
+        """语义命中：累计命中次数，达到阈值提升为精确缓存（O(1)，无需二次查找）。"""
+        entry.hits += 1
+        if entry.hits >= self.promote_frequency:
+            self._exact[query] = entry
+            self._semantic.pop(idx)
+            self._sem_vecs.pop(query, None)
+            self._sem_norms.pop(query, None)
 
     # ---- 写入 ----
     def put(self, query: str, result: Dict[str, Any]):
@@ -114,8 +124,11 @@ class RouterCache:
             if len(self._semantic) >= self.max_semantic:
                 old_q, _ = self._semantic.pop(0)
                 self._sem_vecs.pop(old_q, None)
+                self._sem_norms.pop(old_q, None)
             self._semantic.append((query, entry))
-            self._sem_vecs[query] = _tf_vector(_ngrams(query))
+            vec = _tf_vector(_ngrams(query))
+            self._sem_vecs[query] = vec
+            self._sem_norms[query] = _norm(vec)
         else:
             self._exact[query] = entry
             if len(self._exact) > self.max_exact:
@@ -137,5 +150,6 @@ class RouterCache:
         self._exact.clear()
         self._semantic.clear()
         self._sem_vecs.clear()
+        self._sem_norms.clear()
         self.hits = {"exact": 0, "semantic": 0}
         self.misses = 0

router_system/classifier.py

@@ -128,7 +128,8 @@ class RuleClassifier(BaseClassifier):
                 matched_rules=[],
             )
 
-        best_domain = max(raw, key=raw.get)
+        # 同分决胜：按领域名字典序，保证与规则表排列顺序无关的确定性
+        best_domain = max(sorted(raw), key=lambda d: raw[d])
         best_score = raw[best_domain]
         confidence = 1.0 - math.exp(-best_score)
 
@@ -138,7 +139,7 @@ class RuleClassifier(BaseClassifier):
 
         # 与次高分的差距影响置信度（区分度）
         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:
                 confidence *= 0.85
 

router_system/router.py

@@ -83,13 +83,7 @@ class Router:
         # 低置信度 -> 直接走大模型
         if self.classifier.should_fallback(classification, self.low_confidence_threshold):
             route.append("direct_fallback")
-            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)
-            self._record(result, latency)
-            return result
+            return await self._fallback_result(query, classification, route, start)
 
         # ---- Step 3: 选择专家 ----
         domain = classification.domain
@@ -106,14 +100,8 @@ class Router:
         except Exception as e:
             self.stats.record_error()
             route.append(f"expert_error:{type(e).__name__}")
-            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=str(e))
-            self._record(result, latency)
-            return result
+            return await self._fallback_result(query, classification, route, start,
+                                               error=str(e))
 
         # ---- Step 5: Judge 评估 ----
         try:
@@ -145,6 +133,22 @@ class Router:
         return result
 
     # ---------------------------------------------------------------
+    async def _fallback_result(self, query: str, classification: Classification,
+                               route: list, start: float,
+                               error: Optional[str] = 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)
+        self._record(result, latency)
+        return result
+
     async def _call_fallback(self, query: str) -> ExpertResponse:
         try:
             return await self.fallback.generate(query)

tests/test_cache.py

@@ -1,6 +1,42 @@
 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():
     c = RouterCache()
     result = {"response": "hello", "domain": "general"}

tests/test_classifier.py

@@ -2,6 +2,25 @@
 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():
     clf = RuleClassifier()
     r = clf.classify("用 Python 写一个快速排序函数")