backend/frontend/runtime: 训练面板 / session 隔离 / agent runtime 调整
- backend/api/server.py：训练 step-detail / pipeline 卡片排序逻辑 - frontend：thread-list / thread / training-pipeline-panel / step-detail-sheet 配套调整，新增 metrics-chart-panel - src/agent_*：tool spec / runtime / prompting 配套 - .gitignore 加 .logs/ Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
hupenglong1
+520
-75
@@ -1696,12 +1696,31 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
async def get_training_step_detail(
|
||||
session_id: str,
|
||||
step: str,
|
||||
run_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
) -> Response:
|
||||
sessions_dir = state.account_paths(account_id)['sessions']
|
||||
state_path = _session_state_path(sessions_dir, session_id)
|
||||
state_entries, _ = _read_program_state(state_path)
|
||||
run_dic = _resolve_run_dic_from_state(state_entries)
|
||||
# Per-round: if frontend passed run_id, use that round's actual runDic
|
||||
# so historical step views (e.g., R1's augment) read the right
|
||||
# workflow<N> directory. Fall back to "max in state" only when run_id
|
||||
# is absent or the round has no resolvable runDic — preserves prior
|
||||
# behaviour for unbundled callers.
|
||||
run_dic: int | None = None
|
||||
if isinstance(run_id, str) and run_id.strip():
|
||||
rid = run_id.strip()
|
||||
per_round = _resolve_run_dic_per_round(state_entries)
|
||||
# augment 处理的是「上一轮评测的错例」，产物按 R{n-1} 的 runDic 命名
|
||||
# （data_clean_<R{n-1}>/、augment_<R{n-1}>.jsonl ...），不是本轮 eval workflow
|
||||
if step == 'augment':
|
||||
m = re.match(r'^[Rr](\d+)$', rid)
|
||||
if m and int(m.group(1)) >= 1:
|
||||
run_dic = per_round.get(f'R{int(m.group(1)) - 1}')
|
||||
if run_dic is None:
|
||||
run_dic = per_round.get(rid)
|
||||
if run_dic is None:
|
||||
run_dic = _resolve_run_dic_from_state(state_entries)
|
||||
runtime = _jupyter_runtime_for_session(state, account_id, session_id)
|
||||
artifact_paths = _step_artifact_paths(step, run_dic, runtime)
|
||||
|
||||
@@ -1850,6 +1869,33 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
'shown_rows': len(table_rows),
|
||||
}
|
||||
)
|
||||
if step == 'dist-analysis':
|
||||
# 软警告：dist-analysis 标 complete 时若 H1 候选 CSV 缺失，把 section
|
||||
# 的兜底 "产物文件不存在" 换成更明确的提示。某些轮次（hparam 类假设、
|
||||
# 无标签问题）确实可能没有候选，所以不 fail 这一步、不阻塞 human-check。
|
||||
target_run = (run_id or '').strip()
|
||||
last_status: str | None = None
|
||||
for entry in state_entries:
|
||||
if not isinstance(entry, dict) or entry.get('step') != 'dist-analysis':
|
||||
continue
|
||||
eid = (entry.get('run_id') or '').strip()
|
||||
if target_run and eid and eid != target_run:
|
||||
continue
|
||||
if 'status' in entry:
|
||||
last_status = entry.get('status')
|
||||
if last_status == 'complete':
|
||||
rd_label = str(run_dic) if run_dic is not None else '<runDic>'
|
||||
for sec in table_sections:
|
||||
if (
|
||||
str(sec.get('title', '')).startswith('H1 预计修改候选')
|
||||
and sec.get('kind') == 'missing'
|
||||
):
|
||||
sec['error'] = (
|
||||
f'未检测到 output/relabel_candidates_{rd_label}.csv —— '
|
||||
f'若本轮无 H1 候选可忽略；如有候选请按 SKILL.md 「阶段一」'
|
||||
f'落盘 file,line,query,old_label,是否改(1/0),建议新label。'
|
||||
)
|
||||
break
|
||||
payload = {
|
||||
'step': step,
|
||||
'run_dic': run_dic,
|
||||
@@ -1923,6 +1969,12 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
metrics_history = await asyncio.to_thread(
|
||||
_read_iteration_log_metrics_history,
|
||||
session_id,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
payload = await asyncio.to_thread(
|
||||
_hardcoded_autoresearch_pipeline,
|
||||
session_id,
|
||||
@@ -1931,6 +1983,8 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
model_config=state.model_config_for(account_id),
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
iteration_log_count=iter_count,
|
||||
metrics_history=metrics_history,
|
||||
)
|
||||
failure_status = await asyncio.to_thread(
|
||||
_read_session_failure_status,
|
||||
@@ -1964,48 +2018,82 @@ def create_app(state: AgentState) -> FastAPI:
|
||||
async def event_gen():
|
||||
last_signature: str | None = None
|
||||
last_heartbeat = time.monotonic()
|
||||
while True:
|
||||
if await request.is_disconnected():
|
||||
break
|
||||
state_entries, state_mtime = _read_program_state(state_path)
|
||||
iter_count = await asyncio.to_thread(
|
||||
_read_iteration_log_count,
|
||||
refresher = asyncio.create_task(
|
||||
_iter_count_refresh_loop(
|
||||
session_id,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
payload = await asyncio.to_thread(
|
||||
_hardcoded_autoresearch_pipeline,
|
||||
)
|
||||
metrics_refresher = asyncio.create_task(
|
||||
_metrics_history_refresh_loop(
|
||||
session_id,
|
||||
sessions_dir=sessions_dir,
|
||||
state_entries=state_entries,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
failure_status = await asyncio.to_thread(
|
||||
_read_session_failure_status,
|
||||
sessions_dir,
|
||||
session_id,
|
||||
)
|
||||
run_active = _is_run_active(state, account_id, session_id)
|
||||
_apply_program_state(
|
||||
payload,
|
||||
state_entries,
|
||||
iteration_log_count=iter_count or 0,
|
||||
session_failure_status=failure_status,
|
||||
run_active=run_active,
|
||||
session_id=session_id,
|
||||
)
|
||||
payload_str = json.dumps(payload, ensure_ascii=False)
|
||||
signature = f'{state_mtime}|{hash(payload_str)}'
|
||||
if signature != last_signature:
|
||||
last_signature = signature
|
||||
yield f'data: {payload_str}\n\n'
|
||||
now = time.monotonic()
|
||||
if now - last_heartbeat > 15:
|
||||
yield ': heartbeat\n\n'
|
||||
last_heartbeat = now
|
||||
await asyncio.sleep(0.5)
|
||||
)
|
||||
try:
|
||||
while True:
|
||||
if await request.is_disconnected():
|
||||
break
|
||||
state_entries, state_mtime = _read_program_state(state_path)
|
||||
iter_count = await asyncio.to_thread(
|
||||
_read_iteration_log_count,
|
||||
session_id,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
metrics_history = await asyncio.to_thread(
|
||||
_read_iteration_log_metrics_history,
|
||||
session_id,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
)
|
||||
payload = await asyncio.to_thread(
|
||||
_hardcoded_autoresearch_pipeline,
|
||||
session_id,
|
||||
sessions_dir=sessions_dir,
|
||||
state_entries=state_entries,
|
||||
agent_state=state,
|
||||
account_id=account_id,
|
||||
iteration_log_count=iter_count,
|
||||
metrics_history=metrics_history,
|
||||
)
|
||||
failure_status = await asyncio.to_thread(
|
||||
_read_session_failure_status,
|
||||
sessions_dir,
|
||||
session_id,
|
||||
)
|
||||
run_active = _is_run_active(state, account_id, session_id)
|
||||
_apply_program_state(
|
||||
payload,
|
||||
state_entries,
|
||||
iteration_log_count=iter_count or 0,
|
||||
session_failure_status=failure_status,
|
||||
run_active=run_active,
|
||||
session_id=session_id,
|
||||
)
|
||||
payload_str = json.dumps(payload, ensure_ascii=False)
|
||||
signature = f'{state_mtime}|{hash(payload_str)}'
|
||||
if signature != last_signature:
|
||||
last_signature = signature
|
||||
yield f'data: {payload_str}\n\n'
|
||||
now = time.monotonic()
|
||||
if now - last_heartbeat > 15:
|
||||
yield ': heartbeat\n\n'
|
||||
last_heartbeat = now
|
||||
await asyncio.sleep(0.5)
|
||||
finally:
|
||||
refresher.cancel()
|
||||
try:
|
||||
await refresher
|
||||
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
||||
pass
|
||||
metrics_refresher.cancel()
|
||||
try:
|
||||
await metrics_refresher
|
||||
except (asyncio.CancelledError, Exception): # noqa: BLE001
|
||||
pass
|
||||
|
||||
return StreamingResponse(
|
||||
event_gen(),
|
||||
@@ -2886,7 +2974,7 @@ def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None:
|
||||
'exceeds_soft_limit',
|
||||
}
|
||||
normalized = {
|
||||
key: _json_safe_limited(value)
|
||||
key: _json_safe_limited(value, parent_key=key)
|
||||
for key, value in event.items()
|
||||
if key in allowed and value is not None
|
||||
}
|
||||
@@ -2894,20 +2982,25 @@ def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None:
|
||||
return normalized
|
||||
|
||||
|
||||
def _json_safe_limited(value: Any, *, depth: int = 0) -> Any:
|
||||
_VERBOSE_STRING_KEYS = {'code', 'command', 'content', 'delta', 'script', 'stdout', 'output'}
|
||||
_VERBOSE_STRING_LIMIT = 20000
|
||||
|
||||
|
||||
def _json_safe_limited(value: Any, *, depth: int = 0, parent_key: str | None = None) -> Any:
|
||||
if depth > 4:
|
||||
return str(value)[:500]
|
||||
if isinstance(value, str):
|
||||
return value[:2000]
|
||||
limit = _VERBOSE_STRING_LIMIT if parent_key in _VERBOSE_STRING_KEYS else 2000
|
||||
return value[:limit]
|
||||
if isinstance(value, (int, float, bool)) or value is None:
|
||||
return value
|
||||
if isinstance(value, dict):
|
||||
return {
|
||||
str(key)[:200]: _json_safe_limited(item, depth=depth + 1)
|
||||
str(key)[:200]: _json_safe_limited(item, depth=depth + 1, parent_key=str(key))
|
||||
for key, item in list(value.items())[:80]
|
||||
}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_json_safe_limited(item, depth=depth + 1) for item in value[:80]]
|
||||
return [_json_safe_limited(item, depth=depth + 1, parent_key=parent_key) for item in value[:80]]
|
||||
try:
|
||||
json.dumps(value)
|
||||
return value
|
||||
@@ -4885,6 +4978,35 @@ class _WatcherManager:
|
||||
account_id=account_id,
|
||||
session_id=session_id,
|
||||
)
|
||||
# R{n>=1} 链路：SFT 训练完后 submit_sft_via_cml.sh 内置 nohup
|
||||
# watcher 自动起 R{n} 评测，agent 不在回路中，没人写
|
||||
# cml=running，UI 会从 sft=complete 直接跳到 cml=complete，
|
||||
# 中间 ~20min 评测期看起来像"流程卡死"。检测到同 session
|
||||
# 已注册 cml watch（说明 agent 提交 SFT 时预注册了下游评测
|
||||
# 的 watch）就顺手补写 cml=running 同 run_id。
|
||||
if (
|
||||
step == 'sft'
|
||||
and session_id is not None
|
||||
and self.is_watching(session_id, 'cml')
|
||||
):
|
||||
await self._write_step(
|
||||
state_path,
|
||||
step='cml',
|
||||
status='running',
|
||||
run_id=run_id,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
session_id=session_id,
|
||||
)
|
||||
await self._write_log(
|
||||
state_path,
|
||||
run_id,
|
||||
'↪ sft 完成自动衔接 → step cml running '
|
||||
'(submit_sft_via_cml.sh 内置 watcher 已触发 R 评测)',
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
session_id=session_id,
|
||||
)
|
||||
return
|
||||
if time.monotonic() - start > timeout:
|
||||
await self._write_step(
|
||||
@@ -5507,13 +5629,27 @@ def _scan_for_watchers(state: AgentState) -> None:
|
||||
status = entry.get('status')
|
||||
if isinstance(step, str) and step and isinstance(status, str):
|
||||
last_status_per_step[step] = status
|
||||
# See note in _scan_for_watchers_async: only the latest watch
|
||||
# entry per step is current.
|
||||
latest_watch_per_step: dict[str, dict[str, Any]] = {}
|
||||
latest_running_run_id: dict[str, str] = {}
|
||||
for entry in entries:
|
||||
watch = entry.get('watch')
|
||||
if not isinstance(watch, dict):
|
||||
continue
|
||||
step = watch.get('step')
|
||||
if not isinstance(step, str) or not step:
|
||||
if isinstance(watch, dict):
|
||||
w_step = watch.get('step')
|
||||
if isinstance(w_step, str) and w_step:
|
||||
latest_watch_per_step[w_step] = watch
|
||||
continue
|
||||
e_step = entry.get('step')
|
||||
e_status = entry.get('status')
|
||||
e_run = entry.get('run_id')
|
||||
if (
|
||||
isinstance(e_step, str) and e_step
|
||||
and e_status == 'running'
|
||||
and isinstance(e_run, str) and e_run
|
||||
):
|
||||
latest_running_run_id[e_step] = e_run
|
||||
for step, watch in latest_watch_per_step.items():
|
||||
if last_status_per_step.get(step) in {
|
||||
'complete',
|
||||
'failed',
|
||||
@@ -5535,6 +5671,9 @@ def _scan_for_watchers(state: AgentState) -> None:
|
||||
interval = 30.0
|
||||
timeout = 3600.0
|
||||
run_id = str(watch.get('run_id') or '?')
|
||||
running_run_id = latest_running_run_id.get(step)
|
||||
if running_run_id and run_id != running_run_id:
|
||||
continue
|
||||
_watcher_manager.register_file_exists(
|
||||
session_id=session_id,
|
||||
state_path=state_path,
|
||||
@@ -5617,13 +5756,30 @@ async def _scan_for_watchers_async(state: AgentState) -> None:
|
||||
status = entry.get('status')
|
||||
if isinstance(step, str) and step and isinstance(status, str):
|
||||
last_status_per_step[step] = status
|
||||
# Only the LAST watch entry per step represents the current round's
|
||||
# intent. Earlier watch entries from previous rounds must NOT be
|
||||
# re-registered when a new round resets the step to running, or
|
||||
# the resulting watcher will fire with a stale run_id and the UI
|
||||
# will never see a complete tagged with the current round.
|
||||
latest_watch_per_step: dict[str, dict[str, Any]] = {}
|
||||
latest_running_run_id: dict[str, str] = {}
|
||||
for entry in entries:
|
||||
watch = entry.get('watch')
|
||||
if not isinstance(watch, dict):
|
||||
continue
|
||||
step = watch.get('step')
|
||||
if not isinstance(step, str) or not step:
|
||||
if isinstance(watch, dict):
|
||||
w_step = watch.get('step')
|
||||
if isinstance(w_step, str) and w_step:
|
||||
latest_watch_per_step[w_step] = watch
|
||||
continue
|
||||
e_step = entry.get('step')
|
||||
e_status = entry.get('status')
|
||||
e_run = entry.get('run_id')
|
||||
if (
|
||||
isinstance(e_step, str) and e_step
|
||||
and e_status == 'running'
|
||||
and isinstance(e_run, str) and e_run
|
||||
):
|
||||
latest_running_run_id[e_step] = e_run
|
||||
for step, watch in latest_watch_per_step.items():
|
||||
if last_status_per_step.get(step) in {
|
||||
'complete',
|
||||
'failed',
|
||||
@@ -5645,6 +5801,13 @@ async def _scan_for_watchers_async(state: AgentState) -> None:
|
||||
interval_s = 30.0
|
||||
timeout_s = 3600.0
|
||||
run_id = str(watch.get('run_id') or '?')
|
||||
# If a newer round has written `step=<step>, status=running`
|
||||
# but the matching watch entry hasn't landed yet, skip and
|
||||
# wait for the next scan rather than spawning a watcher tied
|
||||
# to the previous round's run_id.
|
||||
running_run_id = latest_running_run_id.get(step)
|
||||
if running_run_id and run_id != running_run_id:
|
||||
continue
|
||||
_watcher_manager.register_file_exists(
|
||||
session_id=session_id,
|
||||
state_path=state_path,
|
||||
@@ -5752,7 +5915,6 @@ _STEP_KIND: dict[str, str] = {
|
||||
'cml': 'analysis',
|
||||
'gold-drift': 'analysis',
|
||||
'dist-analysis': 'analysis',
|
||||
'report': 'analysis',
|
||||
'hypothesis': 'analysis',
|
||||
'log': 'analysis',
|
||||
'augment': 'train',
|
||||
@@ -5780,8 +5942,7 @@ _CARD_META: dict[tuple[str, str], dict[str, Any]] = {
|
||||
# R0·Baseline
|
||||
('cml', 'baseline'): {'icon': 'bar-chart', 'title': 'Baseline 评测', 'subtitle': 'CML workflow metric_diff'},
|
||||
('gold-drift', 'baseline'): {'icon': 'alert-triangle', 'title': 'Gold Drift 检查', 'subtitle': 'drift_<runDic>.json'},
|
||||
('dist-analysis', 'baseline'): {'icon': 'trending-up', 'title': '分层结果分析', 'subtitle': '需求集合 / 车载 / specific'},
|
||||
('report', 'baseline'): {'icon': 'file-text', 'title': '问题分析 & 报告', 'subtitle': 'workflow<runDic>.md'},
|
||||
('dist-analysis', 'baseline'): {'icon': 'trending-up', 'title': '分层结果分析 & 报告', 'subtitle': '根因归类 + workflow<runDic>.md'},
|
||||
('hypothesis', 'baseline'): {'icon': 'dna', 'title': '形成假设', 'subtitle': 'iteration_log hypothesis 字段'},
|
||||
('log', 'baseline'): {'icon': 'clipboard-list', 'title': '记录迭代日志', 'subtitle': 'iteration_log.jsonl'},
|
||||
# R{n}·Train
|
||||
@@ -5791,8 +5952,7 @@ _CARD_META: dict[tuple[str, str], dict[str, Any]] = {
|
||||
# R{n}·Analysis
|
||||
('cml', 'analysis'): {'icon': 'bar-chart', 'title': 'CML 评测', 'subtitle': 'workflow metric_diff'},
|
||||
('gold-drift', 'analysis'): {'icon': 'alert-triangle', 'title': 'Gold Drift 检查', 'subtitle': 'drift_<runDic>.json'},
|
||||
('dist-analysis', 'analysis'): {'icon': 'trending-up', 'title': '分层结果分析', 'subtitle': '需求集合 / 车载 / specific'},
|
||||
('report', 'analysis'): {'icon': 'file-text', 'title': '问题分析 & 报告', 'subtitle': 'workflow<runDic>.md'},
|
||||
('dist-analysis', 'analysis'): {'icon': 'trending-up', 'title': '分层结果分析 & 报告', 'subtitle': '根因归类 + workflow<runDic>.md'},
|
||||
('hypothesis', 'analysis'): {'icon': 'dna', 'title': '形成假设', 'subtitle': 'iteration_log hypothesis 字段'},
|
||||
('log', 'analysis'): {'icon': 'clipboard-list', 'title': '记录迭代日志', 'subtitle': 'iteration_log.jsonl'},
|
||||
}
|
||||
@@ -5803,9 +5963,8 @@ _CARD_ORDER: dict[tuple[str, str], int] = {
|
||||
('cml', 'baseline'): 0,
|
||||
('gold-drift', 'baseline'): 1,
|
||||
('dist-analysis', 'baseline'): 2,
|
||||
('report', 'baseline'): 3,
|
||||
('hypothesis', 'baseline'): 4,
|
||||
('log', 'baseline'): 5,
|
||||
('hypothesis', 'baseline'): 3,
|
||||
('log', 'baseline'): 4,
|
||||
# train
|
||||
('augment', 'train'): 0,
|
||||
('verify', 'train'): 1,
|
||||
@@ -5814,9 +5973,8 @@ _CARD_ORDER: dict[tuple[str, str], int] = {
|
||||
('cml', 'analysis'): 0,
|
||||
('gold-drift', 'analysis'): 1,
|
||||
('dist-analysis', 'analysis'): 2,
|
||||
('report', 'analysis'): 3,
|
||||
('hypothesis', 'analysis'): 4,
|
||||
('log', 'analysis'): 5,
|
||||
('hypothesis', 'analysis'): 3,
|
||||
('log', 'analysis'): 4,
|
||||
}
|
||||
|
||||
_SECTION_LABELS: dict[str, str] = {
|
||||
@@ -6514,6 +6672,13 @@ def _scan_run_history_max() -> int | None:
|
||||
return best if best >= 0 else None
|
||||
|
||||
|
||||
_ITER_COUNT_CACHE: dict[str, int | None] = {}
|
||||
|
||||
|
||||
def _iter_count_cache_key(session_id: str | None, account_id: str | None) -> str:
|
||||
return f'{account_id or "_"}|{session_id or "_"}'
|
||||
|
||||
|
||||
def _read_iteration_log_count(
|
||||
session_id: str | None = None,
|
||||
*,
|
||||
@@ -6523,10 +6688,53 @@ def _read_iteration_log_count(
|
||||
"""Count non-empty lines in iteration_log.jsonl.
|
||||
|
||||
Chat-root lives under the remote workspace_cwd, so reads MUST go through
|
||||
the bound jupyter runtime — backend host typically can't read this path
|
||||
locally. Falls back to the legacy global path on the local fs only when
|
||||
no runtime is bound (session never connected, or transient lookup miss).
|
||||
the bound jupyter runtime — the round-trip is multi-second. To keep the
|
||||
SSE pipeline loop responsive, this function returns the last cached value
|
||||
when one exists; only a cold cache pays the synchronous remote cost. The
|
||||
cache is refreshed in the background by `_iter_count_refresh_loop`, which
|
||||
the SSE handler spawns alongside its event loop.
|
||||
"""
|
||||
cache_key = _iter_count_cache_key(session_id, account_id)
|
||||
if cache_key in _ITER_COUNT_CACHE:
|
||||
return _ITER_COUNT_CACHE[cache_key]
|
||||
value = _read_iteration_log_count_uncached(
|
||||
session_id, agent_state=agent_state, account_id=account_id,
|
||||
)
|
||||
_ITER_COUNT_CACHE[cache_key] = value
|
||||
return value
|
||||
|
||||
|
||||
async def _iter_count_refresh_loop(
|
||||
session_id: str | None,
|
||||
*,
|
||||
agent_state: 'AgentState | None',
|
||||
account_id: str | None,
|
||||
interval_seconds: float = 3.0,
|
||||
) -> None:
|
||||
"""Background task that re-reads iteration_log_count every `interval_seconds`
|
||||
and updates the in-memory cache. Cancelled when the caller (SSE handler)
|
||||
exits."""
|
||||
cache_key = _iter_count_cache_key(session_id, account_id)
|
||||
while True:
|
||||
try:
|
||||
value = await asyncio.to_thread(
|
||||
_read_iteration_log_count_uncached,
|
||||
session_id,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
)
|
||||
_ITER_COUNT_CACHE[cache_key] = value
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
await asyncio.sleep(interval_seconds)
|
||||
|
||||
|
||||
def _read_iteration_log_count_uncached(
|
||||
session_id: str | None = None,
|
||||
*,
|
||||
agent_state: 'AgentState | None' = None,
|
||||
account_id: str | None = None,
|
||||
) -> int | None:
|
||||
runtime = (
|
||||
_jupyter_runtime_for_session(agent_state, account_id, session_id)
|
||||
if session_id and agent_state is not None
|
||||
@@ -6565,6 +6773,150 @@ def _read_iteration_log_count(
|
||||
return None
|
||||
|
||||
|
||||
_METRICS_HISTORY_CACHE: dict[str, dict[str, Any] | None] = {}
|
||||
|
||||
|
||||
def _read_iteration_log_metrics_history(
|
||||
session_id: str | None = None,
|
||||
*,
|
||||
agent_state: 'AgentState | None' = None,
|
||||
account_id: 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)
|
||||
if cache_key in _METRICS_HISTORY_CACHE:
|
||||
return _METRICS_HISTORY_CACHE[cache_key]
|
||||
value = _read_iteration_log_metrics_history_uncached(
|
||||
session_id, agent_state=agent_state, account_id=account_id,
|
||||
)
|
||||
_METRICS_HISTORY_CACHE[cache_key] = value
|
||||
return value
|
||||
|
||||
|
||||
async def _metrics_history_refresh_loop(
|
||||
session_id: str | None,
|
||||
*,
|
||||
agent_state: 'AgentState | None',
|
||||
account_id: str | None,
|
||||
interval_seconds: float = 3.0,
|
||||
) -> None:
|
||||
cache_key = _iter_count_cache_key(session_id, account_id)
|
||||
while True:
|
||||
try:
|
||||
value = await asyncio.to_thread(
|
||||
_read_iteration_log_metrics_history_uncached,
|
||||
session_id,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
)
|
||||
_METRICS_HISTORY_CACHE[cache_key] = value
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
await asyncio.sleep(interval_seconds)
|
||||
|
||||
|
||||
def _read_iteration_log_metrics_history_uncached(
|
||||
session_id: str | None = None,
|
||||
*,
|
||||
agent_state: 'AgentState | None' = None,
|
||||
account_id: str | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Parse iteration_log.jsonl into {'metrics': [keys], 'rounds': [...]}.
|
||||
|
||||
Each entry's `results` dict contributes one round point; metric key union
|
||||
is taken across all rounds (sorted by first-appearance for stable column
|
||||
order). Missing values produce nulls so frontend line charts break the
|
||||
series naturally."""
|
||||
runtime = (
|
||||
_jupyter_runtime_for_session(agent_state, account_id, session_id)
|
||||
if session_id and agent_state is not None
|
||||
else None
|
||||
)
|
||||
text: str | None = None
|
||||
if runtime is not None:
|
||||
remote_root = _chat_root_for_runtime(runtime)
|
||||
if remote_root is not None:
|
||||
remote_path = f'{remote_root}/results/iteration_log.jsonl'
|
||||
cmd = (
|
||||
f'test -f {shlex.quote(remote_path)} '
|
||||
f'&& cat {shlex.quote(remote_path)} '
|
||||
f'|| echo __MISSING__'
|
||||
)
|
||||
try:
|
||||
result = runtime.run_command(
|
||||
cmd, timeout_seconds=10.0, max_output_chars=400_000,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
return None
|
||||
stdout = (result.stdout or '').strip() if result is not None else ''
|
||||
if stdout and stdout != '__MISSING__':
|
||||
text = stdout
|
||||
if text is None:
|
||||
legacy = _autoresearch_root() / 'results' / 'iteration_log.jsonl'
|
||||
try:
|
||||
if legacy.is_file():
|
||||
text = legacy.read_text(encoding='utf-8', errors='replace')
|
||||
except OSError:
|
||||
return None
|
||||
if not text:
|
||||
return None
|
||||
|
||||
metrics_order: list[str] = []
|
||||
metrics_seen: set[str] = set()
|
||||
rounds: list[dict[str, Any]] = []
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
obj = json.loads(line)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
if not isinstance(obj, dict):
|
||||
continue
|
||||
results = obj.get('results')
|
||||
if not isinstance(results, dict):
|
||||
continue
|
||||
values: dict[str, float] = {}
|
||||
for k, v in results.items():
|
||||
if not isinstance(k, str):
|
||||
continue
|
||||
try:
|
||||
fv = float(v)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
values[k] = fv
|
||||
if k not in metrics_seen:
|
||||
metrics_seen.add(k)
|
||||
metrics_order.append(k)
|
||||
if not values:
|
||||
continue
|
||||
iteration = obj.get('iteration')
|
||||
if isinstance(iteration, str) and iteration.lstrip('-').isdigit():
|
||||
iteration = int(iteration)
|
||||
if not isinstance(iteration, int):
|
||||
iteration = len(rounds)
|
||||
run_dic = obj.get('runDic') or obj.get('run_dic')
|
||||
if isinstance(run_dic, str) and run_dic.isdigit():
|
||||
run_dic = int(run_dic)
|
||||
if not isinstance(run_dic, int):
|
||||
run_dic = 0
|
||||
timestamp = obj.get('timestamp')
|
||||
if not isinstance(timestamp, str):
|
||||
timestamp = ''
|
||||
rounds.append({
|
||||
'iteration': iteration,
|
||||
'runDic': run_dic,
|
||||
'label': f'R{iteration}',
|
||||
'timestamp': timestamp,
|
||||
'values': values,
|
||||
})
|
||||
|
||||
if not rounds:
|
||||
return None
|
||||
return {'metrics': metrics_order, 'rounds': rounds}
|
||||
|
||||
|
||||
def _read_iteration_log_h1_label(
|
||||
runtime: 'JupyterRuntimeSession | None',
|
||||
run_dic: int | None,
|
||||
@@ -7019,6 +7371,8 @@ def _compute_kpis(
|
||||
*,
|
||||
agent_state: 'AgentState | None' = None,
|
||||
account_id: str | None = None,
|
||||
iteration_log_count: int | None = None,
|
||||
metrics_history: dict[str, Any] | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Compute the 7 top KPIs from authoritative sources. Falls back to '—'.
|
||||
|
||||
@@ -7054,17 +7408,33 @@ def _compute_kpis(
|
||||
target_metric = _format_metric_value(metrics, 'target_set_pass_rate')
|
||||
overall_metric = _format_metric_value(metrics, 'overall_car_pass_rate')
|
||||
specific_metric = _format_metric_value(metrics, 'specific_test_pass_rate')
|
||||
iter_count = _read_iteration_log_count(
|
||||
session_id,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
)
|
||||
if iter_count is None:
|
||||
iteration = 'R0-baseline'
|
||||
elif iter_count == 0:
|
||||
iteration = 'R0-baseline'
|
||||
# Round number = MAX numeric `iteration` field across iteration_log.jsonl
|
||||
# entries, NOT the line count. iteration_log.jsonl carries multiple rows
|
||||
# per round (separate hypothesis-only / results / final-summary entries),
|
||||
# so line count overcounts. Falls back to line-count for backwards
|
||||
# compatibility when metrics_history isn't available.
|
||||
max_iteration: int | None = None
|
||||
if metrics_history and isinstance(metrics_history.get('rounds'), list):
|
||||
for r in metrics_history['rounds']:
|
||||
it = r.get('iteration') if isinstance(r, dict) else None
|
||||
if isinstance(it, int):
|
||||
if max_iteration is None or it > max_iteration:
|
||||
max_iteration = it
|
||||
if max_iteration is not None:
|
||||
iteration = 'R0-baseline' if max_iteration == 0 else f'R{max_iteration}'
|
||||
else:
|
||||
iteration = f'R{iter_count}'
|
||||
if iteration_log_count is not None:
|
||||
iter_count = iteration_log_count
|
||||
else:
|
||||
iter_count = _read_iteration_log_count(
|
||||
session_id,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
)
|
||||
if iter_count is None or iter_count == 0:
|
||||
iteration = 'R0-baseline'
|
||||
else:
|
||||
iteration = f'R{iter_count}'
|
||||
elapsed = _format_elapsed(_compute_step0_start_seconds(state_entries))
|
||||
target_value = (
|
||||
f'{target_set} · {target_metric}' if target_metric != '—' else target_set
|
||||
@@ -7134,6 +7504,79 @@ def _resolve_run_dic_from_state(
|
||||
return best
|
||||
|
||||
|
||||
def _resolve_run_dic_per_round(
|
||||
state_entries: list[dict[str, Any]],
|
||||
) -> dict[str, int]:
|
||||
"""Map each run_id (e.g., 'R2') to the runDic that round actually used.
|
||||
|
||||
Same priority as ``_resolve_run_dic_from_state`` but bucketed per run_id
|
||||
so per-round step-detail views read the right artifact directory instead
|
||||
of always using the latest workflow id seen anywhere in state."""
|
||||
workflow_re = re.compile(r'workflow(\d+)')
|
||||
rundic_re = re.compile(r'runDic\s*[=:取]?\s*(\d+)')
|
||||
|
||||
def _scan(obj: Any) -> int | None:
|
||||
if isinstance(obj, str):
|
||||
cands = [int(x) for x in workflow_re.findall(obj)]
|
||||
cands += [int(x) for x in rundic_re.findall(obj)]
|
||||
return max(cands) if cands else None
|
||||
if isinstance(obj, dict):
|
||||
best: int | None = None
|
||||
for v in obj.values():
|
||||
hit = _scan(v)
|
||||
if hit is not None and (best is None or hit > best):
|
||||
best = hit
|
||||
return best
|
||||
if isinstance(obj, list):
|
||||
best = None
|
||||
for v in obj:
|
||||
hit = _scan(v)
|
||||
if hit is not None and (best is None or hit > best):
|
||||
best = hit
|
||||
return best
|
||||
return None
|
||||
|
||||
def _entry_run_id(entry: dict[str, Any]) -> str | None:
|
||||
# 顶层 run_id 优先；否则从 watch.run_id / log.iter 兜底，让 watcher/log
|
||||
# 这种 agent 不写顶层 run_id 但嵌套里带轮次信号的条目也能贡献 runDic。
|
||||
rid = entry.get('run_id')
|
||||
if isinstance(rid, str) and rid.strip():
|
||||
return rid.strip()
|
||||
watch = entry.get('watch')
|
||||
if isinstance(watch, dict):
|
||||
wrid = watch.get('run_id')
|
||||
if isinstance(wrid, str) and wrid.strip():
|
||||
return wrid.strip()
|
||||
log = entry.get('log')
|
||||
if isinstance(log, dict):
|
||||
lit = log.get('iter')
|
||||
if isinstance(lit, str) and lit.strip():
|
||||
return lit.strip()
|
||||
return None
|
||||
|
||||
explicit: dict[str, int] = {}
|
||||
fallback: dict[str, int] = {}
|
||||
for entry in state_entries:
|
||||
run_id = _entry_run_id(entry)
|
||||
if not run_id:
|
||||
continue
|
||||
rd = entry.get('runDic') or entry.get('run_dic') or entry.get('rundic')
|
||||
if isinstance(rd, str) and rd.isdigit():
|
||||
rd = int(rd)
|
||||
if isinstance(rd, int):
|
||||
explicit[run_id] = rd # latest wins
|
||||
continue
|
||||
hit = _scan(entry)
|
||||
if hit is not None:
|
||||
cur = fallback.get(run_id)
|
||||
if cur is None or hit > cur:
|
||||
fallback[run_id] = hit
|
||||
|
||||
out = dict(fallback)
|
||||
out.update(explicit)
|
||||
return out
|
||||
|
||||
|
||||
def _step_artifact_paths(
|
||||
step: str,
|
||||
run_dic: int | None,
|
||||
@@ -7176,9 +7619,6 @@ def _step_artifact_paths(
|
||||
paths.append(
|
||||
f'{runtime.binding.workspace_cwd}/output/relabel_candidates_{run_dic}.csv'
|
||||
)
|
||||
elif step == 'report':
|
||||
if run_dic is not None:
|
||||
paths.append(f'{chat_root}/results/workflow{run_dic}.md')
|
||||
elif step == 'hypothesis':
|
||||
paths.append(f'{chat_root}/results/iteration_log.jsonl')
|
||||
elif step == 'augment':
|
||||
@@ -7342,6 +7782,8 @@ def _hardcoded_autoresearch_pipeline(
|
||||
*,
|
||||
agent_state: 'AgentState | None' = None,
|
||||
account_id: str | None = None,
|
||||
iteration_log_count: int | None = None,
|
||||
metrics_history: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Canonical autoresearch model-iteration pipeline structure.
|
||||
|
||||
@@ -7359,6 +7801,8 @@ def _hardcoded_autoresearch_pipeline(
|
||||
model_config=model_config,
|
||||
agent_state=agent_state,
|
||||
account_id=account_id,
|
||||
iteration_log_count=iteration_log_count,
|
||||
metrics_history=metrics_history,
|
||||
)
|
||||
else:
|
||||
# Defensive default if caller didn't pass sessions_dir.
|
||||
@@ -7385,6 +7829,7 @@ def _hardcoded_autoresearch_pipeline(
|
||||
'items': [],
|
||||
'in_flight': None,
|
||||
'logs': [],
|
||||
'metrics_history': metrics_history,
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user