Claude Cookbooks 实践指南：用 Contextual Embeddings 构建与优化 Contextual Retrieval 系统

传统 RAG 把文档切成小块后单独嵌入，孤立 chunk 往往丢失文档级上下文，导致检索失败。本篇基于 claude-cookbooks 仓库中的 Contextual Embeddings 指南 及其主教程 guide.ipynb，完整讲解如何：搭建带基线评测的 RAG 检索管线、用 Claude 为每个 chunk 生成"定位上下文"并借助 prompt caching 把一次性摄入成本压低 6 成以上、叠加 Contextual BM25 混合检索与 reranking，最终把 Pass@10 从 87.15% 提升到 95.26%；同时给出面向 AWS Bedrock Knowledge Base 的 Lambda 自定义分块函数实现，可直接用于生产。

一、问题背景：为什么 chunk 单独嵌入会失效

在典型 RAG 流程中，文档先被切分为小 chunk 以便高效检索。但当某个 chunk 本身信息不完整时——比如只含一段函数定义而缺少"它属于哪个模块、为什么存在"的信息——基于该 chunk 生成的向量就难以与语义相关的查询对齐，检索质量随之下降。

Contextual Embeddings（上下文嵌入）的做法是：在嵌入之前，用 Claude 为每个 chunk 生成一段简短的"定位描述"（situating context），把这段描述拼在 chunk 前面一起嵌入。教程中的实验数据（9 个代码库语料、248 条评测查询）显示：

• 平均而言，Contextual Embeddings 将 top-20 chunk 的检索失败率降低了 35%；
• 在仓库自带数据集上，Pass@10 从约 87% 提升到约 95%（结合 reranking 后）。

同一份 chunk 级上下文还可以直接用于 BM25 关键词检索，形成 Contextual BM25 混合检索，这是后文第四节的内容。

二、实验数据集与评测方法

教程使用的评测资产都在 data/ 目录：

文件	内容	实测规模
data/codebase_chunks.json	预分块的语料库	90 个文档（来自 9 个代码库），共 737 个 chunk
data/evaluation_set.jsonl	评测查询集	248 条查询，每条含"golden chunk"标注

codebase_chunks.json 中每个文档的结构为 doc_id / original_uuid / content / chunks，每个 chunk 含 chunk_id / original_index / content；所有 chunk 已按基础的字符切分机制预分块。评测查询集每条记录包含 query、answer、golden_chunk_uuids、golden_documents、golden_chunks 等字段，用于标注"正确答案所在 chunk"。

评测指标是 Pass@k：对每条查询，检查 golden chunk 是否出现在检索结果的前 k 位。教程统一在 k = 5 / 10 / 20 三档评测。

前置环境要求（来自指南 Setup 章节）：

• Python 3.8+，4GB+ 内存，向量库约需 5–10 GB 磁盘空间；
• BM25 环节需要本地 Docker 运行 Elasticsearch（可选）；
• 需要三组 API Key：ANTHROPIC_API_KEY（Claude 生成上下文）、VOYAGE_API_KEY（voyage-2 嵌入）、COHERE_API_KEY（reranking）；
• 核心依赖：anthropic、voyageai、cohere、elasticsearch、pandas、numpy、matplotlib、scikit-learn，通过 pip install --upgrade anthropic voyageai cohere elasticsearch pandas numpy 安装；
• 指南预估：跑完全流程约 30–45 分钟，API 成本约 $5–10。

教程开头统一声明模型名，便于随新模型发布切换：

MODEL_NAME = "claude-haiku-4-5"

三、基线管线：Naive RAG 与 VectorDB 类

基线系统即业界常说的 "Naive RAG"，三步走：

1. 按标题/字符切分文档，每个 chunk 只含自身内容；
2. 直接嵌入每个 chunk；
3. 查询时用余弦相似度检索。

