From ad81e1afb72f36a4f295bafac420848947466bce Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Sun, 30 Aug 2026 21:12:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T6=20CollaborativePipeline=20?= =?UTF-8?q?=E7=BC=96=E6=8E=92=EF=BC=88=E5=BF=AB=E8=B7=AF=E5=BE=84/?= =?UTF-8?q?=E5=8D=8F=E4=BD=9C=E5=BE=AA=E7=8E=AF/=E7=86=94=E6=96=AD/?= =?UTF-8?q?=E7=BB=88=E5=AE=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- router_system/pipeline.py | 298 +++++++++++++++++++++++++++++++++++++ router_system/worker.py | 16 +- router_system/workspace.py | 12 ++ tests/test_pipeline.py | 161 ++++++++++++++++++++ 任务拆解与执行计划.md | 2 +- 5 files changed, 484 insertions(+), 5 deletions(-) create mode 100644 router_system/pipeline.py create mode 100644 tests/test_pipeline.py diff --git a/router_system/pipeline.py b/router_system/pipeline.py new file mode 100644 index 0000000..e4d030a --- /dev/null +++ b/router_system/pipeline.py @@ -0,0 +1,298 @@ +"""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"), + ) diff --git a/router_system/worker.py b/router_system/worker.py index 98aed8c..4179db1 100644 --- a/router_system/worker.py +++ b/router_system/worker.py @@ -68,6 +68,12 @@ class WorkerLoop: self.max_fix_attempts = max_fix_attempts self.model_used = model_used + async def direct_answer(self, query: str) -> str: + """快路径直答：让 Worker 直接生成用户回答（非 JSON、无围栏）。""" + prompt = ("请直接回答下面这个问题，输出对用户有用的正文" + "（不要输出 JSON，不要加代码块围栏）。问题：" + query) + return await self.generate(prompt) + def _domain_from(self, ws: Workspace) -> str: tags = (ws.get("brief") or {}).get("tags") or [] for t in tags: @@ -76,8 +82,8 @@ class WorkerLoop: return "general" async def run_step(self, ws: Workspace, step_id: str, - existing_artifact: str = "") -> StepOutcome: - """执行单个 step。existing_artifact 为该步当前已有工件全文（若有）。""" + existing_artifact: str = "", hint: str = "") -> StepOutcome: + """执行单个 step。existing_artifact 为该步当前已有工件全文；hint 为 Architect 裁决提示。""" domain = self._domain_from(ws) brief = ws.get("brief") or {} plan = brief.get("plan") or [] @@ -89,7 +95,7 @@ class WorkerLoop: details: List[str] = [] for attempt in range(1, self.max_fix_attempts + 1): - prompt = self._build_prompt(ws, step_id, current, attempt, done_criteria) + prompt = self._build_prompt(ws, step_id, current, attempt, done_criteria, hint) out = await self.generate(prompt) if domain == "code": candidate = extract_code_block(out) @@ -134,13 +140,15 @@ class WorkerLoop: ) def _build_prompt(self, ws: Workspace, step_id: str, current: str, - attempt: int, done_criteria: str) -> str: + attempt: int, done_criteria: str, hint: str = "") -> str: base = ws.render_for_worker(step_id, artifact_text=current or None) if attempt > 1: base += ( "\n\n[注意] 上次生成的工件未通过接地验证。请修正以下问题后重新输出" f"完整工件。本次为第 {attempt} 次尝试。" ) + if hint: + base += "\n\n[架构师裁决] " + hint return base diff --git a/router_system/workspace.py b/router_system/workspace.py index fd08d39..9fea750 100644 --- a/router_system/workspace.py +++ b/router_system/workspace.py @@ -355,6 +355,18 @@ class Workspace: }) self._commit(new) + def revise_plan(self, updates: Dict[str, str]) -> None: + """按 decision.patch_plan 修订既有 step 的 task（不改结构/顺序）。""" + if self._data.get("brief") is None: + raise ValueError("brief 尚未写入，无法修订 plan") + new = copy.deepcopy(self._data) + for pid, task in updates.items(): + for p in new["brief"]["plan"]: + if p["id"] == pid: + p["task"] = task + break + self._commit(new) + def mark_round(self) -> None: self._data["meta"]["round"] += 1 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..2638fa4 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,161 @@ +"""T6 CollaborativePipeline 编排单测（封闭：MockTransport + 假 generate）。""" +import json + +import httpx +import pytest + +from router_system.architect import ArchitectClient +from router_system.pipeline import CollaborativePipeline +from router_system.verifier import Verifier +from router_system.worker import WorkerLoop + +BRIEF_JSON = json.dumps({ + "goal": "用 Python 实现一个函数", + "constraints": ["标准库"], + "tags": ["code"], + "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}], + "plan": [{"id": "s1", "task": "实现函数", "deps": [], "done_criteria": "函数可运行"}], +}, ensure_ascii=False) + +DECIDE_JSON = json.dumps({"reply": "改用更简单的实现", "patch_plan": [{"id": "s1", "task": "简化实现"}]}, + ensure_ascii=False) +REVIEW_DONE = json.dumps({"verdict": "done", "notes": "通过", "fix_issues": []}, ensure_ascii=False) + + +def _architect_handler(brief=BRIEF_JSON, decide=DECIDE_JSON, review=REVIEW_DONE): + def handler(request): + body = json.loads(request.content) + msg = (body.get("messages") or [{}])[-1].get("content", "") + if "任务 brief" in msg or "任务分析" in msg: + return _resp(brief) + if "裁决" in msg: + return _resp(decide) + if "终审" in msg: + return _resp(review) + # 兜底（breach:architect_do） + return _resp("（兜底答案）") + return handler + + +def _resp(content): + return httpx.Response(200, json={ + "choices": [{"message": {"content": content}}], + "usage": {"prompt_tokens": 100, "completion_tokens": 20}, + }) + + +def _make_architect(handler): + return ArchitectClient(model="deepseek-chat", base_url="https://api.deepseek.com/v1", + api_key="sk-test", transport=httpx.MockTransport(handler)) + + +def _fake_generate(texts): + it = iter(texts) + + async def _g(prompt): + try: + return next(it) + except StopIteration: + return "" + + return _g + + +def _make_worker(generate): + return WorkerLoop(generate=generate, + verifier=Verifier(sandbox_runner=lambda c, e, t: (0, "", "")), + kb=None, max_fix_attempts=2) + + +def _pipeline(architect, worker, **kw): + return CollaborativePipeline(architect=architect, worker=worker, + fast_path=kw.get("fast_path", False), + rounds_cap=kw.get("rounds_cap", 6), + api_token_cap=kw.get("api_token_cap", 8000), + breach_policy=kw.get("breach_policy", "architect_do"), + run_dir=kw.get("run_dir", "runs")) + + +def asyncio_run(coro): + import asyncio + return asyncio.run(coro) + + +# ---------- 快路径 ---------- +def test_fast_path_returns_direct(tmp_path): + architect = _make_architect(_architect_handler()) + worker = _make_worker(_fake_generate(["这是一段足够长的、非占位的直接回答内容，用于快路径。"])) + pipe = _pipeline(architect, worker, fast_path=True, run_dir=str(tmp_path / "runs")) + res = asyncio_run(pipe.run("讲一下排序")) + assert res.status == "fast_path" + assert res.fast_path is True + assert res.api_input_tokens == 0 # 未调用 Architect + assert "快路径" in res.response or "直接" in res.response + + +# ---------- 协作：一次通过 ---------- +def test_collab_one_pass(tmp_path): + architect = _make_architect(_architect_handler()) + worker = _make_worker(_fake_generate(["def my_func():\n return 42"])) + pipe = _pipeline(architect, worker, fast_path=False, run_dir=str(tmp_path / "runs")) + res = asyncio_run(pipe.run("写个函数")) + assert res.status == "done" + assert res.fast_path is False + assert res.api_input_tokens >= 100 # brief + 终审 用量已计量 + assert res.workspace_path is not None + # 交流文本已落盘 + import os + assert os.path.exists(res.workspace_path) + assert "done" in res.route + + +# ---------- 协作：带 issue 修复 ---------- +def test_collab_issue_fix(tmp_path): + architect = _make_architect(_architect_handler()) + # 前两次失败（占位）-> issue；decide 后重跑成功 + worker = _make_worker(_fake_generate([ + "待实现", "待实现", "def ok():\n return 1", + ])) + pipe = _pipeline(architect, worker, fast_path=False, run_dir=str(tmp_path / "runs")) + res = asyncio_run(pipe.run("写个函数")) + assert res.status == "done" + assert any("issue" in r for r in res.route) + assert any("decide" in r for r in res.route) + # 至少有 1 条 decision + import json as _json + ws = json.load(open(res.workspace_path, encoding="utf-8")) + assert len(ws["decisions"]) >= 1 + + +# ---------- 熔断兜底：本地降级 ---------- +def test_breach_local_degrade(tmp_path): + architect = _make_architect(_architect_handler()) + worker = _make_worker(_fake_generate(["不通过"])) + pipe = _pipeline(architect, worker, fast_path=False, api_token_cap=1, + breach_policy="local_only", run_dir=str(tmp_path / "runs")) + res = asyncio_run(pipe.run("写个函数")) + assert res.status == "failed" + assert "本地降级" in res.response + assert any("breach" in r for r in res.route) + + +# ---------- 熔断兜底：Architect 代做 ---------- +def test_breach_architect_do(tmp_path): + # review 一直打回 fix（制造压力），最终走 breach:architect_do + def handler(request): + body = json.loads(request.content) + msg = (body.get("messages") or [{}])[-1].get("content", "") + if "任务 brief" in msg or "任务分析" in msg: + return _resp(BRIEF_JSON) + if "自动升级" in msg: + return _resp("（Architect 兜底答案）") + return _resp(DECIDE_JSON) + architect = _make_architect(handler) + worker = _make_worker(_fake_generate(["待实现"])) + pipe = _pipeline(architect, worker, fast_path=False, rounds_cap=1, + api_token_cap=8000, breach_policy="architect_do", + run_dir=str(tmp_path / "runs")) + res = asyncio_run(pipe.run("写个函数")) + # rounds_cap=1：一轮后预算未满但回合触顶 -> breach -> architect_do + assert "breach" in " ".join(res.route) + assert res.status in ("escalated", "failed") diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 58b313b..dc79bec 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -72,7 +72,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯 | T3 | ArchitectClient（DeepSeek API，JSON 约束） | ✅ 完成 | T3 | | T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ✅ 完成 | T4 | | T5 | WorkerLoop + 接地验证 | ✅ 完成 | T5 | -| T6 | CollaborativePipeline 编排 | ⬜ | | +| T6 | CollaborativePipeline 编排 | ✅ 完成 | T6 | | T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | | | T8 | 人工检验队列 ReviewQueue | ⬜ | | | T9 | token 计量与账单 | ⬜ | |