检索排序正确，不代表可以把结果原样交给模型。一段高分材料可能属于其他租户；同一段落可能重复出现；前几个长段落可能占满预算；回答还可能引用根本没有进入本次上下文的片段。

这篇接在混合检索与重排之后，研究“候选已经找到了，怎样变成生成输入”。用纯 Python 标准库把权限过滤、精确去重、预算装箱和引用回查拆开。实验不调用模型，不能据此宣称回答质量提升，也不把示范过滤器当成生产身份认证系统。

前置知识

候选与重排

知道排名是相关性的线索，不是权限、内容真实性或完整性的保证。

本次实现

冻结证据集合

保存片段身份、文档版本与原文，限定本轮能被引用的材料。

验证边界

有出处不等于有依据

字面引文检查能查出伪引用，却不能证明回答被原文蕴含。

先把四个环节分开

环节 回答的问题 不应替它承担的任务
权限过滤 当前用户可以看到哪些材料？ 用相似度猜测是否有权访问
精确去重 同一身份的片段是否重复返回？ 把不同文档的相同文字误合为同一出处
预算组装 哪些完整片段能进入本轮输入？ 把字节数直接当作模型 token 数
引用回查 引用是否来自本轮证据，原文是否包含引文？ 证明整个回答的语义一定正确

权限应该在检索服务返回候选之前生效，不能先把未授权文档交给外部重排服务，再在生成阶段删掉。本地示例再次过滤是防御检查，不替代检索端的访问控制。

微软的安全过滤文档明确区分字符串过滤模式与真正的身份认证、授权：身份字段只负责匹配，不能证明调用者是谁。生产中的用户、租户和角色应来自已认证服务端会话，不能相信请求体里自填的角色。

片段身份必须连着文档版本

示例片段包含 id、tenant、doc、version、text、score，权限表以 (tenant, doc, version) 为键，值是允许访问的角色集合。表中没有记录、角色不相交、租户不相同，全部拒绝。

id 在同一请求的租户内必须唯一。同一 id 重复返回时，文档、版本或正文有冲突就报错；仅分数不同则视为重复，排序靠前的一份保留。生产可使用全局唯一片段键、版本号与内容摘要，并保留页码或原文偏移，不能依靠一个易重复的自然语言标题。

这里仅做身份去重。不对相似段落做语义合并：相同句子在两份不同文件中仍可能承担不同证据角色。近重复、重叠窗口与同文档配额是后续可比较的策略，不应悄悄混进本次基线。

预算的单位比数字更重要

实际模型输入需要使用目标模型匹配的 tokenizer，并预留系统规则、问题、对话历史、工具定义及输出空间。不同接口的窗口与输出计费规则要按对应官方资料核对，本文不提供通用模型窗口数字。

为了让实验无依赖、可精确复算，这里使用 序列化证据 JSON 的 UTF-8 字节预算，包括字段名、标点、引号、转义与正文；它不是 token 预算，也不覆盖完整模型请求。空列表 [] 本身需要 2 字节。中文字符、表情与引号都会影响真实字节数，不能拿 len(text) 代替。

序列化使用 json.dumps(..., ensure_ascii=False, separators=(",", ":")) 再编码为 UTF-8；相关参数见 Python 官方 JSON 文档。生产替换计量器时，必须重新计量实际发送的完整结构，不能只把字节预算变量改名为 token 预算。

按排名扫描不等于最优装箱

策略是：权限过滤后按分数降序排列；相同分数按文档、版本和片段 id 排列，确保可复现；重复身份只处理一次；每次尝试加入一个完整片段，若超过预算就跳过，继续检查后面的片段。

这比“遇到首个超长片段就停止”更容易保留后面较短的材料，但不是背包最优解，也没有证明它最大化回答相关性。分数不是收益的精确值，多段证据之间还可能存在互补或重复。

原创示例中的候选 处理 理由
其他租户的高分片段 丢弃 分数再高也不能越过租户边界
c1 排障说明 尝试加入 有权访问且身份有效
同一 c1 再次返回 跳过 同一原文身份，不重复占预算
c2 很长的背景材料 可能跳过 超过剩余预算，不能截成不完整句子
c3 较短的重试说明 继续尝试 前面超长不代表后面也超长

没有任何片段入选时，应走“证据不足”分支，而不是强迫生成一个有出处的答案。

