EvidenceFlow/engine.py at master · dll61/EvidenceFlow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
from __future__ import annotations

import hashlib
import json
from datetime import UTC, datetime
from typing import Any

from .citation import CrossrefClient
from .claims import split_claims
from .evidence import verify_claim
from .repro import lint_reproducibility
from .sources import MultiSourceCitationClient, OpenAlexClient


class AuditEngine:
    def __init__(self, citation_client: Any | None = None) -> None:
        self.citation_client = citation_client or MultiSourceCitationClient(
            (CrossrefClient(), OpenAlexClient())
        )

    def audit(self, payload: dict[str, Any]) -> dict[str, Any]:
        document_text = str(payload.get("document_text") or "").strip()
        references = payload.get("references") or []
        evidence_items = payload.get("evidence") or []
        live_resolution = bool(payload.get("live_resolution", True))

        if not document_text:
            raise ValueError("document_text 为必填字段")
        if not isinstance(references, list) or len(references) > 50:
            raise ValueError("references 必须是最多 50 条的数组")
        if not isinstance(evidence_items, list) or len(evidence_items) > 100:
            raise ValueError("evidence 必须是最多 100 条的数组")

        claims = split_claims(document_text)
        citations = (
            [self.citation_client.verify(str(reference)) for reference in references]
            if live_resolution
            else [
                {
                    "reference": str(reference),
                    "status": "not_checked",
                    "reason": "live_resolution=false，未请求外部数据源",
                    "requires_human_review": True,
                }
                for reference in references
            ]
        )

        claim_by_id = {claim["claim_id"]: claim for claim in claims}
        evidence_results: list[dict[str, Any]] = []
        for item in evidence_items:
            if not isinstance(item, dict):
                continue
            claim_id = str(item.get("claim_id") or "")
            evidence_text = str(item.get("text") or "")
            claim = claim_by_id.get(claim_id)
            if not claim:
                evidence_results.append(
                    {
                        "claim_id": claim_id,
                        "label": "invalid_claim_id",
                        "reason": "找不到对应论断",
                        "requires_human_review": True,
                    }
                )
                continue
            result = verify_claim(claim["text"], evidence_text)
            result.update({"claim_id": claim_id, "evidence_text": evidence_text})
            evidence_results.append(result)

        repro = lint_reproducibility(document_text)
        high_risk = sum(
            citation["status"]
            in {"metadata_conflict", "not_found_in_crossref", "not_found_in_sources"}
            for citation in citations
        )
        high_risk += sum(result.get("label") == "contradict" for result in evidence_results)
        review_items = sum(
            bool(item.get("requires_human_review")) for item in citations + evidence_results
        )
        canonical_payload = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")

        return {
            "audit_id": hashlib.sha256(canonical_payload).hexdigest()[:16],
            "created_at": datetime.now(UTC).isoformat(),
            "privacy": {
                "stored": False,
                "note": "服务默认不持久化请求正文；调用方仍需避免上传未授权或保密内容",
            },
            "claims": claims,
            "citations": citations,
            "evidence_checks": evidence_results,
            "reproducibility": repro,
            "summary": {
                "claim_count": len(claims),
                "citation_count": len(citations),
                "high_risk_count": high_risk,
                "human_review_item_count": review_items,
                "repro_missing_count": repro["missing_count"],
            },
            "limitations": [
                "Crossref 未检索到不等于文献必然不存在",
                "当前论断—证据判断是透明词项基线，必须人工复核",
                "系统不替代导师、编辑、法务或同行评审",
            ],
        }