教程实现了一个轻量的内存向量库类 VectorDB（guide.ipynb 中的 "Initialize a Vector DB Class" 小节），它承担三个职责：

1. 嵌入生成：用 Voyage AI 的 voyage-2 模型把文本转为向量；
2. 存储与缓存：嵌入结果 pickle 序列化落盘（./data/{name}/vector_db.pkl），避免重复支付嵌入 API 费用；
3. 相似度检索：search(query, k) 用 np.dot 计算相似度、np.argsort 取 top-k，并对查询向量做内存缓存加速重复评测。

关键实现要点：批量嵌入每批 128 个 chunk（batch_size = 128），tqdm 跟踪进度。加载数据时按文档遍历 chunks，把 doc_id、original_uuid、chunk_id、original_index、content 一并写入 metadata：

texts_to_embed.append(chunk["content"])
metadata.append({
    "doc_id": doc["doc_id"],
    "original_uuid": doc["original_uuid"],
    "chunk_id": chunk["chunk_id"],
    "original_index": chunk["original_index"],
    "content": chunk["content"],
})

基线评测函数 evaluate_retrieval 的逻辑是：从 evaluation_set.jsonl 取出 golden chunk 的原始文本（strip 后精确比对），调用检索函数取 top-k，统计命中比例，最终输出 pass_at_n / average_score / total_queries 汇总，并按 k 打印结果表。

运行基线评测：

results = evaluate_and_display(
    base_db, "data/evaluation_set.jsonl", k_values=[5, 10, 20], db_name="Baseline RAG"
)

教程记录的基线结果：

指标	Pass 率
Pass@5	80.92%
Pass@10	87.15%
Pass@20	90.06%

这就是后续所有优化的对比基准。

四、Contextual Embeddings：原理、成本与实现

4.1 原理与成本模型

对每个 chunk，把整份源文档和该 chunk一起交给 Claude，让它生成"这段 chunk 在整份文档中处于什么位置、讲的是什么"的简短说明，然后把这段说明拼在 chunk 前面一起嵌入。

成本特性（"Cost and Latency Considerations" 一节的核心结论）：

• 成本只发生在摄入期一次，不在每次查询时发生。与 HyDE 等给每次查询增加延迟/成本的技术不同，这是建库时的一次性投入；
• Prompt caching 使其在生产上可行：同一文档的所有 chunk 顺序处理时，第一个 chunk 把整份文档写入缓存（付出少量溢价），后续 chunk 全部从缓存读取（该部分 token 享 90% 折扣），而缓存有效期为 5 分钟，足够处理完一个文档的所有 chunk；
• 教程给出的成本示例：800 token 的 chunk、8k token 的文档、生成约 100 token 的上下文时，总成本约 $1.02 / 百万文档 token；
• 一个容易踩的坑：部分嵌入模型有固定输入 token 上限，如果上下文嵌入后效果反而变差，可能是"上下文 + chunk"被截断，应换用上下文窗口更大的嵌入模型。

4.2 situate_context：带缓存控制的提示词

教程中的提示词由两段组成：<document>...</document> 包裹全文，<chunk>...</chunk> 包裹目标 chunk，并要求"只输出简短上下文，不要输出其他内容"：

DOCUMENT_CONTEXT_PROMPT = """
<document>
{doc_content}
</document>
"""

CHUNK_CONTEXT_PROMPT = """
Here is the chunk we want to situate within the whole document
<chunk>
{chunk_content}
</chunk>

Please give a short succinct context to situate this chunk within the overall document for the purposes of improving search retrieval of the chunk.
Answer only with the succinct context and nothing else.
"""

调用时关键有两处配置：文档段落的 content block 上挂 "cache_control": {"type": "ephemeral"} 声明提示词缓存，以及请求头 extra_headers={"anthropic-beta": "prompt-caching-2024-07-31"}；另外 max_tokens=1000、temperature=0.0 保证输出稳定简短。

