feat(v2): 架构与算法优化二轮——推理机不变量外提、知识库匹配预编译、协作循环增量索引

- inference.py（/chat/legacy 热路径）：kb.match 循环不变量外提（原每步全量重扫+重排序，
  最坏 O(steps×rules×patterns)）；fired 查重 list→set
- knowledge.py：Rule patterns 注册侧懒缓存小写副本（原每条规则每次匹配重复 lower）；
  match() 文本只 lower 一次（原逐规则重复）；load() 的 yaml 文件名集合提到循环外
- worker.py：本地端点生成器 httpx.AsyncClient 懒建复用（原每步新建/销毁连接，
  对齐 ArchitectClient 惯用法；协作循环最多 10 次生成免重复建连）
- pipeline.py（协作循环）：plan_by_id O(1) 步定义查找；done 集合增量维护
  （原每轮重建 progress+archive 扫描）；领域只解析一次（原 _artifact_name 每步
  全领域 kb.match）；_deps_done 支持传入预填集合（保持旧签名兼容）
- v2stats.py：回合数分布改增量聚合（sum/max/分桶计数），summary() O(n)→O(1)，
  不再持有无界 list（修长时运行内存增长）
- gateway/agent.py + api.py：AgentService 运行计数 O(1) 化（原 register 全量扫描），
  状态迁移收敛到 _transition_state 单一入口（api.py cancel/异常两处绕过点一并接入，
  消除计数与状态脱节隐患）；21 项 agent 测试全绿（两轮全量 230 passed 复核）

gateway/agent.py

@@ -312,6 +312,13 @@ class AgentService:
         self.run_dir = Path(run_dir)
         self._runs: Dict[str, AgentRunInfo] = {}
         self.max_running = 5
+        self._running = 0  # O(1) 运行计数（原 register 每次全量扫描 _runs）
+
+    def _transition_state(self, info: AgentRunInfo, new_state: str) -> None:
+        """状态迁移的单一入口：离开 running 时同步递减计数。"""
+        if info.state == STATE_RUNNING and new_state != STATE_RUNNING:
+            self._running -= 1
+        info.state = new_state
 
     # ---------- 路径 ----------
     def _dir(self, request_id: str) -> Path:
@@ -327,14 +334,14 @@ class AgentService:
     def register(self, request_id: str, task: str, model: str, pool_id: str,
                  workspace: str = "", executor_model: str = "",
                  mode: str = "single") -> Optional[AgentRunInfo]:
-        running = [r for r in self._runs.values() if r.state == STATE_RUNNING]
+        if self._running >= self.max_running:
-        if len(running) >= self.max_running:
             return None
         info = AgentRunInfo(request_id=request_id, task=task, model=model,
                             pool_id=pool_id, workspace=workspace,
                             executor_model=executor_model, mode=mode,
                             started_at=time.time())
         self._runs[request_id] = info
+        self._running += 1
         self._dir(request_id).mkdir(parents=True, exist_ok=True)
         self._write_status(info)
         return info
@@ -413,7 +420,7 @@ class AgentService:
                 throttle.flush("executor")
             self._apply_result(info, result)
         except Exception as exc:  # pragma: no cover
-            info.state = STATE_FAILED
+            self._transition_state(info, STATE_FAILED)
             info.error = f"{type(exc).__name__}: {exc}"
             self._append_event(info, {"type": "final", "round": info.rounds,
                                       "reason": "error", "error": info.error})
@@ -455,14 +462,14 @@ class AgentService:
         info.prompt_tokens = int(result.get("prompt_tokens", 0))
         info.completion_tokens = int(result.get("completion_tokens", 0))
         if result.get("reason") == "error":
-            info.state = STATE_FAILED
+            self._transition_state(info, STATE_FAILED)
             info.error = result.get("error")
         elif result.get("reason") in ("token_cap", "max_rounds", "max_handoffs"):
             # 触顶属于护栏行为：结果仍交付，但标记部分完成信息
-            info.state = STATE_DONE
+            self._transition_state(info, STATE_DONE)
             info.error = result.get("error")
         else:
