☰
Elasticsearch 预计算上下文:降低 agent 成本的 TaoToken 实践
2026/10/3 16:29:13 网站建设 项目流程

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

第二步,在控制台的 API Keys 页面生成一个 Key,复制保存。这个 Key 后面会同时用在抽取脚本和查询接口里。

# API Keys 管理页(deep link): # https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite

第三步,确认 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_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

这里有个容易踩的坑:抽取阶段和查询阶段建议用不同的模型 ID。抽取是批量离线任务,追求吞吐和成本,用轻量模型就够;查询重写是实时链路,追求准确理解意图,用能力强的模型。两个阶段共用同一个 Key,但 model 字段分开配。这样既统一了计费入口,又保留了灵活性。

如果你后面要做长期编码或 Agent 类任务,可以考虑 Coding Plan,它更适合持续性的模型调用场景:

# Coding Plan(长期编码/Agent 场景): # https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite

Key 准备好之后,就可以进入 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(docid=docid, url=url, text=text[:8000]) resp = requests.post( f"{TAOTOKEN_BASE_URL}/v1/chat/completions", headers={ "Authorization": f"Bearer {TAOTOKEN_API_KEY}", "Content-Type": "application/json", }, json={ "model": "gemini-flash", "messages": [{"role": "user", "content": prompt}], "response_format": {"type": "json_object"}, "temperature": 0.2, }, timeout=120, ) 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": [f"index://{index_name}"], "tags": fact.get("tags", []) + [f"doc:{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"}, data=body.encode("utf-8"), timeout=120, ) 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 排序取前 10:

FROM 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 对比。这是整篇文章最该动手做的验证。方法很简单:同一批问题,分别用 baseline(search-and-fetch)和 with-context(KI 查询)跑一遍,记录输入 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 pp
F10.5610.624+0.063
输入 token174.8M48.3M−72%
输出 token373k345k−7%
超时数(43 步限制)28 / 9637 / 96+9

输入 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_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

5. 本篇常见错误排查

预计算上下文这套链路,出错的地方比较集中。下面按真实报错逐个说。

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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

整套方案跑下来,核心不是某个单点配置,而是三个环节咬合:抽取要针对领域调优,检索要支持混合匹配和聚合,反馈循环要持续把失败转成新事实。缺任何一个,预计算上下文都只能停在 demo 阶段。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询