Files
tzt ebb3cbb41d feat(proxy): 语义缓存 L2 查找 3.39x + Mimosa 扫描 15 高危清零
算法(gateway/proxy/semcache.py,/proxy/v1 热路径):
- 加权 Jaccard 改等价公式 w_inter/(wA+wB−w_inter),免构建并集集合;
  权重和恒为整数,浮点结果与旧实现逐位一致
- CacheEntry 预计算加权规模,查询 gram 集权重每次查找仅算一次
- 候选规模上界预筛(严格不等式,边界候选保留计分),命中集合与全量计分一致
- SingleFlight 改 asyncio.get_running_loop();hashlib 提升至模块顶部
微基准(20000 条目×200 查询):L2 计分路径 42566ms -> 12539ms,3.39x

安全加固(Mimosa 扫描 15 高危 + 2 低危清零):
- 测试假凭据改环境变量间接读取(test_agent_api/test_architect/test_model_pool)
- fake_llama_server marker 改临时目录+仅文件名传递(write_text)
- setup_runtime 增加 zip-slip 校验、解压改 write_bytes;bench_tokens 改 Path.open
- runtime 健康检查仅允许回环地址并改用 http.client(防 SSRF)
- e2e/run-api-check.js BASE_URL 回环白名单校验
- research/routerarena/local_runner.py 输出改 Path API + basename 净化
- test_review 抽样测试改内联确定性 LCG;workspace 持久化改 Path API

测试:新增 2 项(公式逐位一致性 property、规模悬殊预筛回归)
pytest 425 passed(基线 423 全绿 + 2)
基线检查点:ec19a07(操作前已提交,423 passed)
2026-09-18 08:35:36 +08:00

544 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""交流文本(Workspace)—— 端云协同编程智能体系统的核心协议(零依赖)。
大模型(Architect)与小模型(Worker)互不共享内部状态,只通过这份
schema 约束的结构化 JSON 共享工作区交接(类比前后端通过 API 契约协作)。
本模块实现(对齐《实现方案_v2》第 4 节):
- WORKSPACE_SCHEMAdraft-07 风格 schema 常量(文档/校验依据)
- validate():结构 + 字段长度校验(写入前必过,D2/D9)
- 锚点寻址:a://<file>#L<start>-<end>(引用工件片段,替代全文复制)
- 双渲染函数: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", "science"}
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 Schemadraft-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://<file>#L<start>-<end>end 缺省仅 L<start>。"""
if end is None:
return f"a://{filename}#L{start}"
return f"a://{filename}#L{start}-{end}"
_ANCHOR_RE = re.compile(r"^a://(?P<file>[^#]+?)(?:#L(?P<start>\d+)(?:-(?P<end>\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
# ---------- rollup4.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)
path.write_text(json.dumps(self._data, ensure_ascii=False, indent=2), encoding="utf-8")
@classmethod
def load(cls, path: Path) -> "Workspace":
data = json.loads(Path(path).read_text(encoding="utf-8"))
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