第七篇中，我把搜索、读取和保存做成 MCP 工具，并让写入经过具体内容确认。现在系统能够生成文件，但还有一个更重要的问题：报告里出现了引用，是否就能说明结论有依据？

不能。模型可能引用一个存在的文档，却把文档里的条件省掉；也可能找到正确的句子，再写出方向相反的结论。工具成功与文件生成，只证明执行过程走到了某一步，并没有证明报告可信。

这一篇我把研究结果拆成来源、证据片段、论断、审阅记录与候选产物，分别检查它们之间的关系。代码先验证引用确实存在，再把支持关系交给独立审阅，最后生成带状态和缺口的 Markdown。

资料核对与本地验证日期：2026-11-06。文内包含六个完整源码与依赖文件，另有 08_evidence_reports/ 运行副本。

基础演示使用人工测试资料与固定审阅替身，不调用模型、不联网检索。另有真实 MCP 读取集成，需要保留第七篇源码。教学资料不能作为技术方案比较的研究结论。

一、检索解决“在哪里找”，证据解决“凭什么说”

检索增强生成，即 RAG，把检索到的外部内容提供给生成过程。原始 RAG 工作将参数化模型与可检索的非参数记忆结合，说明了获取外部资料的一条路线。资料 1 本系列采用更宽泛的工程分层：先获取授权资料，再让模型基于证据生成候选论断。

第七篇的字面搜索只是获取候选来源的一种方法，不能等同于向量检索。以后换成全文索引、混合检索或向量数据库，仍需要保留来源映射、权限过滤和引用校验。

阶段	回答的问题	常见误判
检索	哪些资料可能相关？	把排序靠前当成权威
证据定位	具体哪段原文？	把搜索摘要当成完整依据
支持关系审阅	原文是否支持这项论断？	原文出现过关键词，就认定结论正确
报告交付	读者能否检查过程和缺口？	文件存在，就宣布研究完成

ALCE 研究把生成文本的正确性与引用质量纳入独立评估维度，为“有引用格式”和“有足够引用支持”提供了可区分的评价方向。资料 2 本篇不复刻其评测指标，也不借用论文实验数字宣称本项目有效。

我先落实一个可检查的最小契约：每项外部事实都有原文位置；每项推断显式说明依据；没有证据的判断保留不足状态；相互矛盾的说法不被悄悄合并成确定结论。

二、把报告拆成五种对象

来源配置由 Host 提供，包括来源 ID、标题、逻辑位置、文件映射与类型。模型不能通过自己填写标题或 URL，把未读取的资料注册成可信来源。

来源快照记录正文、版本、内容指纹与获取时间。引用只保存版本引用、字符区间和必要引文，避免每项论断复制整篇文档。报告再通过引用 ID 连接到这些片段。

论断的 fact 类型表示“来源事实陈述”，不表示事实已经证实。例如“说明书称支持离线整理”与“产品实际能完成所有离线任务”是两个不同判断。推断使用 inference，还必须提供 rationale，说明从证据到建议的推导。

最后，审阅记录由图外可信入口创建，绑定用户范围、报告结构摘要、论断 ID、审阅结论与时间。它不属于模型能提交的 Report 字段。模型即使在正文写“已审核”，也不能获得这个状态。

三、引用的位置和版本如何保持一致

我先统一正文换行：把 CRLF 与 CR 转成 LF，然后计算 UTF-8 正文的 SHA-256。引用的 start 和 end 都相对于这个标准化正文，采用 Python 字符索引与左闭右开区间 [start, end)。

这里的字符位置不是 UTF-8 字节偏移，也不是 token 数。引用“方案 A 支持离线整理。”时，程序检查：

当前来源版本 == 引用版本
当前正文指纹 == 引用指纹
正文[start:end] == quote

三项检查各有作用。版本避免来源更新后继续使用旧片段，指纹发现内容与元数据不一致，区间校验防止模型只填写正确来源 ID，却编造原文。

相同标准化内容再次进入存储时不创建新版本，获取时间也保持为该版本首次进入本存储的时间。这个时间不是网页发布日期，不能用它证明信息没有过时。生产研究任务还应根据资料类型制定有效期，保留发布与修订时间。