可直接运行的 Python 标准库实验

使用 Python 3.9 或更新版本，保存为 context_demo.py 后运行 python context_demo.py。正文与身份均为合成数据，无网络请求、不读凭据、不调用语言模型。Principal 和权限表在实验中直接构造，生产必须改由可信服务端提供。

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
107
108
109
110
111
112
113
114
115
from dataclasses import dataclass
import json
import math


@dataclass(frozen=True)
class Principal:
    tenant: str
    roles: frozenset[str]


@dataclass(frozen=True)
class Chunk:
    id: str
    tenant: str
    doc: str
    version: str
    text: str
    score: float


def allowed(chunk, principal, acl):
    grants = acl.get((chunk.tenant, chunk.doc, chunk.version), frozenset())
    return chunk.tenant == principal.tenant and bool(principal.roles & grants)


def wire(chunks):
    records = [dict(id=c.id, doc=c.doc, version=c.version, text=c.text)
               for c in chunks]
    return json.dumps(records, ensure_ascii=False,
                      separators=(",", ":")).encode("utf-8")


def pack(candidates, principal, acl, budget_bytes):
    if type(budget_bytes) is not int or budget_bytes < 2:
        raise ValueError("budget must be an integer of at least 2 bytes")
    eligible = []
    for c in candidates:
        if not allowed(c, principal, acl):
            continue
        if any(not isinstance(v, str) or not v.strip()
               for v in (c.id, c.tenant, c.doc, c.version, c.text)):
            raise ValueError("invalid authorized chunk metadata")
        if type(c.score) not in (int, float):
            raise ValueError("score must be finite")
        try:
            finite = math.isfinite(c.score)
        except OverflowError:
            finite = False
        if not finite:
            raise ValueError("score must be finite")
        eligible.append(c)
    eligible.sort(key=lambda c: (-c.score, c.doc, c.version, c.id))
    selected, seen = [], {}
    for c in eligible:
        identity = (c.doc, c.version, c.text)
        if c.id in seen:
            if seen[c.id] != identity:
                raise ValueError("conflicting chunk identity")
            continue
        seen[c.id] = identity
        if len(wire(selected + [c])) <= budget_bytes:
            selected.append(c)
    frozen = tuple(selected)
    return frozen, wire(frozen)


def check_citations(citations, selected, principal, current_acl):
    evidence = {c.id: c for c in selected}
    issues = []
    if not isinstance(citations, list) or not citations:
        return dict(ok=False, issues=["missing_citations"])
    for ref in citations:
        if not isinstance(ref, dict) or set(ref) != {"id", "version", "quote"}:
            issues.append("schema")
            continue
        if any(not isinstance(v, str) or not v.strip() for v in ref.values()):
            issues.append("schema")
            continue
        c = evidence.get(ref["id"])
        if c is None:
            issues.append("not_in_context")
        elif not allowed(c, principal, current_acl):
            issues.append("permission_revoked")
        elif ref["version"] != c.version:
            issues.append("version_mismatch")
        elif ref["quote"] not in c.text:
            issues.append("quote_not_found")
    return dict(ok=not issues, issues=issues)


def demo():
    user = Principal("t1", frozenset({"ops"}))
    acl = {("t1", "guide", "v1"): frozenset({"ops"}),
           ("t2", "guide", "v1"): frozenset({"ops"})}
    candidates = [
        Chunk("other", "t2", "guide", "v1", "其他租户的材料", 0.99),
        Chunk("c1", "t1", "guide", "v1", "缓存有效期为 30 分钟。", 0.95),
        Chunk("c1", "t1", "guide", "v1", "缓存有效期为 30 分钟。", 0.94),
        Chunk("c2", "t1", "guide", "v1", "背景说明。" * 100, 0.90),
        Chunk("c3", "t1", "guide", "v1", "超时后最多重试 2 次。", 0.85),
    ]
    selected, payload = pack(candidates, user, acl, 240)
    good = [dict(id="c1", version="v1", quote="缓存有效期为 30 分钟。")]
    outside = [dict(id="c2", version="v1", quote="背景说明。")]
    fake = [dict(id="c1", version="v1", quote="缓存永不失效。")]
    return dict(ids=[c.id for c in selected], evidence_bytes=len(payload),
                good=check_citations(good, selected, user, acl),
                outside=check_citations(outside, selected, user, acl),
                fake=check_citations(fake, selected, user, acl),
                revoked=check_citations(good, selected, user, {}))


