基于 Instructor 实现 Anthropic Contextual Retrieval异步 RAG 上下文增强实战指南【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor本文围绕 Anthropic 提出的 Contextual Retrieval上下文检索技术讲解如何在 RAG 系统中为文档分块附加情境化上下文并使用 Instructor 的AsyncInstructor Pydantic 结构化输出 Jinja2 模板与 Anthropic prompt caching实现一个高效、类型安全、可并发处理的实战方案。读完本文你将掌握 Contextual Retrieval 的核心原理、完整可运行代码以及基于仓库源码理解模板注入与异步编排的底层机制。BackgroundRAG 系统中的上下文丢失问题Anthropic 在其关于 Contextual Retrieval 的技术文章中指出了传统 RAG 系统的一个关键痛点文档被切分成块chunk后检索单元丢失了它在原始文档中的语境。设想你的知识库中存放了一批美国 SEC 财报文件用户提出这样的问题ACME Corp 在 2023 年 Q2 的营收增长是多少一个相关度很高的 chunk 可能包含这样的文本The companys revenue grew by 3% over the previous quarter.但这块文本单独存在时并不足以让嵌入模型embedding model或 BM25 检索器判断它属于哪家公司、对应哪个时间段。检索阶段丢失的主语和时间锚点正是导致召回失败的重要原因。Anthropic 的解决方案Contextual RetrievalContextual Retrieval 的思路非常直接在将 chunk 送去嵌入之前先用 LLM 为每个 chunk 生成一段简短的、指向整篇文档的解释性上下文并把这段上下文与 chunk 拼接在一起形成情境化 chunk再进入索引。Anthropic 给出的示例对比original_chunk The companys revenue grew by 3% over the previous quarter. contextualized_chunk This chunk is from an SEC filing on ACME corps performance in Q2 2023; the previous quarters revenue was $314 million. The companys revenue grew by 3% over the previous quarter.可以看到拼接后的 chunk 补全了公司是谁时间是什么时候等信息后续无论做向量检索还是 BM25 检索都能获得更精确的匹配基础。用 Claude 生成情境化上下文官方提示模板Anthropic 使用 Claude 生成上下文其核心提示模板将整篇文档与待处理的 chunk同时提供给模型并要求模型只输出简洁的上下文、不要输出其他内容document {{WHOLE_DOCUMENT}} /document 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.这个模板正是后续 Instructor 实现中 prompt 的雏形我们会在下面的实现里看到它如何与 Jinja2 变量注入结合。Instructor 异步实现完整可运行方案仓库中的可运行示例位于 examples/situate_context/run.py比博客文档中的版本更完整示例文档定义、参数化 chunk 函数与耗时统计。我们以此为主线拆解整个实现。1. 初始化带 Prompt Caching 的异步客户端from instructor import AsyncInstructor, Mode, patch from anthropic import AsyncAnthropic from pydantic import BaseModel, Field client AsyncInstructor( clientAsyncAnthropic(), createpatch( createAsyncAnthropic().beta.prompt_caching.messages.create, modeMode.TOOLS, ), modeMode.TOOLS, )要点说明AsyncInstructorInstructor 的异步客户端包装器其类定义位于 instructor/v2/core/client.py以client、create、mode、provider、hooks为构造参数为底层客户端叠加结构化输出能力。patch(...)将 Anthropic 的beta.prompt_caching.messages.create端点包装为支持结构化输出的创建函数。这里直接使用 beta 提示缓存端点属于 Anthropic prompt caching 的接入方式详见博客 anthropic-prompt-caching.md。modeMode.TOOLS指示 Instructor 通过工具调用function calling协议解析结构化响应这是 Anthropic 支持的结构化输出模式。2. 定义 Pydantic 响应模型class SituatedContext(BaseModel): The context to situate the chunk within the document. The situated context should be as long as the original chunk. Example: - original chunk: The companys revenue grew by 3% over the previous quarter. - situated context: This chunk is from an SEC filing on ACME corps performance in Q2 2023; the previous quarters revenue was $314 million. The companys revenue grew by 3% over the previous quarter. situated_context: str Field( ..., descriptionThe situated context of the chunk within the document. )使用 PydanticBaseModel声明结构化输出契约字段situated_context承载模型生成的情境化上下文。Field(..., description...)中的描述会被 Instructor 用于生成工具/JSON Schema从而引导模型输出符合预期的内容。3. 核心函数为单个 chunk 生成情境化上下文async def situate_context(doc: str, chunk: str) - SituatedContext: response await client.chat.completions.create( modelclaude-3-haiku-20240307, max_tokens1024, temperature0.0, messages[ { role: user, content: [ { type: text, text: document {{doc}} /document , cache_control: {type: ephemeral}, }, { type: text, text: Here is the chunk we want to situate within the whole document chunk {{chunk}} /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. , }, ], } ], response_modelSituatedContext, context{ doc: doc, chunk: chunk, }, ) return response这段代码同时体现了 Instructor 的四个关键特性结构化输出传入response_modelSituatedContext返回对象被自动校验并转换为 Pydantic 模型实例在本例中即SituatedContext其situated_context字段可直接取用。Jinja2 模板注入消息文本中的{{doc}}、{{chunk}}占位符通过context{doc: doc, chunk: chunk}注入。这是 Instructor 内置的 Jinja 模板机制——context参数会以字典形式传给模板引擎渲染出最终 prompt参见 docs/concepts/templating.md。底层实现中Instructor 使用jinja2.sandbox.SandboxedEnvironment渲染模板见 instructor/v2/core/templating.py并在handle_templating中按消息格式Anthropic/OpenAI/Cohere/VertexAI/Gemini递归处理每条消息的 contentinstructor/v2/core/templating.py。Prompt Caching文档块消息上设置了cache_control: {type: ephemeral}即把整篇文档标记为可缓存内容。由于整篇文档对每个 chunk 都是相同的多个 chunk 调用之间可以直接命中缓存显著降低重复读取大文档的成本与延迟。仓库的 Anthropic 处理器对cache_control字段有原生支持见 instructor/v2/providers/anthropic/handlers.py。确定性输出temperature0.0保证上下文生成行为稳定max_tokens1024为输出留足空间。需要说明的是示例源码返回的是整个SituatedContext模型见 examples/situate_context/run.py而博客文档版本返回response.context字符串——两者都可取前者还能保留结构化对象供下游使用。4. 带重叠的文档分块def chunking_function( doc: str, chunk_size: int 1000, overlap: int 200 ) - list[str]: Chunk the document into chunk_size character segments with overlap overlap. chunks [] start 0 while start len(doc): end start chunk_size chunks.append(doc[start:end]) start chunk_size - overlap return chunks这是一个简单但实用的分块函数以chunk_size为块长、以overlap为相邻块重叠长度滑动切分。示例通过参数化chunk_size与overlap便于实验不同的切分策略。重叠的存在是为了避免把一句话或一个语义单元恰好切在边界上。5. 异步并发编排import asyncio async def process_chunk(doc: str, chunk: str) - dict[str, str]: Process a single chunk by situating it within the context of the full document. context await situate_context(doc, chunk) return {chunk: chunk, context: context} async def process( doc: str, chunk_size: int 1000, overlap: int 200 ) - list[dict[str, str]]: Process the document by chunking it and situating each chunk within the context of the full document. Uses asyncio.gather for concurrent processing. chunks chunking_function(doc, chunk_size, overlap) tasks [process_chunk(doc, chunk) for chunk in chunks] results await asyncio.gather(*tasks) return results这里使用了asyncio.gather将全部 chunk 的上下文生成任务并发执行这是博客 learn-async.md 中介绍的异步并发模式asyncio.gather并发执行所有任务并按输入顺序返回结果。结合 Anthropic prompt caching所有任务共享同一份文档缓存IO 密集的 LLM 调用得以并行摊销。process_chunk返回{chunk: ..., context: ...}字典方便下游拼装成情境化 chunk 并入库。6. 入口与运行if __name__ __main__: # Example usage document ACME Corporation Financial Report for Fiscal Year 2023 ... async def main(): import time start_time time.time() processed_chunks await process(document, chunk_size800, overlap200) end_time time.time() print(fTime taken: {end_time - start_time} seconds) for i, item in enumerate(processed_chunks): print(fChunk {i 1}:) print(fText: {item[chunk]}...) print(fContext: {item[context]}) print() asyncio.run(main())示例自带一份 ACME 公司 2023 财年财务报告作为演示文档见 examples/situate_context/run.py运行后会输出每个 chunk 对应的情境化上下文并打印整体耗时。直接执行即可验证完整流程python examples/situate_context/run.py注意示例使用的模型标识为claude-3-haiku-20240307需要有效的 Anthropic API 凭证同时 prompt caching 需要升级到支持该特性的 Anthropic SDK 版本参考 anthropic-prompt-caching.md 中关于 SDK 版本与最小缓存 token 数的限制说明。该实现的关键特性异步处理基于asyncio与asyncio.gather对全部 chunk 并发生成上下文避免串行等待放大 LLM 延迟。结构化输出Pydantic 模型SituatedContext保证返回内容类型安全、可校验response_model直接驱动工具调用的 schema 生成。Prompt Caching利用 Anthropic 的cache_control: ephemeral缓存整篇文档多个 chunk 请求复用同一份文档前缀。分块策略带重叠的滑动窗口切分chunk 大小与重叠度可参数化调优。Jinja2 模板context参数将doc、chunk注入 prompt实现 prompt 结构与业务数据分离且底层使用SandboxedEnvironment渲染instructor/v2/core/templating.py避免在模板中执行任意 Python 代码同时仍支持列表、条件等 Jinja 语法详见 docs/concepts/templating.md。性能收益与实施考虑Anthropic 报告的检索失败率改善Anthropic 在其文章中对 Contextual Retrieval 报告了如下改善以 top-20 chunk 检索失败率为基准方案失败率变化Contextual Embeddings单独下降 35%5.7% → 3.7%Contextual Embeddings Contextual BM25下降 49%5.7% → 2.9%再叠加 reranking下降 67%5.7% → 1.9%这些数字来自 Anthropic 官方实验可作为方案选型的参考依据实际效果仍需在自己数据集上验证。Anthropic 提出的实施注意点chunk 边界实验 chunk 大小、边界与重叠度的组合找到检索质量与成本的最佳平衡点。嵌入模型Anthropic 发现 Gemini 与 Voyage 嵌入模型效果较好——嵌入模型的选型会影响上下文拼接后的表征质量。自定义 contextualizer prompt针对特定领域如法律、医疗、金融定制上下文生成提示可获得更贴合领域语义的上下文。chunk 数量实验中一次提供给上下文生成器的 chunk 数Anthropic 发现使用 20 个 chunk 效果最佳——即并非一次性处理全部文档而是分组批量 situate以控制每次调用的输入规模。评估务必针对你的具体使用场景运行评测如检索命中率、下游问答准确率不要直接套用公开基准结论。进一步的增强方向基于上述方案可以继续演进动态 chunk 大小根据内容复杂度标题层级、段落语义动态决定切分粒度替代固定字符长度切分。接入向量数据库将{chunk, context}拼装为情境化 chunk 后写入向量库同时可保留 BM25 索引做混合检索对应 Contextual BM25 的组合收益。错误处理与重试为situate_context增加异常捕获与重试逻辑Instructor 的max_retries与 retrying.md 提供了现成机制。嵌入模型与 prompt 实验对比不同嵌入模型与不同 contextualizer 措辞的召回效果。增加 reranking 步骤在检索结果之上再做一次重排进一步压缩失败率。总结Contextual Retrieval 通过用 LLM 为每个 chunk 补写情境化上下文再入索引的方式直击 RAG 检索阶段的信息丢失问题。本文的实现将 Instructor 的AsyncInstructor结构化输出、Jinja2 模板注入docs/concepts/templating.md、Anthropic prompt caching 与asyncio.gather异步编排组合在一起形成了一套生产可参考的上下文增强管线。完整可运行代码位于 examples/situate_context/run.py异步并发范式可进一步参考 learn-async.md提示缓存的限制与配置细节可参考 anthropic-prompt-caching.md。【免费下载链接】instructorstructured outputs for llms项目地址: https://gitcode.com/GitHub_Trending/in/instructor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考