删除来源时保留版本计数；删除后即使重新写入相同内容，也不能复活旧引用。来源失效后，当前候选报告不能继续导出，需要重新读取、生成引用和审阅。已经导出的文件则是一份历史产物，撤销内存来源不会自动删除它。

本篇使用小型同线程内存存储，最多登记十六个来源，每份正文最多 24,000 字节。每次任务创建独立存储，结束时关闭，避免持续积累其他任务的数据。接口沿用第五篇身份范围与版本引用的含义，但没有复制完整 SQLite 工程，也不支持持久历史、TTL 或进程恢复。

四、引用存在性不等于支持关系

假设原文写着“方案 A 支持离线整理”，模型引用了这句话，却声称“方案 A 不支持离线整理”。原文位置、版本和指纹都能通过，但论断方向明显不对。

我不会用“关键词重合”冒充语义验证。代码把它保留为 pending_review，可信审阅者可以记录 contradicted。审阅需要检查条件、范围、时效、否定关系和是否存在反证；数值结论还需要核对单位与口径。

报告状态	含义
invalid	引用失效或不匹配，拒绝导出
needs_review	引用有效，但仍有支持关系待审阅
partial	存在证据不足、反驳、显式冲突或已登记缺口
reviewed_candidate	本候选的论断均有支持审阅，且未登记冲突或缺口

最后一个状态仍然叫候选。它表示通过了当前契约，不表示世界中的事实已被形式化证明，也不表示检索范围已经穷尽。即使有人错误地把空引用论断标成支持，程序也仍保留证据不足。

显式冲突通过 Conflict 关联多个论断，原文和相反说法都会留在产物中。本篇没有自动发现矛盾；候选生成过程或审阅者需要登记冲突。若来源实际讨论不同版本或条件，应补充解释，而不是简单选多数。

同一个来源的两个片段也不等于两份独立证据。转载、镜像和相互引用的材料可能来自同一个原始出处。当前结构不自动识别来源独立性，真实研究必须回溯原始资料。

五、候选产物要能回溯，也要能绑定确认

我区分三种摘要：

摘要	绑定内容	用途
来源 digest	标准化正文	检查引用对应的来源内容
report_digest	候选结构的确定性 JSON	防止报告变化后复用旧审阅
content_digest	最终 Markdown 的完整 UTF-8 字节	绑定即将保存的具体文件内容

报告结构变了，旧审阅记录会失效。重新渲染会产生新的导出时间，最终字节摘要也可能改变，所以确认时必须冻结一个 Artifact，审核后保存相同正文。

保存前还应重新检查来源与审阅。如果来源变化，重新生成候选内容并重新确认。第七篇的凭据绑定文件名与正文，尚未实现来源版本和文件写入之间的原子事务；检查之后仍可能发生并发更新。严格一致性需要快照归档或明确的版本提交协议。

产物包含逐项论断状态、推导说明、审阅记录、冲突与缺口，以及引用标题、逻辑位置、版本、正文摘要、获取时间、字符区间和行号。信息多于一份普通回答，但读者可以沿引用回到具体依据。

外部正文按字面文本展示，HTML 与 Markdown 特殊字符被转义。这样避免引用内容在预览中变成外部图片或伪装链接；它不能阻止模型在阅读资料时受到提示注入影响。OWASP 将远程资料中的间接注入与 RAG 污染列为需要处理的攻击路径，并强调文字分隔和关键词筛查不是权限边界。资料 3

因此，第五篇上下文应继续区分资料与可信指令，第七篇工具权限和确认也继续执行。本篇没有调用模型，不能用转义检查证明模型具备提示注入防御能力。

六、完整源码

以下六个文件放在同一目录，基础演示可以独立运行。mcp_demo.py 额外依赖第七篇的 07_mcp_document_tools/ 服务端源码，路径可以从命令行指定。

08_evidence_reports/
├── contracts.py
├── evidence_store.py
├── reporting.py
├── demo.py
├── mcp_demo.py
└── requirements.txt

6.1 contracts.py：来源、引用与候选报告