if __name__ == "__main__":
    print(json.dumps(demo(), ensure_ascii=False, indent=2))

样例会保留 c1 与 c3；c2 没有进入本轮证据，因此引用 c2 仍不合法，即使它确实来自知识库。代码记录错误类别，不把未授权正文或引文写入日志。

实际运行项 输出 解释
ids ["c1", "c3"] 去重并跳过超长材料后仍继续扫描
evidence_bytes 164 完整证据 JSON 的 UTF-8 字节数，低于 240
good ok=true 原文确实包含本轮引文
outside not_in_context c2 虽在候选里，却不在本轮上下文
fake quote_not_found 伪造句子不是 c1 的子串
revoked permission_revoked 复核时权限表已不再允许访问

引文存在不代表回答被支持

当前校验只检查结构、片段入选、权限仍有效、版本一致，以及引文是否为正文的精确子串。它不处理同义改写、空白规范化或 Unicode 等价形式，也没有把答案正文传入校验器。

例如原文说“缓存有效期为 30 分钟”，回答却说“缓存每 30 分钟必然自动刷新”。即便引用原句，字面检查仍会通过，答案却增加了原文没有说的自动刷新机制。要检查这类错误，需要拆分回答断言、核对适用范围与限定词，并做人工抽检或经过独立评估的蕴含校验。

错误 这个实验能否发现 需要补充的环节
引用未入选的 c2 能 本轮证据集合回查
引用旧版本号 能 版本身份校验
编造一句不存在的原文 能 精确子串检查
正文加上来源没有说的机制 不能 断言与来源的语义核验
来源本身已经过时或错误 不能 数据治理、时效与来源可靠性检查
缺少引用的拒答 不能按引用成功计入 拒答应有独立响应与评测分支

权限撤销与缓存不能只看第一次请求

示例在引用检查时重新读取当前权限表。如果权限已撤销，不继续展示依赖该证据的回答。生产还必须把权限版本或身份范围纳入检索、上下文和答案缓存策略，避免不同租户或角色共享敏感结果；对整个响应作权限复核，而不只是对引用标签作复核。

这个 Python 片段只是静态快照实验，没有解决权限变化与响应发送之间的竞态。应在系统中定义复核边界和失效机制，不能从一次函数调用推导出完整的访问控制保证。

检索材料仍然是不可信输入

片段中可能出现“忽略上面的规则”等指令。结构化 JSON、分隔符与引用检查都不是提示注入的完整防护；字面引用检查甚至会允许引用这段指令，只要它真实存在。OWASP 提示注入资料说明外部内容也可能成为注入入口。

工程上应限制工具权限与副作用，将外部文本作为证据而非更高优先级指令，验证结构化输出，建立注入测试集，并隔离高风险操作。本篇只实现输入组装与引用回查，不宣称已经防住提示注入。

实践自测 引用回查通过后还能发生什么错误

把示例答案改成“缓存每 30 分钟必然自动刷新”，保留真实引文。字面检查依然可能通过，这正好说明它没有读取、理解答案断言。

分别为身份伪造、上下文越权、伪造引文、错误推断与过期来源建立测试，不要用一个“引用准确率”合并所有失败。

回归指标应分别报告

指标 本例或生产检查对象
未授权入选数 必须为 0，覆盖跨租户、缺 ACL、角色不匹配
预算超限数 按实际序列化结果计量，必须为 0
身份冲突拒绝数 同一 id 的不同正文、文档或版本不能静默覆盖
引用结构通过率 不等于语义支持率，也不是模型真实效果
证据覆盖率 需要独立标注相关证据，本例没有估计
空上下文与拒答表现 不能通过随意补一句话假装任务成功

预算组装每次重新序列化已选列表，最坏存在二次累计成本；这是可读性优先的小实验，不是高吞吐实现。优化时可以增量计算字节增量，但必须覆盖 JSON 转义、数组逗号与 UTF-8，保持与完整序列化一致。

与已有 AI 工程文章接起来

资料核对日期为 2026-10-05。本文中的缓存、重试、权限与分数都是原创合成示例，不能当作任何真实系统的配置或效果数据。