"""交流文本（Workspace）—— 端云协同 LLM 协作系统的核心协议（零依赖）。 大模型（Architect）与小模型（Worker）互不共享内部状态，只通过这份 schema 约束的结构化 JSON 共享工作区交接（类比前后端通过 API 契约协作）。 本模块实现（对齐《实现方案_v2》第 4 节）： - WORKSPACE_SCHEMA：draft-07 风格 schema 常量（文档/校验依据） - validate()：结构 + 字段长度校验（写入前必过，D2/D9） - 锚点寻址：a://#L-（引用工件片段，替代全文复制） - 双渲染函数：render_for_architect（≤1200 token）、render_for_worker（≤8K token） - rollup()：已完成步骤折叠为 archive 摘要行；超限压缩（只减不删，4.5） - 状态机：draft -> in_progress -> reviewing -> done / escalated / failed """ from __future__ import annotations import copy import json import re from pathlib import Path from typing import Any, Dict, List, Optional VERSION = "1.0" # --------------------------------------------------------------------------- # 字段长度上限（同时是 rollup 依据，4.2） # --------------------------------------------------------------------------- LIMITS = { "goal": 500, # brief.goal 字数 "task": 300, # brief.plan[].task 字数 "summary": 200, # progress[].summary 字数 "issue_text": 300, # issues[].observed/expected/tried/ask 字数 "reply": 600, # decisions[].reply 字数 "archive": 160, # archive[] 每条字数 "constraints": 8, # brief.constraints 上限条数 "plan_steps": 5, # brief.plan 上限步数 "acceptance": 20, # brief.acceptance 上限条数 "query_truncate": 200, # render_for_architect 中 query 截断 } # 允许的领域标签（4.2 brief.tags；仅用于安全标记与验证接地，不做路由 D3） ALLOWED_TAGS = {"code", "math", "legal", "medical", "finance", "life", "education", "general", "safety"} STATUS_FLOW = { "draft": {"in_progress"}, "in_progress": {"reviewing", "escalated", "failed", "in_progress"}, "reviewing": {"done", "in_progress", "failed"}, "escalated": {"reviewing", "done", "failed"}, "done": set(), "failed": set(), } # --------------------------------------------------------------------------- # JSON Schema（draft-07 风格，draft-07 依赖内嵌；供校验与文档参考） # --------------------------------------------------------------------------- WORKSPACE_SCHEMA: Dict[str, Any] = { "$schema": "http://json-schema.org/draft-07/schema#", "title": "Communication Workspace", "type": "object", "required": ["version", "request_id", "query", "meta"], "additionalProperties": False, "properties": { "version": {"const": VERSION}, "request_id": {"type": "string", "minLength": 1}, "query": {"type": "string", "minLength": 1}, "meta": { "type": "object", "required": ["status", "round", "budget"], "properties": { "status": {"enum": ["draft", "in_progress", "reviewing", "escalated", "done", "failed"]}, "round": {"type": "integer", "minimum": 0}, "budget": { "type": "object", "required": ["api_input_tokens", "api_output_tokens", "api_token_cap", "rounds_cap"], "properties": { "api_input_tokens": {"type": "integer", "minimum": 0}, "api_output_tokens": {"type": "integer", "minimum": 0}, "api_token_cap": {"type": "integer", "minimum": 1}, "rounds_cap": {"type": "integer", "minimum": 1}, }, }, }, }, "brief": { "type": "object", "required": ["goal", "constraints", "tags", "acceptance", "plan"], "properties": { "goal": {"type": "string"}, "constraints": {"type": "array", "items": {"type": "string"}}, "tags": {"type": "array", "items": {"type": "string"}}, "acceptance": {"type": "array", "items": {"type": "object"}}, "plan": {"type": "array", "items": {"type": "object"}}, }, }, "progress": {"type": "array", "items": {"type": "object"}}, "issues": {"type": "array", "items": {"type": "object"}}, "decisions": {"type": "array", "items": {"type": "object"}}, "archive": {"type": "array", "items": {"type": "string"}}, }, } # --------------------------------------------------------------------------- # 工具 # --------------------------------------------------------------------------- _TOKEN_PER_CHAR_ZH = 1 / 1.6 # 中文约 1.6 字/token _TOKEN_PER_CHAR_EN = 1 / 4.0 # 英文约 4 字/token def estimate_tokens(text: str) -> int: """粗略 token 估算（中英混合，用于渲染预算校验）。""" if not text: return 0 zh = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff") en = len(text) - zh return max(1, int(zh * _TOKEN_PER_CHAR_ZH + en * _TOKEN_PER_CHAR_EN)) def build_anchor(filename: str, start: int = 1, end: Optional[int] = None) -> str: """构造锚点：a://#L-；end 缺省仅 L。""" if end is None: return f"a://{filename}#L{start}" return f"a://{filename}#L{start}-{end}" _ANCHOR_RE = re.compile(r"^a://(?P[^#]+?)(?:#L(?P\d+)(?:-(?P\d+))?)?$") def parse_anchor(anchor: str) -> Optional[Dict[str, Any]]: """解析锚点为 {file, start, end}；非法返回 None。""" m = _ANCHOR_RE.match(anchor) if not m: return None start = int(m.group("start")) if m.group("start") else 1 end = int(m.group("end")) if m.group("end") else start return {"file": m.group("file"), "start": start, "end": end} def _clip(text: str, limit: int) -> str: """按字数截断（中文按字符）。""" if len(text) <= limit: return text return text[:limit] + "…" # --------------------------------------------------------------------------- # 校验 # --------------------------------------------------------------------------- def validate(ws: Dict[str, Any]) -> List[str]: """校验 workspace 结构 + 字段长度。返回错误列表（空 = 合法）。""" errors: List[str] = [] if not isinstance(ws, dict): return ["workspace 必须是 object"] if ws.get("version") != VERSION: errors.append(f"version 必须是 {VERSION}") if not isinstance(ws.get("request_id"), str) or not ws["request_id"]: errors.append("request_id 必须是非空字符串") if not isinstance(ws.get("query"), str) or not ws["query"]: errors.append("query 必须是非空字符串") meta = ws.get("meta") if not isinstance(meta, dict): errors.append("meta 必须是 object") else: if meta.get("status") not in STATUS_FLOW: errors.append(f"meta.status 非法: {meta.get('status')}") budget = meta.get("budget") if not isinstance(budget, dict): errors.append("meta.budget 必须是 object") else: for k in ("api_input_tokens", "api_output_tokens", "api_token_cap", "rounds_cap"): if not isinstance(budget.get(k), int) or budget.get(k) < 0: errors.append(f"meta.budget.{k} 必须是非负整数") brief = ws.get("brief") if brief is not None: if not isinstance(brief, dict): errors.append("brief 必须是 object") else: if not isinstance(brief.get("goal"), str): errors.append("brief.goal 必须是字符串") elif len(brief["goal"]) > LIMITS["goal"]: errors.append(f"brief.goal 超长（>{LIMITS['goal']}字）") if not isinstance(brief.get("constraints"), list): errors.append("brief.constraints 必须是数组") elif len(brief["constraints"]) > LIMITS["constraints"]: errors.append(f"brief.constraints 超过 {LIMITS['constraints']} 条") if not isinstance(brief.get("tags"), list): errors.append("brief.tags 必须是数组") for t in brief.get("tags", []) or []: if t not in ALLOWED_TAGS: errors.append(f"brief.tags 含非法标签: {t}") acc = brief.get("acceptance") if not isinstance(acc, list) or len(acc) > LIMITS["acceptance"]: errors.append(f"brief.acceptance 需为 ≤{LIMITS['acceptance']} 的数组") plan = brief.get("plan") if not isinstance(plan, list) or len(plan) > LIMITS["plan_steps"]: errors.append(f"brief.plan 需为 ≤{LIMITS['plan_steps']} 步的数组") else: ids = [p.get("id") for p in plan if isinstance(p, dict)] if len(set(ids)) != len(ids): errors.append("brief.plan 存在重复 step id") for p in plan: if not isinstance(p, dict): errors.append("brief.plan 元素必须是 object") continue if not isinstance(p.get("task"), str): errors.append(f"brief.plan[{p.get('id')}].task 必须是字符串") elif len(p["task"]) > LIMITS["task"]: errors.append(f"brief.plan[{p.get('id')}].task 超长（>{LIMITS['task']}字）") for i, entry in enumerate(ws.get("progress", []) or []): if not isinstance(entry, dict): errors.append(f"progress[{i}] 必须是 object"); continue if entry.get("status") not in ("done", "failed", "blocked"): errors.append(f"progress[{i}].status 非法") if not isinstance(entry.get("summary"), str) or len(entry["summary"]) > LIMITS["summary"]: errors.append(f"progress[{i}].summary 非法或超长") for i, entry in enumerate(ws.get("issues", []) or []): if not isinstance(entry, dict): errors.append(f"issues[{i}] 必须是 object"); continue for k in ("observed", "expected", "tried", "ask"): if isinstance(entry.get(k), str) and len(entry[k]) > LIMITS["issue_text"]: errors.append(f"issues[{i}].{k} 超长（>{LIMITS['issue_text']}字）") for i, entry in enumerate(ws.get("decisions", []) or []): if not isinstance(entry, dict): errors.append(f"decisions[{i}] 必须是 object"); continue if isinstance(entry.get("reply"), str) and len(entry["reply"]) > LIMITS["reply"]: errors.append(f"decisions[{i}].reply 超长（>{LIMITS['reply']}字）") for i, line in enumerate(ws.get("archive", []) or []): if not isinstance(line, str) or len(line) > LIMITS["archive"]: errors.append(f"archive[{i}] 非法或超长") return errors # --------------------------------------------------------------------------- # Workspace # --------------------------------------------------------------------------- class Workspace: """交流文本对象：持有状态、执行写入前校验、渲染、rollup、持久化。""" def __init__(self, data: Dict[str, Any]): errors = validate(data) if errors: raise ValueError("workspace 校验失败: " + "; ".join(errors[:5])) self._data = data self._brief_locked = False # ---------- 构造 ---------- @classmethod def new(cls, request_id: str, query: str, api_token_cap: int = 8000, rounds_cap: int = 6) -> "Workspace": data = { "version": VERSION, "request_id": request_id, "query": query, "meta": { "status": "draft", "round": 0, "budget": { "api_input_tokens": 0, "api_output_tokens": 0, "api_token_cap": api_token_cap, "rounds_cap": rounds_cap, }, }, "brief": None, "progress": [], "issues": [], "decisions": [], "archive": [], } return cls(data) # ---------- 访问 ---------- @property def request_id(self) -> str: return self._data["request_id"] @property def status(self) -> str: return self._data["meta"]["status"] @property def data(self) -> Dict[str, Any]: return copy.deepcopy(self._data) def get(self, key: str, default: Any = None) -> Any: return self._data.get(key, default) def __getitem__(self, key: str) -> Any: return self._data[key] def meta(self) -> Dict[str, Any]: return self._data["meta"] def budget(self) -> Dict[str, int]: return self._data["meta"]["budget"] # ---------- 写入（均先校验） ---------- def _commit(self, data: Dict[str, Any]) -> None: errors = validate(data) if errors: raise ValueError("写入校验失败: " + "; ".join(errors[:5])) self._data = data def transition(self, new_status: str) -> None: cur = self.status if new_status == cur: return if new_status not in STATUS_FLOW.get(cur, set()): raise ValueError(f"非法状态迁移: {cur} -> {new_status}") self._data["meta"]["status"] = new_status def apply_brief(self, brief: Dict[str, Any]) -> None: """写入 brief（写一次后锁定，D2：brief 恒定位于文档前部，prefix cache 友好）。""" if self._data.get("brief") is not None or self._brief_locked: raise ValueError("brief 已写入，不可重复") new = copy.deepcopy(self._data) new["brief"] = brief new["meta"]["status"] = "in_progress" self._commit(new) self._brief_locked = True def add_progress(self, step: str, status: str, summary: str, artifact: Optional[str] = None) -> None: new = copy.deepcopy(self._data) entry: Dict[str, Any] = {"step": step, "status": status, "summary": summary} if artifact: entry["artifact"] = artifact new["progress"].append(entry) self._commit(new) def add_issue(self, step: str, anchor: str, observed: str, expected: str, tried: str, ask: str) -> str: new = copy.deepcopy(self._data) iid = f"i{len(new['issues']) + 1}" entry = { "id": iid, "step": step, "anchor": anchor, "observed": observed, "expected": expected, "tried": tried, "ask": ask, } new["issues"].append(entry) self._commit(new) return iid def add_decision(self, ref: str, reply: str, patch_plan: Optional[List[Dict[str, str]]] = None) -> None: new = copy.deepcopy(self._data) new["decisions"].append({ "ref": ref, "reply": reply, "patch_plan": patch_plan or [], }) 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 def add_budget(self, input_tokens: int = 0, output_tokens: int = 0) -> None: b = self._data["meta"]["budget"] b["api_input_tokens"] += int(input_tokens) b["api_output_tokens"] += int(output_tokens) self._commit(self._data) def exhausted(self) -> bool: """预算熔断判定：API token 或回合任一触顶（D6）。""" b = self._data["meta"]["budget"] used = b["api_input_tokens"] + b["api_output_tokens"] if b["api_token_cap"] and used >= b["api_token_cap"]: return True if b["rounds_cap"] and self._data["meta"]["round"] >= b["rounds_cap"]: return True return False # ---------- rollup（4.5） ---------- def rollup(self) -> int: """把 done 的 progress 折叠为 archive 摘要行，并清理解析完成的问题。 只减不删 archive 历史；progress 中 done 的条目折叠后移除（保留 failed/blocked）。 返回本次折叠的条目数。 """ folded = 0 new_progress: List[Dict[str, Any]] = [] for entry in self._data.get("progress", []): if entry.get("status") == "done" and entry.get("step"): line = _clip(f"{entry['step']}: {entry.get('summary', '')}", LIMITS["archive"]) if line not in self._data["archive"]: self._data["archive"].append(line) folded += 1 else: new_progress.append(entry) resolved = {d.get("ref") for d in self._data.get("decisions", [])} kept_issues: List[Dict[str, Any]] = [] for iss in self._data.get("issues", []): if iss.get("id") in resolved: line = _clip(f"{iss['id']}: {iss.get('expected', '')[:60]}", LIMITS["archive"]) if line not in self._data["archive"]: self._data["archive"].append(line) else: kept_issues.append(iss) self._data["issues"] = kept_issues self._data["progress"] = new_progress return folded # ---------- 渲染 ---------- def render_for_architect(self) -> str: """渲染 Architect 输入（D7）：meta+query(截断)+全部 issues+最近3条 decisions +最近回合 progress 摘要。目标 ≤1200 token。""" d = self._data parts: List[str] = [] m = d["meta"] parts.append("== meta ==") parts.append(f"status={m['status']} round={m['round']} " f"budget={json.dumps(m['budget'], ensure_ascii=False)}") parts.append("== query ==") parts.append(_clip(d["query"], LIMITS["query_truncate"])) if d.get("brief"): b = d["brief"] parts.append("== brief(锁定) ==") parts.append(f"goal: {_clip(b['goal'], 120)}") parts.append(f"plan: {[p['id'] for p in b.get('plan', [])]}") parts.append(f"acceptance: {[a.get('id') for a in b.get('acceptance', [])]}") parts.append("== issues ==") for iss in d.get("issues", []): parts.append(f"{iss['id']} step={iss.get('step')} anchor={iss.get('anchor')} " f"ask={_clip(iss.get('ask', ''), 80)}") parts.append("== 最近 3 条 decisions ==") for dec in d.get("decisions", [])[-3:]: parts.append(f"ref={dec.get('ref')} reply={_clip(dec.get('reply', ''), 80)}") parts.append("== progress 摘要 ==") for p in d.get("progress", [])[-5:]: parts.append(f"{p.get('step')} [{p.get('status')}] {_clip(p.get('summary', ''), 40)}") parts.append("== archive ==") for line in d.get("archive", [])[-8:]: parts.append(line) # token 预算：超限先截断最旧 archive（已只保留 3 条 decisions） out = "\n".join(parts) while estimate_tokens(out) > 1200 and len(d.get("archive", [])) > 4: d = copy.deepcopy(d) d["archive"] = d["archive"][4:] out = "\n".join(_rerender(self, d)) return out def render_for_worker(self, step_id: str, artifact_text: Optional[str] = None) -> str: """渲染 Worker 输入（4.4）：brief 全文+该 step 定义+依赖 step 的 archive 摘要行 +该 step 现有工件全文+验收标准。目标 ≤8K token。""" d = self._data b = d.get("brief") parts: List[str] = [] if b: parts.append("== 任务目标 (goal) ==") parts.append(b["goal"]) parts.append("== 约束 (constraints) ==") parts.extend(f"- {c}" for c in b.get("constraints", [])) parts.append("== 全部步骤 (plan) ==") for p in b.get("plan", []): mark = " <-- 当前步" if p.get("id") == step_id else "" parts.append(f"{p['id']}: {p.get('task', '')}{mark}") parts.append(f" done_criteria: {p.get('done_criteria', '')}") parts.append("== 依赖步摘要 (archive) ==") for line in d.get("archive", [])[-6:]: parts.append(line) if artifact_text: parts.append(f"== 当前步已有工件（{step_id}） ==") parts.append(artifact_text) parts.append("== 验收标准 ==") if b: for a in b.get("acceptance", []): parts.append(f"- {a.get('id')}: {a.get('check', '')} " f"(machine_checkable={a.get('machine_checkable', False)})") parts.append("== 要求 ==") parts.append("请实现当前步，并用可执行验证/事实对照/结构检查自验证；" "通过则写 progress(done)，失败自修 ≤2 次，仍失败则写 issue。") return "\n".join(parts) # ---------- 持久化 ---------- def save(self, path: Path) -> None: path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(self._data, f, ensure_ascii=False, indent=2) @classmethod def load(cls, path: Path) -> "Workspace": with open(path, "r", encoding="utf-8") as f: data = json.load(f) return cls(data) def prefix_signature(self) -> str: """返回稳定前缀的签名（T10 prefix cache）。 交流文本的"恒定位于前部"的部分（version + request_id + query + meta + brief） 应不随 progress/issues/decisions 追加而变化，从而使 llama-server 的 --cache-reuse 能命中该前缀、降低 prefill 开销。用紧凑 JSON 的哈希度量稳定性。 """ stable = { "version": self._data.get("version"), "request_id": self._data.get("request_id"), "query": self._data.get("query"), "brief": self._data.get("brief"), } import hashlib s = json.dumps(stable, ensure_ascii=False, sort_keys=True) return hashlib.sha256(s.encode("utf-8")).hexdigest()[:16] def to_dict(self) -> Dict[str, Any]: return self.data def _rerender(ws: "Workspace", d: Dict[str, Any]) -> List[str]: """用裁剪后的数据重建 Architect 渲染（供超限压缩内部用）。""" parts: List[str] = [] m = d["meta"] parts.append("== meta ==") parts.append(f"status={m['status']} round={m['round']}") parts.append("== query ==") parts.append(_clip(d["query"], LIMITS["query_truncate"])) parts.append("== issues ==") for iss in d.get("issues", []): parts.append(f"{iss['id']} step={iss.get('step')} ask={_clip(iss.get('ask', ''), 80)}") parts.append("== decisions(最近3) ==") for dec in d.get("decisions", [])[-3:]: parts.append(f"ref={dec.get('ref')} reply={_clip(dec.get('reply', ''), 80)}") parts.append("== progress ==") for p in d.get("progress", [])[-5:]: parts.append(f"{p.get('step')} [{p.get('status')}] {_clip(p.get('summary', ''), 40)}") parts.append("== archive ==") for line in d.get("archive", [])[-6:]: parts.append(line) return parts