"""候选报告由模型提出，来源配置和审阅记录由可信 Host 管理。"""

from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

Identifier = Annotated[str, Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")]
Digest = Annotated[str, Field(pattern=r"^[a-f0-9]{64}$")]
Text = Annotated[str, Field(min_length=1, max_length=1000)]


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)


class Scope(StrictModel):
    tenant_id: Identifier
    user_id: Identifier


class SourceSpec(StrictModel):
    source_id: Identifier
    title: Annotated[str, Field(min_length=1, max_length=120)]
    locator: Annotated[str, Field(min_length=1, max_length=300)]
    filename: Annotated[str, Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}\.md$")]
    kind: Literal["fixture", "public"]


class SourceRef(StrictModel):
    source_id: Identifier
    revision: Annotated[int, Field(ge=1)]
    digest: Digest


class ReadPayload(StrictModel):
    filename: Annotated[str, Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}\.md$")]
    content: Annotated[str, Field(min_length=1, max_length=24_000)]


class ReadResult(StrictModel):
    status: Literal["ok"]
    message: Annotated[str, Field(max_length=1000)] = ""
    data: ReadPayload


class Citation(StrictModel):
    citation_id: Identifier
    ref: SourceRef
    start: Annotated[int, Field(ge=0)]
    end: Annotated[int, Field(ge=1)]
    quote: Annotated[str, Field(min_length=1, max_length=800)]

    @model_validator(mode="after")
    def check_range(self):
        if self.end <= self.start:
            raise ValueError("引用区间必须非空，采用左闭右开区间。")
        return self


class Claim(StrictModel):
    claim_id: Identifier
    text: Text
    kind: Literal["fact", "inference"]
    rationale: Annotated[str, Field(max_length=1000)] = ""
    citations: Annotated[tuple[Citation, ...], Field(max_length=4)] = ()

    @model_validator(mode="after")
    def check_claim(self):
        if not self.text.strip() or (self.kind == "inference" and not self.rationale.strip()):
            raise ValueError("论断不能为空；推断必须说明推导依据。")
        return self


class Conflict(StrictModel):
    claim_ids: Annotated[tuple[Identifier, ...], Field(min_length=2, max_length=4)]
    reason: Text


class Report(StrictModel):
    title: Annotated[str, Field(min_length=1, max_length=120)]
    claims: Annotated[tuple[Claim, ...], Field(min_length=1, max_length=12)]
    conflicts: Annotated[tuple[Conflict, ...], Field(max_length=6)] = ()
    gaps: Annotated[tuple[Text, ...], Field(max_length=8)] = ()

    @model_validator(mode="after")
    def check_links(self):
        ids = [claim.claim_id for claim in self.claims]
        if len(set(ids)) != len(ids):
            raise ValueError("论断标识重复。")
        citations = {}
        for claim in self.claims:
            local_ids = [citation.citation_id for citation in claim.citations]
            if len(set(local_ids)) != len(local_ids):
                raise ValueError("同一论断重复引用。")
            for citation in claim.citations:
                previous = citations.setdefault(citation.citation_id, citation)
                if previous != citation:
                    raise ValueError("同一引用标识对应不同片段。")
        for conflict in self.conflicts:
            if len(set(conflict.claim_ids)) != len(conflict.claim_ids):
                raise ValueError("冲突成员重复。")
            if not set(conflict.claim_ids) <= set(ids):
                raise ValueError("冲突关联了不存在的论断。")
        return self

6.2 evidence_store.py：版本与原文位置

"""小型同线程内存存储；保持与第五篇版本引用一致的含义。"""

import hashlib
from dataclasses import dataclass
from datetime import datetime, timezone

from contracts import Citation, ReadResult, Scope, SourceRef, SourceSpec

MAX_SOURCE_BYTES = 24_000


def digest_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass(frozen=True)
class Snapshot:
    spec: SourceSpec
    ref: SourceRef
    content: str
    fetched_at: str