教程先对单个 chunk 演示，示例输出是一段说明"该 chunk 包含差分模糊测试执行器的模块文档与 DiffExecutor 结构体定义……"的上下文，并打印 usage 中的 input_tokens / output_tokens / cache_creation_input_tokens / cache_read_input_tokens 以观察缓存命中情况。

4.3 ContextualVectorDB：并行摄入 + token 统计

ContextualVectorDB 类在基线 VectorDB 之上扩展了摄入期的上下文生成，核心特性：

• 并行处理：load_data(dataset, parallel_threads=1) 用 ThreadPoolExecutor 并发处理 chunk，线程数可配——线程调大可提速，担心 API 限流时调小；
• 自动 prompt caching：按文档顺序处理 chunk，最大化缓存命中；
• token 跟踪：用 threading.Lock 保护 token_counts 字典，累计 input / output / cache_read / cache_creation 四类 token，处理完打印缓存命中率与节省比例；
• 持久化：嵌入与上下文化 metadata 一并 pickle 到 ./data/{name}/contextual_vector_db.pkl。

metadata 中同时保留 original_content（原始 chunk 文本，供评测与 BM25 用）和 contextualized_content（Claude 生成的上下文），实际嵌入文本为 f"{contextualized_text}\n\n{chunk['content']}"。

运行摄入（教程用 5 线程）：

contextual_db = ContextualVectorDB("my_contextual_db")
contextual_db.load_data(transformed_dataset, parallel_threads=5)

教程记录的实测 token 数据（737 个 chunk）：

• 输入 token（未计缓存）：1,223,730；输出 token：58,161
• 写入缓存：176,079；从缓存读取：2,267,069
• 61.83% 的输入 token 命中缓存，按 90% 折扣折算，成本从约 $9.20 降到约 $2.85（约 69% 节省）

缓存命中率取决于每个文档的 chunk 数量：chunk 越多的文件，"写一次、读多次"的收益越大，这也解释了为什么必须按文档顺序（而非随机打乱）处理 chunk。

4.4 评测结果

results = evaluate_and_display(
    contextual_db, "data/evaluation_set.jsonl",
    k_values=[5, 10, 20], db_name="Contextual Embeddings"
)

指标	基线 RAG	+ Contextual Embeddings	提升
Pass@5	80.92%	88.12%	+7.20pp
Pass@10	87.15%	92.34%	+5.19pp
Pass@20	90.06%	94.29%	+4.23pp

教程指出提升在 Pass@5 上最显著——说明上下文化后的 chunk 不仅是"更容易被召回"，相关时还排得更靠前。

五、Contextual BM25：混合检索再进一步

语义检索擅长理解含义与同义改写，但会漏掉精确的关键词/函数名匹配；BM25（考虑文档长度与词频饱和的概率式关键词排序算法）恰好补上这一面。Contextual BM25 的妙处在于：BM25 同时检索原始 chunk 文本和 Claude 生成的上下文描述，两个字段都能命中关键词。

5.1 启动本地 Elasticsearch

docker run -d --name elasticsearch -p 9200:9200 -p 9300:9300 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  elasticsearch:9.2.0

排障三件套：docker ps | grep elasticsearch 确认进程；9200 端口被占则 docker stop elasticsearch && docker rm elasticsearch；出问题时看 docker logs elasticsearch。

5.2 索引与融合检索

教程中的 ElasticsearchBM25 类创建索引 contextual_bm25_index，索引设置中显式指定 "similarity": {"default": {"type": "BM25"}} 与 english 分析器；mapping 里 content 与 contextualized_content 均为可检索 text 字段，doc_id / chunk_id / original_index 为不可索引的 keyword/integer 字段（仅返回不检索）。写入用 elasticsearch.helpers.bulk 批量上传并 refresh。

查询用 multi_match 同时命中两个字段：

response = self.es_client.search(
    index=self.index_name,
    query={
        "multi_match": {
            "query": query,
            "fields": ["content", "contextualized_content"],
        }
    },
    size=k,
)

