"""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 @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", ): 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) # --------------------------------------------------------------- # 入口 # --------------------------------------------------------------- async def run(self, query: str) -> PipelineResult: 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 # ---- 快路径：小模型直答 + 自验证（省 API 钱，D3） ---- if self.fast_path: 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] 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), {}) if not self._deps_done(ws, step.get("deps") or []): 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 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: done = {p["step"] for p in ws.get("progress", []) if p.get("status") == "done"} 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 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" / 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"), )