对语料库切块。不要过度设计。

在做任何比子串匹配更聪明的事之前，我们需要把文档拆分成可检索的单元。这就是"切块"（chunking）。很容易想得太复杂——语义切块、延迟切块、递归 AST 拆分——但现在，先做最简单的事。只有当评估（eval）结果表明有必要时，我们才去改进。

经验法则

• 目标大小：每块约 500 个令牌（token）（大约 2000 个英文字符）。
• 重叠：相邻块之间约 50 个令牌（token）的重叠。用于捕获答案跨越块边界的情况。
• 按自然边界切分：优先在段落处断开，其次是句子。不要在句子中间截断。
• 保留元数据：每块都知道自己的 doc_id 和文档内的 chunk_idx。我们需要这些来反向引用。

注意：以下代码与提供商无关，切块只是 Python。

# retrieval/chunk.py
import tiktoken  # works for both providers as a counter
from dataclasses import dataclass

enc = tiktoken.get_encoding("cl100k_base")

def n_tokens(s: str) -> int:
    return len(enc.encode(s))

@dataclass
class Chunk:
    chunk_id: str   # f"{doc_id}::{idx}"
    doc_id: str
    idx: int
    text: str
    n_tok: int

def chunk_doc(doc_id: str, text: str,
              target: int = 500,
              overlap: int = 50) -> list[Chunk]:
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, buf, buf_tok = [], [], 0

    def flush():
        nonlocal buf, buf_tok
        if not buf: return
        joined = "\n\n".join(buf)
        chunks.append(Chunk(
            chunk_id=f"{doc_id}::{len(chunks)}",
            doc_id=doc_id,
            idx=len(chunks),
            text=joined,
            n_tok=n_tokens(joined),
        ))
        # Keep last paragraph for overlap
        buf = [buf[-1]] if overlap and n_tokens(buf[-1]) <= overlap else []
        buf_tok = n_tokens(buf[0]) if buf else 0

    for p in paragraphs:
        t = n_tokens(p)
        if buf_tok + t > target and buf:
            flush()
        buf.append(p)
        buf_tok += t
    flush()
    return chunks

运行并检查结果

>>> from retrieval.chunk import chunk_doc
>>> from pathlib import Path
>>> text = Path("corpus/pgbouncer-modes.md").read_text()
>>> chunks = chunk_doc("pgbouncer-modes", text)
>>> for c in chunks[:3]:
...     print(c.chunk_id, c.n_tok, c.text[:50])

pgbouncer-modes::0 487 # PgBouncer Pool Modes\n\nPgBouncer supports
pgbouncer-modes::1 503 ## Session Pooling\n\nIn session pooling mode,
pgbouncer-modes::2 491 ## Transaction Pooling\n\nThis is the most c

三个块，令牌数都接近 500，每块保留了章节标题。形状不错。

上下文切块技巧

在嵌入切块之前，在每块前面添加一句摘要，描述该块在其父文档中的内容。这能显著提升检索效果——Anthropic 的检索增强（retrieval-augmented）上下文检索研究报告称可提升约 35%。

成本很小：在索引阶段每块调用一次廉价模型，只需一次。用 Haiku 或 GPT-4o-mini，并批量处理。

# retrieval/contextualize.py
from anthropic import Anthropic
client = Anthropic()

CTX_PROMPT = """Document title: {doc_id}
Full document:
<document>{full_text}</document>

Here is a chunk from the document:
<chunk>{chunk_text}</chunk>

In one sentence, situate this chunk in the document
(what section, what topic, what role). No preamble.
Output just the sentence."""

def contextualize(chunk_text, doc_id, full_text):
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=100,
        messages=[{"role": "user",
                   "content": CTX_PROMPT.format(
                       doc_id=doc_id,
                       full_text=full_text,
                       chunk_text=chunk_text)}],
    )
    return response.content[0].text.strip()

# retrieval/contextualize.py
from openai import OpenAI
client = OpenAI()