retrieve_advanced(query, db, es_bm25, k, semantic_weight=0.8, bm25_weight=0.2) 实现三步融合：

1. 各召回 150 条：语义检索与 BM25 各取 top 150 候选；
2. 加权倒数排名融合：每个候选按 score += semantic_weight * (1/(rank+1)) 与 score += bm25_weight * (1/(rank+1)) 累加（默认语义 80% / BM25 20%，权重可调，可按数据特性做实验）；
3. 取 top-k：按融合分排序后只返回前 k 条，并统计结果中各来源的占比（同时命中的按 0.5/0.5 分摊）。

评测函数 evaluate_db_advanced 会先做 10 条预热查询、按 k 输出结果与 Semantic/BM25 来源占比，并在 finally 中删除 Elasticsearch 索引，避免残留。教程记录的混合检索结果：Pass@5 88.86%（Semantic 54.6% / BM25 45.4%）、Pass@10 92.31%、Pass@20 95.23%。

六、Reranking：两阶段检索的最后一段精度

Reranking 是典型的两阶段检索：第一阶段宽召回（多取候选），第二阶段用专门的 rerank 模型对候选做更精细的相关性打分，只保留 top-k。第一阶段检索优化的是海量文档下的速度，rerank 模型较慢但更准，用少量候选换取精度上的收益。

教程用 Cohere 的 rerank-english-v3.0，流程为：

1. 过度召回：取 k * 10 条语义检索结果（例如需要 10 条就取 100 条）；
2. rerank：每条候选文档拼为 原文\n\nContext: 上下文描述——rerank 模型因此同时看到原始文本与 Claude 生成的上下文；然后调用 co.rerank(model="rerank-english-v3.0", query=query, documents=documents, top_n=k)，代码中 time.sleep(0.1) 做简单限流；
3. 选 top-k：按 relevance_score 返回最终结果，再用同一套 golden chunk 精确比对逻辑计分。

教程记录的 reranking 结果：Pass@5 92.15%、Pass@10 95.26%、Pass@20 97.45%。代价是每次查询增加约 100–200ms 延迟与少量按次计费（教程估计约 $0.002/查询）。

七、四种方案对比与选型建议

教程总结的完整对比表：

方案	Pass@5	Pass@10	Pass@20
基线 RAG	80.92%	87.15%	90.06%
+ Contextual Embeddings	88.12%	92.34%	94.29%
+ Hybrid Search (BM25)	86.43%	93.21%	94.99%
+ Reranking	92.15%	95.26%	97.45%

关键结论（教程 "Key Takeaways"）：

1. Contextual Embeddings 是单项提升最大的技术（+5~7 个百分点），仅靠它就能拿到约 90% 的最优效果；
2. Reranking 绝对性能最高：Pass@10 达 95.26%，相对基线的检索失败率从 12.85% 降到 4.74%，即失败率下降约 47%；
3. 每种技术都有成本：上下文嵌入是约 $3 的一次性摄入成本（本数据集、启用 prompt caching）；混合检索需要维护 Elasticsearch 基础设施；reranking 增加查询延迟与按次 API 费用。

按需求选型的建议：

• 高并发、成本敏感：只用 Contextual Embeddings（92.34% Pass@10，无按次查询成本）；
• 追求最高精度、对延迟不敏感：完整 reranking 管线（95.26% Pass@10）；
• 均衡的生产系统：混合检索（约 93% Pass@10，无按次查询成本，但有基础设施开销）。

教程还强调：本例用代码库语料演示，但方法同样适用于企业内部知识库、金融法律文档、教育内容等其他数据类型。

八、生产落地：Bedrock Knowledge Base 的 Lambda 自定义分块

