【task002_m0_secondary_fixes】M0 review 次优先级 3 项修复

1. #7 Hermes 多轮：convert_hermes_conversations 不再在第一条 assistant
   时 break，把所有 assistant + tool 轮收进 extra_env_info.expected_trajectory；
   新增 expected_final_content / expected_turn_count。HERMES_ROLE_MAP 支持
   tool / function / function_response / observation 四种 role。input_messages
   仍只包含首条 assistant 前的上下文（不泄漏期望答案）；expected_tool_calls
   / expected_assistant_content 保留首轮值，向后兼容。

2. #8 hermes hf_config：显式锁定 func_calling_singleturn config（与
   general_tool_calling 环境 max_turns=2 一致）；data_registry 注释解释了
   为何 hermes 没有 hf_val_split（singleturn config 没有 holdout split，
   继续走 fallback 警告路径）。

3. #15 aggregate 重复执行：拆出 score_rows（per-row 打分）与
   aggregate_scored_rows（合并指标）两个纯函数；summarize_baselines 缓存
   split 级 scored 行，aggregate 直接拼接 split 结果。对
   code_execution_python 来说 subprocess fork 数量从 2× 降到 1×。
   evaluate_policy 改成 score_rows + aggregate 的薄封装，签名不变。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: intern_nemontron_review_cc

src/nemotron/recipes/super3/milestones/m0_data_env/data_registry.yaml

@@ -60,8 +60,19 @@ datasets:
     environment: general_tool_calling
     domain: general_tool_calling
     hf_dataset: NousResearch/hermes-function-calling-v1
-    hf_config: null
+    # Hermes ships multiple configs (func_calling_singleturn, func_calling,
+    # glaive_func_calling, json_mode_singleturn, json_mode_agentic). Without an
+    # explicit config, `load_dataset` is ambiguous on this dataset. M0 targets
+    # the single-turn function-calling subset because the `general_tool_calling`
+    # environment registry currently runs with max_turns=2. Multi-turn config
+    # (`func_calling`) is reserved for M1 when the agent loop is wired up.
+    hf_config: func_calling_singleturn
     hf_split: train
+    # NousResearch/hermes-function-calling-v1 does not ship train/val splits per
+    # config — there is only `train`. We intentionally leave hf_val_split unset
+    # so prepare_m0_assets falls back to its sequential-slice path and records
+    # the "not a true holdout" warning in manifest.warnings. This is the same
+    # signal the README surfaces.
     hf_revision: dae3e1d28cfbcf4b915c04ea1e072030529b4bda
     source_url: https://huggingface.co/datasets/NousResearch/hermes-function-calling-v1
     license: apache-2.0

src/nemotron/recipes/super3/milestones/m0_data_env/prepare_m0_assets.py

@@ -262,36 +262,78 @@ def strip_tool_call_blocks(text: str) -> str:
     return TOOL_CALL_RE.sub("", text).strip()
 
 
+HERMES_ROLE_MAP = {
+    "system": "system",
+    "human": "user",
+    "user": "user",
+    "gpt": "assistant",
+    "assistant": "assistant",
+    "tool": "tool",
+    "function": "tool",
+    "function_response": "tool",
+    "observation": "tool",
+}
+
+
 def convert_hermes_conversations(conversations: Iterable[Mapping[str, Any]]) -> tuple[list[JsonDict], JsonDict]:
-    role_map = {
-        "system": "system",
-        "human": "user",
-        "user": "user",
-        "gpt": "assistant",
-        "assistant": "assistant",
-    }
+    """Split a Hermes function-calling conversation into model input vs expected trajectory.
+
+    `input_messages` covers the prompt context up to (but not including) the first
+    assistant turn — that is what the policy sees at inference time. Everything from
+    the first assistant turn onward (assistant tool calls, tool observations,
+    follow-up assistant turns, final answer) is captured in `expected_trajectory`
+    so that downstream verifiers can score multi-turn behavior, not just the first
+    tool emission.
+    """
     input_messages: list[JsonDict] = []
+    expected_trajectory: list[JsonDict] = []
     expected_tool_calls: list[JsonDict] = []
     expected_assistant_content = ""
