feat(evolution): diagnose.py metrics + attribution (Stage 1, #8)

tests/unit/test_diagnose.py

@@ -0,0 +1,508 @@
+"""core/evolution/diagnose.py 单元测试。
+
+覆盖：
+- 7 个规则指标（空输入、边界、典型值）
+- extract_json_from_response（三策略 + 拒绝非 dict + 垃圾输入）
+- attribute_error 瀑布（4 条路径）
+- question_soft_score（空→None）
+- aggregate_soft（跳过 None、全 None→None）
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+
+from core.evolution.diagnose import (
+    _trigrams,
+    aggregate_soft,
+    attribute_error,
+    calc_budget_usage,
+    calc_confidence_calibration,
+    calc_format_compliance,
+    calc_level_jump_pattern,
+    calc_repeat_visit_rate,
+    calc_search_keyword_repetition,
+    calc_tool_usage,
+    extract_json_from_response,
+    extract_rule_metrics,
+    question_soft_score,
+)
+from core.evolution.types import (
+    QuestionMetrics,
+    SpanMetrics,
+)
+
+# =========================================================================
+# 工厂函数
+# =========================================================================
+
+
+def _make_qm(
+    question_id: str = "q1",
+    video_id: str = "v1",
+    task_type: str = "Action Reasoning",
+    correct: bool = False,
+    format_compliance: float = 1.0,
+    budget_usage: float = 0.5,
+    confidence_calibration: str = "calibrated",
+    repeat_visit_rate: float = 0.0,
+    search_keyword_repetition: float = 0.0,
+    level_jump_pattern: str = "",
+    tool_usage: dict[str, int] | None = None,
+    span_metrics: list[SpanMetrics] | None = None,
+    missed_nodes: list[str] | None = None,
+    skill_adherence: list[Any] | None = None,
+    confirmation_bias: bool | None = None,
+    evidence_sufficient: bool | None = None,
+    degraded: bool = False,
+) -> QuestionMetrics:
+    """构造 QuestionMetrics，非关键字段使用合理默认值。"""
+    return QuestionMetrics(
+        question_id=question_id,
+        video_id=video_id,
+        task_type=task_type,
+        correct=correct,
+        format_compliance=format_compliance,
+        budget_usage=budget_usage,
+        confidence_calibration=confidence_calibration,
+        repeat_visit_rate=repeat_visit_rate,
+        search_keyword_repetition=search_keyword_repetition,
+        level_jump_pattern=level_jump_pattern,
+        tool_usage=tool_usage or {},
+        span_metrics=span_metrics or [],
+        missed_nodes=missed_nodes or [],
+        skill_adherence=skill_adherence or [],
+        confirmation_bias=confirmation_bias,
+        evidence_sufficient=evidence_sufficient,
+        degraded=degraded,
+    )
+
+
+def _make_span(
+    step: int = 0,
+    tool_name: str = "view_node",
+    extraction_completeness: float = 0.8,
+    hallucination_rate: float = 0.1,
+) -> SpanMetrics:
+    """构造 SpanMetrics 快捷工厂。"""
+    return SpanMetrics(
+        step=step,
+        tool_name=tool_name,
+        extraction_completeness=extraction_completeness,
+        hallucination_rate=hallucination_rate,
+    )
+
+
+# =========================================================================
+# A. 规则指标测试
+# =========================================================================
+
+
+class TestCalcFormatCompliance:
+    """calc_format_compliance 测试。"""
+
+    def test_empty_returns_one(self) -> None:
+        """空列表返回 1.0。"""
+        assert calc_format_compliance([]) == 1.0
+
+    def test_all_compliant(self) -> None:
+        """全部合规返回 1.0。"""
+        raw = json.dumps({"reflect": {}, "plan": {}, "action": {}})
+        assert calc_format_compliance([raw, raw]) == 1.0
+
+    def test_none_compliant(self) -> None:
+        """全部不合规返回 0.0。"""
+        assert calc_format_compliance(["not json", '{"foo": 1}']) == 0.0
+
+    def test_partial_compliance(self) -> None:
+        """部分合规返回正确比例。"""
+        good = json.dumps({"reflect": {}, "plan": {}, "action": {}})
+        bad = json.dumps({"reflect": {}, "plan": {}})
+        assert calc_format_compliance([good, bad]) == 0.5
+
+
+class TestCalcBudgetUsage:
+    """calc_budget_usage 测试。"""
+
+    def test_typical(self) -> None:
+        """典型值。"""
+        assert calc_budget_usage(5, 10) == 0.5
+
+    def test_full_budget(self) -> None:
+        """用满预算。"""
+        assert calc_budget_usage(10, 10) == 1.0
+
+    def test_zero_steps(self) -> None:
+        """未使用步数。"""
+        assert calc_budget_usage(0, 10) == 0.0
+
+    def test_zero_max_steps_raises(self) -> None:
+        """max_steps=0 应抛出 ZeroDivisionError（P5: 不掩盖错误）。"""
+        with pytest.raises(ZeroDivisionError):
+            calc_budget_usage(5, 0)
+
+
+class TestCalcConfidenceCalibration:
+    """calc_confidence_calibration 测试。"""
+
+    def test_high_conf_wrong(self) -> None:
+        """高置信度答错。"""
+        assert calc_confidence_calibration(0.7, correct=False) == "high_conf_wrong"
+        assert calc_confidence_calibration(0.9, correct=False) == "high_conf_wrong"
+
+    def test_low_conf_right(self) -> None:
+        """低置信度答对。"""
+        assert calc_confidence_calibration(0.3, correct=True) == "low_conf_right"
+        assert calc_confidence_calibration(0.49, correct=True) == "low_conf_right"
+
+    def test_calibrated(self) -> None:
+        """正常校准。"""
+        assert calc_confidence_calibration(0.5, correct=True) == "calibrated"
+        assert calc_confidence_calibration(0.7, correct=True) == "calibrated"
+        assert calc_confidence_calibration(0.3, correct=False) == "calibrated"
+
+    def test_boundary_high(self) -> None:
+        """边界值: 0.7 答错。"""
+        assert calc_confidence_calibration(0.7, correct=False) == "high_conf_wrong"
+
+    def test_boundary_low(self) -> None:
+        """边界值: 0.5 答对不算 low_conf_right。"""
+        assert calc_confidence_calibration(0.5, correct=True) == "calibrated"
+
+
+class TestCalcRepeatVisitRate:
+    """calc_repeat_visit_rate 测试。"""
+
+    def test_empty(self) -> None:
+        """空列表返回 0.0。"""
+        assert calc_repeat_visit_rate([]) == 0.0
+
+    def test_no_repeats(self) -> None:
+        """无重复。"""
+        assert calc_repeat_visit_rate(["a", "b", "c"]) == 0.0
+
+    def test_all_same(self) -> None:
+        """全重复。"""
+        rate = calc_repeat_visit_rate(["a", "a", "a"])
+        assert abs(rate - (1 - 1 / 3)) < 1e-9
+
+    def test_partial_repeats(self) -> None:
+        """部分重复。"""
+        rate = calc_repeat_visit_rate(["a", "b", "a"])
+        assert abs(rate - (1 - 2 / 3)) < 1e-9
+
+
+class TestTrigrams:
+    """_trigrams 辅助函数测试。"""
+
+    def test_short_string(self) -> None:
+        """不足 3 字符返回空集。"""
+        assert _trigrams("ab") == set()
+        assert _trigrams("") == set()
+
+    def test_exact_three(self) -> None:
+        """恰好 3 字符。"""
+        assert _trigrams("abc") == {"abc"}
+
+    def test_longer(self) -> None:
+        """多字符。"""
+        assert _trigrams("abcd") == {"abc", "bcd"}
+
+
+class TestCalcSearchKeywordRepetition:
+    """calc_search_keyword_repetition 测试。"""
+
+    def test_single_query(self) -> None:
+        """单条查询返回 0.0。"""
+        assert calc_search_keyword_repetition(["hello"]) == 0.0
+
+    def test_empty(self) -> None:
+        """空列表返回 0.0。"""
+        assert calc_search_keyword_repetition([]) == 0.0
+
+    def test_identical_queries(self) -> None:
+        """完全相同的连续查询，Jaccard=1.0。"""
+        assert calc_search_keyword_repetition(["hello world", "hello world"]) == 1.0
+
+    def test_disjoint_queries(self) -> None:
+        """完全不同的查询，Jaccard 接近 0。"""
+        score = calc_search_keyword_repetition(["aaa", "zzz"])
+        assert score == 0.0
+
+    def test_takes_max_across_pairs(self) -> None:
+        """取连续对的最大值。"""
+        score = calc_search_keyword_repetition(["aaa", "zzz", "zzz"])
+        assert score == 1.0  # 第二对完全相同
+
+
+class TestCalcLevelJumpPattern:
+    """calc_level_jump_pattern 测试。"""
+
+    def test_empty(self) -> None:
+        """空列表返回空字符串。"""
+        assert calc_level_jump_pattern([]) == ""
+
+    def test_typical(self) -> None:
+        """典型节点 ID 序列。"""
+        result = calc_level_jump_pattern(["vid_L1_001", "vid_L2_003", "vid_L3_005"])
+        assert result == "L1→L2→L3"
+
+    def test_no_match(self) -> None:
+        """无匹配的节点 ID 被跳过。"""
+        assert calc_level_jump_pattern(["no_level_here"]) == ""
+
+    def test_mixed(self) -> None:
+        """混合匹配和非匹配。"""
+        result = calc_level_jump_pattern(["vid_L2_001", "bad_id", "vid_L1_003"])
+        assert result == "L2→L1"
+
+
+class TestCalcToolUsage:
+    """calc_tool_usage 测试。"""
+
+    def test_empty(self) -> None:
+        """空列表返回空字典。"""
+        assert calc_tool_usage([]) == {}
+
+    def test_counts(self) -> None:
+        """正确计数。"""
+        result = calc_tool_usage(["view_node", "search_similar", "view_node"])
+        assert result == {"view_node": 2, "search_similar": 1}
+
+
+class TestExtractRuleMetrics:
+    """extract_rule_metrics 测试。"""
+
+    def test_basic_extraction(self) -> None:
+        """基本规则指标提取。"""
+        prediction = {
+            "steps_json": [
+                {
+                    "tool_call": {
+                        "tool": "view_node",
+                        "args": {"node_id": "vid_L1_001"},
+                    }
+                },
+                {
+                    "tool_call": {
+                        "tool": "search_similar",
+                        "args": {"query": "test query"},
+                    }
+                },
+            ],
+            "correct": True,
+        }
+        result = extract_rule_metrics(prediction, [], max_steps=10)
+        assert result["budget_usage"] == 0.2
+        assert result["tool_usage"] == {"view_node": 1, "search_similar": 1}
+        assert "L1" in result["level_jump_pattern"]
+
+    def test_empty_prediction(self) -> None:
+        """空预测。"""
+        result = extract_rule_metrics({}, [], max_steps=10)
+        assert result["format_compliance"] == 1.0
+        assert result["budget_usage"] == 0.0
+
+
+# =========================================================================
+# B. JSON 提取测试
+# =========================================================================
+
+
+class TestExtractJsonFromResponse:
+    """extract_json_from_response 测试。"""
+
+    def test_fenced_block(self) -> None:
+        """从 markdown 代码块提取。"""
+        raw = '```json\n{"key": "value"}\n```'
+        assert extract_json_from_response(raw) == {"key": "value"}
+
+    def test_fenced_block_no_json_tag(self) -> None:
+        """从无 json 标签的代码块提取。"""
+        raw = '```\n{"key": "value"}\n```'
+        assert extract_json_from_response(raw) == {"key": "value"}
+
+    def test_outermost_braces(self) -> None:
+        """从最外层花括号提取。"""
+        raw = 'Some text before {"result": 42} and after'
+        assert extract_json_from_response(raw) == {"result": 42}
+
+    def test_non_dict_raises(self) -> None:
+        """非 dict 类型抛出 ValueError。"""
+        raw = "[1, 2, 3]"
+        with pytest.raises(ValueError, match="无法从 LLM 回复中提取 JSON"):
+            extract_json_from_response(raw)
+
+    def test_garbage_raises(self) -> None:
+        """完全无法解析的输入抛出 ValueError。"""
+        with pytest.raises(ValueError, match="无法从 LLM 回复中提取 JSON"):
+            extract_json_from_response("this is not json at all !!!")
+
+    def test_nested_braces(self) -> None:
+        """嵌套花括号正确处理。"""
+        inner = {"nested": {"deep": True}}
+        raw = f"Result: {json.dumps(inner)}"
+        assert extract_json_from_response(raw) == inner
+
+    def test_fenced_block_non_dict_falls_through(self) -> None:
+        """代码块中是列表时，回退到后续策略。"""
+        raw = '```json\n[1,2,3]\n``` {"fallback": true}'
+        result = extract_json_from_response(raw)
+        assert result == {"fallback": True}
+
+
+# =========================================================================
+# C. question_soft_score / aggregate_soft 测试
+# =========================================================================
+
+
+class TestQuestionSoftScore:
+    """question_soft_score 测试。"""
+
+    def test_empty_returns_none(self) -> None:
+        """空 span 列表返回 None。"""
+        assert question_soft_score([]) is None
+
+    def test_single_span(self) -> None:
+        """单个 span 的计算。"""
+        span = _make_span(extraction_completeness=0.8, hallucination_rate=0.2)
+        score = question_soft_score([span])
+        # (0.8 + (1.0 - 0.2)) / 2 = (0.8 + 0.8) / 2 = 0.8
+        assert score is not None
+        assert abs(score - 0.8) < 1e-9
+
+    def test_multiple_spans(self) -> None:
+        """多个 span 取均值。"""
+        span1 = _make_span(extraction_completeness=1.0, hallucination_rate=0.0)
+        span2 = _make_span(extraction_completeness=0.6, hallucination_rate=0.4)
+        score = question_soft_score([span1, span2])
+        # span1: (1.0 + 1.0) / 2 = 1.0
+        # span2: (0.6 + 0.6) / 2 = 0.6
+        # mean: (1.0 + 0.6) / 2 = 0.8
+        assert score is not None
+        assert abs(score - 0.8) < 1e-9
+
+    def test_perfect_span(self) -> None:
+        """完美 span。"""
+        span = _make_span(extraction_completeness=1.0, hallucination_rate=0.0)
+        score = question_soft_score([span])
+        assert score == 1.0
+
+
+class TestAggregateSoft:
+    """aggregate_soft 测试。"""
+
+    def test_all_none_returns_none(self) -> None:
+        """全部 None 返回 None。"""
+        assert aggregate_soft([None, None, None]) is None
+
+    def test_empty_returns_none(self) -> None:
+        """空列表返回 None。"""
+        assert aggregate_soft([]) is None
+
+    def test_skip_none(self) -> None:
+        """跳过 None 计算均值。"""
+        result = aggregate_soft([0.8, None, 0.6])
+        assert result is not None
+        assert abs(result - 0.7) < 1e-9
+
+    def test_all_valid(self) -> None:
+        """全部有效。"""
+        result = aggregate_soft([0.5, 0.7, 0.9])
+        assert result is not None
+        assert abs(result - 0.7) < 1e-9
+
+
+# =========================================================================
+# D. attribute_error 瀑布测试
+# =========================================================================
+
+
+class TestAttributeError:
+    """attribute_error 瀑布规则测试。"""
+
+    def test_extraction_failure_low_completeness(self) -> None:
+        """avg completeness < 0.5 → extraction_failure。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.3, hallucination_rate=0.1),
+                _make_span(extraction_completeness=0.4, hallucination_rate=0.1),
+            ],
+            missed_nodes=["node_1"],  # 有遗漏，但 extraction 优先
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "extraction_failure"
+        assert result.question_id == "q1"
+
+    def test_extraction_failure_high_hallucination(self) -> None:
+        """max hallucination > 0.5 → extraction_failure。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.9, hallucination_rate=0.6),
+            ],
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "extraction_failure"
+
+    def test_search_failure(self) -> None:
+        """有遗漏节点 → search_failure。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.8, hallucination_rate=0.1),
+            ],
+            missed_nodes=["node_1", "node_2"],
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "search_failure"
+
+    def test_reasoning_failure(self) -> None:
+        """evidence_sufficient=True → reasoning_failure。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.8, hallucination_rate=0.1),
+            ],
+            missed_nodes=[],
+            evidence_sufficient=True,
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "reasoning_failure"
+
+    def test_mixed_evidence_none(self) -> None:
+        """evidence_sufficient=None → mixed。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.8, hallucination_rate=0.1),
+            ],
+            missed_nodes=[],
+            evidence_sufficient=None,
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "mixed"
+
+    def test_mixed_evidence_false(self) -> None:
+        """evidence_sufficient=False → mixed。"""
+        qm = _make_qm(
+            span_metrics=[
+                _make_span(extraction_completeness=0.8, hallucination_rate=0.1),
+            ],
+            missed_nodes=[],
+            evidence_sufficient=False,
+        )
+        result = attribute_error(qm)
+        assert result.error_type == "mixed"
+
+    def test_no_spans_extraction(self) -> None:
+        """无 span 时 avg_completeness=0 (< 0.5) → extraction_failure。"""
+        qm = _make_qm(span_metrics=[], missed_nodes=["node_x"])
+        result = attribute_error(qm)
+        # _mean([]) = 0.0 < 0.5 → extraction_failure
+        assert result.error_type == "extraction_failure"
+
+    def test_reasoning_failure_type_is_none(self) -> None:
+        """attribute_error 不设 reasoning_failure_type（由后续阶段补充）。"""
+        qm = _make_qm(evidence_sufficient=True)
+        result = attribute_error(qm)
+        assert result.reasoning_failure_type is None