CTX_PROMPT = """Document title: {doc_id}
Full document:
<document>{full_text}</document>

Here is a chunk from the document:
<chunk>{chunk_text}</chunk>

In one sentence, situate this chunk in the document
(what section, what topic, what role). No preamble.
Output just the sentence."""

def contextualize(chunk_text, doc_id, full_text):
    response = client.responses.create(
        model="gpt-5-mini",
        input=CTX_PROMPT.format(
            doc_id=doc_id,
            full_text=full_text,
            chunk_text=chunk_text),
    )
    return response.output_text.strip()

输出长什么样？以 PgBouncer 事务池化章节的某个块为例：

"This chunk explains transaction pooling in PgBouncer,
describing how server connections are released after
each transaction commits — a constraint that breaks
features requiring session state."

这一句话现在会随每个块在检索（retrieval）全程中携带。当用户问"为什么我的预备语句在 PgBouncer 中会失败？"时，嵌入中包含了"约束会破坏需要会话状态的特性"——即使该块里没有"预备"这个确切词，语义上也与问题接近。

另外两种值得了解的方案

上下文切块技巧并不是赋予块文档级上下文的唯一办法。还有两种主流方案，用不同的取舍解决同一问题：

• 父文档检索（parent-document retrieval）。用小块来建索引和搜索（检索精准），但当某个小块命中时，喂给 LLM 的是它所在的更大的父级章节，而不是小块本身。简单，无需额外模型调用——当你的块小到无法单独推理时很合适。
• 延迟切块（late chunking）。先在令牌（token）级别嵌入整篇文档，之后再把令牌嵌入池化成块向量。每个块向量从周围令牌中继承了文档级的上下文——完全不需要逐块调用 LLM。索引阶段比上下文检索更便宜，后者每块要花一次小模型调用。

粗略法则：上下文检索在密集技术文档上最强（它写出一条真正、贴合查询形态的摘要）；当索引阶段成本或语料规模要紧时，延迟切块以低得多的代价拿到大部分收益；父文档最简单，在块单独太小无法作答时表现突出。在第 4 阶段用评估来选——别想当然。

构建混合搜索：BM25 + 稠密 + RRF。

现在我们用真正理解查询意图的东西替换玩具版的 search_docs。三个组件，分别构建，然后融合。

BM25 — 词法基线

BM25 是一种 1990 年代的统计关键词匹配算法，在技术文档上，它往往单独击败花哨的基于嵌入的搜索。请重读这句话。人们习惯性地直接用嵌入而跳过 BM25——这是个错误。BM25 是你的底线：对于用户使用了文档里相同词汇的查询，它绝不应该输。

pip install bm25s

# retrieval/bm25_index.py
import bm25s

class BM25Index:
    def __init__(self, chunks):
        self.chunks = chunks
        corpus = [c.text for c in chunks]
        self.retriever = bm25s.BM25()
        self.retriever.index(bm25s.tokenize(corpus, stopwords="en"))

    def search(self, query: str, top_k: int = 50) -> list[str]:
        q_tok = bm25s.tokenize(query, stopwords="en")
        results, scores = self.retriever.retrieve(q_tok, k=top_k)
        return [self.chunks[i].chunk_id for i in results[0]]

稠密嵌入 — 语义层

稠密检索将块和查询都嵌入同一个向量空间，然后找出最接近查询的块。能捕获 BM25 漏掉的同义转述（"how to vacuum a table" 匹配 "running VACUUM on a relation"）。

对于嵌入模型，两家提供商都有不错的选择。我们将使用 Voyage AI（Anthropic 推荐的嵌入合作伙伴）和 OpenAI 的 text-embedding-3-large。向量数据库两者相同——我们在本地使用 ChromaDB。

pip install chromadb voyageai

# retrieval/dense_index.py — Voyage embeddings
import chromadb, voyageai
voyage = voyageai.Client()
chroma = chromadb.PersistentClient(path=".chroma")

