diff --git a/gateway/agent.py b/gateway/agent.py index b860a9c..99bd8ef 100644 --- a/gateway/agent.py +++ b/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 len(running) >= self.max_running: + if self._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, diff --git a/gateway/api.py b/gateway/api.py index dbd7fd7..59788db 100644 --- a/gateway/api.py +++ b/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) diff --git a/router_system/inference.py b/router_system/inference.py index 7d7d826..c92a922 100644 --- a/router_system/inference.py +++ b/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 diff --git a/router_system/knowledge.py b/router_system/knowledge.py index cce20a0..4c140c6 100644 --- a/router_system/knowledge.py +++ b/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 any(p.lower() in q for p in self.patterns) + return self._match_lower(text.lower()) + + 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 diff --git a/router_system/pipeline.py b/router_system/pipeline.py index 68080b9..1ad9a3e 100644 --- a/router_system/pipeline.py +++ b/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: - # progress 中 done 的条目；done 条目 rollup 后移到 archive（字符串格式如 "s1: ..."） + @staticmethod + 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) - return all(d in done for d in deps) + done.add(sid) + return done + + 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。""" diff --git a/router_system/v2stats.py b/router_system/v2stats.py index 7a6f832..3a5fe5d 100644 --- a/router_system/v2stats.py +++ b/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, - "max": max(rounds) if rounds else 0, - "distribution": _histogram(rounds), + "avg": round(self._rounds_sum / rounds_n, 2) if rounds_n else 0.0, + "max": self._rounds_max if rounds_n else 0, + "distribution": dict(self._rounds_hist), }, "api_tokens": { "input": self._api_input_tokens, diff --git a/router_system/worker.py b/router_system/worker.py index 65401ca..5b391ec 100644 --- a/router_system/worker.py +++ b/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: - resp = await client.post( - f"{base_url}/chat/completions", - json={"model": model, "messages": [{"role": "user", "content": prompt}], - "temperature": temperature, "max_tokens": 4096}, - ) - resp.raise_for_status() - return resp.json()["choices"][0]["message"]["content"] + client = client_holder["client"] + if client is None or client.is_closed: + client = httpx.AsyncClient(timeout=timeout_s) + client_holder["client"] = client + resp = await client.post( + f"{base_url}/chat/completions", + json={"model": model, "messages": [{"role": "user", "content": prompt}], + "temperature": temperature, "max_tokens": 4096}, + ) + resp.raise_for_status() + return resp.json()["choices"][0]["message"]["content"] except Exception as e: # noqa: BLE001 # 连不上本地模型 -> 优雅降级（不抛 500），提示用户检查模型端点 return ("（本地降级）无法连接本地模型端点，未能生成该步骤内容。"