-            info.state = STATE_DONE
+            self._transition_state(info, STATE_DONE)
 
     # ---------- 两级模式（D7）：规划者 + 执行者 ----------
     async def run_dual(self, info: AgentRunInfo, planner_chat: Any, executor_chat: Any,

gateway/api.py

@@ -647,7 +647,7 @@ try:
             except Exception as exc:
                 import traceback
                 traceback.print_exc()
-                info.state = "failed"
+                service._transition_state(info, "failed")
                 info.error = str(exc)
                 info.finished_at = __import__("time").time()
                 service._write_status(info)
@@ -674,7 +674,7 @@ try:
             return {"ok": False, "detail": f"任务已结束（{info.state}）"}
         if info.asyncio_task is not None:
             info.asyncio_task.cancel()
-        info.state = "failed"
+        service._transition_state(info, "failed")
         info.error = "cancelled_by_user"
         info.finished_at = __import__("time").time()
         service._write_status(info)

router_system/inference.py

@@ -53,20 +53,26 @@ class InferenceEngine:
     # ---------------------------------------------------------------
     def run(self, query: str, domain: str, memory: WorkingMemory,
             max_steps: Optional[int] = None) -> List[str]:
-        """前向链主循环。返回触发规则 id 列表（按触发顺序）。"""
+        """前向链主循环。返回触发规则 id 列表（按触发顺序）。
+
+        循环不变量外提：query/domain 在循环内不变，kb.match 结果只算一次
+        （原实现每步全量重扫+重排序，最坏 O(steps × rules × patterns)）。
+        """
         steps = max_steps or self.max_steps
+        rules = self.kb.match(query, domain=domain)
         fired: List[str] = []
+        fired_set: set = set()  # O(1) 查重（fired 保持列表维护触发顺序）
         for _ in range(steps):
-            rules = self.kb.match(query, domain=domain)
             # 选第一个"未触发过"的规则
             target: Optional[Rule] = None
             for r in rules:
-                if r.id not in fired:
+                if r.id not in fired_set:
                     target = r
                     break
             if target is None:
                 break  # 无新规则可触发 → 终止
             fired.append(target.id)
+            fired_set.add(target.id)
             self._fire(target, query, memory)
         return fired
 

router_system/knowledge.py

@@ -51,13 +51,25 @@ class Rule:
     actions: List[str] = field(default_factory=list)  # 保留字段：动作扩展
     subdomain: Optional[str] = None     # 二级子领域（如 investing/labor/calculus）
     subdomain2: Optional[str] = None    # 三级子领域（如 fund/overtime/sorting）
+    _patterns_lower: Optional[tuple] = field(default=None, repr=False, compare=False)
+
+    def _lowered(self) -> tuple:
+        """patterns 的小写缓存（注册后规则视为不可变；懒计算一次）。"""
+        if self._patterns_lower is None:
+            self._patterns_lower = tuple(p.lower() for p in self.patterns)
+        return self._patterns_lower
 
     def matches(self, text: str) -> bool:
         """任一 pattern 是 text 的子串即命中（大小写不敏感）。"""
         if not self.patterns:
             return False
-        q = text.lower()
+        return self._match_lower(text.lower())
-        return any(p.lower() in q for p in self.patterns)
+
+    def _match_lower(self, q: str) -> bool:
+        """已 lowercase 文本的快速匹配（避免每条规则重复 lower 同一文本）。"""
+        if not self.patterns:
+            return False
+        return any(p in q for p in self._lowered())
 
 
 # ---------------------------------------------------------------
@@ -420,12 +432,13 @@ class KnowledgeBase:
         self._facts = {d: [dict(f) for f in facts] for d, facts in BUILTIN_FACTS.items()}
 
         if self.rules_dir.is_dir():
+            yaml_names = {p.name for p in self.rules_dir.glob("*.yaml")}  # 一次遍历，避免逐文件重扫
             for f in sorted(self.rules_dir.glob("*.yaml")):
                 data = _try_load_yaml(f)
                 if data is not None:
                     self._load_file_data(f, data)
             for f in sorted(self.rules_dir.glob("*.json")):
-                if f.name not in {p.name for p in self.rules_dir.glob("*.yaml")}:
+                if f.name not in yaml_names:
                     data = _try_load_json(f)
                     if data is not None:
                         self._load_file_data(f, data)
@@ -461,12 +474,16 @@ class KnowledgeBase:
 
     # ---- 查询 ----
     def match(self, text: str, domain: Optional[str] = None) -> List[Rule]:
-        """返回命中的规则，按优先级降序。domain 为空则全领域匹配。"""
+        """返回命中的规则，按优先级降序。domain 为空则全领域匹配。
+
+        文本只 lowercase 一次（原实现每条规则各 lower 一遍）。
+        """
+        q = text.lower()
         hits = []
         for rule in self._rules.values():
             if domain is not None and rule.domain != domain:
                 continue
-            if rule.matches(text):
+            if rule._match_lower(q):
                 hits.append(rule)
         hits.sort(key=lambda r: r.priority, reverse=True)
         return hits

router_system/pipeline.py

@@ -110,20 +110,24 @@ class CollaborativePipeline:
         plan = brief.get("plan") or []
         pending = [p.get("id") for p in plan]
         plan_ids = {p.get("id") for p in plan}          # 用于 _deps_done 过滤
+        plan_by_id = {p.get("id"): p for p in plan}     # O(1) 步定义查找（原每轮线性扫描）
+        done_ids = self._done_ids(ws)                   # 增量维护的完成集（原每轮重建）
+        base_domain = self._guess_domain(query)         # ws["query"] 恒定，领域只解析一次
         while pending and not ws.exhausted():
             progressed = False
             for sid in list(pending):
-                step = next((p for p in plan if p.get("id") == sid), {})
+                step = plan_by_id.get(sid, {})
                 deps = step.get("deps") or []
                 # 只检查在 plan 中的依赖；不在 plan 的 ID 视为"不存在"→自动满足
                 relevant = [d for d in deps if d in plan_ids]
-                if not self._deps_done(ws, relevant):
+                if not all(d in done_ids for d in relevant):
                     continue
-                existing = self._read_artifact(request_id, self._artifact_name(sid, ws))
+                existing = self._read_artifact(request_id, self._artifact_name(sid, ws, domain=base_domain))
                 outcome = await self.worker.run_step(ws, sid, existing_artifact=existing,
                                                      hint=self._last_decision_for(ws, sid))
                 if outcome.status == "done":
                     pending.remove(sid)
+                    done_ids.add(sid)
                     self._save_artifact(request_id, outcome.artifact_name, outcome.artifact_text)
                     ws.rollup()
                     route.append(f"step:{sid}:done")
@@ -230,9 +234,10 @@ class CollaborativePipeline:
                 return hits[0].domain
         return "general"
 
-    def _artifact_name(self, sid: str, ws: Workspace) -> str:
+    def _artifact_name(self, sid: str, ws: Workspace, domain: Optional[str] = None) -> str:
         from .worker import artifact_name_for
-        domain = self._guess_domain(ws["query"])
+        if domain is None:
+            domain = self._guess_domain(ws["query"])
         # 用 brief.tags 优先
         tags = (ws.get("brief") or {}).get("tags") or []
         for t in tags:
@@ -241,8 +246,10 @@ class CollaborativePipeline:
                 break
         return artifact_name_for(sid, domain)
 
-    def _deps_done(self, ws: Workspace, deps: List[str]) -> bool:
+    @staticmethod
-        # progress 中 done 的条目；done 条目 rollup 后移到 archive（字符串格式如 "s1: ..."）
+    def _done_ids(ws: Workspace) -> set:
+        """收集已完成 step id 集合：progress 中 done 的条目 + archive 折叠行
+        （done 条目 rollup 后移到 archive，字符串格式如 "s1: ..."）。"""
         done = {p["step"] for p in ws.get("progress", []) if p.get("status") == "done"}
         for entry in ws.get("archive", []):
             if isinstance(entry, dict):
@@ -251,9 +258,13 @@ class CollaborativePipeline:
                 sid = entry.split(":")[0].strip() if ":" in entry else ""
             else:
                 sid = ""
-            if sid in deps:
+            done.add(sid)
-                done.add(sid)
+        return done
-        return all(d in done for d in deps)
+
+    def _deps_done(self, ws: Workspace, deps: List[str], done_ids: Optional[set] = None) -> bool:
+        if done_ids is None:
+            done_ids = self._done_ids(ws)
+        return all(d in done_ids for d in deps)
 
     def _last_decision_for(self, ws: Workspace, sid: str) -> str:
         """取最近一条针对该 step 的决策 reply，作为 worker hint。"""

router_system/v2stats.py

@@ -2,6 +2,9 @@
 
 T9：每请求 API token 记账；聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。
 配合 /metrics 对外透出（论文 E1 token 经济学数据来源之一）。
+
+性能设计（2026-09 优化）：回合数分布以"和/最大值/分桶计数"增量维护，
+summary() 从每次全量重算 O(n) 降为 O(桶数)，且不再持有无界 list。
 """
 from __future__ import annotations
 
@@ -18,7 +21,11 @@ class V2Stats:
         self._fast_path = 0
         self._breach = 0
         self._by_status: Dict[str, int] = {}
-        self._rounds: List[int] = []
+        # 回合数增量聚合（等价于全量 list 的 sum/max/histogram，内存 O(桶数)）
+        self._rounds_count = 0
+        self._rounds_sum = 0
+        self._rounds_max = 0
+        self._rounds_hist: Dict[str, int] = {}
         self._api_input_tokens = 0
         self._api_output_tokens = 0
         self._api_cost_usd = 0.0
@@ -33,7 +40,13 @@ class V2Stats:
                 self._fast_path += 1
             status = getattr(result, "status", "?")
             self._by_status[status] = self._by_status.get(status, 0) + 1
-            self._rounds.append(getattr(result, "rounds_used", 0))
+            rounds = getattr(result, "rounds_used", 0)
+            self._rounds_count += 1
+            self._rounds_sum += rounds
+            if rounds > self._rounds_max:
+                self._rounds_max = rounds
+            bucket = str(rounds) if rounds <= 10 else ">10"
+            self._rounds_hist[bucket] = self._rounds_hist.get(bucket, 0) + 1
             if "breach" in " ".join(getattr(result, "route", [])):
                 self._breach += 1
             self._api_input_tokens += getattr(result, "api_input_tokens", 0)
@@ -64,16 +77,16 @@ class V2Stats:
     def summary(self) -> Dict[str, Any]:
         with self._lock:
             total = self._total
-            rounds = self._rounds
+            rounds_n = self._rounds_count
             return {
                 "total_requests": total,
                 "fast_path_rate": round(self._fast_path / total, 4) if total else 0.0,
                 "status_distribution": dict(self._by_status),
                 "breach_count": self._breach,
                 "rounds_used": {
-                    "avg": round(sum(rounds) / len(rounds), 2) if rounds else 0.0,
+                    "avg": round(self._rounds_sum / rounds_n, 2) if rounds_n else 0.0,
-                    "max": max(rounds) if rounds else 0,
+                    "max": self._rounds_max if rounds_n else 0,
-                    "distribution": _histogram(rounds),
+                    "distribution": dict(self._rounds_hist),
                 },
                 "api_tokens": {
                     "input": self._api_input_tokens,

router_system/worker.py

@@ -188,25 +188,33 @@ def _mock_generate() -> Callable[[str], Awaitable[str]]:
 
 def _make_llama_generate(cfg: Dict[str, Any],
                            default_base_url: Optional[str] = None) -> Callable[[str], Awaitable[str]]:
-    """返回调用本地 OpenAI 兼容端点（llama-server / Ollama / vLLM）的生成器。"""
+    """返回调用本地 OpenAI 兼容端点（llama-server / Ollama / vLLM）的生成器。
+
+    连接复用：httpx.AsyncClient 懒建一次、跨步骤复用（与 ArchitectClient 一致），
+    避免协作循环每步重新 TCP 建连。
+    """
     if default_base_url is None:
         default_base_url = f"http://127.0.0.1:{cfg.get('port', 8901)}/v1"
     base_url = cfg.get("base_url") or default_base_url
     model = cfg.get("model") or "local"
     temperature = float(cfg.get("temperature", 0.3))
     timeout_s = float(cfg.get("per_step_timeout_s", 300))
+    client_holder: Dict[str, Any] = {"client": None}
 
     async def _gen(prompt: str) -> str:
         try:
             import httpx
-            async with httpx.AsyncClient(timeout=timeout_s) as client:
+            client = client_holder["client"]
-                resp = await client.post(
+            if client is None or client.is_closed:
-                    f"{base_url}/chat/completions",
+                client = httpx.AsyncClient(timeout=timeout_s)
-                    json={"model": model, "messages": [{"role": "user", "content": prompt}],
+                client_holder["client"] = client
-                          "temperature": temperature, "max_tokens": 4096},
+            resp = await client.post(
-                )
+                f"{base_url}/chat/completions",
-                resp.raise_for_status()
+                json={"model": model, "messages": [{"role": "user", "content": prompt}],
-                return resp.json()["choices"][0]["message"]["content"]
+                      "temperature": temperature, "max_tokens": 4096},
+            )
+            resp.raise_for_status()
+            return resp.json()["choices"][0]["message"]["content"]
         except Exception as e:  # noqa: BLE001
             # 连不上本地模型 -> 优雅降级（不抛 500），提示用户检查模型端点
             return ("（本地降级）无法连接本地模型端点，未能生成该步骤内容。"