"""诊断引擎 — 指标计算、judge 辅助、聚合、案例包构建与入口。 两阶段诊断管线的可提取内核。包含： - 7 个规则指标的纯函数计算 - JSON 提取工具 - 5 个 LLM judge 评估函数（async） - 单题指标编排 compute_question_metrics - 错误归因瀑布 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, 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 # ========================================================================= # 常量 # ========================================================================= _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, ) # ========================================================================= # 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, )