def embed(texts, input_type="document"):
    r = voyage.embed(texts, model="voyage-3",
                     input_type=input_type)
    return r.embeddings

class DenseIndex:
    def __init__(self, chunks, name="corpus"):
        self.col = chroma.get_or_create_collection(name)
        if self.col.count() == 0:
            embeddings = embed([c.text for c in chunks])
            self.col.add(
                ids=[c.chunk_id for c in chunks],
                embeddings=embeddings,
                documents=[c.text for c in chunks],
            )

    def search(self, query, top_k=50):
        q_emb = embed([query], input_type="query")[0]
        r = self.col.query(query_embeddings=[q_emb],
                           n_results=top_k)
        return r["ids"][0]

# retrieval/dense_index.py — OpenAI embeddings
import chromadb
from openai import OpenAI
oai = OpenAI()
chroma = chromadb.PersistentClient(path=".chroma")

def embed(texts):
    r = oai.embeddings.create(
        model="text-embedding-3-large",
        input=texts,
    )
    return [e.embedding for e in r.data]

class DenseIndex:
    def __init__(self, chunks, name="corpus"):
        self.col = chroma.get_or_create_collection(name)
        if self.col.count() == 0:
            embeddings = embed([c.text for c in chunks])
            self.col.add(
                ids=[c.chunk_id for c in chunks],
                embeddings=embeddings,
                documents=[c.text for c in chunks],
            )

    def search(self, query, top_k=50):
        q_emb = embed([query])[0]
        r = self.col.query(query_embeddings=[q_emb],
                           n_results=top_k)
        return r["ids"][0]

互惠排名融合 — 合并两路结果

BM25 和稠密检索都返回按排名排列的块 ID 列表。RRF 用一个简单公式将它们合并为单一排名：对每个结果，在所有排名中按 1 / (k + rank) 计分，累加得分，降序排列。k=60 是标准默认值。

# retrieval/hybrid.py
def rrf(rankings: list[list[str]], k: int = 60) -> list[str]:
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, cid in enumerate(ranking):
            scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank + 1)
    return sorted(scores, key=scores.get, reverse=True)

class HybridSearch:
    def __init__(self, bm25: BM25Index, dense: DenseIndex):
        self.bm25 = bm25
        self.dense = dense

    def search(self, query: str, top_k: int = 20) -> list[str]:
        bm = self.bm25.search(query, top_k=50)
        ds = self.dense.search(query, top_k=50)
        return rrf([bm, ds])[:top_k]

快速对比：改进了什么？

用同一个查询——"when to use VACUUM versus VACUUM FULL"——分别跑各种检索方法：

SUBSTRING (Phase 1):  found 0 docs ─ phrase doesn't appear literally

BM25 alone:
  1. routine-vacuuming::3     0.71   (contains "VACUUM FULL")
  2. sql-vacuum::0            0.65
  3. runtime-config-vacuum::1 0.41

DENSE alone:
  1. routine-vacuuming::3     0.89   (matches semantically)
  2. routine-vacuuming::4     0.86
  3. sql-vacuum::2            0.81

HYBRID (RRF):
  1. routine-vacuuming::3     0.0328
  2. routine-vacuuming::4     0.0163
  3. sql-vacuum::0            0.0162
  4. sql-vacuum::2            0.0161

在前 20 个结果上加重排序器。

混合搜索让你获得不错的召回率——正确的块通常在前 20 内。但它可能在第 8 位，而你只有时间给智能体看 5 个。重排序器通过使用更贵但更精准的模型对前 20 重新打分来解决这个问题。

架构分两阶段：混合搜索快速宽泛地运行（从整个语料库取前 20 个候选），然后重排序器慢速精准地运行（对这 20 个重新排序，得出最优的前 5 个）。这是一种标准模式，叫做"先检索再重排"。

重排序器与嵌入的区别

