299 lines
13 KiB
Python
299 lines
13 KiB
Python
"""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"),
|
||
)
|