From 2d2c2184b9660962d0708bcf71e1dc808463b3da Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Sat, 19 Sep 2026 09:40:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T-M3=20=E9=87=87=E7=BA=B3=20llmrout?= =?UTF-8?q?er=20pinch=20=E4=B8=89=E6=A1=A3=E8=A3=81=E5=89=AA=E2=80=94?= =?UTF-8?q?=E2=80=94Architect=20=E6=B8=B2=E6=9F=93=E8=B6=85=E9=A2=84?= =?UTF-8?q?=E7=AE=97=E6=8C=89=E7=9B=B8=E5=85=B3=E6=80=A7=E5=88=86=E6=A1=A3?= =?UTF-8?q?=E4=BF=9D=E7=95=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Workspace.render_for_architect：超 1200 token 时对 archive 行按与 query+goal 的字符 2-gram Dice 相似度分档——sim<0.25 先丢、sim<0.55 截断为 40% 摘要、 仍超限从低到高继续丢；保留行维持原插入序（前缀稳定），最近 4 行下限不变 （相比旧'从最旧整段丢'，高相关事实在预算内留存更久） - 顺带消除两项旧债：压缩态与首次渲染共用同一 parts 结构（格式漂移）、 裁剪循环不再每轮 deepcopy 全文档 - 修复本轮引入的缺陷：会话轮次 tool_calls 为 int 计数时清单提取迭代崩溃 （test_session_multi_turn 抓出，已加类型防御） - 新增 tests/test_pinch_trim.py 4 项；全量 244 passed ×2（基线 230） --- gateway/agent.py | 2 +- router_system/workspace.py | 105 +++++++++++++++++++++++++++---------- tests/test_pinch_trim.py | 60 +++++++++++++++++++++ 3 files changed, 137 insertions(+), 30 deletions(-) create mode 100644 tests/test_pinch_trim.py diff --git a/gateway/agent.py b/gateway/agent.py index 9321a51..fe8382c 100644 --- a/gateway/agent.py +++ b/gateway/agent.py @@ -461,7 +461,7 @@ class AgentService: files = sorted({ str((a.get("arguments") or {}).get("path") or "").strip() for t in session.data.get("turns", []) - for a in (t.get("tool_calls") or []) + for a in (t.get("tool_calls") if isinstance(t.get("tool_calls"), list) else []) if isinstance(a, dict) and str(a.get("name") or "") in _FILE_TOOLS and (a.get("arguments") or {}).get("path") diff --git a/router_system/workspace.py b/router_system/workspace.py index 99f69e6..4baccff 100644 --- a/router_system/workspace.py +++ b/router_system/workspace.py @@ -450,15 +450,17 @@ class Workspace: 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） + archive = d.get("archive", []) or [] + shown = list(archive[-8:]) + parts.extend(shown) + # token 预算：超限按 pinch 三档裁剪 archive（T-M3，采纳 llmrouter pinch + # keep/summarize/drop 分档思想），决策见 _trim_archive_lines 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 + if estimate_tokens(out) <= 1200 or len(shown) <= 4: + return out # 预算内 / 已在下限（与旧实现的 4 行下限一致） + head = parts[:-len(shown)] + ref_text = d.get("query", "") + " " + (d.get("brief") or {}).get("goal", "") + return _trim_archive_lines(head, archive, shown, ref_text) def render_for_worker(self, step_id: str, artifact_text: Optional[str] = None) -> str: @@ -525,24 +527,69 @@ class Workspace: 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 +def _bigrams(text: str) -> Dict[str, int]: + """字符 2-gram 计数（去空白、小写；零依赖轻量相关性度量）。""" + t = "".join(text.lower().split()) + out: Dict[str, int] = {} + for i in range(len(t) - 1): + g = t[i:i + 2] + out[g] = out.get(g, 0) + 1 + return out + + +def _dice(a: Dict[str, int], b: Dict[str, int]) -> float: + """Dice 系数：2 * 交集 / (|a| + |b|)，空集返回 0。""" + if not a or not b: + return 0.0 + inter = sum(min(v, b.get(k, 0)) for k, v in a.items()) + return 2.0 * inter / (sum(a.values()) + sum(b.values())) + + +def _trim_archive_lines(head: List[str], archive: List[str], + shown: List[str], ref_text: str) -> str: + """Architect 渲染超预算时的 archive 三档裁剪（T-M3，采纳 llmrouter pinch + keep/summarize/drop 分档思想）。 + + 相关度 = 行与 query+goal 的字符 2-gram Dice 系数： + 第一档：丢 sim<0.25 的行（最低相关先丢）； + 第二档：把 sim<0.55 的行截断为 40% 摘要； + 第三档：仍超限则从低相关到高相关继续丢。 + 保留行维持原插入序（前缀稳定）；任何情况下至少保留 4 行， + 达到下限仍超限则接受溢出（与旧实现一致）。 + 相比旧"从最旧起整段丢弃"，高相关事实在预算内留存得更久； + 且压缩态与首次渲染共用同一 parts 结构（消除格式漂移与每轮 deepcopy）。 + """ + ref = _bigrams(ref_text) + start = len(archive) - len(shown) + idx = list(range(start, len(archive))) + sims = {i: _dice(_bigrams(archive[i]), ref) for i in idx} + order = sorted(idx, key=lambda i: (sims[i], i)) + + kept = set(idx) + clips: Dict[int, str] = {} + + def _over_budget() -> bool: + lines = [clips.get(i, archive[i]) for i in sorted(kept)] + return estimate_tokens("\n".join(head + lines)) > 1200 + + # 第一档：丢低相关 + for i in order: + if not _over_budget() or len(kept) <= 4: + break + if sims[i] < 0.25: + kept.discard(i) + # 第二档：中相关截断为摘要（40% 长度） + for i in order: + if not _over_budget(): + break + if i in kept and sims[i] < 0.55: + clips[i] = _clip(archive[i], max(20, int(len(archive[i]) * 0.4))) + # 第三档：仍超限从低到高继续丢 + for i in order: + if not _over_budget() or len(kept) <= 4: + break + kept.discard(i) + clips.pop(i, None) + + lines = [clips.get(i, archive[i]) for i in sorted(kept)] + return "\n".join(head + lines) diff --git a/tests/test_pinch_trim.py b/tests/test_pinch_trim.py new file mode 100644 index 0000000..811f38e --- /dev/null +++ b/tests/test_pinch_trim.py @@ -0,0 +1,60 @@ +"""Architect 渲染三档裁剪（T-M3，采纳 llmrouter pinch keep/summarize/drop 设计）。""" +from router_system.workspace import Workspace + +_GOAL = "用 Python 实现快速排序并分析时间复杂度" + +_BRIEF = { + "goal": _GOAL, "constraints": ["标准库"], "tags": ["code"], + "acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}], + "plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "过"}], +} + + +def _ws_with_archive(): + ws = Workspace.new(request_id="abc123def456", query=_GOAL, + api_token_cap=8000, rounds_cap=6) + ws.apply_brief(_BRIEF) + relevant = ("s1: 完成快速排序实现，基准选取三数取中，递归深度优化，" + "平均时间复杂度 O(n log n) 分析见附件。" + "快排边界处理细节" * 50) + lines = [f"sX{i}: 无关行{i}，今日食堂菜单有红烧肉、番茄炒蛋与清炒时蔬，" + f"周末计划去郊外徒步露营并整理旅行照片。" + "生活琐事记录" * 50 + + f"独特结尾标记{i}号" # 唯一尾串：判断"该行是否还被输出" + for i in range(7)] + # 相关行放在中间：不在最新 4 行下限保护区内，检验相关性而非新近度 + ws._data["archive"] = lines[:3] + [relevant] + lines[3:] + return ws, relevant, lines + + +def test_trim_keeps_relevant_line_full(): + """高相关行全文保留，低相关行被丢弃/截断（尾段不再出现在输出）。""" + ws, relevant, lines = _ws_with_archive() + out = ws.render_for_architect() + # 相关行全文（含尾段独特内容）仍在 + assert relevant[-40:] in out + # 至少一条低相关行的尾段消失（被丢或被截为摘要） + assert lines[0][-40:] not in out + + +def test_trim_respects_floor_of_four_lines(): + """任何情况下至少保留 4 行（与旧实现下限一致）。""" + ws, _, lines = _ws_with_archive() + out = ws.render_for_architect() + archive_shown = out.split("== archive ==")[1].strip().splitlines() + assert len(archive_shown) >= 4 + + +def test_render_deterministic(): + """同输入两次渲染逐字节一致（裁剪决策确定性）。""" + ws1, _, _ = _ws_with_archive() + ws2, _, _ = _ws_with_archive() + assert ws1.render_for_architect() == ws2.render_for_architect() + + +def test_under_budget_untouched(): + """预算内不触发裁剪：archive 行原样出现、无截断省略号。""" + ws = Workspace.new(request_id="abc123def456", query="写个快排", + api_token_cap=8000, rounds_cap=6) + ws.apply_brief(_BRIEF) + ws._data["archive"] = ["s1: 完成（产出: a://s1.py）"] + out = ws.render_for_architect() + assert "s1: 完成（产出: a://s1.py）" in out