嵌入模型分别将查询和块编码为向量，再比较——速度快但失去了两者之间的交叉注意力。重排序器（"交叉编码器"）同时编码两者并给出一个分数——每对比较更慢，但精度高得多，因为它能看到每个块如何匹配查询的每个部分。

重排序器有三种选项——选一个：

• Cohere Rerank — 托管 API，20 个块约 50ms。生产环境最流行的选择。
• Voyage rerank-2 — 托管 API，质量和价格类似。
• BGE-reranker-v2 — 开放权重模型，本地运行。配置慢一些；推理时免费。

pip install cohere

# retrieval/rerank.py — Cohere
import cohere
co = cohere.Client()

def rerank(query: str, chunks: list[Chunk],
           top_k: int = 5) -> list[Chunk]:
    if not chunks: return []
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=[c.text for c in chunks],
        top_n=top_k,
    )
    return [chunks[r.index] for r in response.results]

整合在一起

完整的检索（retrieval）管道，封装在智能体调用的单一函数后面：

# retrieval/__init__.py
from retrieval.bm25_index import BM25Index
from retrieval.dense_index import DenseIndex
from retrieval.hybrid import HybridSearch
from retrieval.rerank import rerank
from retrieval.chunk import chunk_doc

# Built once at import time
ALL_CHUNKS = load_or_build_chunks()
BY_ID = {c.chunk_id: c for c in ALL_CHUNKS}
hybrid = HybridSearch(BM25Index(ALL_CHUNKS), DenseIndex(ALL_CHUNKS))

def retrieve(query: str, top_k: int = 5) -> list[Chunk]:
    candidate_ids = hybrid.search(query, top_k=20)
    candidates = [BY_ID[cid] for cid in candidate_ids]
    return rerank(query, candidates, top_k=top_k)

将其重新接入智能体

替换 agent/tools.py 中的玩具版 search_docs：

from retrieval import retrieve

def search_docs(query: str) -> list[dict]:
    chunks = retrieve(query, top_k=5)
    return [
        {
            "chunk_id": c.chunk_id,
            "doc_id": c.doc_id,
            "snippet": c.text[:300],
        }
        for c in chunks
    ]

def fetch_doc(chunk_id: str) -> dict:
    # Now we fetch a CHUNK, not a whole document.
    # Chunks are bounded ~500 tokens — safe for context.
    c = BY_ID.get(chunk_id)
    if not c: return {"error": "not found"}
    return {"chunk_id": chunk_id,
            "doc_id": c.doc_id,
            "text": c.text}

注意变化：fetch_doc 现在返回单个块而不是整篇文档。我们选择 500 令牌的块是有原因的——它们足够小，可以舒适地放入上下文而不会占主导地位。智能体可以在多步骤调查中获取跨文档的多个块。

现在用第 1 阶段同样的 VACUUM 查询运行一遍

$ python scripts/run.py "When should I VACUUM versus VACUUM FULL?"

