"""阶段 3 支持分析案例导入、逐案评审和退出指标计算。""" import json from pathlib import Path from django.utils import timezone from apps.tickets.models import ( SupportEvaluationCase, SupportEvaluationResult, TicketSupportAnalysis, ) SOURCE_KINDS = {"anonymized_real", "synthetic"} class SupportEvaluationError(ValueError): pass def _solution_ids(value, *, label: str) -> list[int]: if not isinstance(value, list): raise SupportEvaluationError(f"{label}必须是整数数组") result = [] for item in value: if isinstance(item, bool) or not isinstance(item, int) or item <= 0: raise SupportEvaluationError(f"{label}必须是正整数数组") if item not in result: result.append(item) return result def import_support_evaluation_cases(path: str | Path) -> tuple[int, int]: """导入阶段 3 支持案例及期望引用、动作和人工降级结果。""" source_path = Path(path) payload = json.loads(source_path.read_text(encoding="utf-8")) if not isinstance(payload, dict) or payload.get("schema_version") != 1: raise SupportEvaluationError("案例文件 schema_version 必须为 1") cases = payload.get("cases") if not isinstance(cases, list): raise SupportEvaluationError("案例文件必须包含 cases 数组") created_count = 0 updated_count = 0 for raw_case in cases: if not isinstance(raw_case, dict): raise SupportEvaluationError("每个案例必须是对象") key = str(raw_case.get("key") or "").strip() source_kind = str(raw_case.get("source_kind") or "").strip() if not key or source_kind not in SOURCE_KINDS: raise SupportEvaluationError("案例必须提供 key 和合法 source_kind") boolean_fields = { "expected_auto_resolve", "expected_manual_fallback", "enabled", } invalid_boolean_fields = sorted( field for field in boolean_fields if field in raw_case and not isinstance(raw_case[field], bool) ) if invalid_boolean_fields: raise SupportEvaluationError( f"案例 {key} 的布尔字段类型错误：{', '.join(invalid_boolean_fields)}" ) input_manifest = raw_case.get("input_manifest") or {} if not isinstance(input_manifest, dict): raise SupportEvaluationError(f"案例 {key} 的 input_manifest 必须是对象") expected_solution_ids = _solution_ids( raw_case.get("expected_solution_ids") or [], label=f"案例 {key} 的 expected_solution_ids", ) expected_auto_resolve = bool(raw_case.get("expected_auto_resolve", False)) expected_manual_fallback = bool(raw_case.get("expected_manual_fallback", True)) if expected_auto_resolve == expected_manual_fallback: raise SupportEvaluationError( f"案例 {key} 必须且只能选择自动解决或人工降级一种期望结果" ) if expected_auto_resolve and not expected_solution_ids: raise SupportEvaluationError(f"自动解决案例 {key} 必须提供期望方案 ID") if source_kind == "anonymized_real": required = { "input_manifest", "expected_solution_ids", "expected_auto_resolve", "expected_manual_fallback", } missing = sorted(required - set(raw_case)) if missing: raise SupportEvaluationError(f"真实案例 {key} 缺少字段：{', '.join(missing)}") if not input_manifest: raise SupportEvaluationError(f"真实案例 {key} 必须提供脱敏工单输入摘要") _, created = SupportEvaluationCase.objects.update_or_create( key=key, defaults={ "name": str(raw_case.get("name") or key), "source_kind": source_kind, "input_manifest": input_manifest, "expected_solution_ids": expected_solution_ids, "expected_auto_resolve": expected_auto_resolve, "expected_manual_fallback": expected_manual_fallback, "notes": str(raw_case.get("notes") or ""), "enabled": bool(raw_case.get("enabled", True)), }, ) created_count += int(created) updated_count += int(not created) return created_count, updated_count def record_support_evaluation_result( case: SupportEvaluationCase, analysis: TicketSupportAnalysis, *, reviewer=None, ) -> SupportEvaluationResult: """记录支持分析与案例期望的逐项评审结果。""" if analysis.run.status != "succeeded" or analysis.processed_at is None: raise SupportEvaluationError("支持分析尚未成功完成后处理，不能进入业务指标") output = analysis.run.final_output if isinstance(analysis.run.final_output, dict) else {} candidate_ids = _solution_ids( output.get("candidate_solution_ids") or [], label="Agent candidate_solution_ids", ) raw_citations = output.get("citations") if not isinstance(raw_citations, list): raw_citations = [] citation_ranks = [ item for item in raw_citations if isinstance(item, int) and not isinstance(item, bool) and item > 0 ] reference_ranks = { item.get("rank") for item in analysis.references if isinstance(item, dict) and isinstance(item.get("rank"), int) and not isinstance(item.get("rank"), bool) } citations_valid = ( len(citation_ranks) == len(raw_citations) and set(citation_ranks).issubset(reference_ranks) and (bool(citation_ranks) if reference_ranks else not citation_ranks) ) expected_ids = set(case.expected_solution_ids) matched_expected_solution = ( bool(expected_ids.intersection(candidate_ids)) if expected_ids else not candidate_ids ) decision = ( analysis.automation_decision if isinstance(analysis.automation_decision, dict) else {} ) # 一次确认流程把“分析判定可自动复用”和“人工批准后真正解决”拆成两个时点。 # 旧字段名继续保留给既有评估数据和接口，但阶段指标按是否到达批准就绪态计算。 actual_auto_resolved = decision.get("status") in { "waiting_approval", "auto_resolved", } and bool( decision.get("eligible") or ( decision.get("applied") is True and analysis.applied_solution_id is not None ) ) actual_manual_fallback = decision.get("status") == "manual_recommendation" evaluated_solution_id = analysis.applied_solution_id or decision.get("solution_id") incorrect_auto_reuse = bool( actual_auto_resolved and evaluated_solution_id not in expected_ids ) outcome_correct = ( actual_auto_resolved == case.expected_auto_resolve and actual_manual_fallback == case.expected_manual_fallback ) passed = ( matched_expected_solution and outcome_correct and not incorrect_auto_reuse and citations_valid ) result, _ = SupportEvaluationResult.objects.update_or_create( case=case, analysis=analysis, defaults={ "actual_candidate_solution_ids": candidate_ids, "actual_citation_ranks": citation_ranks, "matched_expected_solution": matched_expected_solution, "actual_auto_resolved": actual_auto_resolved, "actual_manual_fallback": actual_manual_fallback, "incorrect_auto_reuse": incorrect_auto_reuse, "citations_valid": citations_valid, "passed": passed, "details": { "expected_solution_ids": list(case.expected_solution_ids), "expected_auto_resolve": case.expected_auto_resolve, "expected_manual_fallback": case.expected_manual_fallback, "applied_solution_id": analysis.applied_solution_id, "evaluated_solution_id": evaluated_solution_id, "automation_decision": decision, "reference_count": len(analysis.references), "outcome_correct": outcome_correct, }, "reviewed_by": reviewer, }, ) return result def _latest_support_evaluation_scope( *, include_synthetic: bool, ) -> tuple[list[SupportEvaluationCase], list[SupportEvaluationResult]]: cases = SupportEvaluationCase.objects.filter(enabled=True) if not include_synthetic: cases = cases.filter(source_kind="anonymized_real") case_list = list(cases.order_by("key")) queryset = ( SupportEvaluationResult.objects.select_related( "case", "analysis__ticket", "analysis__run", "reviewed_by", ) .filter(case__in=case_list) .order_by("case_id", "-updated_at", "-id") ) latest_by_case = {} for result in queryset: latest_by_case.setdefault(result.case_id, result) return case_list, list(latest_by_case.values()) def build_support_quality_metrics(*, include_synthetic: bool = False) -> dict: """按每个案例最新结果计算方案匹配、错误复用和人工降级指标。""" cases, results = _latest_support_evaluation_scope( include_synthetic=include_synthetic ) expected_match_results = [ result for result in results if result.case.expected_solution_ids ] expected_auto_results = [ result for result in results if result.case.expected_auto_resolve ] expected_manual_results = [ result for result in results if result.case.expected_manual_fallback ] actual_auto_results = [result for result in results if result.actual_auto_resolved] def rate(numerator: int, denominator: int) -> float | None: return numerator / denominator if denominator else None solution_match_count = sum( result.matched_expected_solution for result in expected_match_results ) auto_resolved_count = sum( result.actual_auto_resolved for result in expected_auto_results ) incorrect_auto_reuse_count = sum( result.incorrect_auto_reuse for result in actual_auto_results ) manual_fallback_count = sum( result.actual_manual_fallback for result in expected_manual_results ) citation_valid_count = sum(result.citations_valid for result in results) enabled_case_count = len(cases) reviewed_case_count = len(results) coverage_complete = ( enabled_case_count > 0 and reviewed_case_count == enabled_case_count ) solution_match_rate = rate(solution_match_count, len(expected_match_results)) eligible_auto_resolve_rate = rate(auto_resolved_count, len(expected_auto_results)) incorrect_auto_reuse_rate = rate( incorrect_auto_reuse_count, len(actual_auto_results), ) manual_fallback_accuracy = rate( manual_fallback_count, len(expected_manual_results), ) citation_validity_rate = rate(citation_valid_count, reviewed_case_count) evidence_ready = coverage_complete and not include_synthetic return { "include_synthetic": include_synthetic, "evaluation": { "enabled_case_count": enabled_case_count, "reviewed_case_count": reviewed_case_count, "coverage_rate": rate(reviewed_case_count, enabled_case_count), "expected_match_case_count": len(expected_match_results), "matched_solution_count": solution_match_count, "solution_match_rate": solution_match_rate, "eligible_auto_resolve_case_count": len(expected_auto_results), "auto_resolved_count": auto_resolved_count, "eligible_auto_resolve_rate": eligible_auto_resolve_rate, "actual_auto_resolved_count": len(actual_auto_results), "incorrect_auto_reuse_count": incorrect_auto_reuse_count, "incorrect_auto_reuse_rate": incorrect_auto_reuse_rate, "expected_manual_fallback_count": len(expected_manual_results), "manual_fallback_count": manual_fallback_count, "manual_fallback_accuracy": manual_fallback_accuracy, "citation_valid_count": citation_valid_count, "citation_validity_rate": citation_validity_rate, "result_selection": "latest_per_case", }, "targets": { "solution_match_rate": 0.50, "eligible_auto_resolve_rate": 0.30, "incorrect_auto_reuse_rate": 0.05, "coverage_complete": coverage_complete, "solution_match_rate_met": ( solution_match_rate >= 0.50 if evidence_ready and solution_match_rate is not None else None ), "eligible_auto_resolve_rate_met": ( eligible_auto_resolve_rate >= 0.30 if evidence_ready and eligible_auto_resolve_rate is not None else None ), "incorrect_auto_reuse_rate_met": ( (incorrect_auto_reuse_rate or 0.0) <= 0.05 if evidence_ready and actual_auto_results else None ), "manual_fallback_accuracy_met": ( manual_fallback_accuracy == 1.0 if evidence_ready and manual_fallback_accuracy is not None else None ), "citation_validity_rate_met": ( citation_validity_rate == 1.0 if evidence_ready and citation_validity_rate is not None else None ), }, } def build_stage_3_acceptance_report(*, include_synthetic: bool = False) -> dict: """生成阶段 3 退出报告；合成案例仅用于开发回归。""" cases, results = _latest_support_evaluation_scope( include_synthetic=include_synthetic ) metrics = build_support_quality_metrics(include_synthetic=include_synthetic) result_by_case = {result.case_id: result for result in results} case_results = [] for case in cases: result = result_by_case.get(case.id) case_results.append( { "key": case.key, "name": case.name, "source_kind": case.source_kind, "reviewed": result is not None, "passed": result.passed if result is not None else None, "analysis_id": result.analysis_id if result is not None else None, "ticket_id": ( result.analysis.ticket_id if result is not None else None ), "matched_expected_solution": ( result.matched_expected_solution if result is not None else None ), "actual_auto_resolved": ( result.actual_auto_resolved if result is not None else None ), "actual_manual_fallback": ( result.actual_manual_fallback if result is not None else None ), "incorrect_auto_reuse": ( result.incorrect_auto_reuse if result is not None else None ), "citations_valid": result.citations_valid if result is not None else None, "reviewed_at": ( result.updated_at.isoformat() if result is not None else None ), } ) reasons = [] if include_synthetic: reasons.append( { "code": "synthetic_scope", "message": "合成案例只验证评估框架，不能作为阶段 3 业务退出证据。", } ) elif not cases: reasons.append( { "code": "missing_real_cases", "message": "尚未导入启用的阶段 3 脱敏真实案例。", } ) elif not metrics["targets"]["coverage_complete"]: reasons.append( { "code": "incomplete_review_coverage", "message": "仍有启用案例没有支持分析评审结果。", } ) if not include_synthetic and cases: expected_auto_count = sum(case.expected_auto_resolve for case in cases) expected_manual_count = sum(case.expected_manual_fallback for case in cases) expected_no_match_count = sum(not case.expected_solution_ids for case in cases) if expected_auto_count == 0: reasons.append( { "code": "missing_auto_resolve_case", "message": "真实案例集未覆盖允许自动解决的场景。", } ) if expected_manual_count == 0: reasons.append( { "code": "missing_manual_fallback_case", "message": "真实案例集未覆盖策略阻断或人工降级场景。", } ) if expected_no_match_count == 0: reasons.append( { "code": "missing_no_match_case", "message": "真实案例集未覆盖无正确方案或错误匹配场景。", } ) targets = metrics["targets"] if targets["coverage_complete"] and not include_synthetic: target_messages = { "solution_match_rate_met": "已发布方案自动匹配率未达到 50%。", "eligible_auto_resolve_rate_met": "符合条件工单自动解决率未达到 30%。", "incorrect_auto_reuse_rate_met": "错误自动复用率超过 5%，或没有自动解决证据。", "manual_fallback_accuracy_met": "人工降级场景未全部按预期阻断自动解决。", "citation_validity_rate_met": "支持分析建议未全部具备有效冻结引用。", } for target, message in target_messages.items(): if targets[target] is not True: reasons.append({"code": target.removesuffix("_met"), "message": message}) exit_ready = not reasons and not include_synthetic status = "passed" if exit_ready else "diagnostic" if include_synthetic else "not_ready" return { "schema_version": 1, "generated_at": timezone.now().isoformat(), "scope": "diagnostic_with_synthetic" if include_synthetic else "anonymized_real", "status": status, "exit_ready": exit_ready, "reasons": reasons, "metrics": metrics, "case_results": case_results, }