+    first_assistant_seen = False
+    last_assistant_content = ""
+    last_assistant_had_tool_calls = False
 
     for turn in conversations:
         raw_role = str(turn.get("from") or turn.get("role") or "").strip()
-        role = role_map.get(raw_role)
+        role = HERMES_ROLE_MAP.get(raw_role)
         if role is None:
             continue
         content = str(turn.get("value") or turn.get("content") or "")
         if role == "assistant":
-            expected_tool_calls = parse_tool_calls(content)
-            expected_assistant_content = strip_tool_call_blocks(content)
-            break
-        input_messages.append({"role": role, "content": content})
+            tool_calls = parse_tool_calls(content)
+            stripped = strip_tool_call_blocks(content)
+            expected_trajectory.append(
+                {
+                    "role": "assistant",
+                    "content": stripped,
+                    "tool_calls": tool_calls,
+                }
+            )
+            last_assistant_content = stripped
+            last_assistant_had_tool_calls = bool(tool_calls)
+            if not first_assistant_seen:
+                expected_tool_calls = tool_calls
+                expected_assistant_content = stripped
+                first_assistant_seen = True
+        elif role == "tool":
+            expected_trajectory.append({"role": "tool", "content": content, "tool_calls": []})
+        else:  # system / user
+            if first_assistant_seen:
+                # Late system/user turn after the assistant has spoken — record in
+                # the trajectory rather than leaking it into model input.
+                expected_trajectory.append({"role": role, "content": content, "tool_calls": []})
+            else:
+                input_messages.append({"role": role, "content": content})
 
     if not any(message["role"] == "system" for message in input_messages):
         input_messages.insert(0, {"role": "system", "content": SYSTEM_PROMPTS["general_tool_calling"]})
 
     return input_messages, {
         "expected_tool_calls": expected_tool_calls,
         "expected_assistant_content": expected_assistant_content,
+        "expected_trajectory": expected_trajectory,
+        "expected_final_content": last_assistant_content if not last_assistant_had_tool_calls else "",
+        "expected_turn_count": len(expected_trajectory),
     }
 
 
