7 Commits
Author SHA1 Message Date
tzt 3f303a69c3 chore(v2): .gitignore 登记 extra/ 参考项目目录(不入库) 2026-09-19 10:18:16 +08:00
tzt 2d2c2184b9 feat(v2): T-M3 采纳 llmrouter pinch 三档裁剪——Architect 渲染超预算按相关性分档保留
- 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)
2026-09-19 09:40:43 +08:00
tzt 184c11199a feat(v2): T-M2 采纳 pi branch-summarization——折叠摘要携带文件清单 + rollup 行带产物锚点
- AgentService._history_from_session:从全部轮次的 write_file/edit_file
  工具调用提取已读写文件清单(去重/单路径裁 120 字符/封顶 20 个),以摘要轮
  注入对话上下文——轮次滑出窗口后事实清单仍在(丢上下文不丢事实)
- Workspace.rollup:done 折叠行追加(产出: <artifact 锚点>),上下文折叠后
  产物事实在 a:// 锚点体系内仍可寻址;既有子串断言全部兼容
- 新增 tests/test_branch_manifest.py 4 项
2026-09-19 09:32:33 +08:00
tzt f7e8273434 feat(v2): T-M1 采纳 pi 会话树设计——AgentSession fork 分叉(parent_id/fork_point 血统 + 值拷贝隔离)
- SessionStore.fork(sid, turn_index, title):值拷贝 turns[:n+1] 生成新会话,
  原会话不变;新会话记录 parent_id/fork_point(pi 的 id/parentId 树思想);
  拷贝式而非共享存储引用——规避并发写冲突,会话体量小冗余可接受
- 旧会话数据无血统字段自然兼容(dict.get 语义,零迁移,对齐 pi 的版本迁移纪律)
- 新增 POST /agent/sessions/{sid}/fork 端点(404 = 会话不存在/分叉点越界)
- 新增 tests/test_session_fork.py 6 项(拷贝正确性/缺省语义/越界/深拷贝/
  旧数据兼容/端点 200+404)
2026-09-19 09:29:57 +08:00
tzt 819c5cceaf chore(v2): 外部优点融入前置锚点——本轮五项目吸收的回滚基线 2026-09-19 09:16:52 +08:00
tzt 8820e8da48 feat(v2): 架构与算法优化二轮——推理机不变量外提、知识库匹配预编译、协作循环增量索引
- inference.py(/chat/legacy 热路径):kb.match 循环不变量外提(原每步全量重扫+重排序,
  最坏 O(steps×rules×patterns));fired 查重 list→set
- knowledge.py:Rule patterns 注册侧懒缓存小写副本(原每条规则每次匹配重复 lower);
  match() 文本只 lower 一次(原逐规则重复);load() 的 yaml 文件名集合提到循环外
- worker.py:本地端点生成器 httpx.AsyncClient 懒建复用(原每步新建/销毁连接,
  对齐 ArchitectClient 惯用法;协作循环最多 10 次生成免重复建连)
- pipeline.py(协作循环):plan_by_id O(1) 步定义查找;done 集合增量维护
  (原每轮重建 progress+archive 扫描);领域只解析一次(原 _artifact_name 每步
  全领域 kb.match);_deps_done 支持传入预填集合(保持旧签名兼容)
- v2stats.py:回合数分布改增量聚合(sum/max/分桶计数),summary() O(n)→O(1),
  不再持有无界 list(修长时运行内存增长)
- gateway/agent.py + api.py:AgentService 运行计数 O(1) 化(原 register 全量扫描),
  状态迁移收敛到 _transition_state 单一入口(api.py cancel/异常两处绕过点一并接入,
  消除计数与状态脱节隐患);21 项 agent 测试全绿(两轮全量 230 passed 复核)
