From 98b3e586de001dc9b34705bbb952aefc17284306 Mon Sep 17 00:00:00 2001 From: hupenglong1 Date: Thu, 28 May 2026 14:35:35 +0800 Subject: [PATCH] fix: subprocess-based remote sync, pipeline status, skill updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace thread-based remote state sync with subprocess (sync_remote_state.py) to prevent thread pool exhaustion hanging the API - Fix pipeline showing 'waiting' when a real step is running alongside a gate - Fix watcher scanner path for linux accounts mode - Disable label-master semantic review (Step A.5 + §4.5) per user request - Correct field names: use is_model_correct_dev for accuracy, origin_predict_dev for model output (not cleaned_predict) - Add prohibition against putting test set data in relabel_candidates Co-Authored-By: Claude Opus 4.6 --- backend/api/server.py | 113 +++--- scripts/sync_remote_state.py | 96 +++++ skills/model-iteration/SKILL.md | 52 +-- skills/model-iteration/references/program.md | 110 ++---- src/agent_runtime.py | 383 ++++++++++++------- src/agent_tools.py | 47 ++- src/jupyter_runtime.py | 21 +- 7 files changed, 492 insertions(+), 330 deletions(-) create mode 100644 scripts/sync_remote_state.py diff --git a/backend/api/server.py b/backend/api/server.py index e4435a1..87745ce 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio from contextlib import asynccontextmanager import csv +import sys import hashlib import shlex import json @@ -2024,9 +2025,9 @@ def create_app(state: AgentState) -> FastAPI: ) -> Response: sessions_dir = state.account_paths(account_id)['sessions'] state_path = _session_state_path(sessions_dir, session_id) - await asyncio.to_thread( - _sync_remote_program_state, state, account_id, session_id, state_path, - ) + binding_path = _jupyter_binding_path(sessions_dir, session_id) if session_id else None + if binding_path and binding_path.is_file(): + await _subprocess_sync_remote(binding_path, state_path) state_entries, _ = _read_program_state(state_path) trigger = _extract_trigger_from_state(state_entries) or _extract_trigger_from_session( sessions_dir, session_id @@ -2118,9 +2119,9 @@ def create_app(state: AgentState) -> FastAPI: break now_mono = time.monotonic() if now_mono - last_remote_sync >= 5.0: - await asyncio.to_thread( - _sync_remote_program_state, state, account_id, session_id, state_path, - ) + _bp = _jupyter_binding_path(sessions_dir, session_id) if session_id else None + if _bp and _bp.is_file(): + await _subprocess_sync_remote(_bp, state_path) last_remote_sync = now_mono state_entries, state_mtime = _read_program_state(state_path) iter_count = await asyncio.to_thread( @@ -5902,9 +5903,29 @@ def _scan_for_watchers(state: AgentState) -> None: ) +_SYNC_SCRIPT = str(Path(__file__).resolve().parent.parent.parent / 'scripts' / 'sync_remote_state.py') + + +async def _subprocess_sync_remote(binding_path: Path, local_state_path: Path) -> None: + """Sync remote program-state via subprocess. Killable on timeout.""" + try: + proc = await asyncio.create_subprocess_exec( + sys.executable, _SYNC_SCRIPT, + str(binding_path), str(local_state_path), + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await asyncio.wait_for(proc.wait(), timeout=10.0) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + except OSError: + pass + + async def _scan_for_watchers_async(state: AgentState) -> None: - """Async-friendly scan: runs blocking I/O (remote sync, LLM target_set - extraction) in worker threads so the event loop stays responsive for SSE.""" + """Scan program-state.jsonl files for watch declarations and register + watchers. Uses subprocess for remote sync (killable, no thread pool).""" accounts_root = state.session_directory.parent / 'accounts' if not accounts_root.is_dir(): return @@ -5916,7 +5937,7 @@ async def _scan_for_watchers_async(state: AgentState) -> None: if not account_dir.is_dir(): continue account_id = account_dir.name - sessions_dir = account_dir / 'sessions' + sessions_dir = state.account_paths(account_id)['sessions'] if not sessions_dir.is_dir(): continue try: @@ -5926,44 +5947,16 @@ async def _scan_for_watchers_async(state: AgentState) -> None: for session_dir in session_dirs: session_id = session_dir.name state_path = session_dir / 'output' / 'program-state.jsonl' - # Remote sync runs HTTP/WS — blocking. Off the loop. - try: - await asyncio.to_thread( - _sync_remote_program_state, - state, - account_id, - session_id, - state_path, - ) - except Exception as exc: # noqa: BLE001 - print( - f'[scanner] remote sync failed for {account_id}/{session_id}: {exc}', - flush=True, - ) + binding_path = session_dir / 'jupyter_workspace.json' + # Subprocess remote sync only for sessions with a jupyter binding + if binding_path.is_file(): + await _subprocess_sync_remote(binding_path, state_path) if not state_path.is_file(): continue try: - entries, _ = await asyncio.to_thread( - _read_program_state, state_path - ) + entries, _ = _read_program_state(state_path) except Exception: # noqa: BLE001 continue - # Refresh LLM-derived target_set cache (also off the loop). - trigger_text = await asyncio.to_thread( - _extract_trigger_from_session, sessions_dir, session_id - ) - if trigger_text: - try: - await asyncio.to_thread( - _refresh_target_set_cache_blocking, - state.model_config_for(account_id), - trigger_text, - ) - except Exception as exc: # noqa: BLE001 - print( - f'[scanner] target_set LLM refresh failed: {exc}', - flush=True, - ) # Watch registration is fast (just spawns asyncio tasks); stay on loop. last_status_per_step: dict[str, str] = {} for entry in entries: @@ -6565,6 +6558,21 @@ def _build_pipeline_items( return all_items +def _has_active_watch_for_step(state_entries: list[dict[str, Any]], step: str) -> bool: + """True if state_entries contain a watch for this step that hasn't resolved.""" + last_status: dict[str, str] = {} + has_watch = False + for entry in state_entries: + s = entry.get('step') + st = entry.get('status') + if isinstance(s, str) and isinstance(st, str): + last_status[s] = st + watch = entry.get('watch') + if isinstance(watch, dict) and watch.get('step') == step: + has_watch = True + return has_watch and last_status.get(step) not in ('complete', 'failed', 'cancelled') + + def _compute_in_flight( state_entries: list[dict[str, Any]], iteration_log_count: int, @@ -6664,7 +6672,8 @@ def _compute_in_flight( if status not in ('complete', 'running'): continue if status == 'running' and gate_running_for_run: - status = 'waiting' + if not _has_active_watch_for_step(state_entries, step): + status = 'waiting' meta = _CARD_META.get((step, target_section), { 'icon': 'clock', 'title': step, @@ -6848,14 +6857,18 @@ def _apply_program_state( statuses = list(latest_per_step.values()) if not statuses: payload['status']['state'] = 'pending' - elif non_gate_running and not gate_running: + elif non_gate_running: payload['status']['state'] = 'running' elif gate_running: # HiTL gate is running — agent has handed control back to the user. - # Even if a parent step (e.g. augment) is still technically 'running', - # the gate blocks progress. Surface as 'waiting' so the header pill - # says "等待人工" instead of the spinning "Running" badge. - payload['status']['state'] = 'waiting' + # Exception: if a non-gate step has an active watcher (e.g. cml eval + # running on remote), the session is still progressing — show running. + watched_running = any( + status == 'running' and _STEP_KIND.get(step) != 'gate' + and _has_active_watch_for_step(state_entries, step) + for step, status in latest_per_step.items() + ) + payload['status']['state'] = 'running' if watched_running else 'waiting' elif all(s == 'complete' for s in statuses): if run_active: payload['status']['state'] = 'running' @@ -6946,7 +6959,7 @@ def _chat_root_for_runtime( ) -> str | None: """Resolve the per-chat autoresearch root as a REMOTE path string. - Returns the NFS path under /mnt/wangsenhao/autoresearch-zk-users/// + Returns the NFS path under /mnt//autoresearch-zk-users// so jupyter pod and SFT training pod both read/write the same directory. Returns None when no runtime is bound — caller decides whether to fall back to the legacy global path.""" @@ -8033,7 +8046,7 @@ def _step_artifact_paths( Per-chat artifacts (results/, ai-planning/, output/) live under the chat workspace root on shared NFS at - `/mnt/wangsenhao/autoresearch-zk-users///`. Reads still + `/mnt//autoresearch-zk-users//`. Reads still go through the bound jupyter runtime since the path is only mounted there. metric_diff (cml step) stays on the global RUN_HISTORY_DIR mount that is shared across chats. When no runtime is bound, falls back to the legacy diff --git a/scripts/sync_remote_state.py b/scripts/sync_remote_state.py new file mode 100644 index 0000000..7c02539 --- /dev/null +++ b/scripts/sync_remote_state.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Subprocess-safe remote program-state sync. + +Reads jupyter binding JSON, fetches program-state.jsonl via jupyter +Contents API (HTTP GET), writes to local path. Designed to be called +via asyncio.create_subprocess_exec with a hard timeout — if the pod is +unreachable, the parent kills this process cleanly. + +Usage: python3 sync_remote_state.py +""" +import json +import sys +import urllib.request +import urllib.error +import ssl + +def main(): + if len(sys.argv) != 3: + sys.exit(1) + binding_path = sys.argv[1] + local_path = sys.argv[2] + + try: + with open(binding_path, encoding='utf-8') as f: + data = json.load(f) + except (OSError, json.JSONDecodeError): + sys.exit(1) + + binding = data.get('binding', {}) + base_url = binding.get('base_url', '').rstrip('/') + api_path = binding.get('workspace_api_path', '') + if not base_url or not api_path: + sys.exit(1) + + cookies_list = data.get('cookies', []) + cookie_str = '; '.join(f"{c['name']}={c['value']}" for c in cookies_list) + xsrf = data.get('xsrf_token', '') + + file_api_path = f"{api_path}/output/program-state.jsonl" + url = f"{base_url}/api/contents/{file_api_path}?content=1&type=file" + + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_peer = False + + req = urllib.request.Request(url) + req.add_header('Cookie', cookie_str) + if xsrf: + req.add_header('X-XSRFToken', xsrf) + + try: + resp = urllib.request.urlopen(req, timeout=8, context=ctx) + body = json.loads(resp.read()) + except (urllib.error.URLError, OSError, json.JSONDecodeError, TimeoutError): + sys.exit(1) + + content = body.get('content', '') + if not content or not content.strip(): + sys.exit(0) + + # Preserve local-only _synthetic entries + local_synthetics = [] + try: + with open(local_path, encoding='utf-8') as f: + for line in f: + if '"_synthetic"' in line: + try: + obj = json.loads(line) + if obj.get('_synthetic'): + local_synthetics.append(line.rstrip('\n')) + except (json.JSONDecodeError, ValueError): + pass + except OSError: + pass + + merged = content.rstrip('\n') + if local_synthetics: + merged += '\n' + '\n'.join(local_synthetics) + merged += '\n' + + # Skip rewrite if unchanged + try: + with open(local_path, encoding='utf-8') as f: + if f.read() == merged: + sys.exit(0) + except OSError: + pass + + import os + os.makedirs(os.path.dirname(local_path), exist_ok=True) + with open(local_path, 'w', encoding='utf-8') as f: + f.write(merged) + + +if __name__ == '__main__': + main() diff --git a/skills/model-iteration/SKILL.md b/skills/model-iteration/SKILL.md index 657fb30..52a9f6d 100644 --- a/skills/model-iteration/SKILL.md +++ b/skills/model-iteration/SKILL.md @@ -7,6 +7,12 @@ when_to_use: | # autoresearch-zk — 小爱中控模型自主迭代框架 +## ⛔ 第一步：读 program.md（强制，任何操作之前） + +收到"开始"触发信号后，**第一个动作**必须是读取 `references/program.md`。本文件（SKILL.md）只是概要和 UI 约定，**完整操作规范全在 program.md 里**——包括 bootstrap 流程（clone ai-planning 仓库）、cml 自动安装、评测集定位逻辑等关键步骤。 + +**不读 program.md 就开始工作 = 必然跑偏。** 已踩坑：agent 只看 SKILL.md 就开始环境检查，发现评测集文件不在本地后停下来问用户路径——而 program.md 明确写了"文件在 git 仓库里，先 clone"。 + ## 🚫 严禁问用户的事（违反任一条 = 违反 skill） 每次问用户都让用户烦。program.md 已经把"什么时候停"写得很死（17 类 HiTL 信号 + 用户主动打断），**这之外一律不准停下来等确认**。下面是被反复踩坑的"擅自暂停"模式，**全部禁止**： @@ -18,11 +24,11 @@ when_to_use: | | "我先把控制权交回，等你来问跑完了吗" | watcher 接手 UI 同步，长任务用 `bash(run_in_background=true, wait_for_completion=true)` 提交后本轮主动结束；后端会在产物落盘后自动起新一轮把结果送回，**直接进 Step 1** | | "这一步是关键决策点，需要你拍板" | program.md 没写就不是。**自主决策 + 落 iteration_log** | | "我把现状停在这里，把决策摘要给你 review" | 不准。摘要可以写，但不准停 | -| "Step 4 augment 完成 ✅，等回调后续做 SFT" | augment=complete 那一刻就是 SFT 启动那一刻——**同一轮 bash** 紧接着跑 §4.5 label-master 复核 + 提交 SFT，不许写简报、不许等回调（详见 program.md "Step 4 → Step 5 硬连接"）| +| "Step 4 augment 完成 ✅，等回调后续做 SFT" | augment=complete 那一刻就是 SFT 启动那一刻——**同一轮 bash** 紧接着跑层1格式校验 + 提交 SFT，不许写简报、不许等回调（详见 program.md "Step 4 → Step 5 硬连接"）| | "R1 评测发现 regression，先把诊断给你看，等你拍板再决定要不要回滚" | dist-analysis 发现 regression **也算 dist-analysis 完成**——**同轮 bash** 紧接着写 `results/workflow.md`（包含完整 delta 表 + new_fail/new_fix 子集 + 病灶定位 + 回滚建议）+ append `iteration_log.jsonl` R{n} entry。文件落完了再用 chat reply 给人提回滚选项。**不许把诊断只写在 chat 里、不落盘**——前端「分层结果分析」卡片读的是 `results/workflow.md`，你不写卡片永远停在上轮。 | | "SFT 用哪个 basemodel？(a) 用户 eval 给的路径 (b) 默认 Qwen3-4B" | program.md §5.0 写死 `--model_path = /mnt/wangsenhao/verl_zk/Qwen3-4B-Instruct-2507`，**每轮强制 basemodel**——这是规则不是选项。eval 阶段的 `model_path_new` 跟 SFT 的 `--model_path` 是两件事，不要混。直接用 Qwen3-4B-Instruct-2507。 | | "zk_trainer 仓库 URL 是什么？" | program.md §5.0 line 1323 写死 `git clone git@git.n.xiaomi.com:wangsenhao/zk_trainer.git`，**直接用**。clone 失败先看 ssh key（program.md 「SSH Key 检查」），**不要**自己脑补 `nlp/`、`xiaoai/`、`autoresearch/` 等命名空间问用户。 | -| "label-master 复核 100 条仿写，5 条全 pass，按比例外推 100 条 OK" | **`X/X pass` 缩写禁用**——§4.5 写死「H2 augment_.jsonl 必须逐条过 label-master」，review jsonl 行数必须 == augment.jsonl 行数（1:1 覆盖）。抽样外推 = 协议违反，§5.0 准入门会用 `wc -l` 拦住。状态日志里写"100 条复核完成"也必须真的是 100 条 review entry。 | +| "格式校验 5 条 pass，按比例外推 100 条 OK" | **抽样外推禁用**——`validate_label_output.py` 必须对全量文件跑，不许抽样 | **合法暂停只有**： @@ -83,20 +89,19 @@ when_to_use: | - [ ] **runDic 必须在当轮 eval `lark_template.json` 落盘后重新 `eval "$(./scripts/resolve_run_ids.sh)"` 取**——禁止复用之前缓存值或手算，eval 期间 resolve 会得到上一轮 workflow id 导致偏移 -1 - [ ] **§4.0 原始训练数据清洗（增强前必做）**——按 4 种动作走完逐 pattern 判定 - [ ] §4.0.1 Step A：候选定位输出 csv，**不直接改** -- [ ] §4.0.1 Step A.5：**先走 label-master 预审**（用推荐标签覆盖 new_label，剔除 label-master 不认同要改的条目） -- [ ] §4.0.1 Step B：量级判定基于 label-master 处理后的**残留候选量**走对应支线（≤50 自动按 label-master 推荐改 / 51-200 全量人审 / >200 触发 #3，全量交付） +- [ ] ~~§4.0.1 Step A.5~~ label-master 预审已禁用，跳过 +- [ ] §4.0.1 Step B：量级判定基于 Step A 候选量走对应支线（≤50 自动改 / 51-200 全量人审 / >200 触发 #3） - [ ] §4.0.1 Step C：备份 `.bak` 文件 - [ ] §4.1 输入准备 / §4.2 GPT 调用 / §4.3 sanity check / §4.4 写入 -- [ ] **§4.5 label-master 标签复核（落盘后必做，强制，H1+H2 都要）**：H1 `modified_samples.jsonl` **和** H2 `augment_.jsonl` 都必须跑两层（`validate_label_output.py` 格式 + Skill 调用 `label-master` 语义），**逐条 1:1 全量覆盖**。§4.0.1 A.5 的预审是粗筛，**不能替代落盘前的 layer2 语义复核**。verdict 写 `results/data_clean_/label_master_review.jsonl`，**不通过必须 = 0**（任何不通过必须修正后重新 review 直到全 pass 才能进 Step 5） -- [ ] **§4.5 层 2 语义复核必须通过子 agent 调用 `Skill(skill="label-master")`，禁止主 agent 直接调、禁止正则/规则脚本替代**：主 agent 把待审列表写入 `scratchpad/lm_input.jsonl`，起子 agent（`delegate_agent(prompt="...", allow_shell=true)`）逐条调 label-master，结果写 `scratchpad/lm_output.jsonl`，主 agent 读取汇总。子 agent prompt 里给**绝对路径**。**禁止**：主 agent 直接调 Skill(skill="label-master")、纯正则/关键词匹配脚本、"target 都一样所以直接 pass"逻辑、批量写 pass 不看 query 内容。 +- [ ] ~~§4.5 label-master 标签复核~~ 已禁用，仅保留层 1 格式校验（`validate_label_output.py`） ### Step 4 → Step 5 边界（**NEVER STOP 硬连接**，反复踩坑） -- [ ] 写完 `augment=complete` 那一刻，**同一轮 bash 不许结束**：紧接着跑 §4.5 label-master 复核 → 写 `sft=running` → 调 `submit_sft.sh` → 挂 watcher（评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不要在 SFT bg 里串接评测） +- [ ] 写完 `augment=complete` 那一刻，**同一轮 bash 不许结束**：紧接着跑层 1 格式校验 → 写 `sft=running` → 调 `submit_sft.sh` → 挂 watcher（评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不要在 SFT bg 里串接评测） - [ ] **不许写"Step 4 完成"进度简报后 turn 结束**，不许"等回调后续做 SFT" - [ ] H2 后台任务回调（`[system] 后台任务 ... exit_code=0`）**不是 turn 结束信号**——它只是 augment 子流程的一个中间节拍，agent 必须在同轮里继续走完 §4.5 → Step 5 ### Step 5 准入（`sft`） -- [ ] §4.5 label-master 复核报告 `label_master_review.jsonl` 存在、行数 = augment 行数（全量不抽样）、不通过 = 0（任何不通过必须修正后重新 review 直到全 pass） +- [ ] 层 1 格式校验通过（`validate_label_output.py`） - [ ] 旧 sft_output 已 `mv sft_output sft_output_r{prev}` 备份 - [ ] 训练参数从 config.yaml 读，model_path = basemodel（**不从上轮 ckpt 续训**） @@ -219,7 +224,7 @@ bg 任务（`bash(run_in_background=true)`）的合法范围是**一个 step 内 |---|---|---| | `cml` | 数分钟到数十分钟 | 远端 workflow 启动 + 评测 | | `sft` | 30 分钟以上 | 训练 pod | -| `augment` | 数分钟 | GPT 仿写 + sanity + label-master 复核 | +| `augment` | 数分钟 | GPT 仿写 + sanity + 格式校验 | | `dist-analysis` | 1-3 分钟 | 读 metric_diff + 跨轮 diff + 写 workflow.md | | `gold-drift` | 1-2 分钟 | drift 检测 | @@ -229,7 +234,7 @@ bg 任务（`bash(run_in_background=true)`）的合法范围是**一个 step 内 **正确做法**：human-review 之后必须按 program.md 逐 step 真跑： 1. 写 `hypothesis=running` → 真写假设到 iteration_log → 写 `hypothesis=complete` -2. 写 `augment=running` → 真跑 §4.0.1 + §4.1 + §4.2 + §4.3 + §4.5 label-master 复核 → 写 `augment=complete`（**这一步至少 5 分钟**） +2. 写 `augment=running` → 真跑 §4.0.1 + §4.1 + §4.2 + §4.3 + 层1格式校验 → 写 `augment=complete`（**这一步至少 5 分钟**） 3. 写 `sft=running` → 提交 cml custom_train + 起 watcher **自查**：连续两个 step 的 `ts` 间隔 < 60 秒，必然有一个是假的。复盘时 grep program-state.jsonl 看相邻 entry 时间戳。 @@ -403,7 +408,7 @@ echo '{"log":{"ts":"11:55:10","iter":"R1","text":"Step 1 dist-analysis: 写 work | 触发信号 | `"开始，<需求集合名>"` | | 评测 workflow | `f-20260408161444-wu3pz`（版本见 `config.yaml`） | | 训练基模 | `/mnt/wangsenhao/verl_zk/Qwen3-4B-Instruct-2507` | -| 工作目录 | `$AUTORESEARCH_CHAT_ROOT`（每 chat 独立，由后端注入；位于共享 NFS：`/mnt/wangsenhao/autoresearch-zk-users///`，jupyter pod 与训练 pod 都能读写） | +| 工作目录 | `$AUTORESEARCH_CHAT_ROOT`（每 chat 独立，由后端注入；位于共享 NFS：`/mnt//autoresearch-zk-users//`，jupyter pod 与训练 pod 都能读写） | | 历史记录目录 | `/mnt/xiaoai-zk-model-train-tj5/workflow5/` | | 需求集合位置 | `https://git.n.xiaomi.com/ai-service/ai-planning/-/tree/autoresearch-v1` 下 `ai-planning/data/specific_test_set/` | @@ -412,7 +417,7 @@ echo '{"log":{"ts":"11:55:10","iter":"R1","text":"Step 1 dist-analysis: 写 work 每个 chat session 一份独立工作区 `$AUTORESEARCH_CHAT_ROOT`（后端在 jupyter 启动时自动注入这个 env，agent 每次 bash 都能拿到）。该目录位于共享 NFS： ``` -/mnt/wangsenhao/autoresearch-zk-users/// +/mnt//autoresearch-zk-users// ``` `` 是用户登录时的小米邮箱前缀（`xxx@xiaomi.com` 取 `xxx`），用作账号根目录；同一用户跨 chat 共享根目录但 chat 之间完全隔离。 @@ -506,7 +511,7 @@ assets/ 写 gate entry / 在 chat 里向用户提问之前，**逐条过这三问**。下面任何一段流程示例都默认你已经过了这三问；过不了，再漂亮的 summary/proposal/ask 也是干扰用户。 **Q1. 是真 §HiTL 信号吗？** -- ✅ 真信号：label-master 预审后残留 > 50/200、Gold drift ≥ 10、跨子集净退步、连续 3 轮无提升 等列表里写明的条件 +- ✅ 真信号：候选 > 50/200、Gold drift ≥ 10、跨子集净退步、连续 3 轮无提升 等列表里写明的条件 - ❌ 凑出来的理由：`第一次跑想让用户校方向` / `我担心副作用` / `想让用户拍板更稳` / `proposal 听起来风险大` —— 这些都是脑补，不是信号 **Q2. `ask` 是真分叉吗？** @@ -534,10 +539,10 @@ assets/ | 时机 | gate 类型 | run_id | |---|---|---| -| R0 baseline 分析完成、命中 §HiTL 信号要让用户拍板（label-master 预审后残留候选超阈值、目标子集异常、跨子集分歧大 等真实信号；⛔ **不是**"第一次跑想让用户校方向" / "我担心 H1 修标的副作用" / "想让用户在激进/保守里选"这种凑出来的理由——先过上面的自检三问） | `human-check` | `R0` | +| R0 baseline 分析完成、命中 §HiTL 信号要让用户拍板（候选超阈值、目标子集异常、跨子集分歧大 等真实信号；⛔ **不是**"第一次跑想让用户校方向" / "我担心 H1 修标的副作用" / "想让用户在激进/保守里选"这种凑出来的理由——先过上面的自检三问） | `human-check` | `R0` | | R{n≥1} analysis 完成、命中 §HiTL 信号（gold drift / regression / 跨子集退步 / 连续 3 轮无提升 等） | `human-review` | `R{n}` | -| §4.0.1 Step B：**label-master 预审后**残留候选 > 200 条命中触发条件 #3（原扫数不算数） | `human-check` | 当前 round | -| 其他 HiTL 信号（label-master 预审后残留 > 50 条、Gold drift ≥ 10 条 等） | `human-check`（开始前）/ `human-review`（结果后） | 当前 round | +| §4.0.1 Step B：候选 > 200 条命中触发条件 #3 | `human-check` | 当前 round | +| 其他 HiTL 信号（候选 > 50 条需人审、Gold drift ≥ 10 条 等） | `human-check`（开始前）/ `human-review`（结果后） | 当前 round | 判断标准就一条：**只要你下一步打算 chat-ask 用户拍板，就先写 gate entry 再问**。 @@ -550,12 +555,10 @@ assets/ # ⛔ ask 必须是真分叉。⛔ "H1+H2 一起 vs 只跑 H1"、"激进 vs 保守"、"先 H1 还是先 H2" 这类 # "假设组合"统统不是分叉——是 agent 自己根据信号决定的，决定完写进 proposal 公布即可。 # R0 baseline 后命中真实 HiTL 信号（这里是候选量超阈值），让用户拍板更精细 pattern -# ⚠️ 给用户排板的候选数量必须是 §4.0.1 Step A.5 label-master 预审之后的残留数（推荐 != old_label 的那部分）， -# 不是 wide pattern 原始扫出来的数。原始数 label-master 还要剔掉一大半，先用原始数找用户=干扰用户判断。 echo '{"step":"human-check","status":"running","run_id":"R0", - "summary":"R0 baseline 完成；复杂导航过召专项0511 = 58.89%(53/90)；wide pattern 原扫 412 条，label-master 预审后残留 287 条（> 200 阈值）", - "proposal":"H1（标签纠错）：按 label-master 推荐改这 287 条 complex=true → complex=false。 H2（数据增强）：仿写 100 条 complex=false 的简单导航 query 补进训练集", - "ask":"残留 287 条仍偏多，是全部按 label-master 推荐改、还是先收窄到 query 含「打开/进入」的子集（约 110 条）单独审一轮？", + "summary":"R0 baseline 完成；复杂导航过召专项0511 = 58.89%(53/90)；候选 412 条（> 200 阈值）", + "proposal":"H1（标签纠错）：改这 412 条 complex=true → complex=false。 H2（数据增强）：仿写 100 条 complex=false 的简单导航 query 补进训练集", + "ask":"412 条偏多，是全部改、还是先收窄到 query 含「打开/进入」的子集（约 110 条）单独审一轮？", "ts":"'$(date -Iseconds)'"}' >> $S # R1 评测发现 regression @@ -566,8 +569,7 @@ echo '{"step":"human-review","status":"running","run_id":"R1", "ts":"'$(date -Iseconds)'"}' >> $S # fallback：只写 reason 也能跑（前端会启发式切分），但不如结构化清晰 -# 注意 reason 里的"412"也是 label-master 预审后残留数，不是原扫数 -echo '{"step":"human-check","status":"running","run_id":"R0","reason":"label-master 预审后残留 412 条仍超阈值，需要人工定更精细 pattern 收窄","ts":"'$(date -Iseconds)'"}' >> $S +echo '{"step":"human-check","status":"running","run_id":"R0","reason":"候选 412 条超阈值，需要人工定更精细 pattern 收窄","ts":"'$(date -Iseconds)'"}' >> $S ``` **顺序不能反**：先 echo gate entry → 再 chat reply 给用户。否则用户先看到聊天问话、UI 里却没卡，会困惑"流程是不是卡死了"。 @@ -581,7 +583,7 @@ echo '{"step":"human-check","status":"running","run_id":"R0","reason":"label-mas | `run_id` | ✅ | 当前所在轮次（决定 gate 插在哪个 section 后） | | `summary` | 🔼 | **现状一句话**：跑了什么、关键数字。例：`R0 baseline 完成；专项 58.89%(53/90)，37 错全为 complex 误判` | | `proposal` | 🔼 | **打算怎么干**：每个 H 单独说"H? (类型)：具体做什么"。详见下面规则 | -| `ask` | 🔼 | **让用户选什么**：必须是真分叉（用户的判断能改变下一步动作）。例：`label-master 预审后残留 287 条偏多，全改、还是收窄到「打开/进入」子集（~110 条）单独审一轮？` ⚠️ 写候选数量必须是 §4.0.1 Step A.5 label-master 预审后的残留数，不是原扫数。⛔ 反例：`H1+H2 一起 vs 只跑 H1`——假设组合是你定的，不甩给用户 | +| `ask` | 🔼 | **让用户选什么**：必须是真分叉（用户的判断能改变下一步动作）。例：`候选 287 条偏多，全改、还是收窄到「打开/进入」子集（~110 条）单独审一轮？` ⛔ 反例：`H1+H2 一起 vs 只跑 H1`——假设组合是你定的，不甩给用户 | | `reason` | ⭕ | 兜底用：没写 summary/proposal/ask 时前端会拿 reason 做启发式切分。但**优先用结构化三段**，别只写 reason | | `ts` | ✅ | ISO 时间戳 | @@ -598,7 +600,7 @@ echo '{"step":"human-check","status":"running","run_id":"R0","reason":"label-mas **不要写进 proposal 的内容**： - §X.X.X 规则引用（`触发 §4.0.1 Step B`、`命中条件 #3`）—— 用户不关心你按哪条规则做的，只关心你要做什么 -- 流程自洽说明（`需要人工逐条审 1/0`、`走 sanity check`、`过 label-master 复核`）—— 这些是 agent 内部流程，对用户决策没用 +- 流程自洽说明（`需要人工逐条审 1/0`、`走 sanity check`、`过格式校验`）—— 这些是 agent 内部流程，对用户决策没用 - 候选量区间括号注释（`100~150 触发 51-200 区间`）—— 数量 OK，区间归属归属是规则细节，删 **`ask` 字段尤其要注意：** diff --git a/skills/model-iteration/references/program.md b/skills/model-iteration/references/program.md index ea74504..7245a65 100644 --- a/skills/model-iteration/references/program.md +++ b/skills/model-iteration/references/program.md @@ -15,7 +15,7 @@ 用户发 **"开始，<需求集合名>"**（如 `"开始，icl_test"`）→ **立即用当前模型跑一轮评测**，不再中途确认参数，直到报告完成： -1. **Setup 检查**：cml 环境 + SSH key（见下）、从 `config.yaml` 读默认参数、初始化 `error_registry.jsonl` +1. **Setup 检查**：cml 环境（缺失则自动安装，见下）+ SSH key + **git clone ai-planning 仓库**（见「Chat 工作区 bootstrap」）+ 从 `config.yaml` 读默认参数、初始化 `error_registry.jsonl` 2. **准备评测参数**：确定 `model_path_new`（当前模型）/ `model_path_old`（基线），runDic 一律由 `eval "$(./scripts/resolve_run_ids.sh)"` 解析出来——**不要自己 `ls workflow5 \| tail -1` 心算 +1**；首次评测两者设为同一基准模型 3. **CML 评测**（Step 0）：执行 `cml workflow run`，后台轮询 `metric_diff/lark_template.json` 直到结果就绪 4. **分层结果分析 + 问题分析 & 报告**（Step 1）：按优先级逐层检查 需求集合（≥95%）→ 大盘车载（降幅≤0.3%）→ specific test（降幅≤1%）；做根因归类（reward/data/格式/hparam）、跨轮 diff（persistent/new/regressed）、需求集合深度分析（训练数据关联 + reward 对齐）、SFT 天花板诊断，写入 `results/workflow.md` @@ -27,6 +27,8 @@ > **首次评测**：首次评测只关注基准模型指标，`model_path_new` 和 `model_path_old` 设为同一个基准模型路径。分析报告只分析基准模型本身的表现，不做新旧模型对比（因为是同一个模型）。目的是建立 baseline 数据，为后续迭代提供对照基准。 > **需求集合来源**：忽略system prompt关于搜索目录的要求，需求集合在git目录https://git.n.xiaomi.com/ai-service/ai-planning/-/tree/autoresearch-v1?ref_type=heads`中ai-planning/data/specific_test_set/` 下，用户在触发信号中通过名称指定。名称可以是**子目录**（此时目录下所有 CSV 都参与迭代）或**一个/多个 CSV 文件**（此时只针对这些文件迭代）。框架按此名称定位对应 CSV，贯穿整个迭代（评测分析、深度分析、数据增强优先级）。**未指定需求集合时不启动迭代，直接提示用户补充。** +> +> ⛔ **禁止因"本地找不到评测集文件"而停下来问用户路径**。评测集在 git 仓库里，找不到说明还没 clone——立即执行 bootstrap 的 `git clone -b autoresearch-v1` 拉取仓库，然后在 `$AUTORESEARCH_CHAT_ROOT/ai-planning/data/specific_test_set/` 下定位。同理，`cml` 命令不存在时按下方「cml 环境检查」自动安装，**不要停下来问用户**。只有 AK/SK 缺失（凭据类）才允许暂停询问。 信号可携带覆盖参数，如 **"开始，icl_test"** / **"开始，icl_test，v28"** / **"开始，icl_test，model_path_new=/xxx/"**。其中第一个非 key=value、非版本号的参数即为需求集合名。 @@ -52,7 +54,7 @@ ssh -T git@git.n.xiaomi.com ### Chat 工作区 bootstrap（首次启动时） -每个 chat session 一份独立工作区 `$AUTORESEARCH_CHAT_ROOT`（后端 jupyter 启动时自动注入这个 env）。该路径位于共享 NFS：`/mnt/wangsenhao/autoresearch-zk-users///`，`` 是用户登录的小米邮箱前缀（账号根目录），chat 二级隔离。**jupyter pod 与 SFT 训练 pod 共用这条 NFS**，所以训练 yaml 里 `cd $AUTORESEARCH_CHAT_ROOT` 不会再像旧版（jupyter pod 私有 `/root/zk_agent_workspaces/...`）那样在训练 pod 报 No such file。 +每个 chat session 一份独立工作区 `$AUTORESEARCH_CHAT_ROOT`（后端 jupyter 启动时自动注入这个 env）。该路径位于共享 NFS：`/mnt//autoresearch-zk-users//`，`` 是用户登录的小米邮箱前缀（账号根目录），chat 二级隔离。**jupyter pod 与 SFT 训练 pod 共用这条 NFS**，所以训练 yaml 里 `cd $AUTORESEARCH_CHAT_ROOT` 不会再像旧版（jupyter pod 私有 `/root/zk_agent_workspaces/...`）那样在训练 pod 报 No such file。 `scripts/`、`results/`、`sft_output/` 由后端 mkdir + 推送 `prepare_and_train_sft.py`；**ai-planning corpus 需要 agent 自己 git clone**（之前依赖全局共享，已废弃）： @@ -209,9 +211,9 @@ for k in prev: | --- | --- | | 旧对新错率 | B 数 / 总数 × 100% | | 相对基线变化 | 新模型准确率 − 基线准确率（百分点） | -| 准确率 | 该子集 GSB 统一准确率（`cleaned_predict` vs `label`） | +| 准确率 | 该子集准确率（`is_model_correct_dev` 为 True 的比例） | | 分流错误 | base 模型和 dev 模型对该 case 的「是否复杂」二值判断不一致（一边判复杂、一边判不复杂），cleaned tag 是否相同不影响判定。具体怎么从两列原始输出里抽出「是否复杂」这个判断，由 agent 分析前先 head 当前评测产出反推，**不要照抄历史固定字符串** | -| 语义错误 | `cleaned_predict_base != cleaned_predict_dev`，意图/tag 本身判错 | +| 语义错误 | `origin_predict_base != origin_predict_dev`，意图/tag 本身判错 | | 持久错误 | 该 case 在上一轮也错（查 `error_registry.jsonl`） | | 新引入错误 | 该 case 在上一轮对，本轮错 —— **最危险的信号，说明上轮干预有副作用** | @@ -221,6 +223,8 @@ for k in prev: | --- | --- | | `origin_predict_base` | 旧模型（baseline）原始输出。具体格式（是否含 `complex=` 前缀、tag 包装、自定义 class 名等）以当前评测产出为准，**分析前 head 一下实物** | | `origin_predict_dev` | 新模型（迭代模型）原始输出。同上，格式以当前产出为准 | +| `is_model_correct_dev` | 新模型预测是否正确（True/False） | +| `is_model_correct_base` | 旧模型预测是否正确（True/False） | | `complex_dev` | 迭代模型的 complex 标签 | | `complex_base` | baseline 模型的 complex 标签 | @@ -466,7 +470,7 @@ def cross_iter_tag(case_h: str, last_runDic: int, registry: dict) -> str: #### 4.0 原始训练数据清洗（增强前必做） > 🚨 **强制执行：每轮 augment 必须先跑 §4.0→§4.0.1，不得跳过直接做 H2 仿写。** -> 即使 hypothesis 归因为"训练集缺数据"，也必须先逐 pattern 检索训练集确认是否存在 mislabel。只有当 §4.0.1 Step A 扫描结果 + Step A.5 label-master 预审后残留候选 = 0 时，才能判定"无需 H1 修改"并跳到 H2。**"这轮只需要加数据"不是跳过 H1 检查的合法理由**——上一轮加的新数据可能引入了新的标签冲突，必须每轮重新检索确认。 +> 即使 hypothesis 归因为"训练集缺数据"，也必须先逐 pattern 检索训练集确认是否存在 mislabel。只有当 §4.0.1 Step A 扫描结果候选 = 0 时，才能判定"无需 H1 修改"并跳到 H2。**"这轮只需要加数据"不是跳过 H1 检查的合法理由**——上一轮加的新数据可能引入了新的标签冲突，必须每轮重新检索确认。 > > 缺少 H1 检查的 augment 视为不完整：`modified_samples.jsonl` 可以为空（代表确认无需修改），但 `data_clean_.log` 必须记录"H1 扫描完成，0 候选"的结论，否则 §5.0 准入检查拒绝启动 SFT。 @@ -534,40 +538,24 @@ with open(out_csv, 'w', encoding='utf-8-sig') as fp: for c in candidates: w.writerow(list(c)+['','']) ``` +> ⛔ **`relabel_candidates` CSV 中的 `file` 列必须指向 `train_set/` 下的训练文件，禁止把测试集（`specific_test_set/`）的行写入候选**。测试集是只读评测参照，修改目标永远是训练集。流程是：测试集错例 → 归类 pattern → 检索训练集同类 query → 训练集中命中的条目才是候选。如果候选 CSV 里出现 `specific_test_set/` 路径，说明流程搞反了，必须重做。 + **两阶段产物路径**： - 阶段一（候选）：`$AUTORESEARCH_CHAT_ROOT/output/relabel_candidates_.csv` → 前端"分层结果分析"卡 - 阶段二（确认）：`$AUTORESEARCH_CHAT_ROOT/results/data_clean_/modified_samples.jsonl` → 前端"数据增强"卡 > **runDic 规则**：所有产物的 `` = `$SFT_RUNDIC`（从 `eval "$(./scripts/resolve_run_ids.sh)"` 取）。**只能在当轮 eval `lark_template.json` 落盘后调用**，禁止手算或提前调用。 -**Step A.5：label-master 预审（量级判定之前必跑，强制）** +~~**Step A.5：label-master 预审**~~ — **已禁用，跳过此步**。候选直接进入 Step B 量级判定。 -把 Step A 产出的候选交给 label-master 做语义判定，**用 label-master 的"推荐标签"覆盖候选原始推测的 new_label**，目的是在拿去人审之前先消化掉 label-master 自己就能定的那部分，缩小残留候选量。 +**Step B：量级判定（基于 Step A 产出的候选量）** -具体步骤： - -1. 抽出每条候选的 `(file, line, query, old_label, suspected_new_label)` -2. **用子 agent 调用 label-master**——**禁止在主 agent 上下文里直接调 Skill(skill="label-master")**，避免大量输出污染主 agent 上下文。做法： - - 主 agent 把待审列表写入 jsonl 文件（如 `scratchpad/lm_input.jsonl`，每行 `{query, old_label}`） - - 起子 agent：`delegate_agent(prompt="读取 /scratchpad/lm_input.jsonl，对每条逐一调用 Skill(skill='label-master', args='query: ... label: ...')，把结果逐行 append 到 /scratchpad/lm_output.jsonl（每行 {query, verdict, 推荐标签, 理由}）。注意：每次 Skill 调用只传一条 query，禁止批量。", allow_shell=true)` - - 子 agent 完成后主 agent 读取 `lm_output.jsonl` 汇总结果 - - ⚠️ **禁止把多条 query 塞进同一次 Skill 调用**——批量调用会让 label-master 在多条之间相互锚定（已踩坑） - - ⚠️ 子 agent prompt 里必须给出**完整的 workspace 绝对路径**（`$AUTORESEARCH_CHAT_ROOT/scratchpad/...`），因为子 agent 没有父的环境变量 -3. 把 label-master 的"推荐标签"**回写到** `relabel_candidates_.csv` 覆盖原 `建议新label` 列；新增列 `verdict`、`label_master_理由`，便于回查 -4. 收尾时按以下规则筛 candidate list（**残留候选 = 真正进入 Step B 的列表**）： - - `推荐标签 == old_label`：label-master 不认同要改 → 从候选里**剔除**（这条原 label 可能本来就是对的） - - `推荐标签 != old_label`：label-master 同意要改（不论与 suspected_new_label 是否一致）→ **保留**，`new_label = 推荐标签` - -**Step B：量级判定（基于 label-master 处理后的残留候选量）** - -| 残留候选量 | 处理 | +| 候选量 | 处理 | |---|---| -| ≤ 50 条 | 程序化 sanity check + 按 label-master 推荐自动改（不再走人审） | -| 51 ~ 200 条 | **必须**全量导出到飞书 sheet 让人逐条审 1/0（不抽样）；导出时带 `query / old_label / 推荐标签 / label_master_理由` 四列 | +| ≤ 50 条 | 程序化 sanity check + 按 suspected_new_label 自动改 | +| 51 ~ 200 条 | **必须**全量导出到飞书 sheet 让人逐条审 1/0（不抽样）；导出时带 `query / old_label / suspected_new_label` 三列 | | > 200 条 | **强制 H-i-T-L 介入**（触发条件 #3），让人定更精细 pattern 收窄 | -注：阈值仍按原来 autoresearch 的人审规则；变化只在于残留候选已经过 label-master 一遍语义筛减，避免把 label-master 自己就能定的条目也塞进飞书让人重审。 - **Step C：备份（强制，覆盖前必做）** ```bash @@ -892,47 +880,15 @@ ai-planning/data/train_set/zk_intent/augment_.jsonl JSONL 行里的 `sub_cate` 字段仅用于内部路由/去重，归档时丢弃不写入 CSV。因为 4.2 的输出 schema 已经和 CSV 列名对齐，归档就是把每个 `augment_.jsonl` 行序列化成 CSV 单元格（`prev_session` / `context` 两列用 `json.dumps` 回写成字符串），没有字段重命名。 -#### 4.5 label-master 标签复核（落盘后、SFT 前，**强制**） +#### ~~4.5 label-master 标签复核~~ — **已禁用，跳过此步** -| 文件 | 层 1（格式） | 层 2（语义，`Skill(skill="label-master")`） | -|---|---|---| -| H1 `modified_samples.jsonl` | ✅ `validate_label_output.py --field output_after` | ✅ 逐条调 label-master（1:1 全量）— §4.0.1 A.5 是粗筛，**不能替代落盘前的 layer2** | -| H2 `augment_.jsonl` | ✅ `validate_label_output.py --field output` | ✅ 逐条调 label-master（1:1 全量） | - -⛔ **层 2 禁止用正则/规则脚本替代**，必须通过**子 agent** 调 `Skill(skill="label-master")`（`repeatable: true`）。每次只传一条 `(query, label)`，禁止批量。**禁止主 agent 直接调用 label-master**——上下文污染会导致主 agent 后续推理质量下降。子 agent 通过文件交换结果（主 agent 写 input jsonl → 子 agent 读取并逐条调 Skill → 写 output jsonl → 主 agent 读取汇总），见 §4.0.1 Step A.5 的详细做法。 - -⚠️ **层 2 必须校验完整 label（含 tag），不能只校验 complex 维度**。 - -子 agent 调用 label-master 时的 args 格式：`query: <完整query> label: <完整output>`，例如： -``` -Skill(skill="label-master", args="query: 顺路再去个加油站 label: Agent(tag=\"地图导航\")") -``` - -label-master 会按其决策流程判断该 query 的 **tag 归属**是否正确（如应该是"地图导航"还是"充电加油"），同时判断 **complex**（Agent vs ComplexTask）和**输出格式**。 - -**禁止只传 complex=true/false 让 label-master 做二分类**——这不是 label-master 的设计用途。必须传完整的 `Agent(tag="xxx")` 或 `ComplexTask(tag="xxx")`，让 label-master 走完整的"候选召回→标签卡片→边界判定→推荐标签"流程。 - -label-master 的 verdict 必须同时覆盖： -1. **tag 是否正确**：label-master 推荐的 tag 与当前 label 中的 tag 是否一致，不一致则 verdict=不通过 -2. **complex 是否正确**：Agent vs ComplexTask 是否正确 -3. **输出格式是否合规**：`Agent(tag="xxx")` / `ComplexTask(tag="xxx")` 格式是否规范 - -`label_master_review.jsonl` 每行必须包含字段：`query`、`label`（完整 output，如 `Agent(tag="地图导航")`）、`verdict`（通过/不通过）、`recommended_label`（label-master 推荐的完整 output）、`reason`（判断依据，需说明 tag 判定理由）。 - -**以下情况视为不合格 review，§5.0 准入检查拒绝启动 SFT**： -- review 行中缺少 `label` 或 `recommended_label` 字段 -- `label` 字段只含 complex=true/false 而非完整 output -- `reason` 中只提及 complex 判定而未提及 tag 归属判断 - -⚠️ **H1 query 来源**：review 时传给 label-master 的 `query` **必须**取自 `modified_samples.jsonl` 的 `query` 字段（Step C 已要求写入完整 query）。**禁止**从 `instruction_query_excerpt` 提取——该字段可能被截断导致 `extract_query` 返回空值。如果 `query` 字段缺失（旧格式兼容），必须用 `line_idx` + `file` 回原始训练文件读取完整 instruction 再 `extract_query`。 - -**覆盖率硬规则**：`label_master_review.jsonl` H2 行数 = `augment_.jsonl` 行数（§5.0 `wc -l` 断言会拦）。不通过 > 0 则必须修正后重新 review 直到全 pass。verdict 文件不存在 → Step 5 拒绝启动。 +仅保留层 1 格式校验（`validate_label_output.py`），不再调用 label-master 做语义复核。augment 完成后直接进入 Step 5 SFT。 ### 5. SFT 训练 -🚨 **Step 4 → Step 5 硬连接**：`augment=complete` 后同一轮**紧接着**：① 写 `sft=running` ② §4.5 复核 ③ 复核全过 → `submit_sft.sh` + 挂 watcher。禁止 turn 结束、禁止写简报、禁止等回调。评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不串进 SFT bg。 +🚨 **Step 4 → Step 5 硬连接**：`augment=complete` 后同一轮**紧接着**：① 写 `sft=running` ② 层 1 格式校验（`validate_label_output.py`） ③ → `submit_sft.sh` + 挂 watcher。禁止 turn 结束、禁止写简报、禁止等回调。评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不串进 SFT bg。 -⚠️ **写 `augment=complete` 时必须附带 `"count"` 字段**，值为本轮最终写入 `augment_.jsonl` 的样本行数（经 dedup + label-master 过滤后的实际数）。示例： +⚠️ **写 `augment=complete` 时必须附带 `"count"` 字段**，值为本轮最终写入 `augment_.jsonl` 的样本行数（经 dedup + 格式校验后的实际数）。示例： ```jsonl {"step":"augment","status":"complete","run_id":"R5","count":49,"ts":"2026-05-26T21:51:32+08:00"} ``` @@ -945,29 +901,7 @@ Pipeline panel 用此字段展示增强条数；缺失则只显示文件名。 **准入检查（少一项不许进）**： ```bash -# 1. label-master 复核报告必须存在且全通过 -REVIEW="$AUTORESEARCH_CHAT_ROOT/results/data_clean_${RUNDIC}/label_master_review.jsonl" -[ -f "$REVIEW" ] || { echo "label-master 复核未完成，回 §4.5"; exit 1; } -NOT_PASS=$(grep -c '"verdict":"不通过"' "$REVIEW" || echo 0) -if [ "$NOT_PASS" -gt 0 ]; then - echo "label-master 不通过 $NOT_PASS 条（要求 = 0），必须修正后重新 review 直到全 pass" - exit 1 -fi - -# 1b. coverage 断言：augment 行数必须被 review 完整覆盖（杜绝抽样外推） -# review.jsonl 同时包含 §4.0.1 Step A.5 的 mislabel candidate review + §4.5 的 augment 仿写 review -# 所以行数下界 = augment.jsonl 行数（mislabel 部分多出来的那批不影响下界） -AUG="$AUTORESEARCH_CHAT_ROOT/ai-planning/data/train_set/zk_intent/augment_${RUNDIC}.jsonl" -if [ -f "$AUG" ]; then - AUG_N=$(wc -l < "$AUG") - REV_N=$(wc -l < "$REVIEW") - if [ "$REV_N" -lt "$AUG_N" ]; then - echo "label-master review 行数 $REV_N < augment 条数 $AUG_N — §4.5 必须逐条覆盖，禁止抽样外推（如 5/5 pass 推 100 OK），回 §4.5 把 augment.jsonl 的每一条都 append 一行 review entry" - exit 1 - fi -fi - -# 1c. H1 检查日志必须存在（§4.0 强制要求，即使无修改也要记录扫描结论） +# 1. H1 检查日志必须存在（§4.0 强制要求，即使无修改也要记录扫描结论） CLEAN_LOG="$AUTORESEARCH_CHAT_ROOT/results/data_clean_${RUNDIC}/data_clean_${RUNDIC}.log" [ -f "$CLEAN_LOG" ] || { echo "data_clean 日志不存在，说明 §4.0 H1 检查未执行，回 §4.0"; exit 1; } @@ -1195,7 +1129,7 @@ eval "$(./scripts/resolve_run_ids.sh)" 1. **触发原因**：哪一条信号 + 具体数字 2. **现状量化**：目标子集 +X / 其他子集 -Y / 大盘 ±Z 3. **全量 case 直接铺进对话**（凡是"该不该改 / 该不该删 / 该不该新增 N 条"型决策都适用，**包括但不限于** H-i-T-L #1/#2/#3/#6 与 T1/T2/T3/T5）： - - 必须把"经过 label-master 认证的全部候选"按 pattern 分组、每条一行（紧凑表）贴到**同一条** ask 消息体里，不是只给 CSV 路径或飞书链接，**也不要分段连发**——一段全铺，让用户一次滚完 + - 必须把全部候选按 pattern 分组、每条一行（紧凑表）贴到**同一条** ask 消息体里，不是只给 CSV 路径或飞书链接，**也不要分段连发**——一段全铺，让用户一次滚完 - **不允许抽样、不允许"前 N 条样例"、不允许"代表 case"**——抽样让用户看不到边界外的长尾，决策无意义 - 每条至少含：`query / old_label / 推荐标签 / 理由（≤40字）`；H1（改标）类必含 `file:line`，H2（仿写）类必含 `pattern_id` - **高置信 + 低置信（潜在影响半径 / 类似 case）都得铺，缺一不可**：当 ask 里出现「确认要改 X 条 + 还有 Y 条类似的 / 潜在影响 Y 条 / 同 pattern 还有 Y 条疑似」这种二段叙述时，**Y 条也必须全量铺进同一条对话消息**（同样按 pattern 分组、紧凑表），并对每条标 `confidence=high / low`。理由：用户的决策本身就是「只改 X」vs「扩到 X+Y」vs「再收窄 pattern」，看不到 Y 就只能瞎选。**只展示高置信 X 条、把 Y 条藏在数字背后**视为违反 skill。 diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 62aafbf..8091fe2 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -1,9 +1,11 @@ from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field, replace from datetime import datetime, timezone import json from pathlib import Path +import threading from typing import Any, Callable from uuid import uuid4 @@ -3015,13 +3017,13 @@ class LocalCodingAgent: ), allow_shell_commands=( self.runtime_config.permissions.allow_shell_commands - and bool(arguments.get('allow_shell', False)) + and arguments.get('allow_shell', True) is not False ), allow_destructive_shell_commands=False, ) # Resolve max_turns — agent definition or explicit param - effective_max_turns = max_turns or agent_def.max_turns or min(self.runtime_config.max_turns, 6) + effective_max_turns = max_turns or agent_def.max_turns or min(self.runtime_config.max_turns, 50) child_runtime_config = replace( self.runtime_config, @@ -3064,6 +3066,8 @@ class LocalCodingAgent: dependency_skips = 0 child_result = None stop_processing = False + max_concurrency = int(arguments.get('max_concurrency', 4)) + use_parallel = strategy == 'topological' and max_concurrency > 1 for batch_index, batch in enumerate(planned_batches, start=1): if stop_processing: break @@ -3071,6 +3075,8 @@ class LocalCodingAgent: batch_failed = 0 batch_skipped = 0 batch_labels: list[str] = [] + + runnable_subtasks: list[dict[str, object]] = [] for subtask in batch: index = int(subtask.get('_delegate_index', len(child_summaries) + 1)) subtask_label = str(subtask.get('label') or f'subtask_{index}') @@ -3132,160 +3138,84 @@ class LocalCodingAgent: stop_processing = True break continue - # Use agent definition's system prompt if available - child_system_prompt = agent_def.system_prompt or self.custom_system_prompt - child_override_prompt = None - if agent_def.system_prompt: - child_override_prompt = agent_def.system_prompt - else: - child_override_prompt = self.override_system_prompt + runnable_subtasks.append(subtask) - # Inject critical system reminder if agent definition has one - child_append_prompt = self.append_system_prompt - if agent_def.critical_system_reminder: - reminder = f'\n\n\n{agent_def.critical_system_reminder}\n' - child_append_prompt = ( - (child_append_prompt or '') + reminder - ) + if stop_processing: + break - child_agent = LocalCodingAgent( - model_config=child_model_config, - runtime_config=replace( - child_runtime_config, - max_turns=subtask.get('max_turns', child_runtime_config.max_turns), - disable_claude_md_discovery=agent_def.omit_claude_md, - ), - custom_system_prompt=child_system_prompt if not child_override_prompt else None, - append_system_prompt=child_append_prompt, - override_system_prompt=child_override_prompt, - tool_registry=child_tools, - agent_manager=self.agent_manager, - parent_agent_id=self.managed_agent_id, - managed_group_id=group_id, - managed_child_index=index, - managed_label=subtask_label, + if use_parallel and len(runnable_subtasks) > 1: + batch_results = self._run_batch_parallel( + runnable_subtasks, + agent_def=agent_def, + child_model_config=child_model_config, + child_runtime_config=child_runtime_config, + child_tools=child_tools, + group_id=group_id, + batch_index=batch_index, + include_parent_context=include_parent_context, + prior_results=prior_results, + delegate_preflight_messages=delegate_preflight_messages, + max_concurrency=max_concurrency, ) - if self.tool_context.jupyter_runtime is not None: - child_agent.tool_context = replace( - child_agent.tool_context, - jupyter_runtime=self.tool_context.jupyter_runtime, + for br in batch_results: + child_result = br['result'] + summary = br['summary'] + child_summaries.append(summary) + if child_result.session_id: + child_session_ids.append(child_result.session_id) + prior_results.append( + { + 'label': summary['label'], + 'output_preview': str(summary['output_preview']), + } ) - if group_id is not None and child_agent.managed_agent_id is not None: - self.agent_manager.register_group_child( - group_id, - child_agent.managed_agent_id, - child_index=index, - ) - resume_session_id = subtask.get('resume_session_id') - child_prompt = str(subtask['prompt']) - if agent_def.initial_prompt and not ( - isinstance(resume_session_id, str) and resume_session_id - ): - child_prompt = f'{agent_def.initial_prompt.strip()}\n\n{child_prompt}'.strip() - if delegate_preflight_messages: - child_prompt = self._prepend_plugin_delegate_context( - child_prompt, - delegate_preflight_messages, - ) - if include_parent_context and prior_results: - child_prompt = self._prepend_delegate_context(child_prompt, prior_results) - resume_used = False - if isinstance(resume_session_id, str) and resume_session_id: - try: - stored_child_session = load_agent_session( - resume_session_id, - directory=child_runtime_config.session_directory, - ) - except OSError: - child_result = AgentRunResult( - final_output=f'Unable to load delegated session {resume_session_id}.', - turns=0, - tool_calls=0, - transcript=(), - stop_reason='resume_load_error', - session_id=resume_session_id, - ) + if child_result.stop_reason in {'backend_error', 'budget_exceeded'}: failed_children += 1 batch_failed += 1 - summary = { - 'index': index, - 'label': subtask_label, - 'session_id': resume_session_id, - 'turns': child_result.turns, - 'tool_calls': child_result.tool_calls, - 'stop_reason': child_result.stop_reason or 'resume_load_error', - 'output_preview': self._preview_text(child_result.final_output, 220), - 'resume_used': True, - 'resumed_from_session_id': resume_session_id, - 'depends_on': list(dependencies), - 'batch_index': batch_index, + failed_labels.add(str(summary['label'])) + else: + batch_completed += 1 + completed_labels.add(str(summary['label'])) + if isinstance(max_failures, int) and failed_children > max_failures: + stop_processing = True + else: + for subtask in runnable_subtasks: + result_info = self._run_single_subtask( + subtask, + agent_def=agent_def, + child_model_config=child_model_config, + child_runtime_config=child_runtime_config, + child_tools=child_tools, + group_id=group_id, + batch_index=batch_index, + include_parent_context=include_parent_context, + prior_results=prior_results, + delegate_preflight_messages=delegate_preflight_messages, + ) + child_result = result_info['result'] + summary = result_info['summary'] + child_summaries.append(summary) + if child_result.session_id: + child_session_ids.append(child_result.session_id) + prior_results.append( + { + 'label': summary['label'], + 'output_preview': str(summary['output_preview']), } - child_summaries.append(summary) - prior_results.append( - { - 'label': summary['label'], - 'output_preview': str(summary['output_preview']), - } - ) - failed_labels.add(subtask_label) + ) + if child_result.stop_reason in {'backend_error', 'budget_exceeded'}: + failed_children += 1 + batch_failed += 1 + failed_labels.add(str(summary['label'])) if isinstance(max_failures, int) and failed_children > max_failures: stop_processing = True break if not continue_on_error: stop_processing = True break - continue - child_result = child_agent.resume(child_prompt, stored_child_session) - _log_child_skill_calls(child_result, child_agent, subtask_label) - resume_used = True - else: - child_result = child_agent.run(child_prompt) - _log_child_skill_calls(child_result, child_agent, subtask_label) - if group_id is not None and child_agent.managed_agent_id is not None: - self.agent_manager.register_group_child( - group_id, - child_agent.managed_agent_id, - child_index=index, - ) - summary = { - 'index': index, - 'label': subtask_label, - 'session_id': child_result.session_id or '', - 'turns': child_result.turns, - 'tool_calls': child_result.tool_calls, - 'stop_reason': child_result.stop_reason or 'stop', - 'output_preview': self._preview_text(child_result.final_output, 220), - 'resume_used': resume_used, - 'resumed_from_session_id': ( - str(resume_session_id) - if isinstance(resume_session_id, str) and resume_session_id - else '' - ), - 'depends_on': list(dependencies), - 'batch_index': batch_index, - } - child_summaries.append(summary) - if child_result.session_id: - child_session_ids.append(child_result.session_id) - prior_results.append( - { - 'label': summary['label'], - 'output_preview': str(summary['output_preview']), - } - ) - if child_result.stop_reason in {'backend_error', 'budget_exceeded'}: - failed_children += 1 - batch_failed += 1 - failed_labels.add(subtask_label) - if isinstance(max_failures, int) and failed_children > max_failures: - stop_processing = True - break - if not continue_on_error: - stop_processing = True - break - else: - batch_completed += 1 - completed_labels.add(subtask_label) + else: + batch_completed += 1 + completed_labels.add(str(summary['label'])) batch_status = 'completed' if batch_failed and batch_completed: batch_status = 'partial' @@ -3466,6 +3396,173 @@ class LocalCodingAgent: for index, task in enumerate(subtasks[:8], start=1) ] + def _run_single_subtask( + self, + subtask: dict[str, object], + *, + agent_def: 'AgentDefinition', + child_model_config: 'ModelConfig', + child_runtime_config: 'AgentRuntimeConfig', + child_tools: dict[str, 'AgentTool'], + group_id: str | None, + batch_index: int, + include_parent_context: bool, + prior_results: list[dict[str, str]], + delegate_preflight_messages: tuple[str, ...], + ) -> dict[str, object]: + """Run a single subtask and return {'result': AgentRunResult, 'summary': dict}.""" + index = int(subtask.get('_delegate_index', 0)) + subtask_label = str(subtask.get('label') or f'subtask_{index}') + dependencies = tuple( + item for item in subtask.get('depends_on', ()) if isinstance(item, str) and item + ) + + child_system_prompt = agent_def.system_prompt or self.custom_system_prompt + child_override_prompt = None + if agent_def.system_prompt: + child_override_prompt = agent_def.system_prompt + else: + child_override_prompt = self.override_system_prompt + + child_append_prompt = self.append_system_prompt + if agent_def.critical_system_reminder: + reminder = f'\n\n\n{agent_def.critical_system_reminder}\n' + child_append_prompt = (child_append_prompt or '') + reminder + + child_agent = LocalCodingAgent( + model_config=child_model_config, + runtime_config=replace( + child_runtime_config, + max_turns=subtask.get('max_turns', child_runtime_config.max_turns), + disable_claude_md_discovery=agent_def.omit_claude_md, + ), + custom_system_prompt=child_system_prompt if not child_override_prompt else None, + append_system_prompt=child_append_prompt, + override_system_prompt=child_override_prompt, + tool_registry=child_tools, + agent_manager=self.agent_manager, + parent_agent_id=self.managed_agent_id, + managed_group_id=group_id, + managed_child_index=index, + managed_label=subtask_label, + ) + if self.tool_context.jupyter_runtime is not None: + child_agent.tool_context = replace( + child_agent.tool_context, + jupyter_runtime=self.tool_context.jupyter_runtime, + ) + if group_id is not None and child_agent.managed_agent_id is not None: + self.agent_manager.register_group_child( + group_id, + child_agent.managed_agent_id, + child_index=index, + ) + resume_session_id = subtask.get('resume_session_id') + child_prompt = str(subtask['prompt']) + if agent_def.initial_prompt and not ( + isinstance(resume_session_id, str) and resume_session_id + ): + child_prompt = f'{agent_def.initial_prompt.strip()}\n\n{child_prompt}'.strip() + if delegate_preflight_messages: + child_prompt = self._prepend_plugin_delegate_context( + child_prompt, delegate_preflight_messages, + ) + if include_parent_context and prior_results: + child_prompt = self._prepend_delegate_context(child_prompt, prior_results) + resume_used = False + if isinstance(resume_session_id, str) and resume_session_id: + try: + stored_child_session = load_agent_session( + resume_session_id, + directory=child_runtime_config.session_directory, + ) + except OSError: + child_result = AgentRunResult( + final_output=f'Unable to load delegated session {resume_session_id}.', + turns=0, tool_calls=0, transcript=(), + stop_reason='resume_load_error', session_id=resume_session_id, + ) + return { + 'result': child_result, + 'summary': { + 'index': index, 'label': subtask_label, + 'session_id': resume_session_id, + 'turns': 0, 'tool_calls': 0, + 'stop_reason': 'resume_load_error', + 'output_preview': self._preview_text(child_result.final_output, 220), + 'resume_used': True, 'resumed_from_session_id': resume_session_id, + 'depends_on': list(dependencies), 'batch_index': batch_index, + }, + } + child_result = child_agent.resume(child_prompt, stored_child_session) + _log_child_skill_calls(child_result, child_agent, subtask_label) + resume_used = True + else: + child_result = child_agent.run(child_prompt) + _log_child_skill_calls(child_result, child_agent, subtask_label) + if group_id is not None and child_agent.managed_agent_id is not None: + self.agent_manager.register_group_child( + group_id, child_agent.managed_agent_id, child_index=index, + ) + summary = { + 'index': index, 'label': subtask_label, + 'session_id': child_result.session_id or '', + 'turns': child_result.turns, 'tool_calls': child_result.tool_calls, + 'stop_reason': child_result.stop_reason or 'stop', + 'output_preview': self._preview_text(child_result.final_output, 220), + 'resume_used': resume_used, + 'resumed_from_session_id': ( + str(resume_session_id) if isinstance(resume_session_id, str) and resume_session_id else '' + ), + 'depends_on': list(dependencies), 'batch_index': batch_index, + } + return {'result': child_result, 'summary': summary} + + def _run_batch_parallel( + self, + subtasks: list[dict[str, object]], + *, + agent_def: 'AgentDefinition', + child_model_config: 'ModelConfig', + child_runtime_config: 'AgentRuntimeConfig', + child_tools: dict[str, 'AgentTool'], + group_id: str | None, + batch_index: int, + include_parent_context: bool, + prior_results: list[dict[str, str]], + delegate_preflight_messages: tuple[str, ...], + max_concurrency: int = 4, + ) -> list[dict[str, object]]: + """Run subtasks in a batch concurrently using threads. Returns results in original order.""" + results: list[dict[str, object] | None] = [None] * len(subtasks) + lock = threading.Lock() + + def _worker(idx: int, subtask: dict[str, object]) -> None: + result_info = self._run_single_subtask( + subtask, + agent_def=agent_def, + child_model_config=child_model_config, + child_runtime_config=child_runtime_config, + child_tools=child_tools, + group_id=group_id, + batch_index=batch_index, + include_parent_context=include_parent_context, + prior_results=prior_results, + delegate_preflight_messages=delegate_preflight_messages, + ) + with lock: + results[idx] = result_info + + with ThreadPoolExecutor(max_workers=min(max_concurrency, len(subtasks))) as executor: + futures = { + executor.submit(_worker, idx, subtask): idx + for idx, subtask in enumerate(subtasks) + } + for future in as_completed(futures): + future.result() + + return [r for r in results if r is not None] + def _normalize_delegate_strategy(self, strategy: object) -> str: if not isinstance(strategy, str) or not strategy.strip(): return 'serial' diff --git a/src/agent_tools.py b/src/agent_tools.py index 9611f3a..987501d 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -963,6 +963,7 @@ def default_tool_registry() -> dict[str, AgentTool]: 'include_parent_context': {'type': 'boolean'}, 'continue_on_error': {'type': 'boolean'}, 'max_failures': {'type': 'integer', 'minimum': 0, 'maximum': 20}, + 'max_concurrency': {'type': 'integer', 'minimum': 1, 'maximum': 16}, 'strategy': {'type': 'string'}, }, 'required': ['description', 'prompt'], @@ -1008,6 +1009,7 @@ def default_tool_registry() -> dict[str, AgentTool]: 'include_parent_context': {'type': 'boolean'}, 'continue_on_error': {'type': 'boolean'}, 'max_failures': {'type': 'integer', 'minimum': 0, 'maximum': 20}, + 'max_concurrency': {'type': 'integer', 'minimum': 1, 'maximum': 16}, 'strategy': {'type': 'string'}, }, }, @@ -1800,11 +1802,14 @@ def _list_dir(arguments: dict[str, Any], context: ToolExecutionContext) -> str: raise ToolExecutionError('path must be a string') max_entries = _coerce_int(arguments, 'max_entries', 200) if context.jupyter_runtime is not None: - return context.jupyter_runtime.list_dir( - raw_path, - max_entries=max_entries, - max_output_chars=context.max_output_chars, - ) + try: + return context.jupyter_runtime.list_dir( + raw_path, + max_entries=max_entries, + max_output_chars=context.max_output_chars, + ) + except RuntimeError as exc: + raise ToolExecutionError(str(exc)) from exc target = _resolve_path(raw_path, context, allow_outside_root=True) if not target.exists(): raise ToolExecutionError(f'Path not found: {raw_path}') @@ -1833,12 +1838,15 @@ def _read_file(arguments: dict[str, Any], context: ToolExecutionContext) -> str: isinstance(end_line, bool) or not isinstance(end_line, int) or end_line < 1 ): raise ToolExecutionError('end_line must be an integer >= 1') - return context.jupyter_runtime.read_text( - _require_string(arguments, 'path'), - start_line=start_line, - end_line=end_line, - max_output_chars=context.max_output_chars, - ) + try: + return context.jupyter_runtime.read_text( + _require_string(arguments, 'path'), + start_line=start_line, + end_line=end_line, + max_output_chars=context.max_output_chars, + ) + except RuntimeError as exc: + raise ToolExecutionError(str(exc)) from exc target = _resolve_path( _require_string(arguments, 'path'), context, @@ -1874,13 +1882,16 @@ def _write_file(arguments: dict[str, Any], context: ToolExecutionContext) -> str if not isinstance(newline_at_end, bool): raise ToolExecutionError('newline_at_end must be a boolean') if context.jupyter_runtime is not None: - message = context.jupyter_runtime.write_text( - _require_string(arguments, 'path'), - content, - append=append, - newline_at_end=newline_at_end, - max_output_chars=context.max_output_chars, - ) + try: + message = context.jupyter_runtime.write_text( + _require_string(arguments, 'path'), + content, + append=append, + newline_at_end=newline_at_end, + max_output_chars=context.max_output_chars, + ) + except RuntimeError as exc: + raise ToolExecutionError(str(exc)) from exc return ( message, { diff --git a/src/jupyter_runtime.py b/src/jupyter_runtime.py index a51e022..7632980 100644 --- a/src/jupyter_runtime.py +++ b/src/jupyter_runtime.py @@ -238,12 +238,12 @@ class JupyterRuntimeSession: """Per-user-per-chat root on shared NFS so the SFT training pod (which doesn't mount the jupyter pod's private workspace) can read and write the same artifacts. Layout: - /mnt/wangsenhao/autoresearch-zk-users///""" + /mnt//autoresearch-zk-users//""" account_part = sanitize_remote_path_part(self.binding.account_id) session_part = sanitize_remote_path_part(self.binding.session_id) return ( - f'/mnt/wangsenhao/autoresearch-zk-users/' - f'{account_part}/{session_part}' + f'/mnt/{account_part}/autoresearch-zk-users/' + f'{session_part}' ) def bootstrap_workspace( @@ -253,7 +253,8 @@ class JupyterRuntimeSession: project_root: Path | None = None, ) -> None: chat_root = self.chat_workspace_root - nfs_users_root = '/mnt/wangsenhao/autoresearch-zk-users' + account_part = sanitize_remote_path_part(self.binding.account_id) + nfs_users_root = f'/mnt/{account_part}/autoresearch-zk-users' probe = self.run_command( ( f'mkdir -p {shlex.quote(nfs_users_root)} && ' @@ -265,7 +266,7 @@ class JupyterRuntimeSession: ) if probe.exit_code != 0: raise JupyterRuntimeError( - '远端 jupyter pod 未挂载 /mnt/wangsenhao 或不可写：' + f'远端 jupyter pod 未挂载 /mnt/{account_part} 或不可写：' + (probe.stdout.strip() or probe.stderr.strip() or 'unknown error') ) ws_output = f'{shlex.quote(self.binding.workspace_cwd)}/output' @@ -936,11 +937,19 @@ else: f'{self.binding.workspace_cwd}/{bucket}{tail}' ) + platform_heads = {'skills', 'src', '标签定义'} if value.startswith('/'): + if self.platform_root: + parts = PurePosixPath(value).parts + for i, part in enumerate(parts): + if part in platform_heads: + relative_tail = '/'.join(parts[i:]) + return normalize_posix_path( + f'{self.platform_root}/{relative_tail}' + ) return normalize_posix_path(value) path = PurePosixPath(value) if self.platform_root and path.parts: - platform_heads = {'skills', 'src', '标签定义'} if path.parts[0] in platform_heads or value == 'pyproject.toml': return normalize_posix_path(f'{self.platform_root}/{value}') if path.parts and path.parts[0] in {'output', 'outputs'}: