555 lines
22 KiB
Python
555 lines
22 KiB
Python
"""阶段 2 智能受理案例导入、逐案评审和质量指标计算。"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from django.utils import timezone
|
|
|
|
from apps.tickets.models import (
|
|
IntakeEvaluationCase,
|
|
IntakeEvaluationResult,
|
|
TicketDraft,
|
|
)
|
|
|
|
SOURCE_KINDS = {"anonymized_real", "synthetic"}
|
|
|
|
|
|
class IntakeEvaluationError(ValueError):
|
|
pass
|
|
|
|
|
|
def _normalize_value(value):
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, str):
|
|
return " ".join(value.casefold().split())
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
return str(value).casefold()
|
|
|
|
|
|
def import_evaluation_cases(path: str | Path) -> tuple[int, int]:
|
|
"""导入版本化案例清单；真实案例必须提供完整期望与材料清单。"""
|
|
|
|
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 IntakeEvaluationError("案例文件 schema_version 必须为 1")
|
|
cases = payload.get("cases") if isinstance(payload, dict) else None
|
|
if not isinstance(cases, list):
|
|
raise IntakeEvaluationError("案例文件必须包含 cases 数组")
|
|
created_count = 0
|
|
updated_count = 0
|
|
for raw_case in cases:
|
|
if not isinstance(raw_case, dict):
|
|
raise IntakeEvaluationError("每个案例必须是对象")
|
|
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 IntakeEvaluationError("案例必须提供 key 和合法 source_kind")
|
|
boolean_fields = {
|
|
"valid_input",
|
|
"expected_auto_created",
|
|
"expected_similar_ticket",
|
|
"enabled",
|
|
}
|
|
invalid_boolean_fields = sorted(
|
|
field_name
|
|
for field_name in boolean_fields
|
|
if field_name in raw_case and not isinstance(raw_case[field_name], bool)
|
|
)
|
|
if invalid_boolean_fields:
|
|
raise IntakeEvaluationError(
|
|
f"案例 {key} 的布尔字段类型错误：{', '.join(invalid_boolean_fields)}"
|
|
)
|
|
input_manifest = raw_case.get("input_manifest") or {}
|
|
expected_fields = raw_case.get("expected_fields") or {}
|
|
expected_timeline = raw_case.get("expected_timeline") or []
|
|
if not isinstance(input_manifest, dict):
|
|
raise IntakeEvaluationError(f"案例 {key} 的 input_manifest 必须是对象")
|
|
if not isinstance(expected_fields, dict):
|
|
raise IntakeEvaluationError(f"案例 {key} 的 expected_fields 必须是对象")
|
|
if not isinstance(expected_timeline, list) or not all(
|
|
isinstance(item, dict) for item in expected_timeline
|
|
):
|
|
raise IntakeEvaluationError(f"案例 {key} 的 expected_timeline 必须是对象数组")
|
|
if source_kind == "anonymized_real":
|
|
required_keys = {
|
|
"valid_input",
|
|
"input_manifest",
|
|
"expected_fields",
|
|
"expected_timeline",
|
|
"expected_auto_created",
|
|
"expected_similar_ticket",
|
|
}
|
|
missing_keys = sorted(required_keys - set(raw_case))
|
|
if missing_keys:
|
|
raise IntakeEvaluationError(
|
|
f"真实案例 {key} 缺少字段：{', '.join(missing_keys)}"
|
|
)
|
|
source_text = str(input_manifest.get("source_text") or "").strip()
|
|
materials = input_manifest.get("materials") or []
|
|
if not isinstance(materials, list) or not all(
|
|
isinstance(item, dict) for item in materials
|
|
):
|
|
raise IntakeEvaluationError(f"真实案例 {key} 的 materials 必须是对象数组")
|
|
if not source_text and not materials:
|
|
raise IntakeEvaluationError(f"真实案例 {key} 必须包含文本或材料")
|
|
_, created = IntakeEvaluationCase.objects.update_or_create(
|
|
key=key,
|
|
defaults={
|
|
"name": str(raw_case.get("name") or key),
|
|
"source_kind": source_kind,
|
|
"valid_input": bool(raw_case.get("valid_input", True)),
|
|
"input_manifest": input_manifest,
|
|
"expected_fields": expected_fields,
|
|
"expected_timeline": expected_timeline,
|
|
"expected_auto_created": bool(raw_case.get("expected_auto_created", True)),
|
|
"expected_similar_ticket": bool(
|
|
raw_case.get("expected_similar_ticket", False)
|
|
),
|
|
"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 _evaluate_timeline(expected_timeline: list, actual_timeline: list) -> tuple[list, int]:
|
|
results = []
|
|
correct_count = 0
|
|
for index, expected in enumerate(expected_timeline):
|
|
actual = actual_timeline[index] if index < len(actual_timeline) else None
|
|
field_results = {}
|
|
correct = isinstance(actual, dict)
|
|
for field_name, expected_value in expected.items():
|
|
actual_value = actual.get(field_name) if isinstance(actual, dict) else None
|
|
field_correct = _normalize_value(actual_value) == _normalize_value(expected_value)
|
|
correct = correct and field_correct
|
|
field_results[field_name] = {
|
|
"expected": expected_value,
|
|
"actual": actual_value,
|
|
"correct": field_correct,
|
|
}
|
|
correct_count += int(correct)
|
|
results.append(
|
|
{
|
|
"sequence": index + 1,
|
|
"correct": correct,
|
|
"field_results": field_results,
|
|
}
|
|
)
|
|
return results, correct_count
|
|
|
|
|
|
def record_evaluation_result(
|
|
case: IntakeEvaluationCase,
|
|
draft: TicketDraft,
|
|
*,
|
|
reviewer=None,
|
|
) -> IntakeEvaluationResult:
|
|
"""把一次受理结果与案例期望逐字段比对，并保留人工评审证据。"""
|
|
|
|
expected_fields = case.expected_fields if isinstance(case.expected_fields, dict) else {}
|
|
actual_fields = draft.current_fields if isinstance(draft.current_fields, dict) else {}
|
|
field_results = {}
|
|
correct_count = 0
|
|
for field_name, expected in expected_fields.items():
|
|
actual = actual_fields.get(field_name)
|
|
correct = _normalize_value(actual) == _normalize_value(expected)
|
|
correct_count += int(correct)
|
|
field_results[field_name] = {
|
|
"expected": expected,
|
|
"actual": actual,
|
|
"correct": correct,
|
|
}
|
|
evaluated_count = len(field_results)
|
|
field_accuracy = correct_count / evaluated_count if evaluated_count else None
|
|
field_evidence = draft.field_evidence if isinstance(draft.field_evidence, dict) else {}
|
|
evidence_results = {}
|
|
evidence_count = 0
|
|
evidence_covered_count = 0
|
|
for field_name, expected in expected_fields.items():
|
|
if not _normalize_value(expected):
|
|
continue
|
|
evidence_count += 1
|
|
evidence = field_evidence.get(field_name)
|
|
covered = isinstance(evidence, dict) and any(
|
|
bool(evidence.get(key))
|
|
for key in (
|
|
"source_refs",
|
|
"excerpt",
|
|
"location",
|
|
"generated_from",
|
|
"raw_value",
|
|
)
|
|
)
|
|
evidence_covered_count += int(covered)
|
|
evidence_results[field_name] = {
|
|
"covered": covered,
|
|
"evidence": evidence if isinstance(evidence, dict) else {},
|
|
}
|
|
expected_timeline = (
|
|
case.expected_timeline if isinstance(case.expected_timeline, list) else []
|
|
)
|
|
actual_timeline = (
|
|
draft.message_timeline if isinstance(draft.message_timeline, list) else []
|
|
)
|
|
timeline_results, correct_timeline_count = _evaluate_timeline(
|
|
expected_timeline,
|
|
actual_timeline,
|
|
)
|
|
timeline_correct = correct_timeline_count == len(expected_timeline)
|
|
actual_auto_created = draft.decision == "auto_created" and bool(draft.confirmed_ticket_id)
|
|
actual_similar_ticket = bool(
|
|
draft.confirmed_ticket_id
|
|
and draft.confirmed_ticket.similarity_suggestions.exists()
|
|
)
|
|
passed = (
|
|
draft.problem_valid is case.valid_input
|
|
and actual_auto_created == case.expected_auto_created
|
|
and actual_similar_ticket == case.expected_similar_ticket
|
|
and correct_count == evaluated_count
|
|
and timeline_correct
|
|
and evidence_covered_count == evidence_count
|
|
)
|
|
result, _ = IntakeEvaluationResult.objects.update_or_create(
|
|
case=case,
|
|
draft=draft,
|
|
defaults={
|
|
"ticket": draft.confirmed_ticket,
|
|
"actual_fields": actual_fields,
|
|
"actual_auto_created": actual_auto_created,
|
|
"correct_field_count": correct_count,
|
|
"evaluated_field_count": evaluated_count,
|
|
"field_accuracy": field_accuracy,
|
|
"passed": passed,
|
|
"details": {
|
|
"field_results": field_results,
|
|
"evidence_results": evidence_results,
|
|
"evidence_covered_count": evidence_covered_count,
|
|
"evidence_field_count": evidence_count,
|
|
"timeline_results": timeline_results,
|
|
"correct_timeline_count": correct_timeline_count,
|
|
"evaluated_timeline_count": len(expected_timeline),
|
|
"timeline_correct": timeline_correct,
|
|
"expected_problem_valid": case.valid_input,
|
|
"actual_problem_valid": draft.problem_valid,
|
|
"expected_similar_ticket": case.expected_similar_ticket,
|
|
"actual_similar_ticket": actual_similar_ticket,
|
|
},
|
|
"reviewed_by": reviewer,
|
|
},
|
|
)
|
|
return result
|
|
|
|
|
|
def _latest_evaluation_scope(*, include_synthetic: bool) -> tuple[list, list]:
|
|
cases = IntakeEvaluationCase.objects.filter(enabled=True)
|
|
if not include_synthetic:
|
|
cases = cases.filter(source_kind="anonymized_real")
|
|
case_list = list(cases.order_by("key"))
|
|
queryset = (
|
|
IntakeEvaluationResult.objects.select_related(
|
|
"case",
|
|
"draft",
|
|
"ticket",
|
|
"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_modality_coverage(cases: list[IntakeEvaluationCase]) -> dict:
|
|
image_types = {"jpg", "jpeg", "png", "webp", "image"}
|
|
document_types = {
|
|
"pdf",
|
|
"txt",
|
|
"log",
|
|
"docx",
|
|
"xlsx",
|
|
"pptx",
|
|
"document",
|
|
"text",
|
|
"spreadsheet",
|
|
"presentation",
|
|
}
|
|
coverage = {
|
|
"text_case_count": 0,
|
|
"image_case_count": 0,
|
|
"multi_image_case_count": 0,
|
|
"document_case_count": 0,
|
|
"mixed_text_image_case_count": 0,
|
|
"valid_input_case_count": 0,
|
|
"invalid_input_case_count": 0,
|
|
}
|
|
for case in cases:
|
|
manifest = case.input_manifest if isinstance(case.input_manifest, dict) else {}
|
|
has_text = bool(str(manifest.get("source_text") or "").strip())
|
|
materials = manifest.get("materials")
|
|
materials = materials if isinstance(materials, list) else []
|
|
material_types = []
|
|
for material in materials:
|
|
if not isinstance(material, dict):
|
|
continue
|
|
raw_type = str(
|
|
material.get("type")
|
|
or material.get("extension")
|
|
or Path(str(material.get("file_name") or "")).suffix.lstrip(".")
|
|
)
|
|
material_types.append(raw_type.casefold())
|
|
image_count = sum(item in image_types for item in material_types)
|
|
has_document = any(item in document_types for item in material_types)
|
|
coverage["text_case_count"] += int(has_text)
|
|
coverage["image_case_count"] += int(image_count > 0)
|
|
coverage["multi_image_case_count"] += int(image_count > 1)
|
|
coverage["document_case_count"] += int(has_document)
|
|
coverage["mixed_text_image_case_count"] += int(has_text and image_count > 0)
|
|
coverage["valid_input_case_count"] += int(case.valid_input)
|
|
coverage["invalid_input_case_count"] += int(not case.valid_input)
|
|
return coverage
|
|
|
|
|
|
def build_intake_quality_metrics(*, include_synthetic: bool = False) -> dict:
|
|
"""按每个案例最新评审结果计算覆盖率、自动建单率和字段准确率。"""
|
|
|
|
operational = TicketDraft.objects.filter(problem_valid=True).exclude(
|
|
status__in={"rejected", "expired"}
|
|
)
|
|
operational_valid_count = operational.count()
|
|
operational_auto_count = operational.filter(
|
|
decision="auto_created",
|
|
confirmed_ticket__isnull=False,
|
|
).count()
|
|
operational_auto_rate = (
|
|
operational_auto_count / operational_valid_count if operational_valid_count else None
|
|
)
|
|
|
|
cases, results = _latest_evaluation_scope(include_synthetic=include_synthetic)
|
|
valid_results = [result for result in results if result.case.valid_input]
|
|
evaluated_valid_count = len(valid_results)
|
|
evaluated_auto_count = sum(result.actual_auto_created for result in valid_results)
|
|
evaluated_auto_rate = (
|
|
evaluated_auto_count / evaluated_valid_count if evaluated_valid_count else None
|
|
)
|
|
correct_fields = sum(result.correct_field_count for result in results)
|
|
evaluated_fields = sum(result.evaluated_field_count for result in results)
|
|
field_accuracy = correct_fields / evaluated_fields if evaluated_fields else None
|
|
evidence_covered_fields = 0
|
|
evidence_fields = 0
|
|
correct_timeline_items = 0
|
|
evaluated_timeline_items = 0
|
|
for result in results:
|
|
details = result.details
|
|
if not isinstance(details, dict):
|
|
continue
|
|
evidence_covered_fields += int(details.get("evidence_covered_count") or 0)
|
|
evidence_fields += int(details.get("evidence_field_count") or 0)
|
|
correct_timeline_items += int(details.get("correct_timeline_count") or 0)
|
|
evaluated_timeline_items += int(details.get("evaluated_timeline_count") or 0)
|
|
reviewed_case_count = len(results)
|
|
enabled_case_count = len(cases)
|
|
coverage_complete = (
|
|
enabled_case_count > 0 and reviewed_case_count == enabled_case_count
|
|
)
|
|
return {
|
|
"include_synthetic": include_synthetic,
|
|
"operational": {
|
|
"valid_input_count": operational_valid_count,
|
|
"auto_created_count": operational_auto_count,
|
|
"auto_creation_rate": operational_auto_rate,
|
|
},
|
|
"evaluation": {
|
|
"enabled_case_count": enabled_case_count,
|
|
"reviewed_case_count": reviewed_case_count,
|
|
"coverage_rate": (
|
|
reviewed_case_count / enabled_case_count if enabled_case_count else None
|
|
),
|
|
"valid_input_count": evaluated_valid_count,
|
|
"auto_created_count": evaluated_auto_count,
|
|
"auto_creation_rate": evaluated_auto_rate,
|
|
"correct_field_count": correct_fields,
|
|
"evaluated_field_count": evaluated_fields,
|
|
"field_accuracy": field_accuracy,
|
|
"evidence_covered_field_count": evidence_covered_fields,
|
|
"evidence_field_count": evidence_fields,
|
|
"evidence_coverage_rate": (
|
|
evidence_covered_fields / evidence_fields if evidence_fields else None
|
|
),
|
|
"correct_timeline_item_count": correct_timeline_items,
|
|
"evaluated_timeline_item_count": evaluated_timeline_items,
|
|
"timeline_accuracy": (
|
|
correct_timeline_items / evaluated_timeline_items
|
|
if evaluated_timeline_items
|
|
else None
|
|
),
|
|
"result_selection": "latest_per_case",
|
|
},
|
|
"targets": {
|
|
"auto_creation_rate": 0.70,
|
|
"field_accuracy": 0.90,
|
|
"coverage_complete": coverage_complete,
|
|
"auto_creation_rate_met": (
|
|
evaluated_auto_rate >= 0.70
|
|
if coverage_complete and evaluated_auto_rate is not None
|
|
else None
|
|
),
|
|
"field_accuracy_met": (
|
|
field_accuracy >= 0.90
|
|
if coverage_complete and field_accuracy is not None
|
|
else None
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def build_stage_2_acceptance_report(*, include_synthetic: bool = False) -> dict:
|
|
"""生成阶段 2 退出报告；默认只允许脱敏真实案例作为验收依据。"""
|
|
|
|
cases, results = _latest_evaluation_scope(include_synthetic=include_synthetic)
|
|
metrics = build_intake_quality_metrics(include_synthetic=include_synthetic)
|
|
modality_coverage = _build_modality_coverage(cases)
|
|
result_by_case = {result.case_id: result for result in results}
|
|
case_results = []
|
|
for case in cases:
|
|
result = result_by_case.get(case.id)
|
|
details = result.details if result is not None and isinstance(result.details, dict) else {}
|
|
case_results.append(
|
|
{
|
|
"key": case.key,
|
|
"name": case.name,
|
|
"source_kind": case.source_kind,
|
|
"valid_input": case.valid_input,
|
|
"reviewed": result is not None,
|
|
"passed": result.passed if result is not None else None,
|
|
"draft_public_id": str(result.draft.public_id) if result is not None else None,
|
|
"ticket_id": result.ticket_id if result is not None else None,
|
|
"field_accuracy": (
|
|
float(result.field_accuracy)
|
|
if result is not None and result.field_accuracy is not None
|
|
else None
|
|
),
|
|
"evidence_coverage_rate": (
|
|
details.get("evidence_covered_count", 0)
|
|
/ details.get("evidence_field_count", 0)
|
|
if details.get("evidence_field_count", 0)
|
|
else None
|
|
),
|
|
"timeline_accuracy": (
|
|
details.get("correct_timeline_count", 0)
|
|
/ details.get("evaluated_timeline_count", 0)
|
|
if details.get("evaluated_timeline_count", 0)
|
|
else None
|
|
),
|
|
"reviewed_at": result.updated_at.isoformat() if result is not None else None,
|
|
}
|
|
)
|
|
|
|
reasons = []
|
|
readiness_state = "failed"
|
|
if include_synthetic:
|
|
readiness_state = "diagnostic"
|
|
reasons.append(
|
|
{
|
|
"code": "synthetic_scope",
|
|
"message": "合成案例只验证评估框架，不能作为阶段 2 业务退出证据。",
|
|
}
|
|
)
|
|
elif not cases:
|
|
readiness_state = "not_ready"
|
|
reasons.append(
|
|
{
|
|
"code": "missing_real_cases",
|
|
"message": "尚未导入启用的脱敏真实案例。",
|
|
}
|
|
)
|
|
elif not metrics["targets"]["coverage_complete"]:
|
|
readiness_state = "not_ready"
|
|
reasons.append(
|
|
{
|
|
"code": "incomplete_review_coverage",
|
|
"message": "仍有启用案例没有评审结果。",
|
|
}
|
|
)
|
|
|
|
if not include_synthetic and cases:
|
|
required_modalities = {
|
|
"image_case_count": "真实案例集未覆盖图片输入。",
|
|
"mixed_text_image_case_count": "真实案例集未覆盖文本加图片混合输入。",
|
|
"document_case_count": "真实案例集未覆盖文档或日志输入。",
|
|
"invalid_input_case_count": "真实案例集未覆盖无效输入与人工追问路径。",
|
|
}
|
|
for key, message in required_modalities.items():
|
|
if modality_coverage[key] == 0:
|
|
if readiness_state == "failed":
|
|
readiness_state = "not_ready"
|
|
reasons.append({"code": f"missing_{key}", "message": message})
|
|
|
|
if metrics["targets"]["coverage_complete"] and not include_synthetic:
|
|
if metrics["targets"]["auto_creation_rate_met"] is not True:
|
|
reasons.append(
|
|
{
|
|
"code": "auto_creation_rate_below_target",
|
|
"message": "有效输入自动建单率未达到 70%。",
|
|
}
|
|
)
|
|
if metrics["targets"]["field_accuracy_met"] is not True:
|
|
reasons.append(
|
|
{
|
|
"code": "field_accuracy_below_target",
|
|
"message": "自动建单字段准确率未达到 90%。",
|
|
}
|
|
)
|
|
evidence_count = metrics["evaluation"]["evidence_field_count"]
|
|
evidence_rate = metrics["evaluation"]["evidence_coverage_rate"]
|
|
if not evidence_count or evidence_rate != 1.0:
|
|
reasons.append(
|
|
{
|
|
"code": "incomplete_field_evidence",
|
|
"message": "非空评估字段尚未全部具备可追溯证据。",
|
|
}
|
|
)
|
|
if metrics["evaluation"]["evaluated_timeline_item_count"] == 0:
|
|
reasons.append(
|
|
{
|
|
"code": "missing_timeline_baseline",
|
|
"message": "真实案例集没有可比对的期望消息时间线。",
|
|
}
|
|
)
|
|
|
|
exit_ready = not reasons and not include_synthetic
|
|
if exit_ready:
|
|
readiness_state = "passed"
|
|
elif readiness_state == "failed" and any(
|
|
reason["code"]
|
|
in {
|
|
"missing_real_cases",
|
|
"incomplete_review_coverage",
|
|
"missing_image_case_count",
|
|
"missing_mixed_text_image_case_count",
|
|
"missing_document_case_count",
|
|
"missing_invalid_input_case_count",
|
|
}
|
|
for reason in reasons
|
|
):
|
|
readiness_state = "not_ready"
|
|
|
|
return {
|
|
"schema_version": 1,
|
|
"generated_at": timezone.now().isoformat(),
|
|
"scope": "diagnostic_with_synthetic" if include_synthetic else "anonymized_real",
|
|
"status": readiness_state,
|
|
"exit_ready": exit_ready,
|
|
"reasons": reasons,
|
|
"modality_coverage": modality_coverage,
|
|
"metrics": metrics,
|
|
"case_results": case_results,
|
|
}
|