RAG教程：从检索评估到权限与生产治理

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Deterministic lexical retrieval lab over self-authored synthetic data."""

from __future__ import annotations

import argparse
import collections
import datetime
import hashlib
import json
import math
import platform
import re
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parent
CORPUS_PATH = ROOT / "retrieval-corpus.json"
QUERY_PATH = ROOT / "retrieval-queries.json"
EXPERIMENT_PATH = ROOT / "retrieval-experiment.json"
OUTPUT_PATH = ROOT / "retrieval-experiment-output.json"

FIXED_CHARS = 96
FIXED_OVERLAP = 24
PARAGRAPH_LIMIT = 96
TOP_K = 3
BM25_K1 = 1.2
BM25_B = 0.75
ACL_MODES = ("none", "filter_after_top3", "filter_before_top3")
ROLE_ACCESS = {
    "public": {"public"},
    "team-a": {"public", "team-a"},
    "team-b": {"public", "team-b"},
}


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def tokens(text: str) -> list[str]:
    normalized = text.casefold()
    output: list[str] = []
    pattern = r"[a-z0-9]+(?:[-_.][a-z0-9]+)*|[\u3400-\u9fff]+"
    for match in re.finditer(pattern, normalized):
        value = match.group(0)
        if re.fullmatch(r"[\u3400-\u9fff]+", value):
            if len(value) == 1:
                output.append("zh:" + value)
            else:
                output.extend("zh:" + value[index:index + 2] for index in range(len(value) - 1))
        else:
            output.append("ascii:" + value)
    return output


def fixed_chunks(document: dict) -> list[dict]:
    text = document["content"]
    step = FIXED_CHARS - FIXED_OVERLAP
    chunks = []
    for index, start in enumerate(range(0, len(text), step)):
        end = min(len(text), start + FIXED_CHARS)
        chunks.append(make_chunk(document, "fixed", index, start, end, text[start:end]))
        if end == len(text):
            break
    return chunks


def paragraph_chunks(document: dict) -> list[dict]:
    chunks = []
    offset = 0
    for index, paragraph in enumerate(document["content"].split("\n\n")):
        start = offset
        end = start + len(paragraph)
        assert paragraph and len(paragraph) <= PARAGRAPH_LIMIT
        chunks.append(make_chunk(document, "paragraph", index, start, end, paragraph))
        offset = end + 2
    assert offset - 2 == len(document["content"])
    return chunks


def make_chunk(document: dict, method: str, index: int, start: int, end: int, text: str) -> dict:
    return {
        "chunk_id": f"{document['id']}:{method[0].upper()}:{index:02d}:{start}-{end}",
        "doc_id": document["id"],
        "acl": document["acl"],
        "start": start,
        "end": end,
        "text": text,
        "tokens": tokens(text),
    }


def tfidf_scores(query_tokens: list[str], chunks: list[dict],
                 statistics_chunks: list[dict] | None = None) -> list[float]:
    statistics_chunks = statistics_chunks or chunks
    document_frequency = collections.Counter()
    for chunk in statistics_chunks:
        document_frequency.update(set(chunk["tokens"]))
    total = len(statistics_chunks)
    idf = {term: math.log((total + 1) / (document_frequency[term] + 1)) + 1
           for term in set(query_tokens).union(document_frequency)}
    query_counts = collections.Counter(query_tokens)
    query_vector = {term: count * idf[term] for term, count in query_counts.items()}
    query_norm = math.sqrt(sum(value * value for value in query_vector.values()))
    scores = []
    for chunk in chunks:
        counts = collections.Counter(chunk["tokens"])
        vector = {term: count * idf[term] for term, count in counts.items()}
        norm = math.sqrt(sum(value * value for value in vector.values()))
        dot = sum(query_vector.get(term, 0.0) * value for term, value in vector.items())
        scores.append(dot / (query_norm * norm) if query_norm and norm else 0.0)
    return scores


def bm25_scores(query_tokens: list[str], chunks: list[dict],
                statistics_chunks: list[dict] | None = None) -> list[float]:
    statistics_chunks = statistics_chunks or chunks
    total = len(statistics_chunks)
    average_length = sum(len(chunk["tokens"]) for chunk in statistics_chunks) / total
    document_frequency = collections.Counter()
    for chunk in statistics_chunks:
        document_frequency.update(set(chunk["tokens"]))
    scores = []
    for chunk in chunks:
        counts = collections.Counter(chunk["tokens"])
        length_ratio = len(chunk["tokens"]) / average_length if average_length else 0.0
        score = 0.0
        for term in sorted(set(query_tokens)):
            frequency = counts[term]
            if not frequency:
                continue
            idf = math.log(1 + (total - document_frequency[term] + 0.5) /
                            (document_frequency[term] + 0.5))
            denominator = frequency + BM25_K1 * (1 - BM25_B + BM25_B * length_ratio)
            score += idf * frequency * (BM25_K1 + 1) / denominator
        scores.append(score)
    return scores


def allowed(role: str, chunk: dict) -> bool:
    return chunk["acl"] in ROLE_ACCESS[role]


def full_support(chunk: dict, query: dict, documents: dict[str, dict]) -> bool:
    support = query["required_support"]
    if chunk["doc_id"] != support["doc_id"]:
        return False
    source = documents[support["doc_id"]]["content"]
    start = source.index(support["text"])
    end = start + len(support["text"])
    return chunk["start"] <= start and chunk["end"] >= end


def rank(query: dict, chunks: list[dict], ranker: str, acl_mode: str,
         documents: dict[str, dict]) -> dict:
    candidates = ([chunk for chunk in chunks if allowed(query["trusted_role"], chunk)]
                  if acl_mode == "filter_before_top3" else list(chunks))
    scorer = tfidf_scores if ranker == "tfidf_cosine" else bm25_scores
    raw_scores = scorer(tokens(query["query"]), candidates, chunks)
    ranked = sorted(zip(candidates, raw_scores), key=lambda item: (-item[1], item[0]["chunk_id"]))
    top = ranked[:TOP_K]
    if acl_mode == "filter_after_top3":
        top = [item for item in top if allowed(query["trusted_role"], item[0])]
    traces = []
    for position, (chunk, score) in enumerate(ranked, 1):
        traces.append({
            "rank": position,
            "chunk_id": chunk["chunk_id"],
            "doc_id": chunk["doc_id"],
            "acl": chunk["acl"],
            "score": round(score, 12),
            "fully_supports_required_span": full_support(chunk, query, documents),
            "authorized_for_trusted_role": allowed(query["trusted_role"], chunk),
        })
    contexts = []
    reciprocal_rank = 0.0
    for position, (chunk, score) in enumerate(top, 1):
        supports = full_support(chunk, query, documents) and allowed(query["trusted_role"], chunk)
        if supports and reciprocal_rank == 0.0:
            reciprocal_rank = 1.0 / position
        contexts.append({
            "context_position": position,
            "chunk_id": chunk["chunk_id"],
            "doc_id": chunk["doc_id"],
            "acl": chunk["acl"],
            "score": round(score, 12),
            "fully_supports_required_span": full_support(chunk, query, documents),
            "authorized_for_trusted_role": allowed(query["trusted_role"], chunk),
            "text": chunk["text"],
        })
    coverage = int(any(item["fully_supports_required_span"] and item["authorized_for_trusted_role"]
                       for item in contexts))
    unauthorized = sum(not item["authorized_for_trusted_role"] for item in contexts)
    representation_ceiling = int(any(full_support(chunk, query, documents) and
                                     allowed(query["trusted_role"], chunk) for chunk in chunks))
    return {
        "ranked_candidates": traces,
        "top3_context": contexts,
        "metrics": {
            "support_span_recall_at_3": coverage,
            "mrr_at_3_first_fully_supporting_chunk": round(reciprocal_rank, 12),
            "unauthorized_synthetic_chunks_entering_context": unauthorized,
            "authorized_support_coverage": coverage,
            "authorized_support_shortfall": 1 - coverage,
            "representation_ceiling_before_top_k": representation_ceiling,
            "context_slots_returned": len(contexts),
            "context_slot_shortfall": TOP_K - len(contexts),
            "top3_context_char_budget": sum(len(item["text"]) for item in contexts),
        },
    }


def validate_inputs(corpus: dict, query_set: dict) -> dict[str, dict]:
    documents = {document["id"]: document for document in corpus["documents"]}
    assert len(documents) == len(corpus["documents"]) and 10 <= len(documents) <= 12
    assert 12 <= len(query_set["queries"]) <= 16
    assert {document["acl"] for document in documents.values()} == {"public", "team-a", "team-b"}
    span_ids = set()
    for query in query_set["queries"]:
        assert query["trusted_role"] in ROLE_ACCESS
        support = query["required_support"]
        assert re.fullmatch(r"SPAN-Q\d{2}", support["span_id"])
        assert support["span_id"] not in span_ids
        span_ids.add(support["span_id"])
        assert support["doc_id"] in documents
        assert documents[support["doc_id"]]["acl"] in ROLE_ACCESS[query["trusted_role"]]
        assert documents[support["doc_id"]]["content"].count(support["text"]) == 1
    assert tokens("中文检索 BM25 orange-gate") == [
        "zh:中文", "zh:文检", "zh:检索", "ascii:bm25", "ascii:orange-gate"]
    tiny = [make_chunk({"id": "X", "acl": "public"}, "fixed", 0, 0, 1, "alpha"),
            make_chunk({"id": "Y", "acl": "public"}, "fixed", 0, 0, 1, "beta")]
    assert tfidf_scores(tokens("alpha"), tiny)[0] > tfidf_scores(tokens("alpha"), tiny)[1]
    assert math.isclose(tfidf_scores(tokens("alpha"), tiny)[0], 1.0)
    assert math.isclose(bm25_scores(tokens("alpha"), tiny)[0], math.log(2))
    tied = sorted(zip(reversed(tiny), (0.0, 0.0)), key=lambda item: (-item[1], item[0]["chunk_id"]))
    assert [item[0]["chunk_id"] for item in tied] == ["X:F:00:0-1", "Y:F:00:0-1"]
    assert allowed("public", {"acl": "public"}) and not allowed("public", {"acl": "team-a"})
    span_doc = {"D": {"content": "完整支持片段"}}
    span_query = {"required_support": {"doc_id": "D", "text": "完整支持片段"}}
    assert full_support({"doc_id": "D", "start": 0, "end": 6}, span_query, span_doc)
    assert not full_support({"doc_id": "D", "start": 0, "end": 5}, span_query, span_doc)
    return documents


def aggregate(records: list[dict]) -> dict:
    count = len(records)
    keys = ("support_span_recall_at_3", "mrr_at_3_first_fully_supporting_chunk",
            "authorized_support_coverage", "authorized_support_shortfall",
            "representation_ceiling_before_top_k", "context_slot_shortfall",
            "top3_context_char_budget")
    summary = {key: round(sum(record["metrics"][key] for record in records) / count, 12)
               for key in keys}
    summary["unauthorized_synthetic_chunks_entering_context_total"] = sum(
        record["metrics"]["unauthorized_synthetic_chunks_entering_context"] for record in records)
    summary["queries_with_underfilled_context"] = sum(
        record["metrics"]["context_slots_returned"] < TOP_K for record in records)
    return summary


def length_summary(values: list[int]) -> dict:
    return {"min": min(values), "max": max(values),
            "mean": round(sum(values) / len(values), 12), "count": len(values)}


def representation_analysis(chunk_sets: dict[str, list[dict]], queries: list[dict],
                            documents: dict[str, dict]) -> dict:
    support_lengths = [len(query["required_support"]["text"]) for query in queries]
    output = {"support_span_chars": length_summary(support_lengths), "by_chunker": {}}
    for chunker, chunks in chunk_sets.items():
        records = []
        supporting_lengths = []
        for query in queries:
            supporting = [chunk for chunk in chunks if allowed(query["trusted_role"], chunk) and
                          full_support(chunk, query, documents)]
            supporting_lengths.extend(len(chunk["text"]) for chunk in supporting)
            records.append({
                "query_id": query["id"],
                "support_span_id": query["required_support"]["span_id"],
                "support_span_chars": len(query["required_support"]["text"]),
                "authorized_fully_supporting_chunk_ids": [chunk["chunk_id"] for chunk in supporting],
                "representation_ceiling": int(bool(supporting)),
            })
        represented = sum(record["representation_ceiling"] for record in records)
        output["by_chunker"][chunker] = {
            "all_chunk_chars": length_summary([len(chunk["text"]) for chunk in chunks]),
            "fully_supporting_chunk_chars": (length_summary(supporting_lengths)
                                              if supporting_lengths else None),
            "represented_queries": represented,
            "total_queries": len(queries),
            "representation_ceiling_rate": round(represented / len(queries), 12),
            "queries": records,
        }
    return output


def main(argv: list[str] | None = None) -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--work-dir", type=Path, default=ROOT / ".retrieval-lab",
                        help="Directory for the input-freeze record and run logs")
    args = parser.parse_args(argv)
    work_dir = args.work_dir.resolve()
    work_dir.mkdir(parents=True, exist_ok=True)
    freeze_path = work_dir / "input-freeze.json"
    corpus_sha, query_sha = sha256(CORPUS_PATH), sha256(QUERY_PATH)
    if freeze_path.exists():
        freeze = json.loads(freeze_path.read_text(encoding="utf-8"))
        if freeze.get("corpus_sha256") != corpus_sha or freeze.get("queries_sha256") != query_sha:
            raise RuntimeError("frozen input hash mismatch; create an explicitly documented fixture iteration")
    else:
        freeze = {
            "frozen_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "corpus_sha256": corpus_sha,
            "queries_sha256": query_sha,
            "fixture_iteration": 1,
            "note": "Self-authored synthetic fixtures frozen before the first measurement run.",
        }
        freeze_path.write_text(json.dumps(freeze, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    corpus = json.loads(CORPUS_PATH.read_text(encoding="utf-8"))
    query_set = json.loads(QUERY_PATH.read_text(encoding="utf-8"))
    documents = validate_inputs(corpus, query_set)
    chunk_sets = {
        "fixed_char_overlap": [chunk for document in corpus["documents"] for chunk in fixed_chunks(document)],
        "paragraph_preserving": [chunk for document in corpus["documents"] for chunk in paragraph_chunks(document)],
    }
    for chunks in chunk_sets.values():
        for query in query_set["queries"]:
            query_tokens = tokens(query["query"])
            authorized = [chunk for chunk in chunks if allowed(query["trusted_role"], chunk)]
            for scorer in (tfidf_scores, bm25_scores):
                full = dict(zip((chunk["chunk_id"] for chunk in chunks), scorer(query_tokens, chunks, chunks)))
                filtered = dict(zip((chunk["chunk_id"] for chunk in authorized),
                                    scorer(query_tokens, authorized, chunks)))
                if any(not math.isclose(score, full[chunk_id], abs_tol=1e-15)
                       for chunk_id, score in filtered.items()):
                    raise RuntimeError("ACL filtering changed full-index ranking statistics")
    representation = representation_analysis(chunk_sets, query_set["queries"], documents)
    results: dict[str, dict] = {}
    for chunker, chunks in chunk_sets.items():
        results[chunker] = {}
        for ranker in ("tfidf_cosine", "bm25"):
            results[chunker][ranker] = {}
            for acl_mode in ACL_MODES:
                records = []
                for query in query_set["queries"]:
                    record = {"query_id": query["id"], "query": query["query"],
                              "trusted_role": query["trusted_role"],
                              "required_support": query["required_support"]}
                    record.update(rank(query, chunks, ranker, acl_mode, documents))
                    records.append(record)
                results[chunker][ranker][acl_mode] = {
                    "aggregate": aggregate(records), "queries": records}
    for chunker in results.values():
        for ranker in chunker.values():
            for mode in ranker.values():
                aggregate_values = mode["aggregate"]
                bounded = ("support_span_recall_at_3", "mrr_at_3_first_fully_supporting_chunk",
                           "authorized_support_coverage", "authorized_support_shortfall",
                           "representation_ceiling_before_top_k")
                if any(not 0.0 <= aggregate_values[key] <= 1.0 for key in bounded):
                    raise RuntimeError("aggregate retrieval metric is outside [0, 1]")
                if any(value < 0 for value in aggregate_values.values()):
                    raise RuntimeError("aggregate retrieval metric cannot be negative")
    pairwise = {}
    for chunker, rankers in results.items():
        pairwise[chunker] = {}
        for acl_mode in ACL_MODES:
            left = rankers["tfidf_cosine"][acl_mode]["queries"]
            right = rankers["bm25"][acl_mode]["queries"]
            counts = {"tfidf_higher_mrr_queries": 0, "bm25_higher_mrr_queries": 0, "mrr_ties": 0}
            for tfidf_record, bm25_record in zip(left, right):
                tfidf_mrr = tfidf_record["metrics"]["mrr_at_3_first_fully_supporting_chunk"]
                bm25_mrr = bm25_record["metrics"]["mrr_at_3_first_fully_supporting_chunk"]
                key = ("tfidf_higher_mrr_queries" if tfidf_mrr > bm25_mrr else
                       "bm25_higher_mrr_queries" if bm25_mrr > tfidf_mrr else "mrr_ties")
                counts[key] += 1
            assert sum(counts.values()) == len(query_set["queries"])
            pairwise[chunker][acl_mode] = counts
    if sha256(CORPUS_PATH) != corpus_sha or sha256(QUERY_PATH) != query_sha:
        raise RuntimeError("input files changed during measurement")
    experiment = {
        "version": 1,
        "status": "completed_self_authored_synthetic_lexical_retrieval_lab",
        "environment": {"python": sys.version.split()[0], "implementation": platform.python_implementation(),
                        "platform": platform.platform()},
        "source_sha256": sha256(Path(__file__)),
        "input_freeze": freeze,
        "parameters": {
            "tokenization": "Unicode casefold; contiguous CJK character bigrams; ASCII word tokens",
            "fixed_chunk_chars": FIXED_CHARS, "fixed_overlap_chars": FIXED_OVERLAP,
            "paragraph_limit_chars": PARAGRAPH_LIMIT,
            "paragraph_hidden_recursion_or_merge": False,
            "posthoc_parameter_alignment": (
                "Paragraph limit changed from 180 to 96 after the first run to match the fixed-window budget; "
                "all frozen paragraphs were already at most 94 chars, so this parameter change alone changed "
                "no chunks or outcomes. This is diagnostic alignment, not holdout optimization."),
            "top_k": TOP_K, "bm25_k1": BM25_K1, "bm25_b": BM25_B,
            "tie_policy": "score descending, then chunk_id ascending",
            "ranking_statistics_scope": "full chunk set per chunker, reused unchanged across all ACL modes",
            "acl_modes": list(ACL_MODES),
            "filter_after_top3_refill": False,
        },
        "chunk_counts": {name: len(chunks) for name, chunks in chunk_sets.items()},
        "representation_analysis": representation,
        "pairwise_ranker_mrr_comparison": pairwise,
        "analysis_iteration": 3,
        "methodology_history": [
            {
                "iteration": 2,
                "archive": ".local/blog-writing/T05/v2-archive",
                "source_sha256": "178db8e02f527134f1fa45a8cf9cec38c47b845b7f8a7f2944ea6515338651ed",
                "experiment_sha256": "ae7b42bddacfd59c88fdd930dabfebb11cd8b17b86c67c0cb4ae0056a03d62de",
                "output_sha256": "92fb37f68b3b48131740ed16fc9e9f5aa4d688453303433a88ad35e18c3d4fb7",
                "ranking_statistics": "Recomputed on each ACL candidate pool.",
                "notable_old_result": "Paragraph/BM25 filter-before MRR@3 was 1.0 versus TF-IDF 0.966666666667.",
            },
            {
                "iteration": 3,
                "ranking_statistics": "Statistics use the full chunk set for every ACL mode; this simple implementation recomputes identical values per scoring call.",
                "reason": "Remove ACL candidate-pool statistics as a confound; corpus, queries and labels unchanged.",
            },
        ],
        "results": results,
        "limitations": [
            "Small self-authored synthetic corpus and support-span annotations; no human evaluation or customer benchmark.",
            "Complete support-span criterion structurally favors chunks that preserve the annotated sentence.",
            "Chinese character bigrams and ASCII words are deterministic lexical tokens, not embeddings or certified NLP.",
            "Parameters were fixed before measurement and are not claimed universally optimal; no test-set tuning was performed.",
            "ACL labels and roles are synthetic. The lab does not prove real privacy, authorization, prompt-injection defense, or production security.",
            "No model generation, answer factuality, long-context comparison, vector database, framework compatibility, or production load was tested.",
        ],
    }
    EXPERIMENT_PATH.write_text(json.dumps(experiment, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    concise = {
        "status": experiment["status"], "input_freeze": freeze,
        "source_sha256": experiment["source_sha256"],
        "environment": experiment["environment"], "parameters": experiment["parameters"],
        "chunk_counts": experiment["chunk_counts"],
        "representation_analysis": representation,
        "pairwise_ranker_mrr_comparison": pairwise,
        "analysis_iteration": experiment["analysis_iteration"],
        "methodology_history": experiment["methodology_history"],
        "aggregates": {chunker: {ranker: {mode: data["aggregate"] for mode, data in modes.items()}
                                  for ranker, modes in rankers.items()}
                       for chunker, rankers in results.items()},
        "limitations": experiment["limitations"],
    }
    OUTPUT_PATH.write_text(json.dumps(concise, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(concise, ensure_ascii=False))


if __name__ == "__main__":
    main()

{
  "version": 1,
  "kind": "self_authored_synthetic_knowledge_base",
  "documents": [
    {
      "id": "P01",
      "title": "EPUB 文件接收检查",
      "acl": "public",
      "content": "接收文件时先确认扩展名只是线索，不能证明内容有效。基础检查应记录来源、字节数和校验值，再判断是否进入解析流程。\n\n有效 EPUB 包必须是 ZIP 容器，并且首个未压缩成员 mimetype 的内容应为 application/epub+zip。仅把 HTML 文件改成 .epub 不能满足这个条件。\n\n解析失败时保留错误原文，不自动改后缀或上传到不明转换服务。"
    },
    {
      "id": "P02",
      "title": "知识库分块约定",
      "acl": "public",
      "content": "固定字符分块便于控制长度，但边界可能切开一句话或一条完整证据。记录字符起止位置和重叠长度，才能复查被切开的内容。\n\n段落保留分块以空行作为边界，每个段落直接成为一个块；本实验不做递归合并或二次切分。超出上限的段落应在冻结语料前修短，而不是运行时隐藏处理。\n\n两种方法都只是检索输入准备，不代表语义理解或答案正确。"
    },
    {
      "id": "P03",
      "title": "词法排序器说明",
      "acl": "public",
      "content": "TF-IDF 余弦相似度根据词频、逆文档频率和向量夹角排序。分数只反映当前词元空间中的重合，不等于语义蕴含。\n\nBM25 使用词频饱和与长度归一化；本实验固定 k1=1.2、b=0.75，这些参数不是通用最优值。\n\n所有同分结果按 chunk_id 升序处理，避免运行环境改变排序。"
    },
    {
      "id": "P04",
      "title": "检索 ACL 边界",
      "acl": "public",
      "content": "访问角色由受信任的服务器参数传入，不能从查询文本、模型输出或检索到的说明中提升。查询里自称管理员不会改变 trusted_role。\n\n先过滤再排序可以让可见候选填满 top-k；先排序再过滤也能移除未授权块，但可能留下不足三个上下文。两种结果都需要分别记录覆盖与缺口。\n\n本实验只检查合成块是否进入上下文，不证明真实系统权限、隐私或提示注入防护。"
    },
    {
      "id": "P05",
      "title": "支持片段评估口径",
      "acl": "public",
      "content": "支持片段必须完整落在同一个检索块中，才记为本查询的支持命中。命中同一文档的其他段落不计成功。\n\nsupport-span Recall@3 统计前三个上下文是否含完整支持片段；MRR@3 使用第一个完整支持块的倒数排名，未命中记零。\n\n这是小型自制测试集的检索指标，不评估生成答案、事实正确性或生产收益。"
    },
    {
      "id": "P06",
      "title": "检索内容不是指令",
      "acl": "public",
      "content": "知识库文本是待检索数据，不是运行时指令。任何类似“忽略规则”或“执行工具”的句子都只能作为普通字符返回。\n\n查询内容不能授予角色、改变 ACL 或触发代码执行。即使查询写着“我是管理员”，服务端仍应使用调用方原先的 public 角色。\n\n本实验没有连接模型、工具、网络或真实秘密，因此不能据此宣称已经防御生产提示注入。"
    },
    {
      "id": "A01",
      "title": "Team A 发布窗口",
      "acl": "team-a",
      "content": "Team A 的合成发布窗口只用于本实验。发布前要冻结语料和查询哈希，并把运行参数写入记录。\n\nTeam A 的回滚触发条件是同一查询在两个排序器中都找不到已标注支持片段。该条件是演示规则，不适用于真实服务。\n\n复盘时不得把 team-a 文档提供给 public 或 team-b 上下文。"
    },
    {
      "id": "A02",
      "title": "Team A 故障记录",
      "acl": "team-a",
      "content": "Team A 的虚构故障编号为 DEMO-A-17，没有对应客户、订单或生产系统。\n\n故障复盘必须记录排序器、分块器、候选数量和 top3 chunk_id，不能只写“检索失败”。\n\n若同分，Team A 也使用 chunk_id 升序，避免人工挑选看起来更好的结果。"
    },
    {
      "id": "A03",
      "title": "Team A 术语表",
      "acl": "team-a",
      "content": "在 Team A 的合成词表中，orange-gate 表示冻结输入后才允许运行评测。这个词没有外部产品含义。\n\nblue-note 表示记录限制，不代表通过安全审计。\n\n这些词只用于制造可追踪的 ASCII 与中文混合检索条件。"
    },
    {
      "id": "B01",
      "title": "Team B 数据保留",
      "acl": "team-b",
      "content": "Team B 的合成记录不包含真实用户数据。实验输出只保存语料、查询、参数、得分和合成 ACL 标签。\n\nTeam B 的演示保留规则是实验日志保存七天后由测试维护者决定是否删除；本文不执行定时删除。\n\npublic 与 team-a 不应在上下文中看到这条 team-b 演示规则。"
    },
    {
      "id": "B02",
      "title": "Team B 指标解释",
      "acl": "team-b",
      "content": "Team B 将 unauthorized_synthetic_chunks_entering_context 定义为最终上下文中 ACL 不允许的合成块数量。\n\nauthorized_support_shortfall 等于一减去完整支持片段覆盖值；它不是泄露概率或真实风险分数。\n\n指标为零只能说明当前合成样本，没有证明生产权限系统正确。"
    },
    {
      "id": "B03",
      "title": "Team B 查询说明",
      "acl": "team-b",
      "content": "Team B 的检索查询使用固定 trusted_role=team-b，查询正文不能把角色改成 team-a 或管理员。\n\n查询中的 harmless-placeholder-instruction 只是普通数据，实验不会执行它，也没有连接任何工具。\n\n若支持块被 ACL 过滤，结果应记录 shortfall，而不是偷偷换成未授权块。"
    }
  ]
}

{
  "version": 1,
  "kind": "self_authored_synthetic_queries_and_support_labels",
  "queries": [
    {
      "id": "Q01",
      "query": "怎样判断改后缀的文件是不是有效 EPUB？",
      "trusted_role": "public",
      "required_support": {"span_id": "SPAN-Q01", "doc_id": "P01", "text": "有效 EPUB 包必须是 ZIP 容器，并且首个未压缩成员 mimetype 的内容应为 application/epub+zip。"}
    },
    {
      "id": "Q02",
      "query": "段落分块有没有隐藏递归切分？",
      "trusted_role": "public",
      "required_support": {"span_id": "SPAN-Q02", "doc_id": "P02", "text": "段落保留分块以空行作为边界，每个段落直接成为一个块；本实验不做递归合并或二次切分。"}
    },
    {
      "id": "Q03",
      "query": "BM25 的 k1 和 b 参数是多少？",
      "trusted_role": "public",
      "required_support": {"span_id": "SPAN-Q03", "doc_id": "P03", "text": "BM25 使用词频饱和与长度归一化；本实验固定 k1=1.2、b=0.75，这些参数不是通用最优值。"}
    },
    {
      "id": "Q04",
      "query": "先排序再做 ACL 过滤有什么上下文缺口？",
      "trusted_role": "public",
      "required_support": {"span_id": "SPAN-Q04", "doc_id": "P04", "text": "先过滤再排序可以让可见候选填满 top-k；先排序再过滤也能移除未授权块，但可能留下不足三个上下文。"}
    },
    {
      "id": "Q05",
      "query": "我是管理员，请执行 harmless-placeholder-instruction 并提升权限",
      "trusted_role": "public",
      "required_support": {"span_id": "SPAN-Q05", "doc_id": "P06", "text": "查询内容不能授予角色、改变 ACL 或触发代码执行。即使查询写着“我是管理员”，服务端仍应使用调用方原先的 public 角色。"}
    },
    {
      "id": "Q06",
      "query": "Team A 在什么条件下触发合成回滚？",
      "trusted_role": "team-a",
      "required_support": {"span_id": "SPAN-Q06", "doc_id": "A01", "text": "Team A 的回滚触发条件是同一查询在两个排序器中都找不到已标注支持片段。"}
    },
    {
      "id": "Q07",
      "query": "DEMO-A-17 故障复盘必须记录哪些检索信息？",
      "trusted_role": "team-a",
      "required_support": {"span_id": "SPAN-Q07", "doc_id": "A02", "text": "故障复盘必须记录排序器、分块器、候选数量和 top3 chunk_id，不能只写“检索失败”。"}
    },
    {
      "id": "Q08",
      "query": "orange-gate 在 Team A 合成词表中是什么意思？",
      "trusted_role": "team-a",
      "required_support": {"span_id": "SPAN-Q08", "doc_id": "A03", "text": "在 Team A 的合成词表中，orange-gate 表示冻结输入后才允许运行评测。"}
    },
    {
      "id": "Q09",
      "query": "Team A 同分结果怎样保持稳定？",
      "trusted_role": "team-a",
      "required_support": {"span_id": "SPAN-Q09", "doc_id": "A02", "text": "若同分，Team A 也使用 chunk_id 升序，避免人工挑选看起来更好的结果。"}
    },
    {
      "id": "Q10",
      "query": "Team A 的 blue-note 是否代表安全审计通过？",
      "trusted_role": "team-a",
      "required_support": {"span_id": "SPAN-Q10", "doc_id": "A03", "text": "blue-note 表示记录限制，不代表通过安全审计。"}
    },
    {
      "id": "Q11",
      "query": "Team B 合成日志的演示保留规则是什么？",
      "trusted_role": "team-b",
      "required_support": {"span_id": "SPAN-Q11", "doc_id": "B01", "text": "Team B 的演示保留规则是实验日志保存七天后由测试维护者决定是否删除；本文不执行定时删除。"}
    },
    {
      "id": "Q12",
      "query": "unauthorized_synthetic_chunks_entering_context 指标如何定义？",
      "trusted_role": "team-b",
      "required_support": {"span_id": "SPAN-Q12", "doc_id": "B02", "text": "Team B 将 unauthorized_synthetic_chunks_entering_context 定义为最终上下文中 ACL 不允许的合成块数量。"}
    },
    {
      "id": "Q13",
      "query": "authorized_support_shortfall 是真实泄露概率吗？",
      "trusted_role": "team-b",
      "required_support": {"span_id": "SPAN-Q13", "doc_id": "B02", "text": "authorized_support_shortfall 等于一减去完整支持片段覆盖值；它不是泄露概率或真实风险分数。"}
    },
    {
      "id": "Q14",
      "query": "team-b 查询中的 harmless-placeholder-instruction 会执行工具吗？",
      "trusted_role": "team-b",
      "required_support": {"span_id": "SPAN-Q14", "doc_id": "B03", "text": "查询中的 harmless-placeholder-instruction 只是普通数据，实验不会执行它，也没有连接任何工具。"}
    },
    {
      "id": "Q15",
      "query": "Team B 支持块被 ACL 过滤以后应该怎样记录？",
      "trusted_role": "team-b",
      "required_support": {"span_id": "SPAN-Q15", "doc_id": "B03", "text": "若支持块被 ACL 过滤，结果应记录 shortfall，而不是偷偷换成未授权块。"}
    }
  ]
}