From 85ce8c0e692301573747e9fabd32a2acb6b1449d Mon Sep 17 00:00:00 2001 From: wangsenhao Date: Wed, 27 May 2026 17:02:24 +0800 Subject: [PATCH] fix: session resume, config_set schema, metrics chart, subagent logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix session resume: remove messages.length>1 guard so sessionStorage ID is used as resumeSessionId on first message (prevents new session creation when user sends "继续") - Fix config_set tool schema: add missing `items: {}` to array type in oneOf (Azure OpenAI strict validation rejects it otherwise) - Metrics chart: use actual target set filename from trigger text instead of generic "target_subset" key - Add sub-agent skill call logging to agent_runtime.py (full prompt/response, no truncation) - Various metric enrichment and dedup fixes in server.py Co-Authored-By: Claude Opus 4.6 --- backend/api/server.py | 192 ++++++++++++++++-- frontend/app/app/assistant.tsx | 3 +- .../training/metrics-chart-panel.tsx | 5 + skills/model-iteration/SKILL.md | 44 +++- skills/model-iteration/references/program.md | 74 ++++++- src/agent_runtime.py | 120 +++++++++++ src/agent_tools.py | 2 +- src/jupyter_runtime.py | 11 +- 8 files changed, 411 insertions(+), 40 deletions(-) diff --git a/backend/api/server.py b/backend/api/server.py index bf357bc..2e57dbc 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -1751,11 +1751,17 @@ def create_app(state: AgentState) -> FastAPI: if isinstance(run_id, str) and run_id.strip(): rid = run_id.strip() per_round = _resolve_run_dic_per_round(state_entries) - # augment 产物的 runDic = resolve_run_ids.sh 在当轮 eval 落盘后取到的 - # max(已落盘 workflow) = 当轮 eval 自身的 workflow id，即 per_round[R{n}] - # （不是 R{n-1}，脚本取的是 eval 之后的 max，此时 max 已经包含当轮 eval） + # 正确标号: R{n}·Train 和 R{n}·Analysis 共用同一 run_id， + # per_round[R{n}] 从 Analysis 的 watch/log 中解析到 runDic。 + # 兼容旧 session 错误标号: agent 每阶段递增 run_id（R7=train, R6=analysis）， + # 导致 per_round[R_train] = None，需 fallback 到 R{n-1}。 if step == 'augment': - pass # 直接走下面 per_round.get(rid) 取本轮 runDic + run_dic = per_round.get(rid) + if run_dic is None: + run_index = _parse_run_index(rid) + if run_index is not None and run_index >= 1: + prev_rid = f'R{run_index - 1}' + run_dic = per_round.get(prev_rid) if run_dic is None: run_dic = per_round.get(rid) if run_dic is None: @@ -2001,7 +2007,14 @@ 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, + ) state_entries, _ = _read_program_state(state_path) + trigger = _extract_trigger_from_state(state_entries) or _extract_trigger_from_session( + sessions_dir, session_id + ) + _target_set_name = _parse_target_set(trigger, model_config=state.model_config_for(account_id)) iter_count = await asyncio.to_thread( _read_iteration_log_count, session_id, @@ -2013,6 +2026,7 @@ def create_app(state: AgentState) -> FastAPI: session_id, agent_state=state, account_id=account_id, + target_set_name=_target_set_name, ) payload = await asyncio.to_thread( _hardcoded_autoresearch_pipeline, @@ -2060,6 +2074,12 @@ def create_app(state: AgentState) -> FastAPI: async def event_gen(): last_signature: str | None = None last_heartbeat = time.monotonic() + last_remote_sync = 0.0 + _state_entries_init, _ = _read_program_state(state_path) + _trigger_init = _extract_trigger_from_state(_state_entries_init) or _extract_trigger_from_session( + sessions_dir, session_id + ) + _stream_target_set = _parse_target_set(_trigger_init, model_config=state.model_config_for(account_id)) refresher = asyncio.create_task( _iter_count_refresh_loop( session_id, @@ -2072,12 +2092,19 @@ def create_app(state: AgentState) -> FastAPI: session_id, agent_state=state, account_id=account_id, + target_set_name=_stream_target_set, ) ) try: while True: if await request.is_disconnected(): 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, + ) + last_remote_sync = now_mono state_entries, state_mtime = _read_program_state(state_path) iter_count = await asyncio.to_thread( _read_iteration_log_count, @@ -2090,6 +2117,7 @@ def create_app(state: AgentState) -> FastAPI: session_id, agent_state=state, account_id=account_id, + target_set_name=_stream_target_set, ) payload = await asyncio.to_thread( _hardcoded_autoresearch_pipeline, @@ -5714,11 +5742,43 @@ def _sync_remote_program_state( new_content = result.stdout or '' if not new_content.strip(): return - # Skip rewrite if local already matches remote (avoid touching mtime, which + # Preserve local-only synthetic gate entries that remote doesn't have. + local_synthetics: list[str] = [] + try: + if local_state_path.is_file(): + # Parse remote gate keys once + remote_gate_keys: set[str] = set() + for raw_line in new_content.splitlines(): + if not raw_line.strip(): + continue + try: + r = json.loads(raw_line) + if r.get('step') and not r.get('_synthetic'): + remote_gate_keys.add(f"{r.get('run_id','')}:{r.get('step','')}") + except (json.JSONDecodeError, ValueError): + pass + for line in local_state_path.read_text(encoding='utf-8').splitlines(): + if '"_synthetic"' not in line: + continue + try: + obj = json.loads(line) + if obj.get('_synthetic'): + gate_key = f"{obj.get('run_id','')}:{obj.get('step','')}" + if gate_key not in remote_gate_keys: + local_synthetics.append(line) + except (json.JSONDecodeError, AttributeError): + pass + except OSError: + pass + merged = new_content.rstrip('\n') + if local_synthetics: + merged += '\n' + '\n'.join(local_synthetics) + merged += '\n' + # Skip rewrite if local already matches (avoid touching mtime, which # SSE uses to detect changes). try: if local_state_path.is_file() and ( - local_state_path.read_text(encoding='utf-8') == new_content + local_state_path.read_text(encoding='utf-8') == merged ): return except OSError: @@ -5726,7 +5786,7 @@ def _sync_remote_program_state( try: local_state_path.parent.mkdir(parents=True, exist_ok=True) tmp = local_state_path.with_suffix('.jsonl.sync-tmp') - tmp.write_text(new_content, encoding='utf-8') + tmp.write_text(merged, encoding='utf-8') tmp.replace(local_state_path) except OSError: return @@ -6055,7 +6115,7 @@ _STEP_KIND: dict[str, str] = { 'dist-analysis': 'analysis', 'hypothesis': 'analysis', 'log': 'analysis', - 'augment': 'train', + 'augment': 'analysis', 'verify': 'train', 'sft': 'train', 'human-check': 'gate', @@ -6083,8 +6143,10 @@ _CARD_META: dict[tuple[str, str], dict[str, Any]] = { ('dist-analysis', 'baseline'): {'icon': 'trending-up', 'title': '分层结果分析 & 报告', 'subtitle': '根因归类 + workflow.md'}, ('hypothesis', 'baseline'): {'icon': 'dna', 'title': '形成假设', 'subtitle': 'iteration_log hypothesis 字段'}, ('log', 'baseline'): {'icon': 'clipboard-list', 'title': '记录迭代日志', 'subtitle': 'iteration_log.jsonl'}, + # augment — lives under previous round's analysis/baseline + ('augment', 'baseline'): {'icon': 'flask', 'title': '数据增强', 'subtitle': 'augment_.jsonl'}, + ('augment', 'analysis'): {'icon': 'flask', 'title': '数据增强', 'subtitle': 'augment_.jsonl'}, # R{n}·Train - ('augment', 'train'): {'icon': 'flask', 'title': '数据增强', 'subtitle': 'augment_.jsonl'}, ('verify', 'train'): {'icon': 'shield', 'title': '修改返回验证', 'subtitle': 'sanity + 人审'}, ('sft', 'train'): {'icon': 'graduation-cap','title': 'SFT 训练', 'subtitle': 'submit_sft_via_cml.sh'}, # R{n}·Analysis @@ -6103,16 +6165,17 @@ _CARD_ORDER: dict[tuple[str, str], int] = { ('dist-analysis', 'baseline'): 2, ('hypothesis', 'baseline'): 3, ('log', 'baseline'): 4, + ('augment', 'baseline'): 5, # train - ('augment', 'train'): 0, - ('verify', 'train'): 1, - ('sft', 'train'): 2, + ('verify', 'train'): 0, + ('sft', 'train'): 1, # analysis ('cml', 'analysis'): 0, ('gold-drift', 'analysis'): 1, ('dist-analysis', 'analysis'): 2, ('hypothesis', 'analysis'): 3, ('log', 'analysis'): 4, + ('augment', 'analysis'): 5, } _SECTION_LABELS: dict[str, str] = { @@ -6253,9 +6316,9 @@ def _derive_section_status(card_statuses: list[str]) -> str: return 'pending' if all(s == 'complete' for s in card_statuses): return 'complete' - if 'failed' in card_statuses and not any(s in ('running', 'complete') for s in card_statuses): + if 'failed' in card_statuses and not any(s in ('running', 'complete', 'waiting') for s in card_statuses): return 'failed' - if 'running' in card_statuses or 'complete' in card_statuses: + if any(s in ('running', 'complete', 'waiting') for s in card_statuses): return 'running' return 'pending' @@ -6315,6 +6378,11 @@ def _build_pipeline_items( if not run_id: continue last_non_gate_run_id = run_id + # augment belongs to the previous round's analysis/baseline section + if step == 'augment': + aug_index = _parse_run_index(run_id) + if aug_index is not None and aug_index >= 1: + run_id = f'R{aug_index - 1}' run_index = _parse_run_index(run_id) if run_index is None: continue @@ -6375,6 +6443,11 @@ def _build_pipeline_items( card['progress'] = max(0.0, min(1.0, value)) except (TypeError, ValueError): pass + if step == 'augment' and 'count' in entry: + try: + card['subtitle'] = f'{int(entry["count"])} 条新增样本' + except (TypeError, ValueError): + pass sections[section_key]['cards'].append(card) # 3. Sort cards within each section. @@ -6451,7 +6524,7 @@ def _build_pipeline_items( 'summary': entry.get('summary'), 'proposal': entry.get('proposal'), 'ask': entry.get('ask'), - '_order_key': (run_index, 99, gate_position_index.get(gate_key, 0)), + '_order_key': entry.get('ts', ''), }) # 6. Filter: drop empty/pending sections (no cards or all pending). Running @@ -6589,12 +6662,18 @@ def _compute_in_flight( progress = max(0.0, min(1.0, value)) except (TypeError, ValueError): pass + subtitle = meta['subtitle'] + if step == 'augment' and 'count' in entry: + try: + subtitle = f'{int(entry["count"])} 条新增样本' + except (TypeError, ValueError): + pass cards.append({ 'key': f'{target_run_id}:{step}', 'step': step, 'icon': meta['icon'], 'title': meta['title'], - 'subtitle': meta['subtitle'], + 'subtitle': subtitle, 'status': status, 'progress': progress, }) @@ -6679,7 +6758,7 @@ def _maybe_inject_gate_fallback( 'status': 'running', 'run_id': run_id or 'R0', 'reason': last_text[-300:], - 'ts': datetime_utc_iso(), + 'ts': time.strftime('%Y-%m-%dT%H:%M:%S+08:00', time.localtime()), '_synthetic': True, } try: @@ -7033,6 +7112,7 @@ def _read_iteration_log_metrics_history( *, agent_state: 'AgentState | None' = None, account_id: str | None = None, + target_set_name: str | None = None, ) -> dict[str, Any] | None: """Cached read of full iteration_log.jsonl as metrics time-series.""" cache_key = _iter_count_cache_key(session_id, account_id) @@ -7040,6 +7120,7 @@ def _read_iteration_log_metrics_history( return _METRICS_HISTORY_CACHE[cache_key] value = _read_iteration_log_metrics_history_uncached( session_id, agent_state=agent_state, account_id=account_id, + target_set_name=target_set_name, ) _METRICS_HISTORY_CACHE[cache_key] = value return value @@ -7050,6 +7131,7 @@ async def _metrics_history_refresh_loop( *, agent_state: 'AgentState | None', account_id: str | None, + target_set_name: str | None = None, interval_seconds: float = 3.0, ) -> None: cache_key = _iter_count_cache_key(session_id, account_id) @@ -7060,6 +7142,7 @@ async def _metrics_history_refresh_loop( session_id, agent_state=agent_state, account_id=account_id, + target_set_name=target_set_name, ) _METRICS_HISTORY_CACHE[cache_key] = value except Exception: # noqa: BLE001 @@ -7072,6 +7155,7 @@ def _read_iteration_log_metrics_history_uncached( *, agent_state: 'AgentState | None' = None, account_id: str | None = None, + target_set_name: str | None = None, ) -> dict[str, Any] | None: """Parse iteration_log.jsonl into {'metrics': [keys], 'rounds': [...]}. @@ -7164,8 +7248,70 @@ def _read_iteration_log_metrics_history_uncached( 'values': values, }) + # Deduplicate: keep only the last entry per (iteration, runDic). + # Agent may write multiple entries for the same round (e.g. re-running + # dist-analysis); the last one has the most complete metrics. + seen_keys: dict[tuple[int, int], int] = {} + for idx, r in enumerate(rounds): + key = (r['iteration'], r['runDic']) + seen_keys[key] = idx + rounds = [rounds[i] for i in sorted(seen_keys.values())] + + # Rebuild metrics_order from surviving rounds only (dedup may have removed + # entries whose keys shouldn't appear in the chart). + metrics_order = [] + metrics_seen = set() + for r in rounds: + for k in r['values']: + if k not in metrics_seen: + metrics_seen.add(k) + metrics_order.append(k) + if not rounds: return None + + # Enrich each round with KPI-equivalent metrics from lark_template.json. + # This ensures dapan_car / specific_test / target_subset are always present + # even if the agent omitted them from iteration_log. + target_key = target_set_name if target_set_name and target_set_name != '—' else 'target_subset' + _KPI_METRIC_MAP = { + 'specific_test_pass_rate': 'specific_test', + 'overall_car_pass_rate': 'dapan_car', + 'target_set_pass_rate': target_key, + } + # Rename agent-written 'target_subset' to the actual target set name first, + # so enrichment doesn't create duplicates. + if target_key != 'target_subset': + if 'target_subset' in metrics_seen: + metrics_order = [target_key if m == 'target_subset' else m for m in metrics_order] + metrics_seen.discard('target_subset') + metrics_seen.add(target_key) + for r in rounds: + if 'target_subset' in r['values']: + r['values'][target_key] = r['values'].pop('target_subset') + + for r in rounds: + run_dic = r.get('runDic') + if not isinstance(run_dic, int) or run_dic == 0: + continue + metric_diff = _read_workflow_metric_diff( + run_dic, + agent_state=agent_state, + account_id=account_id, + session_id=session_id, + ) + if metric_diff is None: + continue + _enrich_metrics_from_eval_output(metric_diff, None) + for src_key, dst_key in _KPI_METRIC_MAP.items(): + if dst_key not in r['values']: + raw = metric_diff.get(src_key) + if isinstance(raw, (int, float)): + r['values'][dst_key] = round(raw * 100, 2) + if dst_key not in metrics_seen: + metrics_seen.add(dst_key) + metrics_order.append(dst_key) + return {'metrics': metrics_order, 'rounds': rounds} @@ -7385,7 +7531,8 @@ def _parse_target_set( ) -> str | None: """Sync, non-blocking. Reads cache only — never makes the LLM call inline. Cache is filled by the watcher scanner background loop via - _refresh_target_set_cache_blocking (which is run in a thread).""" + _refresh_target_set_cache_blocking (which is run in a thread). + Falls back to regex extraction of .csv filenames from trigger text.""" if not trigger: return None text = trigger.strip() @@ -7394,7 +7541,14 @@ def _parse_target_set( if len(text) > 4000: text = text[:4000] cache_key = hashlib.sha256(text.encode('utf-8')).hexdigest() - return _TARGET_SET_LLM_CACHE.get(cache_key) + cached = _TARGET_SET_LLM_CACHE.get(cache_key) + if cached is not None: + return cached + # Regex fallback: extract .csv filename from trigger text. + m = re.search(r'(?:(?<=[的/，,\s])|(?<=^))([^\s，,。；的中是在]+\.csv)', text) + if m: + return m.group(1) + return None def _refresh_target_set_cache_blocking( diff --git a/frontend/app/app/assistant.tsx b/frontend/app/app/assistant.tsx index 8e3c00f..e21b107 100644 --- a/frontend/app/app/assistant.tsx +++ b/frontend/app/app/assistant.tsx @@ -66,8 +66,7 @@ export const Assistant = ({ initialSessionId }: AssistantProps = {}) => { const lastSessionId = getLastSessionId(messages); const pendingWorkspaceSessionId = messages.length <= 1 ? readPendingWorkspaceSessionId() : null; - const selectedResumeSessionId = - messages.length > 1 ? selectedSessionId : null; + const selectedResumeSessionId = selectedSessionId; const outgoingSessionId = lastSessionId ?? selectedResumeSessionId ?? diff --git a/frontend/app/components/training/metrics-chart-panel.tsx b/frontend/app/components/training/metrics-chart-panel.tsx index ef1d465..642979e 100644 --- a/frontend/app/components/training/metrics-chart-panel.tsx +++ b/frontend/app/components/training/metrics-chart-panel.tsx @@ -28,9 +28,14 @@ const METRIC_LABELS: Record = { req_set_car: "需求集合(车载)", dapan_car: "大盘车载", specific_test: "Specific Test", + target_subset: "目标集合", + icl_test: "ICL Test", + overall: "大盘整体", triage_err_rate: "Triage 错误率", bvt_nav: "导航BVT", talkable_controllable: "可聊可控", + badcase_2025: "Badcase 2025", + autotask: "自动任务", target_subset_count: "目标子集数量", target_subset_wrong: "目标子集错误数", target_subset_correct: "目标子集正确数", diff --git a/skills/model-iteration/SKILL.md b/skills/model-iteration/SKILL.md index e062dac..657fb30 100644 --- a/skills/model-iteration/SKILL.md +++ b/skills/model-iteration/SKILL.md @@ -87,8 +87,8 @@ when_to_use: | - [ ] §4.0.1 Step B：量级判定基于 label-master 处理后的**残留候选量**走对应支线（≤50 自动按 label-master 推荐改 / 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 `modified_samples.jsonl` 只跑层 1（格式）—— 语义已在 §4.0.1 Step A.5 完成；H2 `augment_.jsonl` 跑两层（`validate_label_output.py` 格式 + Skill 调用 `label-master` 语义），**逐条 1:1 全量覆盖**（`augment` 多少行 `label_master_review.jsonl` H2 部分就多少行，禁止抽样 / spot-check / X/X pass 外推），verdict 写 `results/data_clean_/label_master_review.jsonl`，**不通过必须 = 0**（任何不通过必须修正后重新 review 直到全 pass 才能进 Step 5） -- [ ] **§4.5 层 2 语义复核必须调用 `Skill(skill="label-master")`，禁止正则/规则脚本替代**：层 2 的本质是"用 label-master 知识体系对每条 (query, label) 做独立语义判定"。label-master 已标记 `repeatable: true`，可在同一 run 内多次调用。**禁止**：纯正则/关键词匹配脚本、"target 都一样所以直接 pass"逻辑、批量写 pass 不看 query 内容。已踩坑：R1 agent 写了个正则脚本充当"layer 2"，完全没走语义判断。 +- [ ] **§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 内容。 ### 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 里串接评测） @@ -128,17 +128,45 @@ when_to_use: | > echo '{"step":"cml","status":"complete","ts":"'$(date -Iseconds)'"}' >> "$SESSION_OUTPUT/program-state.jsonl" > ``` -### R1.5 — 推荐每条 state entry 带 run_id（让 UI 准确归位 round） +### R1.5 — 每条 state entry 必须带 run_id（让 UI 准确归位 round） -新版 UI 按 R0 / R1 / R2 切分 sections。**强烈建议**每行加 `run_id`： +新版 UI 按 R0 / R1 / R2 切分 sections。每个 R{n>=1} 同时包含 **Train** 和 **Analysis** 两个 section。**必须**每行带 `run_id`。 + +⚠️ **核心规则：run_id 只在 hypothesis=complete 之后递增。** 一个完整迭代（augment → sft → eval → analysis → hypothesis）全程使用同一个 run_id。 + +完整示例（一轮完整迭代 R1）： ```bash -echo '{"step":"cml","status":"running","run_id":"R0","ts":"'$(date -Iseconds)'"}' >> $S -echo '{"step":"hypothesis","status":"complete","run_id":"R0","ts":"..."}' >> $S # hypothesis 跟当前轮（R0），不是 R1 -echo '{"step":"augment","status":"running","run_id":"R1","ts":"..."}' >> $S # augment 才是 R1·Train 起点 +# ─── R0·Baseline ─── +echo '{"step":"cml","status":"running","run_id":"R0"}' >> $S +echo '{"step":"dist-analysis","status":"complete","run_id":"R0"}' >> $S +echo '{"step":"hypothesis","status":"complete","run_id":"R0"}' >> $S +# ← hypothesis=complete → 递增 run_id + +# ─── R1·Train ─── +echo '{"step":"augment","status":"running","run_id":"R1"}' >> $S +echo '{"step":"augment","status":"complete","run_id":"R1","count":80}' >> $S +echo '{"step":"sft","status":"running","run_id":"R1"}' >> $S +echo '{"step":"sft","status":"complete","run_id":"R1"}' >> $S + +# ─── R1·Analysis（SFT 后的 eval 仍然是 R1，不是 R2！）─── +echo '{"step":"cml","status":"running","run_id":"R1"}' >> $S +echo '{"step":"cml","status":"complete","run_id":"R1"}' >> $S +echo '{"step":"dist-analysis","status":"complete","run_id":"R1"}' >> $S +echo '{"step":"hypothesis","status":"complete","run_id":"R1"}' >> $S +# ← hypothesis=complete → 递增 run_id + +# ─── R2·Train ─── +echo '{"step":"augment","status":"running","run_id":"R2"}' >> $S +# ... ``` -run_id 缺省时后端会按 `iteration_log.jsonl` 行数自动推断（analysis 类 step 含 hypothesis → R{count}，train 类 step augment/verify/sft → R{count+1}），但显式写更准——尤其是并发跨轮、补写历史 entry、或要在 R0 强制注入 baseline 时。 +**错误（禁止）**：eval 后递增 run_id，导致 train 和 analysis 分属不同 round： +``` +R1: augment+sft → R2: cml+analysis+hypothesis → R3: augment+sft → R4: cml+analysis ... +``` + +**为什么重要**：后端通过 `per_round[R{n}]` 从 R{n}·Analysis 的 watch/log entry 解析 runDic，再用这个 runDic 去找 R{n}·Train 的 augment 产物文件。如果 train 和 analysis 用了不同 run_id，augment 的产物文件找不到。 ### R2 — 写入即生效，最后一行覆盖前面 diff --git a/skills/model-iteration/references/program.md b/skills/model-iteration/references/program.md index 2a9f5fb..ea74504 100644 --- a/skills/model-iteration/references/program.md +++ b/skills/model-iteration/references/program.md @@ -465,6 +465,11 @@ 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 检查的合法理由**——上一轮加的新数据可能引入了新的标签冲突，必须每轮重新检索确认。 +> +> 缺少 H1 检查的 augment 视为不完整：`modified_samples.jsonl` 可以为空（代表确认无需修改），但 `data_clean_.log` 必须记录"H1 扫描完成，0 候选"的结论，否则 §5.0 准入检查拒绝启动 SFT。 + 数据增强不仅仅是加数据，还必须对原数据集中的错误/不一致标签进行清洗，否则新增数据和旧数据矛盾，模型学不好。 **清洗原则（case 驱动，逐类分析）**： @@ -542,9 +547,12 @@ with open(out_csv, 'w', encoding='utf-8-sig') as fp: 具体步骤： 1. 抽出每条候选的 `(file, line, query, old_label, suspected_new_label)` -2. **逐条**调 `Skill(skill="label-master", args=...)`——每次只传**一条** `(query, old_label)`（**不**给 suspected_new_label，避免锚定），让 label-master 按 §决策流程 / §候选召回索引 / §高频混淆边界 输出 → `{verdict: 通过 | 不通过, 推荐标签, 排除理由, 易混淆边界}`。 - - ⚠️ **禁止把多条 query 塞进同一次 Skill 调用**——批量调用会让 label-master 在多条之间相互锚定，分错率显著升高（已踩坑） - - 可在同一轮 assistant response 里并发多次 Skill 调用（建议 ≤ 8 路并发）以提升吞吐 +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 可能本来就是对的） @@ -571,7 +579,13 @@ done 同时落归档到 `results/data_clean_/`： - `deleted_samples.jsonl`：被删样本原文 -- `modified_samples.jsonl`：被改样本（含 before/after output） +- `modified_samples.jsonl`：被改样本，**必须包含以下字段**： + - `line_idx`：原文件行号 + - `file`：原文件路径 + - `query`：**完整 query**（用 `extract_query(instruction)` 从原始 instruction 提取，**禁止截断**） + - `old_output`：修改前 output + - `new_output`：修改后 output + - 可选：`instruction_query_excerpt`（仅供人工快速浏览，允许截断，但**不得作为 §4.5 复核的 query 来源**） - `data_clean_.log`：摘要 + 影响 pattern + 样本数 **Step D：修改执行（按文件批量，避免重复读写）** @@ -882,10 +896,35 @@ JSONL 行里的 `sub_cate` 字段仅用于内部路由/去重，归档时丢弃 | 文件 | 层 1（格式） | 层 2（语义，`Skill(skill="label-master")`） | |---|---|---| -| H1 `modified_samples.jsonl` | ✅ `validate_label_output.py --field output_after` | ❌ 已在 §4.0.1 A.5 完成 | +| 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 禁止用正则/规则脚本替代**，必须调 `Skill(skill="label-master")`（`repeatable: true`）。每次只传一条 `(query, label)`，禁止批量。 +⛔ **层 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 拒绝启动。 @@ -893,6 +932,12 @@ JSONL 行里的 `sub_cate` 字段仅用于内部路由/去重，归档时丢弃 🚨 **Step 4 → Step 5 硬连接**：`augment=complete` 后同一轮**紧接着**：① 写 `sft=running` ② §4.5 复核 ③ 复核全过 → `submit_sft.sh` + 挂 watcher。禁止 turn 结束、禁止写简报、禁止等回调。评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不串进 SFT bg。 +⚠️ **写 `augment=complete` 时必须附带 `"count"` 字段**，值为本轮最终写入 `augment_.jsonl` 的样本行数（经 dedup + label-master 过滤后的实际数）。示例： +```jsonl +{"step":"augment","status":"complete","run_id":"R5","count":49,"ts":"2026-05-26T21:51:32+08:00"} +``` +Pipeline panel 用此字段展示增强条数；缺失则只显示文件名。 + 使用 `prepare_and_train_sft.py` 完成数据组装和训练。 #### 5.0 环境准备 @@ -922,6 +967,10 @@ if [ -f "$AUG" ]; then fi fi +# 1c. 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; } + # 2. zk_trainer 仓库已 clone（URL 写死，不要换命名空间，clone 失败先看 ssh key） cd "$AUTORESEARCH_CHAT_ROOT" [ -d zk_trainer ] || git clone git@git.n.xiaomi.com:wangsenhao/zk_trainer.git @@ -1029,6 +1078,14 @@ eval "$(./scripts/resolve_run_ids.sh)" 每轮写入 `results/iteration_log.jsonl` 一行，schema： +> 🚨 **`results` 字段必须包含以下固定指标（每轮都要，不可遗漏）**： +> - `specific_test`：Specific Test 准确率 +> - `dapan_car`：大盘车载准确率 +> - `target_subset`：目标集合准确率（= 当前需求子集） +> - `icl_test`：ICL Test 准确率 +> +> 这些指标从 `lark_template.json` 的 metric_diff 中提取。如果某个指标在评测结果中确实不存在（如首次评测缺少某子集），写 `null` 而不是省略 key——前端 metrics 折线图依赖每轮 key 的一致性，缺 key 会导致数据点丢失。 + ```json { "iteration": 12, @@ -1041,8 +1098,9 @@ eval "$(./scripts/resolve_run_ids.sh)" }, "prediction": {"req_set_car": 93.0, "triage_err_rate": 0.15}, "results": { - "req_set_car": 92.1, "dapan_car": 96.30, - "specific_test": 95.10, "triage_err_rate": 0.18 + "specific_test": 95.10, "dapan_car": 96.30, + "target_subset": 92.1, "icl_test": 86.11, + "triage_err_rate": 0.18 }, "verdict": "partial", "root_cause_findings": [ diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 7db63f8..62aafbf 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -154,6 +154,119 @@ def _prepend_runtime_context(prompt: str, runtime_context: str | None) -> str: ) +def _log_child_skill_calls( + child_result: 'AgentRunResult', + child_agent: 'LocalCodingAgent', + subtask_label: str, +) -> None: + """Extract Skill tool calls from a child agent's transcript and write to a log file. + + Logs each Skill invocation with the args sent and the agent's response, + so we can audit whether sub-agents actually called skills like label-master. + """ + from .agent_types import AgentRunResult # noqa: F811 + + transcript = child_result.transcript + if not transcript: + return + log_dir = child_agent.runtime_config.session_directory / '_subagent_logs' + try: + log_dir.mkdir(parents=True, exist_ok=True) + except OSError: + return + session_id = child_result.session_id or 'unknown' + log_path = log_dir / f'{session_id}.jsonl' + + skill_calls: list[dict[str, object]] = [] + # Build a map of tool_call_id -> skill args from assistant messages with tool_calls + pending_skills: dict[str, dict[str, str]] = {} + for entry in transcript: + if not isinstance(entry, dict): + continue + role = entry.get('role') + if role == 'assistant': + tool_calls = entry.get('tool_calls') + if isinstance(tool_calls, list): + for tc in tool_calls: + if not isinstance(tc, dict): + continue + fn = tc.get('function') or {} + if not isinstance(fn, dict): + continue + if fn.get('name') == 'Skill': + tc_id = tc.get('id', '') + raw_args = fn.get('arguments', '{}') + try: + args_dict = json.loads(raw_args) if isinstance(raw_args, str) else raw_args + except (json.JSONDecodeError, TypeError): + args_dict = {'raw': raw_args} + pending_skills[tc_id] = { + 'skill': args_dict.get('skill', ''), + 'args': args_dict.get('args', ''), + } + elif role == 'tool': + tc_id = entry.get('tool_call_id', '') + if tc_id in pending_skills: + skill_info = pending_skills.pop(tc_id) + skill_calls.append({ + 'skill': skill_info['skill'], + 'args': skill_info['args'], + 'prompt_injected': entry.get('content') or '', + 'ts': datetime.now(timezone.utc).isoformat(), + }) + + # Also capture the assistant responses that follow skill injections + # by looking at assistant messages after tool messages + assistant_after_skill: list[str] = [] + saw_skill_tool = False + for entry in transcript: + if not isinstance(entry, dict): + continue + role = entry.get('role') + if role == 'tool' and entry.get('tool_call_id', '') in [ + tc.get('id', '') for tc in _all_skill_tool_call_ids(transcript) + ]: + saw_skill_tool = True + elif role == 'assistant' and saw_skill_tool: + content = entry.get('content', '') + if content: + assistant_after_skill.append(content) + saw_skill_tool = False + + # Merge assistant responses into skill_calls + for i, response_text in enumerate(assistant_after_skill): + if i < len(skill_calls): + skill_calls[i]['agent_response'] = response_text + + if not skill_calls: + return + try: + with open(log_path, 'a', encoding='utf-8') as f: + for call in skill_calls: + call['subtask_label'] = subtask_label + f.write(json.dumps(call, ensure_ascii=False) + '\n') + except OSError: + pass + + +def _all_skill_tool_call_ids(transcript: tuple[dict[str, object], ...]) -> list[dict[str, str]]: + """Collect all tool_call entries for Skill from assistant messages.""" + results = [] + for entry in transcript: + if not isinstance(entry, dict) or entry.get('role') != 'assistant': + continue + tool_calls = entry.get('tool_calls') + if not isinstance(tool_calls, list): + continue + for tc in tool_calls: + if not isinstance(tc, dict): + continue + fn = tc.get('function') or {} + if isinstance(fn, dict) and fn.get('name') == 'Skill': + results.append(tc) + return results + + @dataclass class LocalCodingAgent: model_config: ModelConfig @@ -3052,6 +3165,11 @@ class LocalCodingAgent: 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, @@ -3118,9 +3236,11 @@ class LocalCodingAgent: 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, diff --git a/src/agent_tools.py b/src/agent_tools.py index c77eeea..9611f3a 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -340,7 +340,7 @@ def default_tool_registry() -> dict[str, AgentTool]: {'type': 'number'}, {'type': 'integer'}, {'type': 'boolean'}, - {'type': 'array'}, + {'type': 'array', 'items': {}}, {'type': 'object'}, {'type': 'null'}, ] diff --git a/src/jupyter_runtime.py b/src/jupyter_runtime.py index 831cd98..a51e022 100644 --- a/src/jupyter_runtime.py +++ b/src/jupyter_runtime.py @@ -268,10 +268,12 @@ class JupyterRuntimeSession: '远端 jupyter pod 未挂载 /mnt/wangsenhao 或不可写：' + (probe.stdout.strip() or probe.stderr.strip() or 'unknown error') ) + ws_output = f'{shlex.quote(self.binding.workspace_cwd)}/output' + chat_output_link = f'{shlex.quote(chat_root)}/output' result = self.run_command( ( 'mkdir -p ' - f'{shlex.quote(self.binding.workspace_cwd)}/output ' + f'{ws_output} ' f'{shlex.quote(self.binding.workspace_cwd)}/input ' f'{shlex.quote(self.binding.workspace_cwd)}/scratchpad ' f'{shlex.quote(self.binding.workspace_root)}/.runtime/uploads ' @@ -279,7 +281,12 @@ class JupyterRuntimeSession: f'{shlex.quote(self.account_runtime_root)}/python ' f'{shlex.quote(chat_root)}/results ' f'{shlex.quote(chat_root)}/sft_output ' - f'{shlex.quote(chat_root)}/scripts' + f'{shlex.quote(chat_root)}/scripts && ' + f'if [ -d {chat_output_link} ] && [ ! -L {chat_output_link} ]; then ' + f'cp -a {chat_output_link}/* {ws_output}/ 2>/dev/null; ' + f'rm -rf {chat_output_link}; ' + f'fi && ' + f'ln -sfnT {ws_output} {chat_output_link}' ), timeout_seconds=timeout_seconds, max_output_chars=4000,