RAG 幻觉治理实战：五层证据控制、拒答与发布门禁

{
  "run_id": "rag-202",
  "query": "外包人员的国际差旅能否报销？",
  "principal": "user-42",
  "tenant": "acme",
  "corpus_revision": "policy-index@sha256:7ad1",
  "retriever_version": "hybrid-v4",
  "reranker_version": "rerank-v3",
  "generator_version": "answer-v8",
  "policy_version": "evidence-policy-v5",
  "evidence_ids": ["travel-2026#p18", "contractors-2026#p4"],
  "decision": "review"
}

{
  "evidence_id": "travel-2026#p18",
  "document_id": "travel-policy",
  "revision": "sha256:3c81",
  "locator": {"page": 18, "section": "International travel"},
  "authority": "finance-policy-owner",
  "valid_from": "2026-01-01",
  "valid_to": null,
  "tenant": "acme",
  "allowed_roles": ["employee", "contractor-manager"],
  "status": "active",
  "content_hash": "sha256:915f"
}

query_id: contractor-international-travel
required_facts:
  - contractor eligibility
  - international travel rule
  - approval authority
required_qualifiers:
  - employment type
  - effective date
unanswerable_when:
  - contractor policy is absent
  - applicable revisions conflict without precedence

<evidence id="travel-2026#p18" status="active" valid_from="2026-01-01">
外包人员国际差旅必须由项目所属总监审批。
</evidence>

<evidence id="travel-2024#p11" status="superseded" valid_to="2025-12-31">
外包人员国际差旅必须由部门经理审批。
</evidence>

{
  "claims": [
    {
      "claim_id": "c1",
      "text": "外包人员国际差旅必须由项目所属总监审批。",
      "evidence_ids": ["travel-2026#p18"]
    }
  ],
  "answer": "外包人员国际差旅必须由项目所属总监审批 [c1]。"
}

from dataclasses import dataclass
from enum import Enum
from typing import Dict, FrozenSet, List, Optional


class Verdict(str, Enum):
    SUPPORTED = "supported"
    CONTRADICTED = "contradicted"
    INSUFFICIENT = "insufficient"


class Outcome(str, Enum):
    RELEASE = "release"
    PARTIAL = "partial"
    ABSTAIN = "abstain"
    REVIEW = "review"
    DENY = "deny"


@dataclass(frozen=True)
class Evidence:
    evidence_id: str
    tenant: str
    allowed_roles: FrozenSet[str]
    status: str
    valid_from: int
    valid_to: Optional[int]


@dataclass(frozen=True)
class Claim:
    claim_id: str
    text: str
    evidence_ids: tuple
    verdict: Verdict
    is_material: bool = True


@dataclass(frozen=True)
class Decision:
    outcome: Outcome
    released_claim_ids: tuple
    reasons: tuple


def is_authorized(
    evidence: Evidence,
    *,
    tenant: str,
    roles: FrozenSet[str],
    now: int,
) -> bool:
    return (
        evidence.tenant == tenant
        and bool(evidence.allowed_roles & roles)
        and evidence.status == "active"
        and evidence.valid_from <= now
        and (evidence.valid_to is None or now <= evidence.valid_to)
    )


def decide(
    claims: List[Claim],
    evidence_by_id: Dict[str, Evidence],
    *,
    tenant: str,
    roles: FrozenSet[str],
    now: int,
    high_risk: bool,
) -> Decision:
    reasons = []
    supported = []

    for claim in claims:
        evidence = [evidence_by_id.get(item) for item in claim.evidence_ids]
        if not evidence or any(item is None for item in evidence):
            reasons.append("%s:unknown_evidence" % claim.claim_id)
            continue
        if not all(
            is_authorized(item, tenant=tenant, roles=roles, now=now)
            for item in evidence
        ):
            return Decision(Outcome.DENY, (), ("%s:unauthorized" % claim.claim_id,))
        if claim.verdict == Verdict.CONTRADICTED:
            return Decision(Outcome.REVIEW, (), ("%s:contradicted" % claim.claim_id,))
        if claim.verdict == Verdict.SUPPORTED:
            supported.append(claim.claim_id)
        elif claim.is_material:
            reasons.append("%s:insufficient" % claim.claim_id)

    material = [claim for claim in claims if claim.is_material]
    if not material or not supported:
        return Decision(Outcome.ABSTAIN, (), tuple(reasons or ["no_material_claims"]))
    if high_risk and reasons:
        return Decision(Outcome.REVIEW, (), tuple(reasons))
    if reasons:
        return Decision(Outcome.PARTIAL, tuple(supported), tuple(reasons))
    return Decision(Outcome.RELEASE, tuple(supported), ())


evidence = {
    "travel-2026#p18": Evidence(
        evidence_id="travel-2026#p18",
        tenant="acme",
        allowed_roles=frozenset({"employee", "contractor-manager"}),
        status="active",
        valid_from=1_767_225_600,
        valid_to=None,
    )
}

supported_claim = Claim(
    claim_id="c1",
    text="Contractors need sponsoring-director approval.",
    evidence_ids=("travel-2026#p18",),
    verdict=Verdict.SUPPORTED,
)
missing_claim = Claim(
    claim_id="c2",
    text="The company always reimburses business-class fares.",
    evidence_ids=("travel-2026#p18",),
    verdict=Verdict.INSUFFICIENT,
)

release = decide(
    [supported_claim],
    evidence,
    tenant="acme",
    roles=frozenset({"contractor-manager"}),
    now=1_800_000_000,
    high_risk=False,
)
partial = decide(
    [supported_claim, missing_claim],
    evidence,
    tenant="acme",
    roles=frozenset({"contractor-manager"}),
    now=1_800_000_000,
    high_risk=False,
)
denied = decide(
    [supported_claim],
    evidence,
    tenant="other",
    roles=frozenset({"contractor-manager"}),
    now=1_800_000_000,
    high_risk=False,
)

assert release.outcome == Outcome.RELEASE
assert partial.outcome == Outcome.PARTIAL
assert partial.released_claim_ids == ("c1",)
assert denied.outcome == Outcome.DENY
print(release.outcome.value, partial.outcome.value, denied.outcome.value)

release partial deny