@@ -329,6 +371,9 @@ def transform_hermes_function_calling(row: Mapping[str, Any], spec: Mapping[str,
         extra_env_info={
             "expected_tool_calls": expected["expected_tool_calls"],
             "expected_assistant_content": expected["expected_assistant_content"],
+            "expected_trajectory": expected["expected_trajectory"],
+            "expected_final_content": expected["expected_final_content"],
+            "expected_turn_count": expected["expected_turn_count"],
             "category": row.get("category"),
             "subcategory": row.get("subcategory"),
             "task": row.get("task"),

src/nemotron/recipes/super3/milestones/m0_data_env/run_m0_health_baseline.py

@@ -220,22 +220,20 @@ def candidates_for_policy(record: Mapping[str, Any], policy: str) -> list[Any]:
     raise ValueError(f"unknown baseline policy: {policy}")
 
 
-def evaluate_policy(
+def score_rows(
     rows: Sequence[Mapping[str, Any]],
     *,
     policy: str,
     best_k: int,
     run_code: bool,
-) -> JsonDict:
-    pass_at_1 = 0
-    best_at_k = 0
-    total_score_at_1 = 0.0
-    total_best_score_at_k = 0.0
-    errors: list[JsonDict] = []
-    skipped_rows = 0
-    scored_rows = 0
-    threshold = 1.0
-
+) -> list[JsonDict]:
+    """Score each row once and return raw per-row results.
+
+    Pulled out of `evaluate_policy` so that aggregate metrics can be derived from
+    the same per-row scores as the split metrics, instead of re-invoking the
+    verifier (and, for python_unit_tests, re-spawning subprocesses).
+    """
+    out: list[JsonDict] = []
     for index, record in enumerate(rows):
         candidates = candidates_for_policy(record, policy)
         scores: list[float] = []
@@ -246,8 +244,36 @@ def evaluate_policy(
             if score is None:
                 continue
             scores.append(score)
+        out.append(
+            {
+                "row_index": index,
+                "scores": scores,
+                "diagnostics": diagnostics,
+                "metadata": record.get("metadata", {}),
+                "verifier": record.get("reward_config", {}).get("verifier"),
+            }
+        )
+    return out
+
+
+def aggregate_scored_rows(
+    scored: Sequence[Mapping[str, Any]],
+    *,
+    policy: str,
+    best_k: int,
+) -> JsonDict:
+    pass_at_1 = 0
+    best_at_k = 0
+    total_score_at_1 = 0.0
+    total_best_score_at_k = 0.0
+    errors: list[JsonDict] = []
+    skipped_rows = 0
+    scored_rows = 0
+    threshold = 1.0
+
+    for entry in scored:
+        scores = entry["scores"]
         if not scores:
-            # Every candidate for this row was skipped (e.g. code execution disabled).
             skipped_rows += 1
             continue
         scored_rows += 1
@@ -258,19 +284,19 @@ def evaluate_policy(
         total_score_at_1 += first_score
         total_best_score_at_k += best_score
         if best_score < threshold:
-            metadata = record.get("metadata", {})
+            metadata = entry["metadata"]
             errors.append(
                 {
-                    "row_index": index,
+                    "row_index": entry["row_index"],
                     "source_dataset": metadata.get("source_dataset"),
                     "source_id": metadata.get("source_id"),
                     "source_row_index": metadata.get("source_row_index"),
-                    "verifier": record.get("reward_config", {}).get("verifier"),
-                    "diagnostics": diagnostics,
+                    "verifier": entry["verifier"],
+                    "diagnostics": entry["diagnostics"],
                 }
             )
 
-    total = len(rows)
+    total = len(scored)
     denominator = scored_rows or 1
     return {
         "policy": policy,
@@ -286,6 +312,20 @@ def evaluate_policy(
     }
 
 
+def evaluate_policy(
+    rows: Sequence[Mapping[str, Any]],
+    *,
+    policy: str,
+    best_k: int,
+    run_code: bool,
+) -> JsonDict:
+    """Score and aggregate in one pass. Kept for callers that don't need
+    the per-row breakdown (existing tests, direct CLI users).
+    """
+    scored = score_rows(rows, policy=policy, best_k=best_k, run_code=run_code)
+    return aggregate_scored_rows(scored, policy=policy, best_k=best_k)
+
+
 def load_environment_specs(path: Path) -> dict[str, JsonDict]:
     registry = load_yaml(path)
     specs = {}
@@ -366,16 +406,26 @@ def summarize_baselines(
 ) -> JsonDict:
     summary: JsonDict = {"best_k": best_k, "policies": list(policies), "environments": {}}
     for env_id, splits in rows_by_env.items():
-        env_summary = {"splits": {}, "aggregate": {}}
-        aggregate_rows = []
+        env_summary: JsonDict = {"splits": {}, "aggregate": {}}
+        # Cache per-(split, policy) scored rows so the aggregate metric reuses
+        # the same scores instead of re-running the verifier (which, for
+        # python_unit_tests, halves the number of subprocess forks).
+        scored_cache: dict[tuple[str, str], list[JsonDict]] = {}
         for split, rows in sorted(splits.items()):
-            aggregate_rows.extend(rows)
-            env_summary["splits"][split] = {
-                policy: evaluate_policy(rows, policy=policy, best_k=best_k, run_code=run_code) for policy in policies
-            }
-        env_summary["aggregate"] = {
-            policy: evaluate_policy(aggregate_rows, policy=policy, best_k=best_k, run_code=run_code) for policy in policies
-        }
+            env_summary["splits"][split] = {}
+            for policy in policies:
+                scored = score_rows(rows, policy=policy, best_k=best_k, run_code=run_code)
+                scored_cache[(split, policy)] = scored
+                env_summary["splits"][split][policy] = aggregate_scored_rows(
+                    scored, policy=policy, best_k=best_k
+                )
+        for policy in policies:
+            combined: list[JsonDict] = []
+            for split in sorted(splits):
+                combined.extend(scored_cache[(split, policy)])
+            env_summary["aggregate"][policy] = aggregate_scored_rows(
+                combined, policy=policy, best_k=best_k
+            )
         summary["environments"][env_id] = env_summary
     return summary