指南的 "Additional Notes" 指出：prompt caching 目前可用在 Anthropic 一方 API 上，第三方伙伴环境（Bedrock、Vertex）需要定制；AWS 团队为此提供了一个 Lambda 函数代码，部署后可在配置 Bedrock Knowledge Base 时选择它作为自定义分块选项。代码位于 contextual-rag-lambda-function/，共三个文件：

• lambda_function.py：主入口；
• inference_adapter.py：Bedrock 调用封装；
• s3_adapter.py：S3 读写封装。

8.1 lambda_function.py 的处理流程

入口 lambda_handler(event, context) 从事件中提取 inputFiles 与 bucketName（缺失即抛 ValueError），随后对每个文件批次的每个 contentBatches 条目：

1. 从 S3 读回 chunk 文件：s3_adapter.read_from_s3(...) 取回该批次的 fileContents（每个 chunk 的 contentBody）；
2. 重建整份文档：把所有 chunk 的 contentBody 用 "".join(...) 拼回原文全文（注释说明也可以直接读原始文件再抽取文本）；
3. 逐 chunk 生成上下文：把文档全文与 chunk 套入 contextual_retrieval_prompt（与笔记本中相同的 <document>/<chunk> 提示词，要求"只输出简短上下文"），经 inference_adapter.invoke_model_with_response_stream(prompt) 流式取回；
4. 写回输出 S3：新 chunk 内容为 chunk_context + "\n\n" + content_body（上下文在前、原文在后，与笔记本的嵌入文本拼接方式一致），连同 contentType、contentMetadata 原样保留，写到 Output/{input_key}；
5. 组装返回：输出 {"outputFiles": [...]}，每个输出文件保留 originalFileLocation 与处理后的 contentBatches。

从源码结构看，该 Lambda 与笔记本的 ContextualVectorDB.situate_context 使用的是同一套提示词模板与 temperature=0 的确定性调用，只是把"向量库摄入前处理"前移到了 Bedrock 的分块环节，从而让 Knowledge Base 入库的每条 chunk 天然带文档级上下文。

8.2 inference_adapter.py 与 s3_adapter.py

InferenceAdapter 通过 boto3 的 bedrock-runtime 客户端（默认 region us-east-1，可按需修改）调用模型 anthropic.claude-haiku-4-5-20251001-v1:0，请求体为 Bedrock 消息格式（anthropic_version: bedrock-2023-05-31、max_tokens=1000、temperature=0.0），用 invoke_model_with_response_stream 流式解析 content_block_delta 事件并 yield 文本，遇到 stop_reason 即停止。

S3Adapter 提供 write_output_to_s3（put_object，校验 HTTP 200）、read_from_s3（get_object 并解析 JSON）与 parse_s3_path（把 s3://bucket/key 拆分为 bucket 与 key）三个方法，均以 boto3 ClientError 兜底。

GCP 侧没有现成代码，教程给出的思路是自建 Cloud Run 实例、按同样的"读全文 → 逐 chunk 生成上下文 → 拼回"模式实现。

九、小结

本教程给出了一条可复现、有量化结论的 RAG 精度优化路径：

1. 先立基线：Naive RAG + Pass@k 评测（本数据集 87.15% Pass@10），任何优化都要对基线说话；
2. Contextual Embeddings 是性价比最高的一步：摄入期一次性成本、prompt caching 把 60%+ 输入 token 打到 90% 折扣、Pass@10 +5.19pp，且无按次查询成本；
3. Contextual BM25 与 reranking 按需叠加：前者用上下文增强关键词召回（需要 Elasticsearch），后者换延迟与按次费用买最后 2–3pp；
4. 平台落地有现成路径：Bedrock 用户可直接部署仓库提供的 Lambda 自定义分块函数，GCP 用户可按同一模式自建服务。

完整可运行的代码、数据集与全部中间结果，见 capabilities/contextual-embeddings/ 目录；若希望进一步理解 RAG 评测体系本身，可参考同仓库的 retrieval_augmented_generation 教程。