class EvidenceStore:
    def __init__(self, specs: tuple[SourceSpec, ...]):
        if len(specs) > 16 or len({spec.source_id for spec in specs}) != len(specs):
            raise ValueError("来源配置重复或超限。")
        self.specs = {spec.source_id: spec for spec in specs}
        self.current: dict[tuple[Scope, str], Snapshot] = {}
        self.revisions: dict[tuple[Scope, str], int] = {}
        self.closed = False

    def ensure_open(self) -> None:
        if self.closed:
            raise ValueError("证据存储已关闭。")

    def ingest(self, scope: Scope, source_id: str, content: str) -> SourceRef:
        self.ensure_open()
        if source_id not in self.specs:
            raise ValueError("来源未在可信目录中登记。")
        if not isinstance(content, str) or not content.strip():
            raise ValueError("正文必须是非空字符串。")
        if len(content.encode("utf-8")) > MAX_SOURCE_BYTES:
            raise ValueError("来源超过字节上限。")
        # 统一换行后计算指纹和位置；不再做其他文本改写。
        content = content.replace("\r\n", "\n").replace("\r", "\n")
        key = (scope, source_id)
        fingerprint = digest_text(content)
        old = self.current.get(key)
        if old and old.ref.digest == fingerprint:
            return old.ref
        revision = self.revisions.get(key, 0) + 1
        ref = SourceRef(source_id=source_id, revision=revision, digest=fingerprint)
        self.current[key] = Snapshot(self.specs[source_id], ref, content, utc_now())
        self.revisions[key] = revision
        return ref

    def ingest_read_result(self, scope: Scope, source_id: str, result: dict) -> SourceRef:
        """接收第七篇已校验的读取业务结果，身份和来源映射由 Host 给定。"""
        if not isinstance(result, dict):
            raise ValueError("读取结果必须是业务结果对象。")
        parsed = ReadResult.model_validate(result)
        spec = self.specs.get(source_id)
        if spec is None or parsed.data.filename != spec.filename:
            raise ValueError("读取失败或来源映射不一致。")
        return self.ingest(scope, source_id, parsed.data.content)

    def get(self, scope: Scope, ref: SourceRef) -> Snapshot:
        self.ensure_open()
        snapshot = self.current.get((scope, ref.source_id))
        if snapshot is None or snapshot.ref != ref:
            raise ValueError("来源不存在、已撤销、越权或版本失效。")
        if digest_text(snapshot.content) != ref.digest:
            raise ValueError("正文指纹不匹配。")
        return snapshot

    def remove(self, scope: Scope, source_id: str) -> None:
        self.ensure_open()
        # 保留版本计数，删除后重建不能复活旧引用。
        self.current.pop((scope, source_id), None)

    def check_citation(self, scope: Scope, citation: Citation) -> Snapshot:
        snapshot = self.get(scope, citation.ref)
        if citation.end > len(snapshot.content):
            raise ValueError("引用区间超出正文。")
        if snapshot.content[citation.start:citation.end] != citation.quote:
            raise ValueError("引文与原文区间不一致。")
        return snapshot

    def quote(self, scope: Scope, ref: SourceRef, text: str, citation_id: str) -> Citation:
        """Host 从唯一出现的片段生成位置；歧义时要求指定更完整的原文。"""
        snapshot = self.get(scope, ref)
        if not text or snapshot.content.count(text) != 1:
            raise ValueError("片段缺失或重复，无法唯一定位。")
        start = snapshot.content.index(text)
        return Citation(citation_id=citation_id, ref=ref, start=start,
                        end=start + len(text), quote=text)

    def close(self) -> None:
        self.current.clear()
        self.revisions.clear()
        self.closed = True

6.3 reporting.py：审阅记录与候选产物

"""存在性与支持关系分开检查；只导出带状态的候选报告。"""

import html
import json
import re
from dataclasses import dataclass
from typing import Literal

from contracts import Report, Scope
from evidence_store import EvidenceStore, digest_text, utc_now


def report_digest(report: Report) -> str:
    encoded = json.dumps(report.model_dump(mode="json"), ensure_ascii=False,
                         sort_keys=True, separators=(",", ":"))
    return digest_text(encoded)


@dataclass(frozen=True)
class Review:
    scope: Scope
    report_digest: str
    claim_id: str
    verdict: Literal["supported", "insufficient", "contradicted"]
    note: str
    reviewed_at: str


