From dc091361c33fe8089cb98be5eb974a6f73abd94d Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 10:04:48 -0400 Subject: [PATCH] feat(evolution): diagnose.py aggregation + case packs + run_diagnosis (#8) --- core/evolution/diagnose.py | 1291 ++++++++++++++++++++++++++++++++++- tests/unit/test_diagnose.py | 266 ++++++++ 2 files changed, 1554 insertions(+), 3 deletions(-) diff --git a/core/evolution/diagnose.py b/core/evolution/diagnose.py index b448cb3..204e40a 100644 --- a/core/evolution/diagnose.py +++ b/core/evolution/diagnose.py @@ -1,6 +1,6 @@ -"""诊断引擎 — 指标计算与 judge 辅助函数。 +"""诊断引擎 — 指标计算、judge 辅助、聚合、案例包构建与入口。 -Stage 1 指标管线的可提取内核。包含： +两阶段诊断管线的可提取内核。包含： - 7 个规则指标的纯函数计算 - JSON 提取工具 - 5 个 LLM judge 评估函数（async） @@ -8,30 +8,43 @@ Stage 1 指标管线的可提取内核。包含： - 错误归因瀑布 attribute_error - defect/lapse 病因判别 classify_defect_vs_lapse - 降级指标生成 _make_degraded_metrics +- D2-D5 聚合函数 +- 案例包构建（skill / system / tool） +- merge 函数 +- run_diagnosis 入口 不依赖 app/ 或 adapters/。所有 LLM 交互通过 LLMProvider Protocol 注入。 """ from __future__ import annotations +import asyncio import json import re -from collections import Counter +from collections import Counter, defaultdict +from statistics import median from typing import TYPE_CHECKING, Any from json_repair import repair_json from loguru import logger from core.evolution.types import ( + CaseSample, DiagnosePrompts, + DiagnosisResult, ErrorAttribution, QuestionMetrics, + SkillCasePack, SkillStepAdherence, SpanMetrics, + SystemCasePack, + ToolCasePack, ) if TYPE_CHECKING: + from core.evolution.protocols import RunLog, SkillStore from core.protocols import LLMProvider + from core.types import GeneratedQuestion # ========================================================================= # 常量 @@ -1018,3 +1031,1275 @@ def _make_degraded_metrics(prediction: dict[str, Any], max_steps: int) -> Questi evidence_sufficient=None, degraded=True, ) + + +# ========================================================================= +# G. 辅助函数 — _percentile +# ========================================================================= + + +def _percentile(values: list[float], pct: float) -> float: + """按线性插值计算分位数。 + + 参数: + values: 浮点值列表。 + pct: 分位数位置 [0.0, 1.0]。 + + 返回: + 分位数值；空列表返回 0.0；单元素返回该元素。 + """ + if not values: + return 0.0 + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + position = pct * (len(ordered) - 1) + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + weight = position - lower + return ordered[lower] * (1 - weight) + ordered[upper] * weight + + +# ========================================================================= +# H. D2-D5 聚合函数 +# ========================================================================= + + +def _parse_level_sequence(level_jump_pattern: str) -> list[str]: + """从层级跳转文本中提取层级序列。 + + 参数: + level_jump_pattern: 层级跳转模式字符串。 + + 返回: + 层级标签列表。 + """ + return re.findall(r"L\d+", level_jump_pattern or "") + + +def _extract_level_from_node(node_id: str) -> str | None: + """从节点 ID 中提取 L1/L2/L3 层级。 + + 参数: + node_id: 节点标识字符串。 + + 返回: + 层级标签或 None。 + """ + match = re.search(r"L([123])", node_id or "") + if match is None: + return None + return f"L{match.group(1)}" + + +def aggregate_d2(all_metrics: list[QuestionMetrics]) -> dict[str, dict]: + """D2: 按工具聚合 span 级质量指标。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + + 返回: + {tool_name: {avg_completeness, avg_hallucination, n_calls, top_missed, top_hallucinated}}。 + """ + grouped: dict[str, list[SpanMetrics]] = defaultdict(list) + for qm in all_metrics: + for span in qm.span_metrics: + grouped[span.tool_name].append(span) + + result: dict[str, dict] = {} + for tool_name, spans in grouped.items(): + missed_counter: Counter[str] = Counter() + hallucinated_counter: Counter[str] = Counter() + for span in spans: + missed_counter.update(span.missed_info_tags) + hallucinated_counter.update(span.hallucination_tags) + result[tool_name] = { + "avg_completeness": _mean([span.extraction_completeness for span in spans]), + "avg_hallucination": _mean([span.hallucination_rate for span in spans]), + "n_calls": len(spans), + "top_missed": [[tag, count] for tag, count in missed_counter.most_common()], + "top_hallucinated": [[tag, count] for tag, count in hallucinated_counter.most_common()], + } + return result + + +def aggregate_d3(all_metrics: list[QuestionMetrics]) -> dict[str, dict]: + """D3: 按题型与正误拆分搜索行为统计。 + + 注意: 键 ``avg_steps`` 实际存储 budget_usage 均值（TRM4 历史命名，保持兼容）。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + + 返回: + {task_type: {correct: {...}, incorrect: {...}}}。 + """ + grouped: dict[str, dict[str, list[QuestionMetrics]]] = defaultdict( + lambda: {"correct": [], "incorrect": []} + ) + for qm in all_metrics: + bucket = "correct" if qm.correct else "incorrect" + grouped[qm.task_type][bucket].append(qm) + + result: dict[str, dict] = {} + for task_type, task_groups in grouped.items(): + task_result: dict[str, Any] = {} + for bucket_name, metrics_group in task_groups.items(): + task_result[bucket_name] = { + "repeat_visit_rate": _mean([qm.repeat_visit_rate for qm in metrics_group]), + "keyword_repetition": _mean([qm.search_keyword_repetition for qm in metrics_group]), + "l3_usage_rate": _mean( + [ + 1.0 if "L3" in _parse_level_sequence(qm.level_jump_pattern) else 0.0 + for qm in metrics_group + ] + ), + "observe_frame_rate": _mean( + [ + 1.0 if qm.tool_usage.get("observe_frame", 0) > 0 else 0.0 + for qm in metrics_group + ] + ), + "avg_steps": _mean([qm.budget_usage for qm in metrics_group]), + "n_questions": len(metrics_group), + } + + incorrect_group = task_groups["incorrect"] + level_counts = {"L1": 0, "L2": 0, "L3": 0} + for qm in incorrect_group: + for node_id in qm.missed_nodes: + level = _extract_level_from_node(node_id) + if level in level_counts: + level_counts[level] += 1 + + task_result["incorrect"]["missed_nodes_rate"] = _mean( + [1.0 if qm.missed_nodes else 0.0 for qm in incorrect_group] + ) + task_result["incorrect"]["missed_node_levels"] = level_counts + result[task_type] = task_result + return result + + +def aggregate_d4(all_metrics: list[QuestionMetrics]) -> dict[str, dict]: + """D4: 按题型聚合 skill step 遵循与收益差异。 + + 除以零时返回 0.0。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + + 返回: + {task_type: {overall_adherence, n_questions, steps: {step_label: {...}}}}。 + """ + grouped: dict[str, list[QuestionMetrics]] = defaultdict(list) + for qm in all_metrics: + grouped[qm.task_type].append(qm) + + result: dict[str, dict] = {} + for task_type, metrics_group in grouped.items(): + total_steps = 0 + adhered_steps = 0 + step_stats: dict[str, dict[str, int]] = defaultdict( + lambda: { + "adhered": 0, + "deviated": 0, + "correct_adhered": 0, + "correct_deviated": 0, + } + ) + + for qm in metrics_group: + for step in qm.skill_adherence: + total_steps += 1 + if step.adhered: + adhered_steps += 1 + step_stats[step.step_label]["adhered"] += 1 + step_stats[step.step_label]["correct_adhered"] += int(qm.correct) + else: + step_stats[step.step_label]["deviated"] += 1 + step_stats[step.step_label]["correct_deviated"] += int(qm.correct) + + task_steps: dict[str, dict[str, float]] = {} + for step_label, stats in step_stats.items(): + adhered_count = stats["adhered"] + deviated_count = stats["deviated"] + total_count = adhered_count + deviated_count + acc_adhered = stats["correct_adhered"] / adhered_count if adhered_count > 0 else 0.0 + acc_deviated = stats["correct_deviated"] / deviated_count if deviated_count > 0 else 0.0 + task_steps[step_label] = { + "adherence_rate": adhered_count / total_count if total_count else 0.0, + "acc_adhered": acc_adhered, + "acc_deviated": acc_deviated, + "delta": acc_adhered - acc_deviated, + } + + result[task_type] = { + "overall_adherence": adhered_steps / total_steps if total_steps else 0.0, + "n_questions": len(metrics_group), + "steps": task_steps, + } + return result + + +def aggregate_d5(all_metrics: list[QuestionMetrics]) -> dict[str, Any]: + """D5: 跨题型聚合决策与校准模式。 + + 空输入返回完整零结构（非空字典）。confirmation_bias_rate 过滤 None。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + + 返回: + 包含各模式比率的字典。 + """ + if not all_metrics: + return { + "format_compliance_rate": 0.0, + "budget_usage_median": 0.0, + "budget_usage_p25": 0.0, + "budget_usage_p75": 0.0, + "early_submit_rate": 0.0, + "high_conf_wrong_rate": 0.0, + "low_conf_right_rate": 0.0, + "confirmation_bias_rate": 0.0, + "per_type_bias": {}, + } + + budget_values = [qm.budget_usage for qm in all_metrics] + wrong_metrics = [qm for qm in all_metrics if not qm.correct] + per_type_groups: dict[str, list[QuestionMetrics]] = defaultdict(list) + for qm in all_metrics: + per_type_groups[qm.task_type].append(qm) + + return { + "format_compliance_rate": _mean([qm.format_compliance for qm in all_metrics]), + "budget_usage_median": median(budget_values), + "budget_usage_p25": _percentile(budget_values, 0.25), + "budget_usage_p75": _percentile(budget_values, 0.75), + "early_submit_rate": ( + sum(1 for qm in wrong_metrics if qm.budget_usage < 0.3) / len(wrong_metrics) + if wrong_metrics + else 0.0 + ), + "high_conf_wrong_rate": _mean( + [1.0 if qm.confidence_calibration == "high_conf_wrong" else 0.0 for qm in all_metrics] + ), + "low_conf_right_rate": _mean( + [1.0 if qm.confidence_calibration == "low_conf_right" else 0.0 for qm in all_metrics] + ), + "confirmation_bias_rate": _mean( + [ + 1.0 if qm.confirmation_bias else 0.0 + for qm in all_metrics + if qm.confirmation_bias is not None + ] + ), + "per_type_bias": { + task_type: _mean( + [ + 1.0 if qm.confirmation_bias else 0.0 + for qm in group + if qm.confirmation_bias is not None + ] + ) + for task_type, group in per_type_groups.items() + }, + } + + +# ========================================================================= +# I. 案例包构建 +# ========================================================================= + + +_SEVERITY_FNS: dict[str, Any] = {} + +_MIN_PATTERN_COUNT = 3 + +_TOOL_TARGET_FILES = { + "view_node": [ + "view_node_extract.md", + "view_node_verify.md", + "view_node_children_extract.md", + "view_node_children_verify.md", + ], + "search_similar": ["search_similar_extract.md", "search_similar_verify.md"], + "observe_frame": ["observe_frame_extract.md", "observe_frame_verify.md"], +} + + +def _calc_adherence_rate(adherence_list: list[SkillStepAdherence]) -> float: + """计算 skill adherence 率。 + + 参数: + adherence_list: 技能步骤遵循判定列表。 + + 返回: + 遵循率；空列表返回 0.0。 + """ + if not adherence_list: + return 0.0 + adhered = sum(1 for s in adherence_list if s.adhered) + return adhered / len(adherence_list) + + +def _severity_search_failure(qm: QuestionMetrics) -> tuple[int, float]: + """search_failure 严重度：(missed_nodes 数降序, budget_usage 降序)。 + + 参数: + qm: 单题指标。 + + 返回: + 严重度排序元组。 + """ + return (len(qm.missed_nodes), qm.budget_usage) + + +def _severity_extraction_failure(qm: QuestionMetrics) -> tuple[float, float]: + """extraction_failure 严重度：(max hallucination 降序, 1-avg completeness 降序)。 + + 参数: + qm: 单题指标。 + + 返回: + 严重度排序元组。 + """ + max_hall = max((s.hallucination_rate for s in qm.span_metrics), default=0.0) + avg_comp = _mean([s.extraction_completeness for s in qm.span_metrics]) + return (max_hall, 1.0 - avg_comp) + + +def _severity_reasoning_failure(qm: QuestionMetrics) -> tuple[int, float]: + """reasoning_failure 严重度：(high_conf_wrong 优先, budget_usage 降序)。 + + 参数: + qm: 单题指标。 + + 返回: + 严重度排序元组。 + """ + is_high_conf = 1 if qm.confidence_calibration == "high_conf_wrong" else 0 + return (is_high_conf, qm.budget_usage) + + +def _severity_mixed(qm: QuestionMetrics) -> tuple[float, int]: + """mixed 严重度：(budget_usage 降序, missed_nodes 数降序)。 + + 参数: + qm: 单题指标。 + + 返回: + 严重度排序元组。 + """ + return (qm.budget_usage, len(qm.missed_nodes)) + + +_SEVERITY_FNS = { + "search_failure": _severity_search_failure, + "extraction_failure": _severity_extraction_failure, + "reasoning_failure": _severity_reasoning_failure, + "mixed": _severity_mixed, +} + + +def _make_case_sample( + qm: QuestionMetrics, + prediction: dict[str, Any], + trace: list[dict[str, Any]], + error_type: str | None, + selection_reason: str, +) -> CaseSample: + """从 QuestionMetrics 和 prediction 构造 CaseSample。 + + 参数: + qm: 单题指标。 + prediction: 单题预测记录。 + trace: 完整推理轨迹。 + error_type: 错误类型；正确题为 None。 + selection_reason: 被选为案例的原因说明。 + + 返回: + CaseSample 实例。 + """ + return CaseSample( + question_id=qm.question_id, + video_id=qm.video_id, + task_type=qm.task_type, + question=prediction.get("question", ""), + options=prediction.get("options", []), + answer=prediction.get("answer", ""), + prediction=prediction.get("prediction"), + correct=qm.correct, + error_type=error_type, + selection_reason=selection_reason, + metrics={ + "correct": qm.correct, + "error_type": error_type, + "budget_usage": qm.budget_usage, + "confidence_calibration": qm.confidence_calibration, + "repeat_visit_rate": qm.repeat_visit_rate, + "tool_usage": qm.tool_usage, + "missed_nodes": qm.missed_nodes, + "adherence_rate": _calc_adherence_rate(qm.skill_adherence), + "confirmation_bias": qm.confirmation_bias, + "evidence_sufficient": qm.evidence_sufficient, + }, + trace=trace, + ) + + +def _build_skill_case_packs( + all_metrics: list[QuestionMetrics], + error_attributions: list[ErrorAttribution], + traces_by_question: dict[tuple[str, str], list[dict[str, Any]]], + predictions: list[dict[str, Any]], + d3_stats: dict[str, dict], + d4_stats: dict[str, dict], +) -> dict[str, SkillCasePack]: + """按题型构建 Skill 案例包。 + + C3 分流：cause_category=='lapse' 路由进 lapse_notes，不进 failure_cases。 + 单例 fallback：仅 1 条 defect → 降级为 lapse_note。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + error_attributions: 错题归因列表。 + traces_by_question: (video_id, question_id) -> trace 列表。 + predictions: 归一化后的 prediction 字典列表。 + d3_stats: D3 搜索有效性聚合。 + d4_stats: D4 技能遵循聚合。 + + 返回: + {task_type: SkillCasePack} 映射。 + """ + attribution_map: dict[str, ErrorAttribution] = {a.question_id: a for a in error_attributions} + prediction_map: dict[str, dict[str, Any]] = {p["question_id"]: p for p in predictions} + by_task: dict[str, list[QuestionMetrics]] = defaultdict(list) + for qm in all_metrics: + by_task[qm.task_type].append(qm) + + packs: dict[str, SkillCasePack] = {} + for task_type, metrics_group in by_task.items(): + target_file = task_type.lower().replace(" ", "-") + ".md" + + # C3 分流 + wrong_by_error: dict[str, list[QuestionMetrics]] = defaultdict(list) + lapse_notes: list[str] = [] + for qm in metrics_group: + if qm.correct: + continue + attr = attribution_map.get(qm.question_id) + if attr is not None and attr.cause_category == "lapse": + if attr.lapse_note and attr.lapse_note.strip(): + lapse_notes.append(attr.lapse_note) + continue + et = attr.error_type if attr else "mixed" + wrong_by_error[et].append(qm) + + # 单条 fallback + n_body_failures = sum(len(group) for group in wrong_by_error.values()) + if n_body_failures == 1: + [lone_qm] = next(iter(wrong_by_error.values())) + wrong_by_error.clear() + lone_attr = attribution_map.get(lone_qm.question_id) + note = lone_attr.lapse_note if lone_attr and lone_attr.lapse_note else None + lapse_notes.append( + note.strip() if note and note.strip() else "复核该类已有规则，避免重复此类单例失败" + ) + + failure_cases: list[CaseSample] = [] + for error_type, wrong_group in wrong_by_error.items(): + severity_fn = _SEVERITY_FNS.get(error_type, _severity_mixed) + sorted_group = sorted(wrong_group, key=severity_fn, reverse=True) + for qm in sorted_group[:2]: + trace = traces_by_question.get((qm.video_id, qm.question_id), []) + pred = prediction_map.get(qm.question_id, {}) + sv = severity_fn(qm) + reason = f"error_type={error_type}, severity={sv}" + failure_cases.append(_make_case_sample(qm, pred, trace, error_type, reason)) + + # 成功案例 + correct_group = [qm for qm in metrics_group if qm.correct] + n_correct = len(correct_group) + n_total = len(metrics_group) + accuracy = n_correct / n_total if n_total > 0 else 0.0 + + n_success = max(2, len(failure_cases) // 2) + low_accuracy = accuracy <= 0.3 + + if low_accuracy: + sorted_correct = sorted(correct_group, key=lambda qm: qm.budget_usage) + else: + sorted_correct = sorted( + correct_group, + key=lambda qm: ( + -_calc_adherence_rate(qm.skill_adherence), + qm.budget_usage, + ), + ) + + success_cases: list[CaseSample] = [] + for qm in sorted_correct[:n_success]: + trace = traces_by_question.get((qm.video_id, qm.question_id), []) + pred = prediction_map.get(qm.question_id, {}) + adh = _calc_adherence_rate(qm.skill_adherence) + reason = f"adherence={adh:.2f}, budget_usage={qm.budget_usage:.2f}" + if low_accuracy: + reason += ", low_accuracy_pool" + success_cases.append(_make_case_sample(qm, pred, trace, None, reason)) + + # D1 按题型拆分 attribution_distribution + attr_dist: dict[str, int] = Counter( + attribution_map[qm.question_id].error_type + for qm in metrics_group + if not qm.correct and qm.question_id in attribution_map + ) + + stats: dict[str, Any] = { + "n_total": n_total, + "n_correct": n_correct, + "accuracy": accuracy, + "attribution_distribution": dict(attr_dist), + } + if task_type in d3_stats: + stats["correct_vs_incorrect"] = d3_stats[task_type] + if task_type in d4_stats: + stats["overall_adherence"] = d4_stats[task_type].get("overall_adherence", 0.0) + stats["steps"] = d4_stats[task_type].get("steps", {}) + + packs[task_type] = SkillCasePack( + task_type=task_type, + target_file=target_file, + stats=stats, + failure_cases=failure_cases, + success_cases=success_cases, + lapse_notes=lapse_notes, + ) + + return packs + + +def _build_system_case_pack( + all_metrics: list[QuestionMetrics], + traces_by_question: dict[tuple[str, str], list[dict[str, Any]]], + predictions: list[dict[str, Any]], + d5_stats: dict[str, Any], +) -> SystemCasePack | None: + """构建跨题型行为模式案例包。 + + 3 个模式：early_submit / high_conf_wrong / confirmation_bias。 + 每个模式 >= _MIN_PATTERN_COUNT 才纳入。全部不达标则返回 None。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + traces_by_question: (video_id, question_id) -> trace 列表。 + predictions: 归一化后的 prediction 字典列表。 + d5_stats: D5 决策模式聚合。 + + 返回: + SystemCasePack 或 None。 + """ + prediction_map: dict[str, dict[str, Any]] = {p["question_id"]: p for p in predictions} + + early_submit = [qm for qm in all_metrics if not qm.correct and qm.budget_usage < 0.3] + high_conf_wrong = [qm for qm in all_metrics if qm.confidence_calibration == "high_conf_wrong"] + confirmation_bias_cases = [ + qm for qm in all_metrics if qm.confirmation_bias is True and not qm.correct + ] + + patterns: list[tuple[str, list[QuestionMetrics], bool]] = [ + ("early_submit", early_submit, True), + ("high_conf_wrong", high_conf_wrong, False), + ("confirmation_bias", confirmation_bias_cases, False), + ] + + failure_cases: list[CaseSample] = [] + for pattern_name, candidates, sort_asc in patterns: + if len(candidates) < _MIN_PATTERN_COUNT: + continue + sorted_cands = sorted(candidates, key=lambda qm: qm.budget_usage, reverse=not sort_asc) + for qm in sorted_cands[:2]: + trace = traces_by_question.get((qm.video_id, qm.question_id), []) + pred = prediction_map.get(qm.question_id, {}) + reason = f"pattern={pattern_name}, budget_usage={qm.budget_usage:.2f}" + failure_cases.append(_make_case_sample(qm, pred, trace, pattern_name, reason)) + + if not failure_cases: + return None + + # 成功案例 + good_candidates = [ + qm + for qm in all_metrics + if qm.correct + and qm.confidence_calibration == "calibrated" + and qm.confirmation_bias is False + and 0.3 <= qm.budget_usage <= 0.8 + ] + sorted_good = sorted(good_candidates, key=lambda qm: abs(qm.budget_usage - 0.5)) + n_success = max(2, len(failure_cases) // 2) + + success_cases: list[CaseSample] = [] + for qm in sorted_good[:n_success]: + trace = traces_by_question.get((qm.video_id, qm.question_id), []) + pred = prediction_map.get(qm.question_id, {}) + reason = f"calibrated, budget_usage={qm.budget_usage:.2f}" + success_cases.append(_make_case_sample(qm, pred, trace, None, reason)) + + stats = dict(d5_stats) + stats["early_submit_count"] = len(early_submit) + stats["high_conf_wrong_count"] = len(high_conf_wrong) + stats["confirmation_bias_count"] = len(confirmation_bias_cases) + + return SystemCasePack( + stats=stats, + failure_cases=failure_cases, + success_cases=success_cases, + ) + + +def _build_tool_case_packs( + all_metrics: list[QuestionMetrics], + traces_by_question: dict[tuple[str, str], list[dict[str, Any]]], + d2_stats: dict[str, dict], + tree_data_by_video: dict[str, dict[str, Any]], +) -> dict[str, ToolCasePack]: + """按工具构建 Tool Prompt 案例包。 + + 失败 span: 低 completeness 优先取 up to 4，高 hallucination 填满到 4。 + 成功 span: completeness>=0.9 且 hallucination==0.0（精确零）。 + + 参数: + all_metrics: 全部题目的 Stage 1 指标。 + traces_by_question: (video_id, question_id) -> trace 列表。 + d2_stats: D2 工具质量聚合。 + tree_data_by_video: {video_id: tree_data} 缓存。 + + 返回: + {tool_name: ToolCasePack} 映射。 + """ + # 收集所有 span 及其来源信息 + all_spans: list[dict[str, Any]] = [] + for qm in all_metrics: + for span in qm.span_metrics: + traces = traces_by_question.get((qm.video_id, qm.question_id), []) + trace_step: dict[str, Any] = {} + for t in traces: + if t.get("step") == span.step and t.get("tool_name") == span.tool_name: + trace_step = t + break + raw_args = trace_step.get("tool_args", {}) + if isinstance(raw_args, str): + try: + raw_args = json.loads(raw_args) + except (json.JSONDecodeError, ValueError): + raw_args = {} + if not isinstance(raw_args, dict): + raw_args = {} + + all_spans.append( + { + "video_id": qm.video_id, + "question_id": qm.question_id, + "step": span.step, + "tool_name": span.tool_name, + "extraction_completeness": span.extraction_completeness, + "hallucination_rate": span.hallucination_rate, + "missed_info_tags": list(span.missed_info_tags), + "hallucination_tags": list(span.hallucination_tags), + "tool_args": raw_args, + "tool_output": str(trace_step.get("tool_output", "")), + "ground_truth": _get_ground_truth_for_trace( + tree_data_by_video.get(qm.video_id, {}), + span.tool_name, + raw_args, + ), + } + ) + + by_tool: dict[str, list[dict[str, Any]]] = defaultdict(list) + for span_record in all_spans: + by_tool[span_record["tool_name"]].append(span_record) + + packs: dict[str, ToolCasePack] = {} + for tool_name, spans in by_tool.items(): + target_files = _TOOL_TARGET_FILES.get(tool_name, []) + if not target_files: + continue + + # 失败 span + by_low_completeness = sorted(spans, key=lambda s: s["extraction_completeness"]) + by_high_hallucination = sorted(spans, key=lambda s: s["hallucination_rate"], reverse=True) + + selected_keys: set[tuple[str, str, int]] = set() + failure_spans: list[dict[str, Any]] = [] + + for source, label in [ + (by_low_completeness, "low_completeness"), + (by_high_hallucination, "high_hallucination"), + ]: + for span_record in source: + key = (span_record["video_id"], span_record["question_id"], span_record["step"]) + if key in selected_keys: + for fs in failure_spans: + if (fs["video_id"], fs["question_id"], fs["step"]) == key: + if label not in fs["selection_reason"]: + fs["selection_reason"] += f", {label}" + break + continue + if len(selected_keys) >= 4 and label == "high_hallucination": + break + selected_keys.add(key) + failure_spans.append( + { + "video_id": span_record["video_id"], + "question_id": span_record["question_id"], + "step": span_record["step"], + "tool_name": tool_name, + "tool_args": span_record["tool_args"], + "tool_output": span_record["tool_output"], + "ground_truth": span_record["ground_truth"], + "extraction_completeness": span_record["extraction_completeness"], + "hallucination_rate": span_record["hallucination_rate"], + "missed_info_tags": span_record["missed_info_tags"], + "hallucination_tags": span_record["hallucination_tags"], + "selection_reason": label, + } + ) + if len(failure_spans) >= 4: + break + + # 成功 span + good_spans = [ + s + for s in spans + if s["extraction_completeness"] >= 0.9 and s["hallucination_rate"] == 0.0 + ] + good_spans.sort(key=lambda s: s["extraction_completeness"], reverse=True) + n_success = max(2, len(failure_spans) // 2) + + success_spans: list[dict[str, Any]] = [] + for span_record in good_spans[:n_success]: + success_spans.append( + { + "video_id": span_record["video_id"], + "question_id": span_record["question_id"], + "step": span_record["step"], + "tool_name": tool_name, + "tool_args": span_record["tool_args"], + "tool_output": span_record["tool_output"], + "ground_truth": span_record["ground_truth"], + "extraction_completeness": span_record["extraction_completeness"], + "hallucination_rate": span_record["hallucination_rate"], + "missed_info_tags": span_record["missed_info_tags"], + "hallucination_tags": span_record["hallucination_tags"], + "selection_reason": "good_quality", + } + ) + + packs[tool_name] = ToolCasePack( + tool_name=tool_name, + target_files=target_files, + stats=d2_stats.get(tool_name, {}), + failure_spans=failure_spans, + success_spans=success_spans, + ) + + return packs + + +# ========================================================================= +# J. Merge 函数 +# ========================================================================= + + +def _collect_step_stats(packs_stats: list[dict[str, Any]]) -> dict[str, Any]: + """将各 step 的 stats 按 step 收集为列表，不做跨 step 数值聚合。 + + 参数: + packs_stats: 各 step pack 的 stats 字典列表。 + + 返回: + {"per_step": [...]}，列表元素为各非空 step 的 stats 浅拷贝。 + """ + return {"per_step": [dict(stats) for stats in packs_stats if stats]} + + +def merge_system_packs(packs: list[SystemCasePack]) -> SystemCasePack | None: + """将多个 step 的 SystemCasePack 累加为单个。 + + 参数: + packs: 一个 epoch 内各 step 产出的 SystemCasePack 列表。 + + 返回: + 累加后的 SystemCasePack；输入为空列表时返回 None。 + """ + if not packs: + return None + + failure_cases: list[CaseSample] = [] + success_cases: list[CaseSample] = [] + for pack in packs: + failure_cases.extend(pack.failure_cases) + success_cases.extend(pack.success_cases) + + return SystemCasePack( + stats=_collect_step_stats([pack.stats for pack in packs]), + failure_cases=failure_cases, + success_cases=success_cases, + ) + + +def merge_tool_packs(packs: list[ToolCasePack]) -> dict[str, ToolCasePack]: + """将多个 step 的 ToolCasePack 按 tool_name 分组累加。 + + 参数: + packs: 一个 epoch 内各 step 产出的 ToolCasePack 列表。 + + 返回: + {tool_name: 合并后的 ToolCasePack}；输入为空列表时返回空字典。 + """ + by_name: dict[str, list[ToolCasePack]] = defaultdict(list) + for pack in packs: + by_name[pack.tool_name].append(pack) + + merged: dict[str, ToolCasePack] = {} + for tool_name, group in by_name.items(): + failure_spans: list[dict[str, Any]] = [] + success_spans: list[dict[str, Any]] = [] + for pack in group: + failure_spans.extend(pack.failure_spans) + success_spans.extend(pack.success_spans) + + merged[tool_name] = ToolCasePack( + tool_name=tool_name, + target_files=list(group[0].target_files), + stats=_collect_step_stats([pack.stats for pack in group]), + failure_spans=failure_spans, + success_spans=success_spans, + ) + return merged + + +# ========================================================================= +# K. 推理失败子分类 +# ========================================================================= + + +async def _classify_reasoning_failure( + llm: LLMProvider, + prompts: DiagnosePrompts, + prediction: dict[str, Any], + traces: list[dict[str, Any]], +) -> str | None: + """调用 judge 模型细分推理失败类型。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + prediction: 单题预测记录。 + traces: 该题执行轨迹。 + + 返回: + 推理失败子类型字符串；解析失败返回 None（不崩溃）。 + """ + trace_text = _format_trace_text_diagnose(traces) + user_prompt = ( + f"## 题目\n{prediction.get('question', '')}\n\n" + f"## 正确答案\n{prediction.get('answer', '')}\n\n" + f"## Agent 错误预测\n{prediction.get('prediction', '')}\n\n" + f"## 执行轨迹\n{trace_text}" + ) + try: + response = await llm.chat( + [ + {"role": "system", "content": prompts.reasoning_sub}, + {"role": "user", "content": user_prompt}, + ], + ) + parsed = extract_json_from_response(response.content) + failure_type = parsed.get("type") + if not isinstance(failure_type, str) or not failure_type.strip(): + return None + return failure_type + except (ValueError, KeyError): + return None + + +# ========================================================================= +# L. 诊断版 trace 格式化（不截断） +# ========================================================================= + + +def _format_trace_text_diagnose(traces: list[dict]) -> str: + """将 trace 列表格式化为完整文本（诊断版，不截断 thought/output）。 + + 与指标版 _format_trace_text 不同：此版本保留全文。 + + 参数: + traces: trace 字典列表。 + + 返回: + 格式化后的多行文本。 + """ + lines: list[str] = [] + for trace in traces: + args = trace.get("tool_args", {}) + if not isinstance(args, str): + args = json.dumps(args, ensure_ascii=False, sort_keys=True) + lines.append( + f"Step {trace.get('step', '')}: thought={trace.get('thought', '')} | " + f"tool={trace.get('tool_name', '')} | args={args} | " + f"output={trace.get('tool_output', '')}" + ) + return "\n".join(lines) + + +# ========================================================================= +# M. Skill 文件解析辅助 +# ========================================================================= + + +def _resolve_skill_file(skill_store: SkillStore, task_type: str) -> str: + """按题型解析对应 skill 文件名并读取内容。 + + 优先精确匹配 ``{task_type}.md``（小写 + 空格转连字符）， + 找不到则回退 ``default-strategy.md``。 + + 注意: 此为临时本地实现。Task 7 将在 evolve.py 中创建规范版本， + Task 9 会统一收口。 + + 参数: + skill_store: 技能文件读取端口。 + task_type: 题目任务类型。 + + 返回: + skill 文件全文。 + """ + task_filename = f"{task_type.lower().replace(' ', '-')}.md" + available = skill_store.list_skill_files() + if task_filename in available: + return skill_store.read_skill(task_filename) + if "default-strategy.md" in available: + return skill_store.read_skill("default-strategy.md") + return "" + + +# ========================================================================= +# N. INFRA 统计 +# ========================================================================= + + +def _count_infra_excluded( + prediction_rows: list[dict[str, Any]], +) -> tuple[int, list[str]]: + """统计因执行/解析层失败（INFRA）被排除的题。 + + 参数: + prediction_rows: 该 run 的预测行。 + + 返回: + (INFRA 题数, question_id 列表)。 + """ + qids = [ + row["question_id"] + for row in prediction_rows + if row.get("stop_reason") in _INFRA_STOP_REASONS + ] + return len(qids), qids + + +# ========================================================================= +# O. run_diagnosis 入口 +# ========================================================================= + + +async def run_diagnosis( + run_id: str, + questions: list[GeneratedQuestion], + tree_data: dict[str, Any], + llm: LLMProvider, + run_log: RunLog, + skill_store: SkillStore, + prompts: DiagnosePrompts, + *, + concurrency: int, + question_ids: list[str] | None = None, + task_types: list[str] | None = None, + only_incorrect: bool = False, +) -> DiagnosisResult: + """执行两阶段诊断流水线。 + + 流程: + 1. 从 RunLog 获取 predictions 和 traces + 2. 按 question_ids / task_types / only_incorrect 过滤，排除 INFRA stop_reasons + 3. Stage 1: 并发计算单题指标 + 错误归因 + defect/lapse 判别 + 4. 推理失败子分类（串行） + 5. Stage 2: D2-D5 聚合，构建案例包 + 6. 计算 INFRA 统计 + 7. 返回 DiagnosisResult + + 参数: + run_id: 本次运行标识。 + questions: 题目列表。 + tree_data: 树结构字典（多视频时为 {video_id: tree_data}， + 单视频时为单棵树）。 + llm: LLM 调用端口。 + run_log: 实验日志查询端口。 + skill_store: 技能文件读取端口。 + prompts: 诊断模板束。 + concurrency: 并发限制。 + question_ids: 可选的题目 ID 过滤列表。 + task_types: 可选的题型过滤列表。 + only_incorrect: 是否仅处理错题。 + + 返回: + DiagnosisResult 实例。 + """ + # Phase 0: 获取 predictions 和 traces + all_predictions = await run_log.get_predictions(run_id, question_ids=question_ids) + all_trace_rows = await run_log.get_traces(run_id, question_ids=question_ids) + + # 构建 question lookup + question_lookup: dict[str, GeneratedQuestion] = {q.question_id: q for q in questions} + + # 构建 tree_data_by_video + # tree_data 可能是 {video_id: {...}} 或单棵树 + tree_data_by_video: dict[str, dict[str, Any]] = {} + if tree_data and "nodes" in tree_data: + # 单棵树：所有视频共用 + for q in questions: + tree_data_by_video[q.video_id] = tree_data + else: + tree_data_by_video = tree_data # type: ignore[assignment] + + # 过滤 predictions + task_type_filter = set(task_types or []) + question_filter = set(question_ids or []) + filtered_predictions: list[dict[str, Any]] = [] + + for row in all_predictions: + stop_reason = row.get("stop_reason") + if stop_reason in _INFRA_STOP_REASONS: + continue + if task_type_filter and row.get("task_type") not in task_type_filter: + continue + if question_filter and row.get("question_id") not in question_filter: + continue + is_correct = row.get("prediction") == row.get("answer") + if only_incorrect and is_correct: + continue + # 补全 question 信息 + q = question_lookup.get(row.get("question_id", "")) + if q is not None: + row.setdefault("question", q.question) + row.setdefault("options", list(q.options)) + row.setdefault("task_type", q.task_type) + row.setdefault("answer", q.answer) + row.setdefault("question", "") + row.setdefault("options", []) + row["correct"] = row.get("prediction") == row.get("answer") + # 解析 steps_json + raw_steps = row.get("steps_json") + if isinstance(raw_steps, str): + try: + row["steps_json"] = json.loads(raw_steps) + except json.JSONDecodeError: + row["steps_json"] = [] + elif not isinstance(raw_steps, list): + row["steps_json"] = [] + filtered_predictions.append(row) + + # 构建 traces_by_question + traces_by_question: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + for trace_row in all_trace_rows: + key = (trace_row.get("video_id", ""), trace_row.get("question_id", "")) + traces_by_question[key].append(trace_row) + + # 推断 max_steps + observed_steps = [ + len(p.get("steps_json", [])) + for p in filtered_predictions + if isinstance(p.get("steps_json"), list) + ] + max_steps = max(max(observed_steps, default=0), 1) + + # 加载 skill 内容 + skill_cache: dict[str, str] = {} + for p in filtered_predictions: + tt = p.get("task_type", "") + if tt and tt not in skill_cache: + skill_cache[tt] = _resolve_skill_file(skill_store, tt) + + # Stage 1: 并发单题指标 + 归因 + C3 + semaphore = asyncio.Semaphore(concurrency) + worker_results: list[dict[str, Any]] = [] + degraded_question_ids: list[str] = [] + + async def _process_question(prediction: dict[str, Any]) -> dict[str, Any]: + """处理单题：计算指标、归因、C3 判别。""" + async with semaphore: + key = (prediction.get("video_id", ""), prediction.get("question_id", "")) + traces = traces_by_question.get(key, []) + vid = prediction.get("video_id", "") + td = tree_data_by_video.get(vid, {}) + skill_content = skill_cache.get(prediction.get("task_type", ""), "") + + try: + qm = await compute_question_metrics( + prediction=prediction, + traces=traces, + tree_data=td, + skill_content=skill_content, + llm=llm, + prompts=prompts, + max_steps=max_steps, + session_id=run_id, + ) + except ValueError: + logger.warning( + "诊断降级: {} / {} — judge JSON 解析失败", + prediction.get("video_id"), + prediction.get("question_id"), + ) + qm = _make_degraded_metrics(prediction, max_steps) + + attribution: ErrorAttribution | None = None + if not qm.correct: + attribution = attribute_error(qm) + try: + category, note = await classify_defect_vs_lapse( + llm, + prompts, + prediction, + traces, + skill_content, + session_id=run_id, + ) + attribution = ErrorAttribution( + question_id=attribution.question_id, + error_type=attribution.error_type, + reasoning_failure_type=attribution.reasoning_failure_type, + cause_category=category, + lapse_note=note if category == "lapse" else None, + ) + except Exception: + logger.warning( + "C3 判别失败: {} / {}", + prediction.get("video_id"), + prediction.get("question_id"), + ) + + return { + "prediction": prediction, + "traces": traces, + "metrics": qm, + "attribution": attribution, + } + + tasks = [_process_question(p) for p in filtered_predictions] + worker_results = list(await asyncio.gather(*tasks)) if tasks else [] + + # 收集降级题 + for item in worker_results: + if item["metrics"].degraded: + degraded_question_ids.append(item["metrics"].question_id) + + # 推理失败子分类（串行） + for item in worker_results: + attribution = item["attribution"] + if attribution is None or attribution.error_type != "reasoning_failure": + continue + reasoning_type = await _classify_reasoning_failure( + llm, prompts, item["prediction"], item["traces"] + ) + if reasoning_type is not None: + item["attribution"] = ErrorAttribution( + question_id=attribution.question_id, + error_type=attribution.error_type, + reasoning_failure_type=reasoning_type, + cause_category=attribution.cause_category, + lapse_note=attribution.lapse_note, + ) + + # Stage 2: 聚合 + all_metrics = [item["metrics"] for item in worker_results] + error_attributions = [ + item["attribution"] for item in worker_results if item["attribution"] is not None + ] + attribution_distribution = dict(Counter(attr.error_type for attr in error_attributions)) + defect_count = sum(1 for a in error_attributions if a.cause_category == "defect") + lapse_count = sum(1 for a in error_attributions if a.cause_category == "lapse") + reasoning_failure_types = dict( + Counter( + attr.reasoning_failure_type + for attr in error_attributions + if attr.reasoning_failure_type + ) + ) + + d2_stats = aggregate_d2(all_metrics) + d3_stats = aggregate_d3(all_metrics) + d4_stats = aggregate_d4(all_metrics) + d5_stats = aggregate_d5(all_metrics) + + # 构建案例包 + prediction_list = [item["prediction"] for item in worker_results] + skill_packs = _build_skill_case_packs( + all_metrics=all_metrics, + error_attributions=error_attributions, + traces_by_question=traces_by_question, + predictions=prediction_list, + d3_stats=d3_stats, + d4_stats=d4_stats, + ) + system_pack = _build_system_case_pack( + all_metrics=all_metrics, + traces_by_question=traces_by_question, + predictions=prediction_list, + d5_stats=d5_stats, + ) + tool_packs = _build_tool_case_packs( + all_metrics=all_metrics, + traces_by_question=traces_by_question, + d2_stats=d2_stats, + tree_data_by_video=tree_data_by_video, + ) + + # INFRA 统计：限定在 filtered scope（过滤 task_type/question_ids），但不过滤 stop_reason + scoped_rows = [ + row + for row in all_predictions + if not (task_type_filter and row.get("task_type") not in task_type_filter) + and not (question_filter and row.get("question_id") not in question_filter) + ] + infra_count, infra_qids = _count_infra_excluded(scoped_rows) + total = len(scoped_rows) + + return DiagnosisResult( + run_id=run_id, + filter_summary={ + "task_types": sorted(task_type_filter), + "question_ids": sorted(question_filter), + "only_incorrect": only_incorrect, + "total_predictions": len(all_predictions), + "selected_predictions": len(filtered_predictions), + }, + error_attributions=error_attributions, + attribution_distribution=attribution_distribution, + defect_count=defect_count, + lapse_count=lapse_count, + reasoning_failure_types=reasoning_failure_types, + tool_quality=d2_stats, + search_effectiveness=d3_stats, + skill_compliance=d4_stats, + decision_patterns=d5_stats, + skill_case_packs=skill_packs, + system_case_pack=system_pack, + tool_case_packs=tool_packs, + infra_excluded_count=infra_count, + infra_excluded_ratio=(infra_count / total if total else 0.0), + infra_question_ids=infra_qids, + degraded_count=len(degraded_question_ids), + degraded_question_ids=degraded_question_ids, + ) diff --git a/tests/unit/test_diagnose.py b/tests/unit/test_diagnose.py index b205f86..cf32e4f 100644 --- a/tests/unit/test_diagnose.py +++ b/tests/unit/test_diagnose.py @@ -12,11 +12,17 @@ from __future__ import annotations import json from typing import Any +from unittest.mock import AsyncMock, MagicMock import pytest from core.evolution.diagnose import ( + _percentile, _trigrams, + aggregate_d2, + aggregate_d3, + aggregate_d4, + aggregate_d5, aggregate_soft, attribute_error, calc_budget_usage, @@ -28,11 +34,20 @@ from core.evolution.diagnose import ( calc_tool_usage, extract_json_from_response, extract_rule_metrics, + merge_system_packs, + merge_tool_packs, question_soft_score, + run_diagnosis, ) from core.evolution.types import ( + CaseSample, + DiagnosePrompts, + DiagnosisResult, QuestionMetrics, + SkillStepAdherence, SpanMetrics, + SystemCasePack, + ToolCasePack, ) # ========================================================================= @@ -506,3 +521,254 @@ class TestAttributeError: qm = _make_qm(evidence_sufficient=True) result = attribute_error(qm) assert result.reasoning_failure_type is None + + +# ========================================================================= +# E. D2-D5 聚合测试 +# ========================================================================= + + +class TestPercentile: + """_percentile 辅助函数测试。""" + + def test_empty_returns_zero(self) -> None: + """空列表返回 0.0。""" + assert _percentile([], 0.5) == 0.0 + + def test_single_element(self) -> None: + """单元素返回该元素。""" + assert _percentile([42.0], 0.5) == 42.0 + + def test_median_two_elements(self) -> None: + """两元素中位数。""" + assert _percentile([1.0, 3.0], 0.5) == 2.0 + + def test_quartiles(self) -> None: + """四分位线性插值。""" + values = [1.0, 2.0, 3.0, 4.0, 5.0] + assert _percentile(values, 0.0) == 1.0 + assert _percentile(values, 1.0) == 5.0 + p25 = _percentile(values, 0.25) + assert abs(p25 - 2.0) < 1e-9 + + +class TestAggregation: + """D2-D5 聚合函数测试。""" + + def test_d2_empty(self) -> None: + """空输入返回空字典。""" + assert aggregate_d2([]) == {} + + def test_d2_groups_by_tool(self) -> None: + """按工具名分组聚合。""" + qm = _make_qm( + span_metrics=[ + _make_span( + step=0, + tool_name="view_node", + extraction_completeness=0.8, + hallucination_rate=0.2, + ), + _make_span( + step=1, + tool_name="view_node", + extraction_completeness=0.6, + hallucination_rate=0.1, + ), + _make_span( + step=2, + tool_name="search_similar", + extraction_completeness=0.9, + hallucination_rate=0.0, + ), + ] + ) + result = aggregate_d2([qm]) + assert "view_node" in result + assert "search_similar" in result + assert result["view_node"]["n_calls"] == 2 + assert result["search_similar"]["n_calls"] == 1 + assert abs(result["view_node"]["avg_completeness"] - 0.7) < 1e-9 + + def test_d3_empty(self) -> None: + """空输入返回空字典。""" + assert aggregate_d3([]) == {} + + def test_d3_correct_vs_incorrect(self) -> None: + """按正误拆分。""" + qm_correct = _make_qm(correct=True, task_type="T1", budget_usage=0.5) + qm_wrong = _make_qm(correct=False, task_type="T1", budget_usage=0.8) + result = aggregate_d3([qm_correct, qm_wrong]) + assert "T1" in result + assert result["T1"]["correct"]["n_questions"] == 1 + assert result["T1"]["incorrect"]["n_questions"] == 1 + # avg_steps 存储 budget_usage 均值 + assert result["T1"]["correct"]["avg_steps"] == 0.5 + assert result["T1"]["incorrect"]["avg_steps"] == 0.8 + + def test_d4_empty(self) -> None: + """空输入返回空字典。""" + assert aggregate_d4([]) == {} + + def test_d4_adherence_rate(self) -> None: + """技能遵循率计算。""" + qm = _make_qm( + task_type="T1", + correct=True, + skill_adherence=[ + SkillStepAdherence(step_label="S1", adhered=True, description=""), + SkillStepAdherence(step_label="S1", adhered=False, description=""), + ], + ) + result = aggregate_d4([qm]) + assert "T1" in result + assert result["T1"]["overall_adherence"] == 0.5 + + def test_d5_empty_returns_zero_structure(self) -> None: + """空输入返回完整零结构。""" + result = aggregate_d5([]) + assert "early_submit_rate" in result + assert result["early_submit_rate"] == 0.0 + assert "format_compliance_rate" in result + assert "budget_usage_median" in result + assert "confirmation_bias_rate" in result + assert "per_type_bias" in result + assert result["per_type_bias"] == {} + + def test_d5_with_data(self) -> None: + """有数据时正确计算。""" + qm1 = _make_qm( + correct=True, + budget_usage=0.5, + format_compliance=1.0, + confidence_calibration="calibrated", + confirmation_bias=False, + ) + qm2 = _make_qm( + correct=False, + budget_usage=0.2, + format_compliance=0.8, + confidence_calibration="high_conf_wrong", + confirmation_bias=True, + ) + result = aggregate_d5([qm1, qm2]) + assert result["format_compliance_rate"] == 0.9 + assert result["high_conf_wrong_rate"] == 0.5 + assert result["early_submit_rate"] == 1.0 # 1 wrong with budget<0.3 + + +# ========================================================================= +# F. Merge 函数测试 +# ========================================================================= + + +class TestMerge: + """merge_system_packs / merge_tool_packs 测试。""" + + def test_merge_system_packs_none_on_empty(self) -> None: + """空列表返回 None。""" + assert merge_system_packs([]) is None + + def test_merge_system_packs_wraps_stats(self) -> None: + """stats 包裹为 per_step 列表。""" + pack = SystemCasePack(stats={"a": 1}, failure_cases=[], success_cases=[]) + merged = merge_system_packs([pack, pack]) + assert merged is not None + assert "per_step" in merged.stats + assert len(merged.stats["per_step"]) == 2 + + def test_merge_system_packs_concats_cases(self) -> None: + """failure/success cases 拼接。""" + case = CaseSample( + question_id="q1", + video_id="v1", + task_type="T1", + question="q", + options=[], + answer="a", + prediction="b", + correct=False, + error_type="mixed", + selection_reason="test", + metrics={}, + trace=[], + ) + p1 = SystemCasePack(stats={}, failure_cases=[case], success_cases=[]) + p2 = SystemCasePack(stats={}, failure_cases=[case], success_cases=[case]) + merged = merge_system_packs([p1, p2]) + assert merged is not None + assert len(merged.failure_cases) == 2 + assert len(merged.success_cases) == 1 + + def test_merge_tool_packs_empty(self) -> None: + """空列表返回空字典。""" + assert merge_tool_packs([]) == {} + + def test_merge_tool_packs_groups_by_name(self) -> None: + """同名工具合并。""" + p1 = ToolCasePack( + tool_name="view_node", + target_files=["f1.md"], + stats={"x": 1}, + failure_spans=[{"a": 1}], + success_spans=[], + ) + p2 = ToolCasePack( + tool_name="view_node", + target_files=["f1.md"], + stats={"x": 2}, + failure_spans=[{"b": 2}], + success_spans=[{"c": 3}], + ) + merged = merge_tool_packs([p1, p2]) + assert "view_node" in merged + vn = merged["view_node"] + assert len(vn.failure_spans) == 2 + assert len(vn.success_spans) == 1 + assert "per_step" in vn.stats + + +# ========================================================================= +# G. run_diagnosis 入口测试 +# ========================================================================= + + +class TestRunDiagnosis: + """run_diagnosis 入口测试。""" + + def test_empty_predictions_returns_empty_result(self) -> None: + """无预测时返回空 DiagnosisResult。""" + import asyncio + + mock_log = AsyncMock() + mock_log.get_predictions.return_value = [] + mock_log.get_traces.return_value = [] + mock_llm = AsyncMock() + mock_store = MagicMock() + mock_store.list_skill_files.return_value = [] + prompts = DiagnosePrompts( + defect_vs_lapse="", + reasoning_sub="", + span_eval_system="", + span_eval_user="", + missed_nodes="", + skill_adherence="", + confirmation_bias="", + evidence_sufficiency="", + ) + result = asyncio.run( + run_diagnosis( + "run1", + [], + {}, + mock_llm, + mock_log, + mock_store, + prompts, + concurrency=1, + ) + ) + assert isinstance(result, DiagnosisResult) + assert result.run_id == "run1" + assert result.error_attributions == [] + assert result.degraded_count == 0