2026-09-18 23:45:35 +08:00
tzt 0767030f02 chore(v2): 优化前置锚点——本轮架构/算法优化的回滚基线 2026-09-18 23:03:04 +08:00
12 changed files with 491 additions and 72 deletions
+3
View File
@@ -38,3 +38,6 @@ agent_workspace/
# 安全扫描器工作目录(不入库)
.mimosa/
# 参考项目目录(不入库,仅作设计吸收来源)
extra/
+67 -7
View File
@@ -15,6 +15,7 @@
from __future__ import annotations
import asyncio
import copy
import json
import time
import uuid
@@ -70,6 +71,9 @@ _REVIEW_SCHEMA_HINT = {
DEFAULT_MAX_HANDOFFS = 2 # 规划者<->执行者交接轮数上限
# 会写文件的工具名(T-M2 折叠摘要的文件清单来源)
_FILE_TOOLS = {"write_file", "edit_file"}
def _parse_json_loose(content: str) -> Dict[str, Any]:
"""宽松解析规划者的 JSON 输出(剥围栏/取首个对象);失败返回 {}。"""
@@ -312,6 +316,13 @@ class AgentService:
self.run_dir = Path(run_dir)
self._runs: Dict[str, AgentRunInfo] = {}
self.max_running = 5
self._running = 0 # O(1) 运行计数(原 register 每次全量扫描 _runs)
def _transition_state(self, info: AgentRunInfo, new_state: str) -> None:
"""状态迁移的单一入口:离开 running 时同步递减计数。"""
if info.state == STATE_RUNNING and new_state != STATE_RUNNING:
self._running -= 1
info.state = new_state
# ---------- 路径 ----------
def _dir(self, request_id: str) -> Path:
@@ -327,14 +338,14 @@ class AgentService:
def register(self, request_id: str, task: str, model: str, pool_id: str,
workspace: str = "", executor_model: str = "",
mode: str = "single") -> Optional[AgentRunInfo]:
running = [r for r in self._runs.values() if r.state == STATE_RUNNING]
if len(running) >= self.max_running:
if self._running >= self.max_running:
return None
info = AgentRunInfo(request_id=request_id, task=task, model=model,
pool_id=pool_id, workspace=workspace,
executor_model=executor_model, mode=mode,
started_at=time.time())
self._runs[request_id] = info
self._running += 1
self._dir(request_id).mkdir(parents=True, exist_ok=True)
self._write_status(info)
return info
@@ -413,7 +424,7 @@ class AgentService:
throttle.flush("executor")
self._apply_result(info, result)
except Exception as exc: # pragma: no cover
info.state = STATE_FAILED
self._transition_state(info, STATE_FAILED)
info.error = f"{type(exc).__name__}: {exc}"
self._append_event(info, {"type": "final", "round": info.rounds,
"reason": "error", "error": info.error})
@@ -437,12 +448,31 @@ class AgentService:
def _history_from_session(session: Optional["AgentSession"],
max_turns: int = 6,
max_chars: int = 1500) -> List[Dict[str, Any]]:
"""把会话既往轮次折叠成对话上下文(不含工具细节)。"""
"""把会话既往轮次折叠成对话上下文(不含工具细节)。
T-M2(采纳 pi branch-summarization 思想):折叠摘要携带本会话
已写入/编辑的文件清单——轮次滑出窗口后,事实清单仍随首条摘要
进入对话,丢上下文不丢事实。
"""
if session is None:
return []
turns = [t for t in session.data.get("turns", [])
if t.get("state") == STATE_DONE and t.get("response")]
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") 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")
})
out: List[Dict[str, Any]] = []
if files:
shown = "、".join(f[:120] for f in files[:20])
more = f"(等共 {len(files)} 个)" if len(files) > 20 else ""
out.append({"role": "user",
"content": f"(上下文摘要)本会话此前已写入/编辑的文件:{shown}{more}"})
out.append({"role": "assistant", "content": "已了解上述文件背景,将继续任务。"})
for t in turns[-max_turns:]:
out.append({"role": "user", "content": str(t["task"])[:max_chars]})
out.append({"role": "assistant", "content": str(t["response"])[:max_chars]})
@@ -455,14 +485,14 @@ class AgentService:
info.prompt_tokens = int(result.get("prompt_tokens", 0))
info.completion_tokens = int(result.get("completion_tokens", 0))
if result.get("reason") == "error":
info.state = STATE_FAILED
self._transition_state(info, STATE_FAILED)
info.error = result.get("error")
elif result.get("reason") in ("token_cap", "max_rounds", "max_handoffs"):
# 触顶属于护栏行为:结果仍交付,但标记部分完成信息
info.state = STATE_DONE
self._transition_state(info, STATE_DONE)
info.error = result.get("error")
else:
info.state = STATE_DONE
self._transition_state(info, STATE_DONE)
# ---------- 两级模式(D7):规划者 + 执行者 ----------
async def run_dual(self, info: AgentRunInfo, planner_chat: Any, executor_chat: Any,
@@ -882,6 +912,36 @@ class SessionStore:
self.save(sess)
return sess
def fork(self, sid: str, turn_index: Optional[int] = None,
title: str = "") -> Optional[AgentSession]:
"""从既有会话分叉新会话(T-M1,采纳 pi 会话树 id/parentId 设计)。
值拷贝 turns[:turn_index+1](缺省 = 全部已完结轮次;-1 = 空会话壳),
原会话不变;新会话记录 parent_id / fork_point 保留血统。采用拷贝式
而非 pi 的共享存储引用——规避并发写冲突,代价是 fork 后磁盘冗余
(会话体量小,可接受)。turn_index 越界返回 None。
"""
src = self.get(sid)
if src is None:
return None
turns = src.data.get("turns", []) or []
n = len(turns) - 1 if turn_index is None else int(turn_index)
if n < -1 or n >= len(turns):
return None
data = copy.deepcopy(src.data)
data["id"] = "as" + uuid.uuid4().hex[:10]
base_title = (title or "").strip() or (src.data.get("title") or "会话") + "·分叉"
data["title"] = base_title[:24]
data["parent_id"] = sid
data["fork_point"] = n
data["created_at"] = time.time()
data["busy"] = False
data["turns"] = copy.deepcopy(turns[:n + 1]) if n >= 0 else []
sess = AgentSession(data)
self._cache[data["id"]] = sess
self._save(sess)
return sess
def save(self, sess: AgentSession) -> None:
self._cache[sess.data["id"]] = sess
self._save(sess)
+20 -2
View File
@@ -647,7 +647,7 @@ try:
except Exception as exc:
import traceback
traceback.print_exc()
info.state = "failed"
service._transition_state(info, "failed")
info.error = str(exc)
info.finished_at = __import__("time").time()
service._write_status(info)
@@ -674,7 +674,7 @@ try:
return {"ok": False, "detail": f"任务已结束({info.state})"}
if info.asyncio_task is not None:
info.asyncio_task.cancel()
info.state = "failed"
service._transition_state(info, "failed")
info.error = "cancelled_by_user"
info.finished_at = __import__("time").time()
service._write_status(info)
@@ -740,6 +740,24 @@ try:
raise HTTPException(status_code=404, detail=f"会话不存在: {sid}")
return sess.view()
@app.post("/agent/sessions/{sid}/fork", tags=["agent"])
async def agent_session_fork(sid: str, req: dict = None):
"""从会话分叉新会话:{"turn_index": 截止轮下标(缺省=全部轮), "title"}。
T-M1(采纳 pi 会话树设计):值拷贝历史轮次,原会话不变;
新会话带 parent_id/fork_point 血统字段。
"""
_check_id(sid, "会话 ID")
from gateway.agent import get_session_store
body = req or {}
turn_index = body.get("turn_index")
sess = get_session_store().fork(sid, turn_index,
str(body.get("title") or ""))
if sess is None:
raise HTTPException(status_code=404,
detail=f"会话不存在或分叉点越界: {sid}")
return sess.view()
@app.delete("/agent/sessions/{sid}", tags=["agent"])
async def agent_session_delete(sid: str):
_check_id(sid, "会话 ID")
+9 -3
View File
@@ -53,20 +53,26 @@ class InferenceEngine:
# ---------------------------------------------------------------
def run(self, query: str, domain: str, memory: WorkingMemory,
max_steps: Optional[int] = None) -> List[str]:
"""前向链主循环。返回触发规则 id 列表(按触发顺序)。"""
"""前向链主循环。返回触发规则 id 列表(按触发顺序)。
循环不变量外提:query/domain 在循环内不变,kb.match 结果只算一次
(原实现每步全量重扫+重排序,最坏 O(steps × rules × patterns))。
"""
steps = max_steps or self.max_steps
rules = self.kb.match(query, domain=domain)
fired: List[str] = []
fired_set: set = set() # O(1) 查重(fired 保持列表维护触发顺序)
for _ in range(steps):
rules = self.kb.match(query, domain=domain)
# 选第一个"未触发过"的规则
target: Optional[Rule] = None
for r in rules:
if r.id not in fired:
if r.id not in fired_set:
target = r
break
if target is None:
break # 无新规则可触发 → 终止
fired.append(target.id)
fired_set.add(target.id)
self._fire(target, query, memory)
return fired
+22 -5
View File
@@ -51,13 +51,25 @@ class Rule:
actions: List[str] = field(default_factory=list) # 保留字段:动作扩展
subdomain: Optional[str] = None # 二级子领域(如 investing/labor/calculus)
subdomain2: Optional[str] = None # 三级子领域(如 fund/overtime/sorting)
_patterns_lower: Optional[tuple] = field(default=None, repr=False, compare=False)
def _lowered(self) -> tuple:
"""patterns 的小写缓存(注册后规则视为不可变;懒计算一次)。"""
if self._patterns_lower is None:
self._patterns_lower = tuple(p.lower() for p in self.patterns)
return self._patterns_lower
def matches(self, text: str) -> bool:
"""任一 pattern 是 text 的子串即命中(大小写不敏感)。"""
if not self.patterns:
return False
q = text.lower()
return any(p.lower() in q for p in self.patterns)
return self._match_lower(text.lower())
def _match_lower(self, q: str) -> bool:
"""已 lowercase 文本的快速匹配(避免每条规则重复 lower 同一文本)。"""
if not self.patterns:
return False
return any(p in q for p in self._lowered())
# ---------------------------------------------------------------
@@ -420,12 +432,13 @@ class KnowledgeBase:
self._facts = {d: [dict(f) for f in facts] for d, facts in BUILTIN_FACTS.items()}
if self.rules_dir.is_dir():
yaml_names = {p.name for p in self.rules_dir.glob("*.yaml")} # 一次遍历,避免逐文件重扫
for f in sorted(self.rules_dir.glob("*.yaml")):
data = _try_load_yaml(f)
if data is not None:
self._load_file_data(f, data)
for f in sorted(self.rules_dir.glob("*.json")):
if f.name not in {p.name for p in self.rules_dir.glob("*.yaml")}:
if f.name not in yaml_names:
data = _try_load_json(f)
if data is not None:
self._load_file_data(f, data)
@@ -461,12 +474,16 @@ class KnowledgeBase:
# ---- 查询 ----
def match(self, text: str, domain: Optional[str] = None) -> List[Rule]:
"""返回命中的规则,按优先级降序。domain 为空则全领域匹配。"""
"""返回命中的规则,按优先级降序。domain 为空则全领域匹配。
文本只 lowercase 一次(原实现每条规则各 lower 一遍)。
"""
q = text.lower()
hits = []
for rule in self._rules.values():
if domain is not None and rule.domain != domain:
continue
if rule.matches(text):
if rule._match_lower(q):
hits.append(rule)
hits.sort(key=lambda r: r.priority, reverse=True)
return hits
+21 -10
View File
@@ -110,20 +110,24 @@ class CollaborativePipeline:
plan = brief.get("plan") or []
pending = [p.get("id") for p in plan]
plan_ids = {p.get("id") for p in plan} # 用于 _deps_done 过滤
plan_by_id = {p.get("id"): p for p in plan} # O(1) 步定义查找(原每轮线性扫描)
done_ids = self._done_ids(ws) # 增量维护的完成集(原每轮重建)
base_domain = self._guess_domain(query) # ws["query"] 恒定,领域只解析一次
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), {})
step = plan_by_id.get(sid, {})
deps = step.get("deps") or []
# 只检查在 plan 中的依赖;不在 plan 的 ID 视为"不存在"→自动满足
relevant = [d for d in deps if d in plan_ids]
if not self._deps_done(ws, relevant):
if not all(d in done_ids for d in relevant):
continue
existing = self._read_artifact(request_id, self._artifact_name(sid, ws))
existing = self._read_artifact(request_id, self._artifact_name(sid, ws, domain=base_domain))
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)
done_ids.add(sid)
self._save_artifact(request_id, outcome.artifact_name, outcome.artifact_text)
ws.rollup()
route.append(f"step:{sid}:done")
@@ -230,9 +234,10 @@ class CollaborativePipeline:
return hits[0].domain
return "general"
def _artifact_name(self, sid: str, ws: Workspace) -> str:
def _artifact_name(self, sid: str, ws: Workspace, domain: Optional[str] = None) -> str:
from .worker import artifact_name_for
domain = self._guess_domain(ws["query"])
if domain is None:
domain = self._guess_domain(ws["query"])
# 用 brief.tags 优先
tags = (ws.get("brief") or {}).get("tags") or []
for t in tags:
@@ -241,8 +246,10 @@ class CollaborativePipeline:
break
return artifact_name_for(sid, domain)
def _deps_done(self, ws: Workspace, deps: List[str]) -> bool:
# progress 中 done 的条目;done 条目 rollup 后移到 archive(字符串格式如 "s1: ...")
@staticmethod
def _done_ids(ws: Workspace) -> set:
"""收集已完成 step id 集合:progress 中 done 的条目 + archive 折叠行
(done 条目 rollup 后移到 archive,字符串格式如 "s1: ...")。"""
done = {p["step"] for p in ws.get("progress", []) if p.get("status") == "done"}
for entry in ws.get("archive", []):
if isinstance(entry, dict):
@@ -251,9 +258,13 @@ class CollaborativePipeline:
sid = entry.split(":")[0].strip() if ":" in entry else ""
else:
sid = ""
if sid in deps:
done.add(sid)
return all(d in done for d in deps)
done.add(sid)
return done
def _deps_done(self, ws: Workspace, deps: List[str], done_ids: Optional[set] = None) -> bool:
if done_ids is None:
done_ids = self._done_ids(ws)
return all(d in done_ids for d in deps)
def _last_decision_for(self, ws: Workspace, sid: str) -> str:
"""取最近一条针对该 step 的决策 reply,作为 worker hint。"""
+19 -6
View File
@@ -2,6 +2,9 @@
T9:每请求 API token 记账;聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。
配合 /metrics 对外透出(论文 E1 token 经济学数据来源之一)。
性能设计(2026-09 优化):回合数分布以"和/最大值/分桶计数"增量维护,
summary() 从每次全量重算 O(n) 降为 O(桶数),且不再持有无界 list。
"""
from __future__ import annotations
@@ -18,7 +21,11 @@ class V2Stats:
self._fast_path = 0
self._breach = 0
self._by_status: Dict[str, int] = {}
self._rounds: List[int] = []
# 回合数增量聚合(等价于全量 list 的 sum/max/histogram,内存 O(桶数))
self._rounds_count = 0
self._rounds_sum = 0
self._rounds_max = 0
self._rounds_hist: Dict[str, int] = {}
self._api_input_tokens = 0
self._api_output_tokens = 0
self._api_cost_usd = 0.0
@@ -33,7 +40,13 @@ class V2Stats:
self._fast_path += 1
status = getattr(result, "status", "?")
self._by_status[status] = self._by_status.get(status, 0) + 1
self._rounds.append(getattr(result, "rounds_used", 0))
rounds = getattr(result, "rounds_used", 0)
self._rounds_count += 1
self._rounds_sum += rounds
if rounds > self._rounds_max:
self._rounds_max = rounds
bucket = str(rounds) if rounds <= 10 else ">10"
self._rounds_hist[bucket] = self._rounds_hist.get(bucket, 0) + 1
if "breach" in " ".join(getattr(result, "route", [])):
self._breach += 1
self._api_input_tokens += getattr(result, "api_input_tokens", 0)
@@ -64,16 +77,16 @@ class V2Stats:
def summary(self) -> Dict[str, Any]:
with self._lock:
total = self._total
rounds = self._rounds
rounds_n = self._rounds_count
return {
"total_requests": total,
"fast_path_rate": round(self._fast_path / total, 4) if total else 0.0,
"status_distribution": dict(self._by_status),
"breach_count": self._breach,
"rounds_used": {
"avg": round(sum(rounds) / len(rounds), 2) if rounds else 0.0,
"max": max(rounds) if rounds else 0,
"distribution": _histogram(rounds),
"avg": round(self._rounds_sum / rounds_n, 2) if rounds_n else 0.0,
"max": self._rounds_max if rounds_n else 0,
"distribution": dict(self._rounds_hist),
},
"api_tokens": {
"input": self._api_input_tokens,
+17 -9
View File
@@ -188,25 +188,33 @@ def _mock_generate() -> Callable[[str], Awaitable[str]]:
def _make_llama_generate(cfg: Dict[str, Any],
default_base_url: Optional[str] = None) -> Callable[[str], Awaitable[str]]:
"""返回调用本地 OpenAI 兼容端点(llama-server / Ollama / vLLM)的生成器。"""
"""返回调用本地 OpenAI 兼容端点(llama-server / Ollama / vLLM)的生成器。
连接复用:httpx.AsyncClient 懒建一次、跨步骤复用(与 ArchitectClient 一致),
避免协作循环每步重新 TCP 建连。
"""
if default_base_url is None:
default_base_url = f"http://127.0.0.1:{cfg.get('port', 8901)}/v1"
base_url = cfg.get("base_url") or default_base_url
model = cfg.get("model") or "local"
temperature = float(cfg.get("temperature", 0.3))
timeout_s = float(cfg.get("per_step_timeout_s", 300))
client_holder: Dict[str, Any] = {"client": None}
async def _gen(prompt: str) -> str:
try:
import httpx
async with httpx.AsyncClient(timeout=timeout_s) as client:
resp = await client.post(
f"{base_url}/chat/completions",
json={"model": model, "messages": [{"role": "user", "content": prompt}],
"temperature": temperature, "max_tokens": 4096},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
client = client_holder["client"]
if client is None or client.is_closed:
client = httpx.AsyncClient(timeout=timeout_s)
client_holder["client"] = client
resp = await client.post(
f"{base_url}/chat/completions",
json={"model": model, "messages": [{"role": "user", "content": prompt}],
"temperature": temperature, "max_tokens": 4096},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
except Exception as e: # noqa: BLE001
# 连不上本地模型 -> 优雅降级(不抛 500),提示用户检查模型端点
return ("(本地降级)无法连接本地模型端点,未能生成该步骤内容。"
+82 -30
View File
@@ -397,7 +397,12 @@ class Workspace:
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"])
# T-M2(采纳 pi branch-summarization 思想):折叠行携带产物锚点,
# 上下文折叠后产物事实仍在 archive 中可寻址(a:// 锚点体系)
base = f"{entry['step']}: {entry.get('summary', '')}"
if entry.get("artifact"):
base += f"(产出: {entry['artifact']})"
line = _clip(base, LIMITS["archive"])
if line not in self._data["archive"]:
self._data["archive"].append(line)
folded += 1
@@ -445,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:
@@ -520,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)
+73
View File
@@ -0,0 +1,73 @@
"""折叠摘要携带文件清单(T-M2,采纳 pi branch-summarization 设计)。"""
import gateway.agent as ag
from gateway.agent import AgentSession, AgentService
from router_system.workspace import Workspace
def _mk_turn(rid, task, tool_calls=None, state="done", response="好"):
return {"request_id": rid, "task": task, "response": response,
"state": state, "tool_calls": tool_calls or [], "tokens": 10,
"error": None, "ts": 0.0}
def _sess(turns):
return AgentSession({"id": "as-x", "title": "s", "turns": turns})
def test_history_carries_file_manifest_from_out_of_window_turns():
"""文件清单来自全部轮次:写文件轮滑出窗口后清单仍在摘要里。"""
turns = [
_mk_turn("t1", "建笔记", tool_calls=[
{"id": "c1", "name": "write_file", "arguments": {"path": "notes.md"}},
]),
_mk_turn("t2", "改笔记", tool_calls=[
{"id": "c2", "name": "edit_file", "arguments": {"path": "notes.md"}},
]),
_mk_turn("t3", "闲聊"),
]
hist = AgentService._history_from_session(_sess(turns), max_turns=1)
flat = "".join(str(m.get("content")) for m in hist)
# 窗口只剩 t3,但文件清单(t1/t2 产生)随摘要进入上下文
assert "notes.md" in flat
assert "上下文摘要" in flat
# 重复路径去重
assert flat.count("notes.md") >= 1
def test_history_without_file_tools_has_no_manifest():
"""无写文件工具调用 -> 不注入摘要轮(零噪音)。"""
turns = [_mk_turn("t1", "问个问题", tool_calls=[
{"id": "c1", "name": "web_fetch", "arguments": {"url": "https://x"}},
])]
hist = AgentService._history_from_session(_sess(turns), max_turns=6)
assert all("上下文摘要" not in str(m.get("content")) for m in hist)
def test_manifest_caps_path_length_and_count():
"""路径裁剪 120 字符、清单封顶 20 个,防摘要膨胀。"""
turns = [_mk_turn(f"t{i}", "写", tool_calls=[
{"id": f"c{i}", "name": "write_file", "arguments": {"path": f"很长的路径{i}" * 30}},
]) for i in range(25)]
hist = AgentService._history_from_session(_sess(turns), max_turns=0)
manifest = next(m for m in hist if "上下文摘要" in str(m.get("content")))
content = str(manifest["content"])
assert "等共 25 个" in content
body = content.split(":", 1)[1] # 剥掉固定前缀后逐段校验单路径裁剪
for seg in body.split("(等共")[0].split("、"):
assert len(seg) <= 120
def test_rollup_line_carries_artifact_anchor():
"""workspace rollup 折叠行携带产物锚点(a:// 体系内可寻址)。"""
ws = Workspace.new(request_id="abc123def456", query="写个快排",
api_token_cap=8000, rounds_cap=6)
ws.apply_brief({
"goal": "快排", "constraints": [], "tags": ["code"],
"acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
"plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "过"}],
})
ws.add_progress("s1", "done", "完成", "a://s1_main.py")
ws.rollup()
archive = "".join(ws["archive"])
assert "s1" in archive
assert "(产出: a://s1_main.py)" in archive
+60
View File
@@ -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
+98
View File
@@ -0,0 +1,98 @@
"""会话树 fork 单元与端点测试(T-M1,采纳 pi 会话树 id/parentId 设计)。"""
import pytest
pytest.importorskip("fastapi")
pytest.importorskip("httpx")
from fastapi.testclient import TestClient
import gateway.agent as ag
import gateway.api as ga
from gateway.agent import AgentSession, SessionStore
def _mk_turn(rid: str, task: str) -> dict:
return {"request_id": rid, "task": task, "response": f"ans-{rid}",
"state": "done", "tool_calls": [], "tokens": 10,
"error": None, "ts": 0.0}
def test_fork_copies_turns_and_keeps_lineage(tmp_path):
"""值拷贝历史轮次 + 记录 parent_id/fork_point 血统。"""
store = SessionStore(root=tmp_path / "sessions")
src = store.create("源会话", "ws")
src.data["turns"] = [_mk_turn("t1", "任务一"), _mk_turn("t2", "任务二")]
store.save(src)
forked = store.fork(src.data["id"], turn_index=0, title="分叉A")
assert forked is not None
assert forked.data["id"] != src.data["id"]
assert forked.data["parent_id"] == src.data["id"]
assert forked.data["fork_point"] == 0
assert [t["request_id"] for t in forked.data["turns"]] == ["t1"]
# 原会话不变
assert len(store.get(src.data["id"]).data["turns"]) == 2
def test_fork_default_takes_all_turns(tmp_path):
"""缺省 turn_index = 全部已完结轮次。"""
store = SessionStore(root=tmp_path / "sessions")
src = store.create("源", "ws")
src.data["turns"] = [_mk_turn("t1", "a"), _mk_turn("t2", "b")]
store.save(src)
forked = store.fork(src.data["id"])
assert len(forked.data["turns"]) == 2
assert forked.data["fork_point"] == 1
def test_fork_out_of_range_returns_none(tmp_path):
"""turn_index 越界 / 源会话不存在 -> None(端点映射 404)。"""
store = SessionStore(root=tmp_path / "sessions")
src = store.create("源", "ws")
src.data["turns"] = [_mk_turn("t1", "a")]
store.save(src)
assert store.fork(src.data["id"], turn_index=5) is None
assert store.fork(src.data["id"], turn_index=-2) is None
assert store.fork("as-not-exist") is None
def test_fork_is_deep_copy(tmp_path):
"""fork 后修改新会话不影响原会话(值拷贝语义)。"""
store = SessionStore(root=tmp_path / "sessions")
src = store.create("源", "ws")
src.data["turns"] = [_mk_turn("t1", "a")]
store.save(src)
forked = store.fork(src.data["id"])
forked.data["turns"][0]["task"] = "被篡改"
forked.data["turns"].append(_mk_turn("t2", "新轮"))
assert store.get(src.data["id"]).data["turns"][0]["task"] == "a"
assert len(store.get(src.data["id"]).data["turns"]) == 1
def test_old_sessions_without_lineage_fields_compatible(tmp_path):
"""旧数据无 parent_id/fork_point 字段自然兼容(dict.get 语义,零迁移)。"""
store = SessionStore(root=tmp_path / "sessions")
sess = AgentSession({"id": "as-legacy", "title": "旧", "turns": []})
store.save(sess)
assert store.get("as-legacy").data.get("parent_id", "") == ""
def test_fork_endpoint(tmp_path, monkeypatch):
"""POST /agent/sessions/{sid}/fork:端到端 200 + 404。"""
ag.reset_session_store()
monkeypatch.setattr(ag, "_session_store", SessionStore(root=tmp_path / "sessions"))
store = ag._session_store
src = store.create("源", "ws")
src.data["turns"] = [_mk_turn("t1", "a")]
store.save(src)
client = TestClient(ga.app)
r = client.post(f"/agent/sessions/{src.data['id']}/fork",
json={"turn_index": 0, "title": "端点分叉"})
assert r.status_code == 200
data = r.json()
assert data["parent_id"] == src.data["id"]
assert data["title"] == "端点分叉"
r404 = client.post("/agent/sessions/as-not-exist/fork", json={})
assert r404.status_code == 404