From 49d6fe8f51eb15169750beb6dce53201086a3f5d Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 09:55:22 -0400 Subject: [PATCH] feat(evolution): diagnose.py metrics + attribution (Stage 1, #8) --- core/evolution/diagnose.py | 1020 +++++++++++++++++++++++++++++++++++ tests/unit/test_diagnose.py | 508 +++++++++++++++++ 2 files changed, 1528 insertions(+) create mode 100644 core/evolution/diagnose.py create mode 100644 tests/unit/test_diagnose.py diff --git a/core/evolution/diagnose.py b/core/evolution/diagnose.py new file mode 100644 index 0000000..b448cb3 --- /dev/null +++ b/core/evolution/diagnose.py @@ -0,0 +1,1020 @@ +"""诊断引擎 — 指标计算与 judge 辅助函数。 + +Stage 1 指标管线的可提取内核。包含： +- 7 个规则指标的纯函数计算 +- JSON 提取工具 +- 5 个 LLM judge 评估函数（async） +- 单题指标编排 compute_question_metrics +- 错误归因瀑布 attribute_error +- defect/lapse 病因判别 classify_defect_vs_lapse +- 降级指标生成 _make_degraded_metrics + +不依赖 app/ 或 adapters/。所有 LLM 交互通过 LLMProvider Protocol 注入。 +""" + +from __future__ import annotations + +import json +import re +from collections import Counter +from typing import TYPE_CHECKING, Any + +from json_repair import repair_json +from loguru import logger + +from core.evolution.types import ( + DiagnosePrompts, + ErrorAttribution, + QuestionMetrics, + SkillStepAdherence, + SpanMetrics, +) + +if TYPE_CHECKING: + from core.protocols import LLMProvider + +# ========================================================================= +# 常量 +# ========================================================================= + +_SPAN_EVAL_TOOLS: frozenset[str] = frozenset({"view_node", "search_similar", "observe_frame"}) +"""span 级评估涵盖的工具集合。""" + +_INFRA_STOP_REASONS: frozenset[str] = frozenset({"error", "parse_error"}) +"""执行/解析层失败导致排除的 stop_reason 集合。""" + + +# ========================================================================= +# A. 规则指标 — 7 个纯函数 + 辅助工具 +# ========================================================================= + + +def _parse_json_object(raw: str) -> dict | None: + """将原始字符串解析为字典；失败时返回 None。 + + 参数: + raw: 待解析的原始字符串。 + + 返回: + 解析成功返回 dict，否则返回 None。 + """ + try: + parsed = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + try: + parsed = json.loads(repair_json(raw)) + except (TypeError, ValueError, json.JSONDecodeError): + return None + + if isinstance(parsed, dict): + return parsed + return None + + +def _trigrams(text: str) -> set[str]: + """返回字符串的字符级 trigram 集合。 + + 参数: + text: 输入文本。 + + 返回: + 长度为 3 的子串集合；文本不足 3 字符时返回空集。 + """ + if len(text) < 3: + return set() + return {text[index : index + 3] for index in range(len(text) - 2)} + + +def _extract_last_confidence(raw_contents: list[str]) -> float: + """从末步 raw_content 提取 reflect.confidence。失败时返回 0.5。 + + 参数: + raw_contents: 各步原始输出内容列表。 + + 返回: + 置信度浮点值，提取失败时返回 0.5。 + """ + try: + parsed = _parse_json_object(raw_contents[-1]) + if parsed is None: + raise ValueError("末步内容不是字典。") + return float(parsed["reflect"]["confidence"]) + except Exception: + return 0.5 + + +def calc_format_compliance(raw_contents: list[str]) -> float: + """每步 JSON 是否包含 reflect/plan/action 三个字段。合规步数/总步数。 + + 参数: + raw_contents: 各步原始输出内容列表。 + + 返回: + 合规比例 [0.0, 1.0]；空列表返回 1.0。 + """ + if not raw_contents: + return 1.0 + + compliant_count = 0 + for raw in raw_contents: + parsed = _parse_json_object(raw) + if parsed is not None and all(key in parsed for key in ("reflect", "plan", "action")): + compliant_count += 1 + + return compliant_count / len(raw_contents) + + +def calc_budget_usage(steps_used: int, max_steps: int) -> float: + """预算使用比例。 + + 参数: + steps_used: 已使用步数。 + max_steps: 最大步数预算。 + + 返回: + steps_used / max_steps。 + + 异常: + ZeroDivisionError: max_steps 为 0 时抛出（P5: 不掩盖错误）。 + """ + return steps_used / max_steps + + +def calc_confidence_calibration(confidence: float, correct: bool) -> str: + """置信度校准分类。 + + 参数: + confidence: 模型置信度 [0.0, 1.0]。 + correct: 是否答对。 + + 返回: + 'high_conf_wrong' | 'low_conf_right' | 'calibrated'。 + """ + if confidence >= 0.7 and not correct: + return "high_conf_wrong" + if confidence < 0.5 and correct: + return "low_conf_right" + return "calibrated" + + +def calc_repeat_visit_rate(view_node_ids: list[str]) -> float: + """重复访问率。 + + 参数: + view_node_ids: 访问的节点 ID 列表。 + + 返回: + 1 - (unique / total)；空列表返回 0.0。 + """ + if not view_node_ids: + return 0.0 + return 1 - (len(set(view_node_ids)) / len(view_node_ids)) + + +def calc_search_keyword_repetition(queries: list[str]) -> float: + """连续 search_similar 查询的最大字符级 trigram Jaccard 相似度。 + + 参数: + queries: 搜索查询列表。 + + 返回: + 连续查询对的最大 Jaccard 值；不足 2 个查询时返回 0.0。 + """ + if len(queries) < 2: + return 0.0 + + max_score = 0.0 + for left, right in zip(queries, queries[1:], strict=False): + left_trigrams = _trigrams(left) + right_trigrams = _trigrams(right) + union = left_trigrams | right_trigrams + score = 0.0 if not union else len(left_trigrams & right_trigrams) / len(union) + if score > max_score: + max_score = score + return max_score + + +def calc_level_jump_pattern(view_node_ids: list[str]) -> str: + """从 node_id 提取层级，拼成 'L1→L2→L3' 格式。 + + 参数: + view_node_ids: 节点 ID 列表。 + + 返回: + 层级跳转模式字符串；无匹配时返回空字符串。 + """ + levels: list[str] = [] + for node_id in view_node_ids: + match = re.search(r"_L(\d+)_", node_id) + if match is not None: + levels.append(f"L{match.group(1)}") + return "→".join(levels) + + +def calc_tool_usage(tool_names: list[str]) -> dict[str, int]: + """按 tool_name 计数。 + + 参数: + tool_names: 工具名称列表。 + + 返回: + {工具名: 调用次数} 映射。 + """ + return dict(Counter(tool_names)) + + +def extract_rule_metrics(prediction: dict, raw_contents: list[str], max_steps: int) -> dict: + """从 prediction 和 raw_contents 提取全部 7 个规则指标。 + + 参数: + prediction: 单题预测记录，含 steps_json / correct / answer_confidence。 + raw_contents: 各步原始输出内容列表。 + max_steps: 最大步数预算。 + + 返回: + 包含 7 个规则指标的字典。 + """ + view_node_ids: list[str] = [] + search_queries: list[str] = [] + tool_names: list[str] = [] + + for step in prediction.get("steps_json", []): + tool_call = step.get("tool_call", {}) + if not isinstance(tool_call, dict): + continue + + tool_name = tool_call.get("tool") + args = tool_call.get("args", {}) + if not isinstance(args, dict): + args = {} + + if isinstance(tool_name, str): + tool_names.append(tool_name) + + if tool_name == "view_node": + node_id = args.get("node_id") + if isinstance(node_id, str): + view_node_ids.append(node_id) + + if tool_name == "search_similar": + query = args.get("query") + if isinstance(query, str): + search_queries.append(query) + + # 置信度优先级：末步 JSON reflect.confidence > prediction["answer_confidence"] + confidence = prediction.get("answer_confidence", 0.5) + if raw_contents: + last_step = _parse_json_object(raw_contents[-1]) + if isinstance(last_step, dict): + confidence = _extract_last_confidence(raw_contents) + + correct = bool(prediction.get("correct", False)) + steps_used = len(prediction.get("steps_json", [])) + + return { + "format_compliance": calc_format_compliance(raw_contents), + "budget_usage": calc_budget_usage(steps_used, max_steps), + "confidence_calibration": calc_confidence_calibration(confidence, correct), + "repeat_visit_rate": calc_repeat_visit_rate(view_node_ids), + "search_keyword_repetition": calc_search_keyword_repetition(search_queries), + "level_jump_pattern": calc_level_jump_pattern(view_node_ids), + "tool_usage": calc_tool_usage(tool_names), + } + + +# ========================================================================= +# B. JSON 提取 +# ========================================================================= + + +def extract_json_from_response(raw: str) -> dict: + """从 LLM 回复中提取 JSON。 + + 三策略依序尝试： + 1. markdown 代码块 ```json ... ``` 或 ``` ... ``` + 2. 最外层花括号 { ... } + 3. json_repair 修复后解析 + + 参数: + raw: LLM 原始回复字符串。 + + 返回: + 解析后的字典。 + + 异常: + ValueError: 三种策略均无法提取合法 JSON 字典时抛出。 + """ + # 策略 1: fenced code block + block_match = re.search(r"```(?:json)?\s*(.*?)\s*```", raw, re.DOTALL) + if block_match is not None: + try: + parsed = json.loads(block_match.group(1)) + except (TypeError, ValueError, json.JSONDecodeError): + pass + else: + if isinstance(parsed, dict): + return parsed + + # 策略 2: outermost braces + start = raw.find("{") + end = raw.rfind("}") + if start != -1 and end != -1 and start <= end: + try: + parsed = json.loads(raw[start : end + 1]) + except (TypeError, ValueError, json.JSONDecodeError): + pass + else: + if isinstance(parsed, dict): + return parsed + + # 策略 3: json_repair + try: + parsed = json.loads(repair_json(raw)) + except (TypeError, ValueError, json.JSONDecodeError) as exc: + raise ValueError("无法从 LLM 回复中提取 JSON。") from exc + + if isinstance(parsed, dict): + return parsed + raise ValueError("无法从 LLM 回复中提取 JSON。") + + +# ========================================================================= +# C. Judge 辅助函数 +# ========================================================================= + + +async def _call_judge( + llm: LLMProvider, + system_prompt: str, + user_prompt: str, + *, + max_retries: int = 2, + session_id: str | None = None, +) -> dict: + """调用 judge 模型，解析 JSON 返回。解析失败时重试。 + + 参数: + llm: LLM 调用端口。 + system_prompt: 系统提示词。 + user_prompt: 用户提示词。 + max_retries: 解析失败后的额外重试次数（默认 2，即总共最多调用 3 次）。 + session_id: 会话标识（可选，传入 LLMProvider 用于遥测）。 + + 返回: + 解析后的 JSON 字典。 + + 异常: + ValueError: 所有尝试均无法从回复中提取合法 JSON 时抛出。 + 其他 API 异常直接传播，不在此处捕获。 + """ + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ] + last_exc: ValueError | None = None + for attempt in range(1 + max_retries): + response = await llm.chat(messages, session_id=session_id) + raw = response.content + try: + return extract_json_from_response(raw) + except ValueError as exc: + last_exc = exc + logger.warning("judge JSON 解析失败 (attempt {}/{})", attempt + 1, 1 + max_retries) + raise last_exc # type: ignore[misc] + + +def question_soft_score(span_metrics: list[SpanMetrics]) -> float | None: + """按题 soft 分 = 各 span 的 mean(completeness, 1-hallucination) 再对 spans 取均值。 + + 参数: + span_metrics: 该题的 SpanMetrics 列表。 + + 返回: + 题级 soft 连续分 [0,1]；无 span_metrics 返回 None（invalid）。 + + 关键实现: + 无 span 时返回 None（invalid），绝不补 0 掩盖（守 P5）—— + 分析阶段按 None 跳过该题，而非把缺失误判为 0 分。 + """ + if not span_metrics: + return None + per_span = [ + (s.extraction_completeness + (1.0 - s.hallucination_rate)) / 2.0 for s in span_metrics + ] + return sum(per_span) / len(per_span) + + +def aggregate_soft(scores: list[float | None]) -> float | None: + """对一组按题 soft 分取均值，跳过 invalid(None)。 + + 参数: + scores: 各题 soft 分，None 表示该题 invalid（无 span）。 + + 返回: + 有效题 soft 均值；全部 invalid 返回 None。 + """ + valid = [s for s in scores if s is not None] + if not valid: + return None + return sum(valid) / len(valid) + + +# ========================================================================= +# D. 5 个 Judge 评估函数（async） +# ========================================================================= + + +def _stringify_tool_args(tool_args: Any) -> str: + """将工具参数转换为紧凑文本。 + + 参数: + tool_args: 工具参数（str 或可序列化对象）。 + + 返回: + 紧凑 JSON 字符串。 + """ + if isinstance(tool_args, str): + return tool_args + return json.dumps(tool_args, ensure_ascii=False, sort_keys=True) + + +def _parse_tool_args(tool_args: Any) -> dict[str, object]: + """解析 trace 中的工具参数。 + + 参数: + tool_args: 原始工具参数（dict 或 JSON 字符串）。 + + 返回: + 解析后的参数字典；解析失败返回空字典。 + """ + if isinstance(tool_args, dict): + return tool_args + if isinstance(tool_args, str): + try: + parsed = json.loads(tool_args) + except json.JSONDecodeError: + logger.warning("tool_args 解析失败，回退为空字典: {}", tool_args) + return {} + if isinstance(parsed, dict): + return parsed + return {} + + +async def evaluate_span( + llm: LLMProvider, + prompts: DiagnosePrompts, + question: str, + tool_name: str, + tool_args: dict, + tool_output: str, + ground_truth: str, + step: int, + *, + session_id: str | None = None, +) -> SpanMetrics: + """评估单次 span 级工具调用质量。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + question: 题目文本。 + tool_name: 工具名称。 + tool_args: 工具参数。 + tool_output: 工具输出。 + ground_truth: 对应节点的 ground truth。 + step: 步骤编号。 + session_id: 会话标识（可选）。 + + 返回: + SpanMetrics 实例。 + """ + user_prompt = ( + f"## 问题\n{question}\n\n" + f"## 工具调用\n工具: {tool_name}\n" + f"参数: {json.dumps(tool_args, ensure_ascii=False)}\n\n" + f"## 工具输出\n{tool_output}\n\n" + f"## 原始数据（ground truth）\n{ground_truth}" + ) + parsed = await _call_judge(llm, prompts.span_eval_system, user_prompt, session_id=session_id) + return SpanMetrics( + step=int(step), + tool_name=tool_name, + extraction_completeness=float(parsed.get("extraction_completeness", 0.0)), + hallucination_rate=float(parsed.get("hallucination_rate", 0.0)), + missed_info_tags=list(parsed.get("missed_info_tags", [])), + hallucination_tags=list(parsed.get("hallucination_tags", [])), + ) + + +async def judge_missed_nodes( + llm: LLMProvider, + prompts: DiagnosePrompts, + question: str, + options: list[str] | str, + answer: str, + tree_content: str, + visited_node_ids: list[str], + *, + session_id: str | None = None, +) -> list[str]: + """评估是否遗漏关键节点。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + question: 题目文本。 + options: 选项列表或文本。 + answer: 正确答案。 + tree_content: 树结构文本。 + visited_node_ids: 已访问的节点 ID 列表。 + session_id: 会话标识（可选）。 + + 返回: + 遗漏的节点 ID 列表。 + """ + options_text = "\n".join(options) if isinstance(options, list | tuple) else str(options) + user_prompt = ( + f"## 问题\n{question}\n\n" + f"## 选项\n{options_text}\n\n" + f"## 答案\n{answer}\n\n" + f"## 树内容\n{tree_content}\n\n" + f"## 已访问节点\n{json.dumps(visited_node_ids, ensure_ascii=False)}" + ) + parsed = await _call_judge(llm, prompts.missed_nodes, user_prompt, session_id=session_id) + missed = parsed.get("missed_nodes", []) + if isinstance(missed, list): + return [str(nid) for nid in missed] + return [] + + +async def judge_skill_adherence( + llm: LLMProvider, + prompts: DiagnosePrompts, + skill_content: str, + trace_text: str, + *, + session_id: str | None = None, +) -> list[SkillStepAdherence]: + """评估技能步骤遵循情况。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + skill_content: 技能文件全文。 + trace_text: 格式化后的执行轨迹文本。 + session_id: 会话标识（可选）。 + + 返回: + SkillStepAdherence 列表。 + """ + user_prompt = f"## Skill 内容\n{skill_content}\n\n## 执行轨迹\n{trace_text}" + parsed = await _call_judge(llm, prompts.skill_adherence, user_prompt, session_id=session_id) + steps = parsed.get("steps", []) + if not isinstance(steps, list): + return [] + + results: list[SkillStepAdherence] = [] + for item in steps: + if not isinstance(item, dict): + continue + results.append( + SkillStepAdherence( + step_label=str(item.get("step_label", "")), + adhered=bool(item.get("adhered", False)), + description=str(item.get("description", "")), + ) + ) + return results + + +async def judge_confirmation_bias( + llm: LLMProvider, + prompts: DiagnosePrompts, + question: str, + options: list[str] | str, + trace_text: str, + *, + session_id: str | None = None, +) -> tuple[bool, str]: + """评估是否存在确认偏误。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + question: 题目文本。 + options: 选项列表或文本。 + trace_text: 格式化后的执行轨迹文本。 + session_id: 会话标识（可选）。 + + 返回: + (has_bias, evidence) 元组。 + """ + options_text = "\n".join(options) if isinstance(options, list | tuple) else str(options) + user_prompt = f"## 问题\n{question}\n\n## 选项\n{options_text}\n\n## 执行轨迹\n{trace_text}" + parsed = await _call_judge(llm, prompts.confirmation_bias, user_prompt, session_id=session_id) + return bool(parsed.get("has_bias", False)), str(parsed.get("evidence", "")) + + +async def judge_evidence_sufficiency( + llm: LLMProvider, + prompts: DiagnosePrompts, + question: str, + options: list[str] | str, + answer: str, + all_tool_outputs: str, + *, + session_id: str | None = None, +) -> tuple[bool, str]: + """评估当前证据是否充足。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + question: 题目文本。 + options: 选项列表或文本。 + answer: 正确答案。 + all_tool_outputs: 全部工具输出拼接文本。 + session_id: 会话标识（可选）。 + + 返回: + (sufficient, reasoning) 元组。 + """ + options_text = "\n".join(options) if isinstance(options, list | tuple) else str(options) + user_prompt = ( + f"## 问题\n{question}\n\n" + f"## 选项\n{options_text}\n\n" + f"## 答案\n{answer}\n\n" + f"## 所有工具输出\n{all_tool_outputs}" + ) + parsed = await _call_judge( + llm, prompts.evidence_sufficiency, user_prompt, session_id=session_id + ) + return bool(parsed.get("sufficient", False)), str(parsed.get("reasoning", "")) + + +# ========================================================================= +# E. compute_question_metrics（async 编排） +# ========================================================================= + + +def _format_trace_text(traces: list[dict]) -> str: + """将 trace 列表格式化为 judge 可读文本（指标版本：截断 thought/tool_output）。 + + 参数: + traces: trace 字典列表。 + + 返回: + 格式化后的多行文本。 + """ + lines: list[str] = [] + for trace in traces: + step = trace.get("step", "") + thought = str(trace.get("thought", ""))[:100] + tool_name = trace.get("tool_name", "") + tool_args = _stringify_tool_args(trace.get("tool_args", {})) + tool_output = str(trace.get("tool_output", ""))[:200] + lines.append( + f'Step {step}: thinking="{thought}" → {tool_name}({tool_args}) → {tool_output}' + ) + return "\n".join(lines) + + +def _load_tree_content(tree_data: dict) -> str: + """将树结构内容整理为文本。 + + 参数: + tree_data: 树结构字典，含 "nodes" 键。 + + 返回: + 格式化后的树结构文本。 + """ + nodes = tree_data.get("nodes", {}) + if not isinstance(nodes, dict): + return "" + + chunks: list[str] = [] + for node_id in sorted(nodes): + node = nodes.get(node_id, {}) + if not isinstance(node, dict): + continue + level = node.get("level", "") + time_range = node.get("time_range", [0, 0]) + if not isinstance(time_range, list | tuple) or len(time_range) < 2: + time_range = [0, 0] + t_start, t_end = time_range[0], time_range[1] + card_json = json.dumps(node.get("card", {}), ensure_ascii=False, sort_keys=True) + chunks.append( + f"### {node_id} | L{level} | {float(t_start):.0f}-{float(t_end):.0f}s\n{card_json}" + ) + return "\n\n".join(chunks) + + +def _get_ground_truth_for_trace(tree_data: dict, tool_name: str, tool_args: dict) -> str: + """按工具类型获取对应节点的 ground truth。 + + 参数: + tree_data: 树结构字典。 + tool_name: 工具名称。 + tool_args: 工具参数字典。 + + 返回: + 节点 card 的 JSON 字符串；无匹配时返回空字符串。 + """ + nodes = tree_data.get("nodes", {}) + if not isinstance(nodes, dict): + return "" + + node_id = "" + if tool_name == "observe_frame": + node_ids = tool_args.get("node_ids", []) + if isinstance(node_ids, list) and node_ids: + node_id = str(node_ids[0]) + else: + node_id = str(tool_args.get("node_id", "")) + if not node_id: + node_ids = tool_args.get("node_ids", []) + if isinstance(node_ids, list) and node_ids: + node_id = str(node_ids[0]) + + node = nodes.get(node_id, {}) + if not isinstance(node, dict): + return "" + return json.dumps(node.get("card", {}), ensure_ascii=False, sort_keys=True) + + +async def compute_question_metrics( + prediction: dict[str, Any], + traces: list[dict[str, Any]], + tree_data: dict[str, Any], + skill_content: str, + llm: LLMProvider, + prompts: DiagnosePrompts, + max_steps: int, + raw_contents: list[str] | None = None, + *, + session_id: str | None = None, +) -> QuestionMetrics: + """编排单题规则指标与 LLM judge 指标。 + + 参数: + prediction: 单题预测记录。 + traces: 该题的执行轨迹列表。 + tree_data: 树结构字典。 + skill_content: 技能文件全文。 + llm: LLM 调用端口。 + prompts: 诊断模板束。 + max_steps: 最大步数预算。 + raw_contents: 各步原始输出（可选，默认从 steps_json 提取）。 + session_id: 会话标识（可选）。 + + 返回: + QuestionMetrics 实例。 + """ + if raw_contents is None: + raw_contents = [ + str(step.get("tool_output", "")) for step in prediction.get("steps_json", []) + ] + + rule_metrics_dict = extract_rule_metrics(prediction, raw_contents, max_steps) + + # Phase 1: span 评估 + 收集已访问节点 + span_evals_list: list[SpanMetrics] = [] + visited_node_ids: list[str] = [] + seen_node_ids: set[str] = set() + + for trace in traces: + tool_name = trace.get("tool_name") + tool_args = _parse_tool_args(trace.get("tool_args", {})) + if tool_name in _SPAN_EVAL_TOOLS: + span_evals_list.append( + await evaluate_span( + llm=llm, + prompts=prompts, + question=prediction.get("question", ""), + tool_name=str(tool_name), + tool_args=tool_args, + tool_output=str(trace.get("tool_output", "")), + ground_truth=_get_ground_truth_for_trace(tree_data, str(tool_name), tool_args), + step=int(trace.get("step", 0)), + session_id=session_id, + ) + ) + + if tool_name == "view_node": + node_id = tool_args.get("node_id") + if isinstance(node_id, str) and node_id and node_id not in seen_node_ids: + seen_node_ids.add(node_id) + visited_node_ids.append(node_id) + + # Phase 2: 全局 judge 评估 + all_tool_outputs = "\n".join( + str(trace.get("tool_output", "")) + for trace in traces + if trace.get("tool_name") in _SPAN_EVAL_TOOLS + ) + options_list = ( + prediction.get("options", "").split("\n") + if isinstance(prediction.get("options"), str) + else prediction.get("options", []) + ) + trace_text = _format_trace_text(traces) + tree_content = _load_tree_content(tree_data) + + missed_nodes_list = await judge_missed_nodes( + llm=llm, + prompts=prompts, + question=prediction.get("question", ""), + options=options_list, + answer=prediction.get("answer", ""), + tree_content=tree_content, + visited_node_ids=visited_node_ids, + session_id=session_id, + ) + skill_adherence_list = await judge_skill_adherence( + llm=llm, + prompts=prompts, + skill_content=skill_content, + trace_text=trace_text, + session_id=session_id, + ) + has_bias, _bias_evidence = await judge_confirmation_bias( + llm=llm, + prompts=prompts, + question=prediction.get("question", ""), + options=options_list, + trace_text=trace_text, + session_id=session_id, + ) + sufficient, _reasoning = await judge_evidence_sufficiency( + llm=llm, + prompts=prompts, + question=prediction.get("question", ""), + options=options_list, + answer=prediction.get("answer", ""), + all_tool_outputs=all_tool_outputs, + session_id=session_id, + ) + + return QuestionMetrics( + question_id=prediction["question_id"], + video_id=prediction["video_id"], + task_type=prediction["task_type"], + correct=bool(prediction.get("correct", False)), + format_compliance=rule_metrics_dict["format_compliance"], + budget_usage=rule_metrics_dict["budget_usage"], + confidence_calibration=rule_metrics_dict["confidence_calibration"], + repeat_visit_rate=rule_metrics_dict["repeat_visit_rate"], + search_keyword_repetition=rule_metrics_dict["search_keyword_repetition"], + level_jump_pattern=rule_metrics_dict["level_jump_pattern"], + tool_usage=rule_metrics_dict["tool_usage"], + span_metrics=span_evals_list, + missed_nodes=missed_nodes_list, + skill_adherence=skill_adherence_list, + confirmation_bias=has_bias, + evidence_sufficient=sufficient, + ) + + +# ========================================================================= +# F. 错误归因 +# ========================================================================= + + +def _mean(values: list[float]) -> float: + """计算均值；空列表返回 0.0。 + + 参数: + values: 浮点值列表。 + + 返回: + 均值或 0.0。 + """ + if not values: + return 0.0 + return sum(values) / len(values) + + +def attribute_error(qm: QuestionMetrics) -> ErrorAttribution: + """按瀑布规则归因单题错误类型。 + + 瀑布顺序: + 1. extraction: completeness<0.5 或 hallucination>0.5 + 2. search: 有遗漏节点 + 3. reasoning: evidence_sufficient=True（证据够但推理错） + 4. mixed: 其余 + + 参数: + qm: 单题指标。 + + 返回: + ErrorAttribution 实例。 + """ + avg_completeness = _mean([span.extraction_completeness for span in qm.span_metrics]) + max_hallucination = max((span.hallucination_rate for span in qm.span_metrics), default=0.0) + + if avg_completeness < 0.5 or max_hallucination > 0.5: + error_type = "extraction_failure" + elif len(qm.missed_nodes) > 0: + error_type = "search_failure" + elif qm.evidence_sufficient is True: + error_type = "reasoning_failure" + else: + error_type = "mixed" + + return ErrorAttribution( + question_id=qm.question_id, + error_type=error_type, + reasoning_failure_type=None, + ) + + +async def classify_defect_vs_lapse( + llm: LLMProvider, + prompts: DiagnosePrompts, + prediction: dict[str, Any], + traces: list[dict[str, Any]], + prompt_content: str, + *, + session_id: str | None = None, +) -> tuple[str, str]: + """判别错题病因：defect（改正文）vs lapse（记提醒）。 + + 判不准默认 lapse（保护正文），这是设计明确的保护性 fallback。 + + 参数: + llm: LLM 调用端口。 + prompts: 诊断模板束。 + prediction: 单题预测记录（含 question/answer/prediction）。 + traces: 该题执行轨迹。 + prompt_content: Agent 当时所用的 prompt 全文。 + session_id: 会话标识（可选）。 + + 返回: + (category, note)；category 取值 'defect' 或 'lapse'， + note 为 lapse 提醒文本。 + + 异常: + API 基础设施异常（网络/超时等）直接传播，不掩盖。 + """ + trace_text = _format_trace_text(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"## 当前 prompt 全文\n{prompt_content}\n\n" + f"## 执行轨迹\n{trace_text}" + ) + # chat() 的基础设施失败（网络/API）刻意不在此捕获——按 P5，应向上传播报错， + # 不能用默认值掩盖。保护性 fallback 只针对"judge 回复无法解析/判不准"这一语义歧义。 + response = await llm.chat( + [ + {"role": "system", "content": prompts.defect_vs_lapse}, + {"role": "user", "content": user_prompt}, + ], + session_id=session_id, + ) + try: + parsed = extract_json_from_response(response.content) + except ValueError: + parsed = None # judge 回复无法解析 → 落入保护性 fallback + category = parsed.get("category") if isinstance(parsed, dict) else None + if category not in ("defect", "lapse"): + category = "lapse" # 保护性 fallback + note = parsed.get("note", "") if isinstance(parsed, dict) else "" + return category, (note if isinstance(note, str) else "") + + +def _make_degraded_metrics(prediction: dict[str, Any], max_steps: int) -> QuestionMetrics: + """生成降级版 QuestionMetrics：规则指标正常计算，judge 指标标记为不可用。 + + 在 judge JSON 解析失败（ValueError）时调用。 + 其他异常类型不由本函数处理，应向上传播。 + + 参数: + prediction: 单题预测记录。 + max_steps: 最大步数预算。 + + 返回: + degraded=True 的 QuestionMetrics，judge 字段置为 None/空。 + """ + raw_contents = [str(step.get("tool_output", "")) for step in prediction.get("steps_json", [])] + rule = extract_rule_metrics(prediction, raw_contents, max_steps) + return QuestionMetrics( + question_id=prediction["question_id"], + video_id=prediction["video_id"], + task_type=prediction["task_type"], + correct=bool(prediction.get("correct", False)), + format_compliance=rule["format_compliance"], + budget_usage=rule["budget_usage"], + confidence_calibration=rule["confidence_calibration"], + repeat_visit_rate=rule["repeat_visit_rate"], + search_keyword_repetition=rule["search_keyword_repetition"], + level_jump_pattern=rule["level_jump_pattern"], + tool_usage=rule["tool_usage"], + span_metrics=[], + missed_nodes=[], + skill_adherence=[], + confirmation_bias=None, + evidence_sufficient=None, + degraded=True, + ) diff --git a/tests/unit/test_diagnose.py b/tests/unit/test_diagnose.py new file mode 100644 index 0000000..b205f86 --- /dev/null +++ b/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