def record_review(scope: Scope, report: Report, claim_id: str, verdict: str, note: str) -> Review:
    """仅供可信审阅入口调用；本函数不负责认证，也不自动判断语义。"""
    if claim_id not in {claim.claim_id for claim in report.claims}:
        raise ValueError("审阅目标不存在。")
    if verdict not in {"supported", "insufficient", "contradicted"}:
        raise ValueError("审阅结论无效。")
    if not isinstance(note, str) or not note.strip() or len(note) > 1000:
        raise ValueError("需要非空、大小受限的审阅说明。")
    return Review(scope, report_digest(report), claim_id, verdict, note, utc_now())


def assess(scope: Scope, report: Report, store: EvidenceStore, reviews: tuple[Review, ...] = ()) -> dict:
    fingerprint = report_digest(report)
    review_map = {}
    for review in reviews:
        if (review.scope != scope or review.report_digest != fingerprint
                or review.claim_id in review_map):
            raise ValueError("审阅范围、报告摘要不匹配或记录重复。")
        if review.claim_id not in {claim.claim_id for claim in report.claims}:
            raise ValueError("审阅目标不在报告中。")
        review_map[review.claim_id] = review
    citations, snapshots, issues, states = {}, {}, [], {}
    disputed = {claim_id for conflict in report.conflicts for claim_id in conflict.claim_ids}
    for claim in report.claims:
        invalid = False
        for citation in claim.citations:
            if citation.citation_id not in citations:
                try:
                    snapshots[citation.citation_id] = store.check_citation(scope, citation)
                    citations[citation.citation_id] = citation
                except ValueError as error:
                    issues.append({"claim_id": claim.claim_id, "message": str(error)})
                    invalid = True
        if invalid:
            states[claim.claim_id] = "invalid"
        elif not claim.citations:
            states[claim.claim_id] = "insufficient"
        elif claim.claim_id in disputed:
            states[claim.claim_id] = "disputed"
        elif claim.claim_id not in review_map:
            states[claim.claim_id] = "pending_review"
        else:
            states[claim.claim_id] = review_map[claim.claim_id].verdict
    if issues:
        status = "invalid"
    elif report.gaps or any(state in {"insufficient", "contradicted", "disputed"} for state in states.values()):
        status = "partial"
    elif "pending_review" in states.values():
        status = "needs_review"
    else:
        status = "reviewed_candidate"
    return {"status": status, "report_digest": fingerprint, "claims": states, "issues": issues,
            "citations": citations, "snapshots": snapshots, "reviews": review_map}


def literal(text: str) -> str:
    # 引文保持原样用于校验；展示时转为字面文本，避免注入 HTML/Markdown。
    text = html.escape(" ".join(text.split()), quote=False)
    return re.sub(r"([\\`*_{}\[\]()#+\-.!|>~])", r"\\\1", text)


@dataclass(frozen=True)
class Artifact:
    content: str
    content_digest: str
    report_digest: str
    status: str