──────────────────── Step 0 ────────────────────
→ search_docs({'query': 'VACUUM versus VACUUM FULL'})
   returned: [
     {chunk_id: 'routine-vacuuming::3', snippet: 'VACUUM
       FULL rewrites the entire table and indexes,
       reclaiming disk space but requiring an ACCESS
       EXCLUSIVE lock. Regular VACUUM cannot...' },
     {chunk_id: 'routine-vacuuming::4', snippet: '...'},
     {chunk_id: 'sql-vacuum::0', snippet: '...'},
     ...
   ]

──────────────────── Step 1 ────────────────────
┌─ thinking ─────────────────────────────────────┐
│ The first snippet has the core distinction.    │
│ I'll fetch the full chunk for detail.          │
└────────────────────────────────────────────────┘
→ fetch_doc({'chunk_id': 'routine-vacuuming::3'})

──────────────────── Step 2 ────────────────────
→ submit_answer({...})

status: answered (3 steps)

添加落地验证器。

检索给了智能体正确的材料。但智能体仍然可能产生幻觉（hallucination）——声称某件事与引用似乎相关，但实际上并不被引用支持。落地验证器可以捕获这种情况。

验证器是一次额外的模型调用，在 submit_answer 最终确定之前运行。它接受答案中的结论和引用的块，返回一个裁决：支持、部分支持或不支持。不受支持的结论要么被重写，要么得到引用纠正，要么触发重试。

验证器提示词（prompt）

两家提供商使用相同的提示词（prompt）；只有 API 调用不同。

VERIFIER_PROMPT = """You are a strict fact-checker.

Given an answer and the sources it cited, decide for each
factual claim in the answer whether the sources SUPPORT,
PARTIALLY support, or DO NOT SUPPORT the claim.

Output JSON only:
{
  "claims": [
    {
      "claim": "the exact claim from the answer",
      "verdict": "SUPPORT" | "PARTIAL" | "NOT_SUPPORTED",
      "evidence": "quote from a source, or empty",
      "reasoning": "one sentence"
    }
  ]
}

Be strict. If a claim is more specific than what
sources actually say, mark it PARTIAL. If a claim
isn't mentioned in sources, mark it NOT_SUPPORTED."""

# grounding/verify.py
import json
from anthropic import Anthropic
client = Anthropic()

def verify(answer: str, sources: list[dict]) -> dict:
    sources_block = "\n\n".join(
        f"[{s['chunk_id']}]\n{s['text']}"
        for s in sources
    )
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        system=VERIFIER_PROMPT,
        messages=[{
            "role": "user",
            "content": f"ANSWER:\n{answer}\n\n"
                       f"SOURCES:\n{sources_block}",
        }],
    )
    return json.loads(response.content[0].text)

# grounding/verify.py
import json
from openai import OpenAI
client = OpenAI()

def verify(answer: str, sources: list[dict]) -> dict:
    sources_block = "\n\n".join(
        f"[{s['chunk_id']}]\n{s['text']}"
        for s in sources
    )
    response = client.responses.create(
        model="gpt-5-mini",
        instructions=VERIFIER_PROMPT,
        input=f"ANSWER:\n{answer}\n\n"
              f"SOURCES:\n{sources_block}",
        text={"format": {"type": "json_object"}},
    )
    return json.loads(response.output_text)

验证器返回什么

{
  "claims": [
    {
      "claim": "Transaction pooling rotates server
                connections between transactions",
      "verdict": "SUPPORT",
      "evidence": "transaction pooling releases server
                   connections back to the pool after
                   each transaction commits",
      "reasoning": "Source directly states this."
    },
    {
      "claim": "Prepared statements are scoped to a
                session",
      "verdict": "SUPPORT",
      "evidence": "PREPARE creates a prepared statement
                   for the current session only",
      "reasoning": "Source explicitly states session scope."
    },
    {
      "claim": "This conflict is the most common cause
                of PgBouncer migration failures",
      "verdict": "NOT_SUPPORTED",
      "evidence": "",
      "reasoning": "Sources discuss the conflict but
                    make no claim about migration
                    failures or their frequency."
    }
  ]
}

接入代理循环（agent loop）

修改循环，在返回成功答案之前进行验证：

# in agent/loop.py, when submit_answer is called:
if tool_name == "submit_answer":
    sources = [BY_ID[cid] for cid in citations
               if cid in BY_ID]
    verdict = verify(answer, [{
        "chunk_id": s.chunk_id,
        "text": s.text,
    } for s in sources])

    unsupported = [c for c in verdict["claims"]
                   if c["verdict"] == "NOT_SUPPORTED"]

    return {
        "status": "answered",
        "answer": answer,
        "citations": citations,
        "verification": verdict,
        "unsupported_claims": len(unsupported),
    }

目前我们只是把裁决附加到结果上——至于怎么处理（重试、警告、剔除），等第 4 阶段的评估结果出来再决定。