vent_ai_platform/app/services/extraction_accuracy.py

from __future__ import annotations

import json
import re
import unicodedata
from dataclasses import dataclass
from typing import Any

from app.agents.runtime import get_agent_runtime


@dataclass
class EditJudgement:
    is_extraction_error: bool
    reason: str
    confidence: float
    source: str


def _parse_json_object(text: str) -> dict[str, Any] | None:
    cleaned = text.strip()
    if cleaned.startswith("```"):
        cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned)
        cleaned = re.sub(r"\s*```$", "", cleaned)
    start = cleaned.find("{")
    end = cleaned.rfind("}")
    if start < 0 or end <= start:
        return None
    try:
        value = json.loads(cleaned[start : end + 1])
    except json.JSONDecodeError:
        return None
    return value if isinstance(value, dict) else None


def _normalize_text(value: str) -> str:
    normalized = unicodedata.normalize("NFKC", str(value or "")).lower()
    return re.sub(r"[\s，。；、,.;:：!！?？\"'“”‘’()（）\[\]【】]+", "", normalized)


def _token_set(value: str) -> set[str]:
    return {
        token
        for token in re.split(r"[\s，。；、,.;:：|/]+", str(value or "").strip())
        if token
    }


def _fallback_judgement(
    original_value: str,
    edited_value: str,
) -> EditJudgement:
    original_normalized = _normalize_text(original_value)
    edited_normalized = _normalize_text(edited_value)
    if original_normalized == edited_normalized:
        return EditJudgement(
            is_extraction_error=False,
            reason="修改仅涉及空格、标点或显示格式，不判定为抽取错误。",
            confidence=0.95,
            source="fallback",
        )
    if (
        original_normalized
        and edited_normalized
        and (
            original_normalized in edited_normalized
            or edited_normalized in original_normalized
        )
    ):
        return EditJudgement(
            is_extraction_error=False,
            reason="修改主要属于内容截断、补充说明或显示长度调整，不判定为抽取错误。",
            confidence=0.8,
            source="fallback",
        )
    original_tokens = _token_set(original_value)
    edited_tokens = _token_set(edited_value)
    if original_tokens and original_tokens == edited_tokens:
        return EditJudgement(
            is_extraction_error=False,
            reason="字段内容仅调整顺序或分隔方式，不判定为抽取错误。",
            confidence=0.85,
            source="fallback",
        )
    return EditJudgement(
        is_extraction_error=True,
        reason="字段值发生了实质变化，暂按原抽取结果错误记录。",
        confidence=0.55,
        source="fallback",
    )


def _normalize_model_confidence(value: Any) -> float:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return 0.7
    if number > 1:
        number /= 100
    return max(0.0, min(number, 1.0))


async def judge_extraction_edit(
    *,
    field_name: str,
    original_value: str,
    edited_value: str,
    evidence_text: str = "",
    document_title: str = "",
    extraction_id: str = "",
) -> EditJudgement:
    fallback = _fallback_judgement(original_value, edited_value)
    runtime = get_agent_runtime()
    if not runtime.enabled:
        return fallback

    prompt = f"""
你是煤矿业务资料字段抽取质量判定器。用户修改了系统抽取出的字段值，
请判断这次修改是否说明“原抽取结果存在抽取错误”。

判定规则：
1. 如果用户只是删除多余内容、截断过长文本、调整换行或空格、统一标点、调整顺序、
   简化表达、补充说明或改变显示格式，则应判定为“非抽取错误”。
2. 如果用户修正了实质性的数值、日期、单位、名称、地点、结论或字段含义，
   则应判定为“抽取错误”。
3. 只判断修改原因是否属于抽取错误，不要评价修改后的文字是否更好。

资料标题：{document_title or "未提供"}
字段名称：{field_name}
原抽取值：{original_value[:1200]}
用户修改后：{edited_value[:1200]}
原文证据：{evidence_text[:1200] or "未提供"}

只输出 JSON：
{{"is_extraction_error": true, "reason": "简短中文理由", "confidence": 0.9}}
""".strip()

    result = await runtime.invoke(
        message=prompt,
        thread_id=f"extraction-edit-judgement-{extraction_id or field_name}",
        context={"purpose": "field_edit_accuracy"},
    )
    if result.status != "success":
        return fallback
    parsed = _parse_json_object(result.output)
    if parsed is None or "is_extraction_error" not in parsed:
        return fallback
    raw_flag = parsed.get("is_extraction_error")
    if isinstance(raw_flag, str):
        is_extraction_error = raw_flag.strip().lower() in {"true", "yes", "1", "是"}
    else:
        is_extraction_error = bool(raw_flag)
    return EditJudgement(
        is_extraction_error=is_extraction_error,
        reason=str(parsed.get("reason") or ("属于抽取错误" if is_extraction_error else "不属于抽取错误")),
        confidence=_normalize_model_confidence(parsed.get("confidence")),
        source="llm",
    )