def render(scope: Scope, report: Report, store: EvidenceStore, reviews: tuple[Review, ...] = ()) -> Artifact:
    assessment = assess(scope, report, store, reviews)
    if assessment["status"] == "invalid":
        raise ValueError("报告引用无效，拒绝导出。")
    citations = assessment["citations"]
    numbers = {citation_id: index for index, citation_id in enumerate(citations, 1)}
    fixture = any(snapshot.spec.kind == "fixture" for snapshot in assessment["snapshots"].values())
    lines = [f"# {literal(report.title)}", "",
             "> 候选产物；存在性校验与审阅记录不等于事实真值证明。", "",
             f"状态：{assessment['status']}；导出时间：{utc_now()}", "",
             f"结构摘要：{assessment['report_digest']}", ""]
    if fixture:
        lines.extend(["> 含人工测试资料，仅用于验证代码，不能作为研究结论。", ""])
    labels = {"supported": "审阅认为支持", "insufficient": "证据不足",
              "contradicted": "审阅认为反驳", "disputed": "存在冲突",
              "pending_review": "支持关系待审阅"}
    lines.extend(["## 逐项论断", ""])
    for claim in report.claims:
        links = " ".join(f"[{numbers[c.citation_id]}](#source-{numbers[c.citation_id]})"
                         for c in claim.citations)
        state = labels[assessment["claims"][claim.claim_id]]
        kind = "来源事实陈述" if claim.kind == "fact" else "推断"
        lines.extend([f"- **{literal(claim.claim_id)} / {kind} / {state}**：{literal(claim.text)} {links}", ""])
        if claim.rationale:
            lines.extend([f"  推导说明：{literal(claim.rationale)}", ""])
        review = assessment["reviews"].get(claim.claim_id)
        if review:
            lines.extend([f"  审阅记录：{literal(review.verdict)}；{literal(review.note)}；"
                          f"{literal(review.reviewed_at)}", ""])
    lines.extend(["## 冲突与缺口", ""])
    for conflict in report.conflicts:
        lines.extend([f"- 冲突 {', '.join(conflict.claim_ids)}：{literal(conflict.reason)}", ""])
    for gap in report.gaps:
        lines.extend([f"- 缺口：{literal(gap)}", ""])
    if not report.conflicts and not report.gaps:
        lines.extend(["未登记显式冲突或缺口；这不表示资料完整。", ""])
    lines.extend(["## 原文引用", ""])
    for citation_id, citation in citations.items():
        snapshot = assessment["snapshots"][citation_id]
        number = numbers[citation_id]
        start_line = snapshot.content.count("\n", 0, citation.start) + 1
        end_line = snapshot.content.count("\n", 0, citation.end - 1) + 1
        lines.extend([f'<a id="source-{number}"></a>', "",
                      f"### 引用 {number}：{literal(snapshot.spec.title)}", "",
                      f"来源：{literal(snapshot.spec.locator)}；类型：{snapshot.spec.kind}", "",
                      f"版本：{citation.ref.revision}；正文 SHA-256：{citation.ref.digest}", "",
                      f"获取时间：{literal(snapshot.fetched_at)}；字符区间："
                      f"[{citation.start}, {citation.end})；行：{start_line}—{end_line}", "",
                      f"> {literal(citation.quote)}", ""])
    content = "\n".join(lines)
    if len(content) > 8000 or len(content.encode("utf-8")) > 24_000:
        raise ValueError("产物超过第七篇保存上限；应拆分报告，不能截断引用。")
    return Artifact(content, digest_text(content), assessment["report_digest"], assessment["status"])

6.4 demo.py：证据不足、冲突与版本失效

"""人工资料与人工审阅替身；不调用模型，不产生研究结论。"""

from contracts import Claim, Conflict, Report, Scope, SourceSpec
from evidence_store import EvidenceStore
from reporting import assess, record_review, render

SPECS = (
    SourceSpec(source_id="plan_a", title="方案 A 教学说明", locator="fixture:plan-a",
               filename="plan_a.md", kind="fixture"),
    SourceSpec(source_id="plan_b", title="方案 B 教学说明", locator="fixture:plan-b",
               filename="plan_b.md", kind="fixture"),
)
TEXTS = {
    "plan_a": "# 人工测试资料\n方案 A 支持离线整理。\n",
    "plan_b": "# 人工测试资料\n方案 B 需要联网检索。\n",
}


def build_example(store: EvidenceStore, scope: Scope, references: dict | None = None) -> Report:
    if references is None:
        references = {}
        for spec in SPECS:
            # 这是读取业务结果替身；真实 MCP 接入另见 mcp_demo.py。
            references[spec.source_id] = store.ingest_read_result(scope, spec.source_id, {
                "status": "ok", "data": {"filename": spec.filename, "content": TEXTS[spec.source_id]},
            })
    cite_a = store.quote(scope, references["plan_a"], "方案 A 支持离线整理。", "citation_a")
    cite_b = store.quote(scope, references["plan_b"], "方案 B 需要联网检索。", "citation_b")
    return Report(title="方案资料的教学核对报告", claims=(
        Claim(claim_id="claim_a", text="教学说明称方案 A 支持离线整理。",
              kind="fact", citations=(cite_a,)),
        Claim(claim_id="claim_b", text="教学说明称方案 B 需要联网检索。",
              kind="fact", citations=(cite_b,)),
        Claim(claim_id="choice", text="若当前任务不能联网，可先核查方案 A。",
              kind="inference", rationale="离线整理说明支持这一候选方向，但没有验证其他前提。",
              citations=(cite_a,)),
    ), gaps=("缺少功能验证和实际使用条件，不能据此宣布某方案整体更优。",))


