775 lines
26 KiB
Python
775 lines
26 KiB
Python
"""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
|
||
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,
|
||
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,
|
||
merge_system_packs,
|
||
merge_tool_packs,
|
||
question_soft_score,
|
||
run_diagnosis,
|
||
)
|
||
from core.evolution.types import (
|
||
CaseSample,
|
||
DiagnosePrompts,
|
||
DiagnosisResult,
|
||
QuestionMetrics,
|
||
SkillStepAdherence,
|
||
SpanMetrics,
|
||
SystemCasePack,
|
||
ToolCasePack,
|
||
)
|
||
|
||
# =========================================================================
|
||
# 工厂函数
|
||
# =========================================================================
|
||
|
||
|
||
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
|
||
|
||
|
||
# =========================================================================
|
||
# 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
|