"""CollaborativePipeline —— 端云协同协作管线编排（v2 核心调度）。 流程（对齐《实现方案_v2》第 2/4.3 节）： 用户 query -> [快路径] Worker 直答+自验证通过即返回 -> 否则 Architect.brief 写交流文本 -> 协作循环（Worker 实现/自验证， issue 时 Architect.decide 裁决）-> 全步完成 -> Architect.final_review -> 交付（入人工检验队列） 护栏（D6）：rounds_cap / api_token_cap 任一触顶即熔断；熔断按 breach_policy 走 Architect 兜底代做（有 key）或本地降级提示（无 key）。 测试封闭性（D11）：architect 用 httpx.MockTransport，worker 用假 generate； 不依赖真实 llama-server 或 API key。 """ from __future__ import annotations import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional from .architect import ArchitectCircuitBreaker, ArchitectClient, ArchitectError from .worker import WorkerLoop from .workspace import Workspace def _safe_artifact_name(name: str) -> str: """工件名单消毒：剥路径成分，只留文件名（工件名来自模型输出，防 ../ 越界写盘）。""" part = str(name or "").replace("\\", "/").split("/")[-1].strip() if not part or part in (".", ".."): return "artifact.bin" return part[:120] @dataclass class PipelineResult: """v2 协作管线一次运行的结果。""" query: str response: str request_id: str status: str # done | fast_path | escalated | failed fast_path: bool rounds_used: int api_input_tokens: int api_output_tokens: int cost_est: float model_used: str latency_ms: float route: List[str] = field(default_factory=list) workspace_path: Optional[str] = None error: Optional[str] = None class CollaborativePipeline: """编排快路径、协作循环、熔断、终审。""" def __init__( self, architect: ArchitectClient, worker: WorkerLoop, fast_path: bool = True, rounds_cap: int = 6, api_token_cap: int = 8000, breach_policy: str = "architect_do", # architect_do | local_only run_dir: str = "runs", tier_fn=None, # D-G5：可选三级分级钩子（唯一受控改动） ): self.architect = architect self.worker = worker self.fast_path = fast_path self.rounds_cap = rounds_cap self.api_token_cap = api_token_cap self.breach_policy = breach_policy self.run_dir = Path(run_dir) self.tier_fn = tier_fn # None = 现行为逐字节不变 # --------------------------------------------------------------- # 入口 # --------------------------------------------------------------- async def run(self, query: str, request_id: Optional[str] = None) -> PipelineResult: if request_id is None: request_id = uuid.uuid4().hex[:12] ws = Workspace.new(request_id, query, self.api_token_cap, self.rounds_cap) route: List[str] = ["v2"] t0 = time.perf_counter() * 1000.0 # ---- 三级分级钩子（D-G5，可选）：T1=fast path / T2=单次云端直答 / T3=完整管线 ---- tier = None if self.tier_fn is not None: try: tier = await self.tier_fn(query) if callable(getattr(self.tier_fn, "__call__", None)) \ and not hasattr(self.tier_fn, "run") else None except Exception: tier = None # 兼容两种 tier_fn：直接可调用，或带 .decide 的对象（grader） if tier is None and self.tier_fn is not None and hasattr(self.tier_fn, "decide"): try: d = await self.tier_fn.decide(query, "pipeline") tier = d.tier if getattr(d, "fallback", False) is not True or d.tier else d.tier route.append(f"tier_fn:{tier}@mode:{getattr(d, 'mode', '')}") except Exception: tier = None # ---- T2 新增档：单次云端直答（跳过 brief/review/循环）---- if tier == "T2": route.append("t2:direct") try: ws2 = Workspace.new(request_id + "-t2", query, self.api_token_cap, self.rounds_cap) answer = await self.architect._chat_once( ws2, [{"role": "user", "content": "直接回答下面的问题（无需计划，一次答完）：" + query}]) return self._finalize(query, ws, response=answer, status="fast_path", fast_path=False, route=route, model_used=self.architect.model, t0=t0, save_ws=False) except Exception: # D-G1：直答失败落回完整管线 route.append("t2:direct:miss") # ---- T3：强制完整管线（跳过快路径） ---- force_full = (tier == "T3") if force_full: route.append("tier_fn:T3") # ---- 快路径：小模型直答 + 自验证（省 API 钱，D3）；T3 时跳过 ---- if self.fast_path and not force_full: direct = await self.worker.direct_answer(query) dom = self._guess_domain(query) passed, _det = self.worker.verifier.verify(dom, "answer.md", direct, query, kb=self.worker.kb) if passed: route.append(f"fast_path@domain:{dom}") self._save_artifact(request_id, "answer.md", direct) return self._finalize(query, ws, response=direct, status="fast_path", fast_path=True, route=route, model_used=self.worker.model_used, t0=t0, save_ws=False) route.append("fast_path:miss") # ---- brief（Architect 一次） ---- try: brief = await self.architect.brief(query, ws) ws.apply_brief(brief) route.append("brief") except (ArchitectError, ArchitectCircuitBreaker) as e: return await self._handle_breach(query, ws, route, t0, exc=e) # ---- 协作循环 ---- route.append("loop") plan = brief.get("plan") or [] pending = [p.get("id") for p in plan] plan_ids = {p.get("id") for p in plan} # 用于 _deps_done 过滤 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), {}) 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): continue existing = self._read_artifact(request_id, self._artifact_name(sid, ws)) 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) self._save_artifact(request_id, outcome.artifact_name, outcome.artifact_text) ws.rollup() route.append(f"step:{sid}:done") progressed = True if not pending: # 所有步骤完成，退出循环 break else: # issue -> Architect 裁决 route.append(f"step:{sid}:issue") try: dec = await self.architect.decide(ws) ws.add_decision(dec.get("ref") or outcome.issue_id or "", dec.get("reply", ""), dec.get("patch_plan")) self._apply_patch_plan(ws, dec.get("patch_plan")) route.append(f"decide:{sid}") except (ArchitectError, ArchitectCircuitBreaker) as e: return await self._handle_breach(query, ws, route, t0, exc=e) progressed = True ws.mark_round() if not progressed: # 死锁（依赖/裁决都推不动）-> 熔断兜底 return await self._handle_breach(query, ws, route, t0, exc=RuntimeError("协作循环死锁：无进度")) if ws.exhausted() and pending: return await self._handle_breach(query, ws, route, t0, exc=RuntimeError("预算/回合触顶")) # ---- 终审 ---- ws.transition("reviewing") route.append("reviewing") try: rev = await self.architect.final_review(ws) if rev.get("verdict") == "done": ws.transition("done") route.append("done") status = "done" else: # reviewing --fail--> in_progress（修正回合）；MVP 返回 escalated 标记打回 ws.transition("in_progress") route.append("review:fix") status = "escalated" except (ArchitectError, ArchitectCircuitBreaker) as e: return await self._handle_breach(query, ws, route, t0, exc=e) response = self._build_response(ws, request_id) return self._finalize(query, ws, response=response, status=status, fast_path=False, route=route, model_used=self.architect.model, t0=t0) # --------------------------------------------------------------- # 熔断兜底 # --------------------------------------------------------------- async def _handle_breach(self, query: str, ws: Workspace, route: List[str], t0: float, exc: BaseException) -> PipelineResult: route.append("breach") if self.breach_policy == "architect_do" and self.architect.api_key: # 有 key：Architect 兜底代做 try: briefish = ws.render_for_architect() answer = await self.architect._chat_once( ws, [{"role": "user", "content": "以下任务自动升级，请直接给出最终可交付答案（非 JSON）：" + briefish}]) route.append("breach:architect_do") return self._finalize(query, ws, response=answer, status="escalated", fast_path=False, route=route, model_used=self.architect.model, t0=t0, error=str(exc)) except Exception as e2: route.append(f"breach:architect_do:fail:{type(e2).__name__}") # 本地降级 route.append("breach:local_deg") msg = ("（本地降级）当前请求超出本地可处理范围，且未配置大模型密钥或预算熔断。" f"原因：{exc}") return self._finalize(query, ws, response=msg, status="failed", fast_path=False, route=route, model_used="none", t0=t0, error=str(exc)) # --------------------------------------------------------------- # 结果组装 # --------------------------------------------------------------- def _finalize(self, query: str, ws: Workspace, response: str, status: str, fast_path: bool, route: List[str], model_used: str, t0: float, save_ws: bool = True, error: Optional[str] = None) -> PipelineResult: if save_ws: path = self.run_dir / ws.request_id / "workspace.json" ws.save(path) ws_path = str(path) else: ws_path = None b = ws.budget() return PipelineResult( query=query, response=response, request_id=ws.request_id, status=status, fast_path=fast_path, rounds_used=ws.meta()["round"], api_input_tokens=b["api_input_tokens"], api_output_tokens=b["api_output_tokens"], cost_est=0.0, model_used=model_used, latency_ms=time.perf_counter() * 1000.0 - t0, route=route, workspace_path=ws_path, error=error, ) # --------------------------------------------------------------- # 辅助 # --------------------------------------------------------------- def _guess_domain(self, query: str) -> str: if self.worker.kb is not None: hits = self.worker.kb.match(query) if hits: return hits[0].domain return "general" def _artifact_name(self, sid: str, ws: Workspace) -> str: from .worker import artifact_name_for domain = self._guess_domain(ws["query"]) # 用 brief.tags 优先 tags = (ws.get("brief") or {}).get("tags") or [] for t in tags: if t != "safety": domain = t break return artifact_name_for(sid, domain) def _deps_done(self, ws: Workspace, deps: List[str]) -> bool: # progress 中 done 的条目；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): sid = entry.get("step", "") elif isinstance(entry, str): 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) def _last_decision_for(self, ws: Workspace, sid: str) -> str: """取最近一条针对该 step 的决策 reply，作为 worker hint。""" step_issues = {i.get("id") for i in ws.get("issues", []) if i.get("step") == sid} for dec in reversed(ws.get("decisions", []) or []): if dec.get("ref") in step_issues: return dec.get("reply", "") return "" def _apply_patch_plan(self, ws: Workspace, patch_plan: Optional[List[Dict[str, Any]]]) -> None: if not patch_plan: return updates = {} for item in patch_plan: if isinstance(item, dict) and item.get("id") and item.get("task"): updates[item["id"]] = item["task"] if updates: ws.revise_plan(updates) def _build_response(self, ws: Workspace, request_id: str) -> str: # 汇总 archive 摘要 + 各已完成步骤工件 lines = list(ws.get("archive", []) or []) brief = ws.get("brief") or {} parts: List[str] = [] if brief.get("goal"): parts.append("任务：" + brief["goal"]) if lines: parts.append("完成情况：") parts.extend(f"- {ln}" for ln in lines) # 附上最后一个已完成步骤的工件全文 progress = ws.get("progress", []) or [] done_steps = [p for p in progress if p.get("status") == "done"] if done_steps: last = done_steps[-1] art = self._read_artifact(request_id, self._artifact_name(last["step"], ws)) if art: parts.append("产出：") parts.append(art) return "\n\n".join(parts) if parts else "（协作管线未产出有效内容）" def _save_artifact(self, request_id: str, name: str, text: str) -> None: if not text: return name = _safe_artifact_name(name) d = self.run_dir / request_id / "artifacts" d.mkdir(parents=True, exist_ok=True) (d / name).write_text(text, encoding="utf-8") def _read_artifact(self, request_id: str, name: str) -> str: p = self.run_dir / request_id / "artifacts" / _safe_artifact_name(name) if p.exists(): return p.read_text(encoding="utf-8") return "" def build_pipeline(cfg: Dict[str, Any], architect: ArchitectClient, worker: WorkerLoop) -> CollaborativePipeline: """cfg 为 config.pipeline 段。""" p = cfg.get("pipeline", {}) return CollaborativePipeline( architect=architect, worker=worker, fast_path=bool(p.get("fast_path", True)), rounds_cap=int(p.get("rounds_cap", 6)), api_token_cap=int(p.get("api_token_cap", 8000)), breach_policy=p.get("breach_policy", "architect_do"), run_dir=p.get("run_dir", "runs"), )