def teaching_reviews(scope: Scope, report: Report) -> tuple:
    # 固定审阅结论只验证状态迁移，不是自动语义判定或真实用户审核。
    return tuple(record_review(scope, report, claim.claim_id, "supported",
                               "教学审阅替身：仅核对本例说明，未验证现实能力。")
                 for claim in report.claims)


def main() -> None:
    scope = Scope(tenant_id="teaching", user_id="reader")
    store = EvidenceStore(SPECS)
    try:
        report = build_example(store, scope)
        before = assess(scope, report, store)
        print("未审阅状态：", before["status"], before["claims"])
        artifact = render(scope, report, store, teaching_reviews(scope, report))
        print("产物字节摘要：", artifact.content_digest)
        print(artifact.content)
        # 同一原文可以被用于相反论断；存在性检查不能识别这类语义错误。
        opposite = Claim(claim_id="opposite", text="方案 A 不支持离线整理。",
                         kind="fact", citations=report.claims[0].citations)
        conflict_report = Report(title="显式冲突教学例", claims=(report.claims[0], opposite),
                                 conflicts=(Conflict(claim_ids=("claim_a", "opposite"),
                                                     reason="两个论断方向相反，需要重新审阅。"),))
        print("冲突状态：", assess(scope, conflict_report, store)["status"])
        store.ingest(scope, "plan_a", "# 更新后的人工资料\n原有说明已修订。\n")
        print("来源更新后的旧报告：", assess(scope, report, store)["status"])
    finally:
        store.close()


if __name__ == "__main__":
    main()

6.5 mcp_demo.py：与第七篇真实读取服务衔接

"""接入第七篇真实 stdio 读取服务；需要保留第七篇源码目录。"""

import asyncio
import json
import os
import secrets
import sys
from pathlib import Path
from tempfile import TemporaryDirectory

from mcp import Client, StdioServerParameters

from contracts import Scope
from demo import SPECS, TEXTS, build_example, teaching_reviews
from evidence_store import EvidenceStore
from reporting import render


async def run(root: Path, server_path: Path) -> None:
    scope = Scope(tenant_id="teaching", user_id="reader")
    store = EvidenceStore(SPECS)
    parameters = StdioServerParameters(
        command=sys.executable, args=[str(server_path)],
        env={**os.environ, "DOCS_ROOT": str(root), "DOCS_SECRET": secrets.token_hex(32),
             "DOCS_SCOPE": "teaching/reader"},
    )
    try:
        async with Client(parameters, read_timeout_seconds=5) as client:
            references = {}
            for spec in SPECS:
                result = await client.call_tool("read_document", {"filename": spec.filename})
                if result.result_type != "complete" or result.is_error:
                    raise ValueError("读取工具没有正常完成。")
                if len(json.dumps(result.model_dump(by_alias=True), ensure_ascii=False).encode()) > 32_000:
                    raise ValueError("读取响应超过本篇 Host 上限。")
                references[spec.source_id] = store.ingest_read_result(
                    scope, spec.source_id, result.structured_content,
                )
            report = build_example(store, scope, references)
            artifact = render(scope, report, store, teaching_reviews(scope, report))
            print("真实 MCP 读取已完成；以下审阅仍为教学替身。")
            print(artifact.content)
            print("候选产物 SHA-256：", artifact.content_digest)
            print("本篇不发起保存；保存仍需第七篇的具体内容确认。")
    finally:
        store.close()


def main() -> None:
    default = Path(__file__).resolve().parent.parent / "07_mcp_document_tools" / "server.py"
    server_path = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else default
    if not server_path.is_file():
        raise SystemExit("请提供第七篇 server.py 的可信路径。")
    with TemporaryDirectory(prefix="evidence-mcp-") as directory:
        root = Path(directory)
        (root / "sources").mkdir()
        (root / "reports").mkdir()
        for spec in SPECS:
            (root / "sources" / spec.filename).write_text(TEXTS[spec.source_id], encoding="utf-8")
        asyncio.run(run(root, server_path))


