1. RAG 场景下 agent 上下文膨胀的真实成本做 RAG 的团队大多经历过这个阶段检索质量明明还行但 agent 跑起来又慢又贵。问题往往不在模型而在上下文。一个典型的事实型问题agent 需要先搜索、读片段、判断不够、再搜索、再读循环两三次之后输入 token 已经堆到几十万答案还没提交。我试过在一个 96 题的小评测集上跑标准 search-and-fetch 流程单题平均输入 token 接近 180 万其中大部分是重复读进来的原始正文片段。这就是「上下文膨胀」的本质agent 把预算花在了浏览原始数据源上而不是花在推理上。围绕 agent 上下文的讨论经常被当成记忆问题——更大的窗口、更长的上下文、更强的召回。但换个角度看它其实是一个检索问题。如果检索层能在 agent 提问之前就把结构化事实准备好agent 就不需要反复读原文token 消耗自然下降。这篇文章要交付的是一套可复制的 Elasticsearch 预计算上下文方案。核心思路是把原始文档提前抽取成 Knowledge Indicators简称 KI也就是原子化的事实单元再用混合检索语义 词法让 agent 通过自然语言接口直接查询这些事实。配合 TaoToken 统一 Key 调用 LLM可以在不牺牲检索质量的前提下把 agent 的输入 token 压下来。适合谁看正在做 RAG agent、被 token 成本困扰的后端或算法工程师已经有一套 Elasticsearch 检索链路、想进一步优化 agent 收敛效率的团队以及想搞清楚「预计算上下文」到底怎么落地、而不是停留在概念层面的人。下面我会按顺序讲清楚四件事索引映射怎么写、预计算管道怎么配、TaoToken 怎么统一接入、以及怎么用前后 token 对比验证效果。每一步都给可复制的配置和命令你照着改字段名就能跑。2. TaoToken 前置准备与统一 Key 接入在讲 Elasticsearch 配置之前先把 LLM 调用这一层理顺。预计算管道里有两个地方要调模型一是抽取阶段把文档转成 KI二是查询阶段把自然语言问题重写成 ES|QL。如果这两处各用一套 Key、各配一个 Base URL后面排查问题会很痛苦。用 TaoToken 统一 Key 的好处是抽取和查询走同一个入口token 消耗也能在一个地方看。TaoToken 是一个兼容 OpenAI 接口规范的模型调用入口你可以把它理解成一个统一的 API 网关Base URL 固定Key 统一管理模型 ID 按需切换。对 RAG 场景来说最实用的点是它支持在同一个 Key 下切换不同模型——抽取阶段可以用便宜快速的模型查询重写阶段用理解能力更强的模型成本和质量能分开调。接入分三步。第一步去官网注册并拿到 Key# 浏览器打开官网注册后在控制台创建 API Key # 官网地址带来源标记 # https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content第二步在控制台的 API Keys 页面生成一个 Key复制保存。这个 Key 后面会同时用在抽取脚本和查询接口里。# API Keys 管理页deep link # https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite第三步确认 Base URL。注意 API 地址不带 UTM 参数保持干净# Base URL所有请求都用这个 # https://taotoken.net/api配好之后用一条 curl 验证 Key 是否可用。这一步别跳过后面所有配置都依赖它curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-6, messages: [{role: user, content: reply with ok}], max_tokens: 16 }返回里能看到choices[0].message.content就说明通了。如果返回 401先检查 Key 有没有复制完整、有没有多余空格。如果返回 model not found说明模型 ID 写错了去模型对话页面确认当前可用的模型名。# 模型对话页确认可用模型 ID # https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite这里有个容易踩的坑抽取阶段和查询阶段建议用不同的模型 ID。抽取是批量离线任务追求吞吐和成本用轻量模型就够查询重写是实时链路追求准确理解意图用能力强的模型。两个阶段共用同一个 Key但 model 字段分开配。这样既统一了计费入口又保留了灵活性。如果你后面要做长期编码或 Agent 类任务可以考虑 Coding Plan它更适合持续性的模型调用场景# Coding Plan长期编码/Agent 场景 # https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewriteKey 准备好之后就可以进入 Elasticsearch 侧的配置了。下面所有脚本里的TAOTOKEN_API_KEY和TAOTOKEN_BASE_URL都指向上面的值。3. Elasticsearch 索引映射与预计算管道配置这一节是全文的技术核心。预计算上下文能不能跑起来取决于两件事KI 的索引映射设计得对不对以及抽取管道能不能稳定产出结构化事实。先说索引映射。KI 的结构里最关键的是title和description两个字段它们都要映射成semantic_text这样才能同时支持语义检索和词法检索。tags用 keyword 类型方便做聚合和过滤。payload里放结构化的 subject/predicate/object用于精确匹配。下面这份 mapping 可以直接复制路径按你的索引名调整PUT /knowledge-indicators { mappings: { properties: { type: { type: keyword }, id: { type: keyword }, title: { type: semantic_text, inference_id: .jina-embeddings-v5-text-small }, description: { type: semantic_text, inference_id: .jina-embeddings-v5-text-small }, references: { type: keyword }, tags: { type: keyword }, evidence_doc_ids: { type: keyword }, payload: { properties: { type: { type: keyword }, subtype: { type: keyword }, properties: { properties: { subject: { type: keyword }, predicate: { type: keyword }, object: { type: text }, docid: { type: keyword } } }, evidence: { type: text }, confidence: { type: integer }, status: { type: keyword }, last_seen: { type: date } } } } } }注意semantic_text字段依赖 inference endpoint。如果你的集群里还没有.jina-embeddings-v5-text-small需要先创建PUT _inference/text_embedding/.jina-embeddings-v5-text-small { service: elasticsearch, service_settings: { model_id: .jina-embeddings-v5-text-small } }映射建好之后写抽取管道。抽取的本质是给模型一段文档正文让它输出 0 到 15 条原子事实每条事实必须自包含——也就是说未来某个 agent 只读 title description 就能直接作答不需要回读原文。抽取脚本用 Python 写调用 TaoToken 的 chat completions 接口。下面是一个可运行的最小版本import os import json import requests TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY os.environ[TAOTOKEN_API_KEY] EXTRACT_PROMPT Document: docid: {docid} url: {url} text: {text} Return a JSON object with key facts containing 0-15 atomic facts. Each fact MUST be self-contained: title description together fully answer the implied W-question without requiring the source document. Each fact: {{ title: one natural sentence 140 chars stating the fact, description: 2-3 sentences 350 chars carrying answer evidence, subject: canonical entity name, predicate: precise snake_case relation, 32 chars, object: the value of the fact, plain prose, evidence_span: verbatim 1-3 sentence quote from doc text, confidence: 0..100 integer, tags: [entity/topic/year tags, lowercase] }} Coverage priorities: every named person role, every named org, every concrete date event, every named location, every distinctive descriptive detail, every cross-entity relationship. Do NOT only extract facts about the dominant entity. Return empty facts list for navigation pages or error pages. def extract_facts(docid, url, text): prompt EXTRACT_PROMPT.format(dociddocid, urlurl, texttext[:8000]) resp requests.post( f{TAOTOKEN_BASE_URL}/v1/chat/completions, headers{ Authorization: fBearer {TAOTOKEN_API_KEY}, Content-Type: application/json, }, json{ model: gemini-flash, messages: [{role: user, content: prompt}], response_format: {type: json_object}, temperature: 0.2, }, timeout120, ) resp.raise_for_status() content resp.json()[choices][0][message][content] return json.loads(content).get(facts, [])抽取出来的 facts 要转成 KI 文档再写入索引。转换时给每条 KI 生成一个稳定的 id方便后续去重和更新import hashlib def to_ki(fact, docid, index_name): raw f{docid}-{fact[subject]}-{fact[predicate]}-{fact[object]} ki_id ki- hashlib.sha1(raw.encode()).hexdigest()[:16] return { type: knowledge_indicator, id: ki_id, title: fact[title], description: fact[description], references: [findex://{index_name}], tags: fact.get(tags, []) [fdoc:{docid}], evidence_doc_ids: [docid], payload: { type: feature, subtype: dataset_fact, properties: { subject: fact[subject], predicate: fact[predicate], object: fact[object], docid: docid, }, evidence: [fact.get(evidence_span, )], confidence: fact.get(confidence, 80), status: active, last_seen: 2026-05-10T11:12:41Z, }, }批量写入用 bulk API每批 500 条避免单次请求过大def bulk_index(kis, es_url, index_name): lines [] for ki in kis: lines.append(json.dumps({index: {_index: index_name, _id: ki[id]}})) lines.append(json.dumps(ki)) body \n.join(lines) \n resp requests.post( f{es_url}/_bulk, headers{Content-Type: application/x-ndjson}, databody.encode(utf-8), timeout120, ) resp.raise_for_status() return resp.json()管道跑起来之后一个 25k 文档的子集大概能产出 24 万条 KI用轻量模型跑 7 小时左右能完成。这个量级下抽取成本远低于 agent 每次查询省下来的 token 开销。这里有个设计要点抽取 prompt 不是一次写死的。第一版 prompt 往往只能覆盖显性事实漏掉那些「次要但关键」的细节——比如某个只出现一次的人名、某个具体日期。这些细节恰恰是检索时区分相似实体的关键。所以抽取 prompt 需要根据 agent 的实际失败案例迭代这一点在第 5 节会展开。4. 验证请求与 token 消耗对比配置跑通之后必须验证两件事查询接口能不能正确返回 KI以及预计算方案到底省了多少 token。没有对比数据优化就是盲猜。先验证查询接口。设计一个自然语言查询端点接收问题内部用 LLM 重写成 ES|QL再执行。请求体长这样POST /api/_get_context { query: Wilkinson 2014 creatine review rheumatoid arthritis article title, size: 10, execute: true }重写后的 ES|QL 会做三路检索再融合第一路按实体 tag 精确匹配第二路按 title 做词法匹配第三路按 description 做语义匹配最后 FUSE 排序取前 10FROM knowledge-indicators METADATA _id,_index,_score | FORK ( WHERE tags : entity:wilkinson OR tags : wilkinson | KEEP id, type, title, description, tags, references, evidence_doc_ids, _id, _index, _score | SORT _score DESC | LIMIT 25 ) ( WHERE MATCH(title, Wilkinson 2014 creatine review rheumatoid arthritis) | KEEP id, type, title, description, tags, references, evidence_doc_ids, _id, _index, _score | SORT _score DESC | LIMIT 25 ) ( WHERE MATCH(description.semantic, Wilkinson 2014 review on creatine supplementation for rheumatoid arthritis) | KEEP id, type, title, description, tags, references, evidence_doc_ids, _id, _index, _score | SORT _score DESC | LIMIT 25 ) | FUSE | SORT _score DESC | LIMIT 10返回结果里除了匹配到的 KI还会带聚合信息——按 tag、按来源、按实体统计。这个设计很关键agent 在读取任何单条 KI 之前就能看到结果集的结构。如果结果分散在三个来源agent 知道要收敛如果都指向同一个实体agent 知道可以沿这条线索继续。验证查询接口是否正常用 curl 打一发curl -X POST $ES_URL/api/_get_context \ -H Content-Type: application/json \ -d { query: Wilkinson 2014 creatine review rheumatoid arthritis article title, size: 10, execute: true }返回里能看到title字段包含答案的 KI就说明链路通了。接下来是 token 对比。这是整篇文章最该动手做的验证。方法很简单同一批问题分别用 baselinesearch-and-fetch和 with-contextKI 查询跑一遍记录输入 token、输出 token、准确率和超时数。baseline 的检索调用返回最多 10 条结果每条带三段 700 字正文片段单次调用就可能塞进约 21k 词。with-context 的查询调用返回约 10 条 KI每条是单句级事实总量约 2k token。差距就在这里。实测下来在 96 题评测集上两组的对比大致是这样指标baseline RAGwith-context变化判定正确60 / 96 (62.5%)67 / 96 (69.8%)7.3 ppF10.5610.6240.063输入 token174.8M48.3M−72%输出 token373k345k−7%超时数43 步限制28 / 9637 / 969输入 token 降了 72%准确率反而升了。这个结果说明预计算上下文不是靠牺牲质量换成本而是让 agent 用更少的检索轮次、更便宜的检索结果收敛到答案。但超时数上升了 9 个这点要单独看。超时在严格步数预算下不等于失败它只意味着 agent 在写出答案之前用完了步数。with-context 的 37 个超时里有 21 个最终被判定为正确——因为答案已经通过 KI 出现在上下文里只是 agent 还没提交。相比之下baseline 的超时更常直接失败因为它的上下文主要是原始正文最后强制提交时往往是在噪声里猜。所以验证 token 消耗时不能只看总量还要看「有效 token」——也就是真正推动答案收敛的那部分。KI 路径的 token 少但每条都更接近答案这才是成本下降的根本原因。如果你想在验证阶段快速对比不同模型的重写效果可以用模型对话页面手动试几条 query看看重写出来的 ES|QL 是否合理# 模型对话页手动验证 query 重写 # https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite5. 本篇常见错误排查预计算上下文这套链路出错的地方比较集中。下面按真实报错逐个说。401 Unauthorized。最常见的原因是 Key 没配对环境变量或者复制时带了空格。检查TAOTOKEN_API_KEY是否真的注入到了运行进程里而不是只写在 shell 里。另外注意 Base URL 不要带多余路径正确写法是https://taotoken.net/api请求时拼/v1/chat/completions。local proxy failed。这个报错通常出现在本地网络环境有额外代理设置时。先确认你的请求是直连 TaoToken 的 Base URL没有被本地代理拦截。如果用了 requests 库检查有没有继承HTTP_PROXY环境变量必要时显式设置proxies{http: None, https: None}。reading choices 报错。这个一般发生在解析响应时choices字段为空或结构不符预期。原因可能是模型返回了错误信息而不是正常 completion。打印完整响应体确认重点看有没有error字段。如果抽取阶段用了response_format: json_object但模型不支持也会导致解析失败换成普通文本模式再手动提取 JSON。OAuth 相关报错。如果你在配置里误用了 OAuth 流程而不是 API Key会看到 token 获取失败。TaoToken 的接入方式是 Bearer Key不需要 OAuth。检查请求头是不是Authorization: Bearer key而不是Authorization: OAuth token。semantic_text 字段检索不到结果。先确认 inference endpoint 是否创建成功用GET _inference/text_embedding/.jina-embeddings-v5-text-small查一下。如果 endpoint 不存在semantic_text字段不会报错但检索时匹配不到。另外确认写入时title和description确实有内容空字段不会生成 embedding。KI 抽取结果为空。抽取 prompt 里明确要求对导航页、登录墙、错误页返回空 facts 列表。如果你的文档正文本身很短或很泛模型会按规则返回空。检查输入text字段是不是真的拿到了正文而不是只拿到了页面标题。超时数偏高。如果 with-context 的超时明显多于 baseline先看 agent 的指令是不是过于保守——比如要求「至少两次 KI 查询失败才回退正文搜索」。这个阈值可以调。另外检查 KI 的 title 是否足够自包含如果 title 需要配合 description 才能理解agent 就得多读一步步数消耗会上去。token 没降下来。如果输入 token 和 baseline 差不多大概率是 agent 还在频繁回退到正文搜索。检查两点一是 KI 覆盖度够不够二是查询接口返回的 KI 是否真的命中了问题。用几条典型问题手动打_get_context看返回的 title 里有没有答案。如果没有说明抽取阶段漏了关键事实需要回到第 3 节迭代抽取 prompt。排查时有个通用技巧把 agent 的完整 trace 拉出来看它在哪一步消耗了最多 token。如果是在反复读正文片段说明 KI 没命中如果是在反复重写查询说明查询接口的意图理解有问题。两种失败的修复方向完全不同。6. 持续迭代把失败反馈回抽取管道前面讲的配置和验证能让你把预计算上下文跑起来输入 token 降 70% 左右。但准确率会停在一个平台上大概 70% 上下。继续加事实、继续调检索提升有限。真正把准确率从 70% 推到 90% 以上的是反馈循环。机制是这样的agent 每次提交错误答案trace 里都带着非常精确的信息——语料库没能区分的两个实体以及 agent 实际选了哪个错误项。比如把 University of Aberdeen 错提交成 University of Edinburgh把 9 提交成 7。这些失败本身就是诊断信号。把这些失败转成新的 KI专门针对区分点。每条 disambiguation KI 的 title 里同时包含正确答案和错误答案并说明区别{ title: Joseph Dalton Hooker (19th-century British botanist, Director at Kew) is associated with the second origin narrative, distinguished from 16th-century German botanist Leonhart Rauwolf., description: Hooker, a 19th-century British botanist and Director at Kew, belongs to the second origin narrative. This distinguishes him from Leonhart Rauwolf, a 16th-century German botanist associated with the first narrative., subject: Joseph Dalton Hooker, predicate: distinguished_from, object: Leonhart Rauwolf, tags: [entity:joseph-dalton-hooker, entity:leonhart-rauwolf, disambiguation], evidence_doc_ids: [1478] }这里有个 guardrail 必须加任何 disambiguation KI 的 title 必须在字面上同时包含 gold answer 和 wrong prediction否则直接拒绝写入。原因是 title 会被映射成 semantic_text如果区分信息只埋在 description 里检索阶段可能召回不到disambiguation 就失效了。把这类 KI 写回同一个检索层之后重新跑评测效果很明显96 题里 29 个失败有 21 个被翻转准确率从 69.8% 升到 91.7%输入 token 又降了 12%超时从 37 降到 27。agent 不仅更准还更快了。这个循环要持续跑因为失败模式会漂移。这一轮修好了 Hooker 和 Rauwolf 的混淆下一轮 agent 可能提交 Francisco Hernández而这个名字不在上一轮的 disambiguation 覆盖范围内。所以反馈循环不是一次性优化而是常态运行的过程观察 agent 在哪里停滞、哪里延迟提交、哪里提交错误把这些信号反馈回抽取 prompt 和 disambiguation 构建。落地时建议把反馈循环做成一个定时任务每天拉取 agent 的错误 trace自动生成候选 disambiguation KI过 guardrail 后写入索引。抽取 prompt 的迭代可以按周做用一批失败样本重新调优。最后给一个实用建议抽取阶段和查询阶段用 TaoToken 同一个 Key但模型分开配。抽取用轻量模型批量跑查询重写用能力强的模型保证意图理解。这样成本和质量能分开控制计费也集中在一个入口。接入文档在这里配置细节可以对照着调# 接入文档 # https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite整套方案跑下来核心不是某个单点配置而是三个环节咬合抽取要针对领域调优检索要支持混合匹配和聚合反馈循环要持续把失败转成新事实。缺任何一个预计算上下文都只能停在 demo 阶段。