if __name__ == "__main__":
    main()

6.6 requirements.txt：本篇依赖

pydantic==2.14.3
mcp==2.3.0

基础演示只使用 Pydantic；MCP 依赖供集成入口使用。第七篇 Server 通过单独的 Python 子进程启动，两篇的同名 contracts.py 不会在一个解释器里相互覆盖。

七、运行、观察与接回状态图

本地验证环境为 Python 3.13.13、Pydantic 2.14.3、MCP 2.3.0。基础源码使用 Python 3.11 及以上语法；其他 Python 版本尚未验证，直接依赖固定版本，传递依赖没有完整锁定。

在本篇源码目录运行：

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python demo.py

输出中，初始论断均为 pending_review。报告已经登记缺口，因此总体状态优先为 partial。加入教学审阅记录后，仍然是部分结果；显式冲突也返回 partial。来源更新后，旧报告变为 invalid。

保留相邻的第七篇完整源码后，再运行：

python mcp_demo.py

如果第七篇位于其他目录，传入其可信服务端路径：

python mcp_demo.py "../07_mcp_document_tools/server.py"

集成入口在临时目录准备教学资料，使用真实 MCP Client 读取两份文档，再校验业务结果、来源映射与大小。候选结构和审阅仍是固定教学替身。它只打印产物，不直接保存；复制正文运行基础演示不需要前篇代码，运行集成则需要第七篇服务端及其同目录依赖文件。

接回第四篇时，执行节点获取资料，观察节点把有效正文写入证据存储，规划节点按缺口决定是否继续检索。第五篇上下文选择器从授权版本中提供必要片段，报告节点生成 Report，校验节点调用 assess()。

invalid 返回修订流程，needs_review 等待独立审阅，partial 按任务约定继续补证或交付明确的部分结果。只有形成可展示的候选正文后，才进入第七篇保存确认。这里的报告状态不等于业务任务终态，不能直接覆盖第四篇预算和停止规则。

真实模型接入还需要限制候选 JSON 大小，再用 Report.model_validate_json() 解析并校验。结构正确之后仍要走本文的证据与审阅流程，不能让模型返回的自评替代它们。

八、验证范围与下一步

本篇通过 42 项规则检查，覆盖来源登记与映射、工具失败、严格类型、租户用户隔离、换行与指纹、引文篡改、字符越界、歧义定位、来源更新撤销、引用关联、推断依据、冲突、审阅绑定、产物摘要、锚点、字面展示、输出预算与关闭。

基础演示与第七篇真实 stdio 读取集成均已运行。42 项检查主要使用人工资料验证确定性规则；教学审阅只是固定记录，没有测量语义判定准确率、检索召回率或真实模型报告质量。

文内六个源码与依赖文件已和运行副本逐字核对；从 Markdown 独立提取的版本再次运行基础演示、真实 MCP 集成和全部 42 项检查，均通过。

内存存储不支持并发共享、持久审阅或历史快照。获取时间与 SHA-256 也不证明来源真实可靠，来源配置、采集路径和审阅者身份仍需可信系统管理。转义只处理本例字面展示路径，没有进行浏览器渲染或模型提示注入评测。

读者应该能从本篇得到一份可回溯的候选产物，并知道哪些检查是程序完成的、哪些判断仍需证据和审阅。下一篇进入 FastAPI 服务化，把任务、会话、产物归属与应用生命周期放到明确的 API 契约中。

参考资料

1. Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks：RAG 原始论文，说明参数化模型与检索记忆结合的生成方式。本篇不复刻其训练与实验。

2. Enabling Large Language Models to Generate Text with Citations：ALCE 研究，区分生成正确性与引用质量评价。论文结果不作为本项目的性能证据。

3. OWASP LLM Prompt Injection Prevention Cheat Sheet：外部资料中的间接注入、RAG 污染，以及工具边界权限与具体操作确认。