1. RAG 場景下 agent 上下文膨脹的真實成本做 RAG 的團隊大多經(jīng)歷過這個階段檢索質(zhì)量明明還行但 agent 跑起來又慢又貴。問題往往不在模型而在上下文。一個典型的事實型問題agent 需要先搜索、讀片段、判斷不夠、再搜索、再讀循環(huán)兩三次之后輸入 token 已經(jīng)堆到幾十萬答案還沒提交。我試過在一個 96 題的小評測集上跑標準 search-and-fetch 流程單題平均輸入 token 接近 180 萬其中大部分是重復(fù)讀進來的原始正文片段。這就是「上下文膨脹」的本質(zhì)agent 把預(yù)算花在了瀏覽原始數(shù)據(jù)源上而不是花在推理上。圍繞 agent 上下文的討論經(jīng)常被當成記憶問題——更大的窗口、更長的上下文、更強的召回。但換個角度看它其實是一個檢索問題。如果檢索層能在 agent 提問之前就把結(jié)構(gòu)化事實準備好agent 就不需要反復(fù)讀原文token 消耗自然下降。這篇文章要交付的是一套可復(fù)制的 Elasticsearch 預(yù)計算上下文方案。核心思路是把原始文檔提前抽取成 Knowledge Indicators簡稱 KI也就是原子化的事實單元再用混合檢索語義 詞法讓 agent 通過自然語言接口直接查詢這些事實。配合 TaoToken 統(tǒng)一 Key 調(diào)用 LLM可以在不犧牲檢索質(zhì)量的前提下把 agent 的輸入 token 壓下來。適合誰看正在做 RAG agent、被 token 成本困擾的后端或算法工程師已經(jīng)有一套 Elasticsearch 檢索鏈路、想進一步優(yōu)化 agent 收斂效率的團隊以及想搞清楚「預(yù)計算上下文」到底怎么落地、而不是停留在概念層面的人。下面我會按順序講清楚四件事索引映射怎么寫、預(yù)計算管道怎么配、TaoToken 怎么統(tǒng)一接入、以及怎么用前后 token 對比驗證效果。每一步都給可復(fù)制的配置和命令你照著改字段名就能跑。2. TaoToken 前置準備與統(tǒng)一 Key 接入在講 Elasticsearch 配置之前先把 LLM 調(diào)用這一層理順。預(yù)計算管道里有兩個地方要調(diào)模型一是抽取階段把文檔轉(zhuǎn)成 KI二是查詢階段把自然語言問題重寫成 ES|QL。如果這兩處各用一套 Key、各配一個 Base URL后面排查問題會很痛苦。用 TaoToken 統(tǒng)一 Key 的好處是抽取和查詢走同一個入口token 消耗也能在一個地方看。TaoToken 是一個兼容 OpenAI 接口規(guī)范的模型調(diào)用入口你可以把它理解成一個統(tǒng)一的 API 網(wǎng)關(guān)Base URL 固定Key 統(tǒng)一管理模型 ID 按需切換。對 RAG 場景來說最實用的點是它支持在同一個 Key 下切換不同模型——抽取階段可以用便宜快速的模型查詢重寫階段用理解能力更強的模型成本和質(zhì)量能分開調(diào)。接入分三步。第一步去官網(wǎng)注冊并拿到 Key# 瀏覽器打開官網(wǎng)注冊后在控制臺創(chuàng)建 API Key # 官網(wǎng)地址帶來源標記 # https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content第二步在控制臺的 API Keys 頁面生成一個 Key復(fù)制保存。這個 Key 后面會同時用在抽取腳本和查詢接口里。# API Keys 管理頁deep link # https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite第三步確認 Base URL。注意 API 地址不帶 UTM 參數(shù)保持干凈# 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 有沒有復(fù)制完整、有沒有多余空格。如果返回 model not found說明模型 ID 寫錯了去模型對話頁面確認當前可用的模型名。# 模型對話頁確認可用模型 ID # https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite這里有個容易踩的坑抽取階段和查詢階段建議用不同的模型 ID。抽取是批量離線任務(wù)追求吞吐和成本用輕量模型就夠查詢重寫是實時鏈路追求準確理解意圖用能力強的模型。兩個階段共用同一個 Key但 model 字段分開配。這樣既統(tǒng)一了計費入口又保留了靈活性。如果你后面要做長期編碼或 Agent 類任務(wù)可以考慮 Coding Plan它更適合持續(xù)性的模型調(diào)用場景# Coding Plan長期編碼/Agent 場景 # https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewriteKey 準備好之后就可以進入 Elasticsearch 側(cè)的配置了。下面所有腳本里的TAOTOKEN_API_KEY和TAOTOKEN_BASE_URL都指向上面的值。3. Elasticsearch 索引映射與預(yù)計算管道配置這一節(jié)是全文的技術(shù)核心。預(yù)計算上下文能不能跑起來取決于兩件事KI 的索引映射設(shè)計得對不對以及抽取管道能不能穩(wěn)定產(chǎn)出結(jié)構(gòu)化事實。先說索引映射。KI 的結(jié)構(gòu)里最關(guān)鍵的是title和description兩個字段它們都要映射成semantic_text這樣才能同時支持語義檢索和詞法檢索。tags用 keyword 類型方便做聚合和過濾。payload里放結(jié)構(gòu)化的 subject/predicate/object用于精確匹配。下面這份 mapping 可以直接復(fù)制路徑按你的索引名調(diào)整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需要先創(chuàng)建PUT _inference/text_embedding/.jina-embeddings-v5-text-small { service: elasticsearch, service_settings: { model_id: .jina-embeddings-v5-text-small } }映射建好之后寫抽取管道。抽取的本質(zhì)是給模型一段文檔正文讓它輸出 0 到 15 條原子事實每條事實必須自包含——也就是說未來某個 agent 只讀 title description 就能直接作答不需要回讀原文。抽取腳本用 Python 寫調(diào)用 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 要轉(zhuǎn)成 KI 文檔再寫入索引。轉(zhuǎn)換時給每條 KI 生成一個穩(wěn)定的 id方便后續(xù)去重和更新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 文檔的子集大概能產(chǎn)出 24 萬條 KI用輕量模型跑 7 小時左右能完成。這個量級下抽取成本遠低于 agent 每次查詢省下來的 token 開銷。這里有個設(shè)計要點抽取 prompt 不是一次寫死的。第一版 prompt 往往只能覆蓋顯性事實漏掉那些「次要但關(guān)鍵」的細節(jié)——比如某個只出現(xiàn)一次的人名、某個具體日期。這些細節(jié)恰恰是檢索時區(qū)分相似實體的關(guān)鍵。所以抽取 prompt 需要根據(jù) agent 的實際失敗案例迭代這一點在第 5 節(jié)會展開。4. 驗證請求與 token 消耗對比配置跑通之后必須驗證兩件事查詢接口能不能正確返回 KI以及預(yù)計算方案到底省了多少 token。沒有對比數(shù)據(jù)優(yōu)化就是盲猜。先驗證查詢接口。設(shè)計一個自然語言查詢端點接收問題內(nèi)部用 LLM 重寫成 ES|QL再執(zhí)行。請求體長這樣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返回結(jié)果里除了匹配到的 KI還會帶聚合信息——按 tag、按來源、按實體統(tǒng)計。這個設(shè)計很關(guān)鍵agent 在讀取任何單條 KI 之前就能看到結(jié)果集的結(jié)構(gòu)。如果結(jié)果分散在三個來源agent 知道要收斂如果都指向同一個實體agent 知道可以沿這條線索繼續(xù)。驗證查詢接口是否正常用 curl 打一發(fā)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、準確率和超時數(shù)。baseline 的檢索調(diào)用返回最多 10 條結(jié)果每條帶三段 700 字正文片段單次調(diào)用就可能塞進約 21k 詞。with-context 的查詢調(diào)用返回約 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%超時數(shù)43 步限制28 / 9637 / 969輸入 token 降了 72%準確率反而升了。這個結(jié)果說明預(yù)計算上下文不是靠犧牲質(zhì)量換成本而是讓 agent 用更少的檢索輪次、更便宜的檢索結(jié)果收斂到答案。但超時數(shù)上升了 9 個這點要單獨看。超時在嚴格步數(shù)預(yù)算下不等于失敗它只意味著 agent 在寫出答案之前用完了步數(shù)。with-context 的 37 個超時里有 21 個最終被判定為正確——因為答案已經(jīng)通過 KI 出現(xiàn)在上下文里只是 agent 還沒提交。相比之下baseline 的超時更常直接失敗因為它的上下文主要是原始正文最后強制提交時往往是在噪聲里猜。所以驗證 token 消耗時不能只看總量還要看「有效 token」——也就是真正推動答案收斂的那部分。KI 路徑的 token 少但每條都更接近答案這才是成本下降的根本原因。如果你想在驗證階段快速對比不同模型的重寫效果可以用模型對話頁面手動試幾條 query看看重寫出來的 ES|QL 是否合理# 模型對話頁手動驗證 query 重寫 # https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite5. 本篇常見錯誤排查預(yù)計算上下文這套鏈路出錯的地方比較集中。下面按真實報錯逐個說。401 Unauthorized。最常見的原因是 Key 沒配對環(huán)境變量或者復(fù)制時帶了空格。檢查TAOTOKEN_API_KEY是否真的注入到了運行進程里而不是只寫在 shell 里。另外注意 Base URL 不要帶多余路徑正確寫法是https://taotoken.net/api請求時拼/v1/chat/completions。local proxy failed。這個報錯通常出現(xiàn)在本地網(wǎng)絡(luò)環(huán)境有額外代理設(shè)置時。先確認你的請求是直連 TaoToken 的 Base URL沒有被本地代理攔截。如果用了 requests 庫檢查有沒有繼承HTTP_PROXY環(huán)境變量必要時顯式設(shè)置proxies{http: None, https: None}。reading choices 報錯。這個一般發(fā)生在解析響應(yīng)時choices字段為空或結(jié)構(gòu)不符預(yù)期。原因可能是模型返回了錯誤信息而不是正常 completion。打印完整響應(yīng)體確認重點看有沒有error字段。如果抽取階段用了response_format: json_object但模型不支持也會導(dǎo)致解析失敗換成普通文本模式再手動提取 JSON。OAuth 相關(guān)報錯。如果你在配置里誤用了 OAuth 流程而不是 API Key會看到 token 獲取失敗。TaoToken 的接入方式是 Bearer Key不需要 OAuth。檢查請求頭是不是Authorization: Bearer key而不是Authorization: OAuth token。semantic_text 字段檢索不到結(jié)果。先確認 inference endpoint 是否創(chuàng)建成功用GET _inference/text_embedding/.jina-embeddings-v5-text-small查一下。如果 endpoint 不存在semantic_text字段不會報錯但檢索時匹配不到。另外確認寫入時title和description確實有內(nèi)容空字段不會生成 embedding。KI 抽取結(jié)果為空。抽取 prompt 里明確要求對導(dǎo)航頁、登錄墻、錯誤頁返回空 facts 列表。如果你的文檔正文本身很短或很泛模型會按規(guī)則返回空。檢查輸入text字段是不是真的拿到了正文而不是只拿到了頁面標題。超時數(shù)偏高。如果 with-context 的超時明顯多于 baseline先看 agent 的指令是不是過于保守——比如要求「至少兩次 KI 查詢失敗才回退正文搜索」。這個閾值可以調(diào)。另外檢查 KI 的 title 是否足夠自包含如果 title 需要配合 description 才能理解agent 就得多讀一步步數(shù)消耗會上去。token 沒降下來。如果輸入 token 和 baseline 差不多大概率是 agent 還在頻繁回退到正文搜索。檢查兩點一是 KI 覆蓋度夠不夠二是查詢接口返回的 KI 是否真的命中了問題。用幾條典型問題手動打_get_context看返回的 title 里有沒有答案。如果沒有說明抽取階段漏了關(guān)鍵事實需要回到第 3 節(jié)迭代抽取 prompt。排查時有個通用技巧把 agent 的完整 trace 拉出來看它在哪一步消耗了最多 token。如果是在反復(fù)讀正文片段說明 KI 沒命中如果是在反復(fù)重寫查詢說明查詢接口的意圖理解有問題。兩種失敗的修復(fù)方向完全不同。6. 持續(xù)迭代把失敗反饋回抽取管道前面講的配置和驗證能讓你把預(yù)計算上下文跑起來輸入 token 降 70% 左右。但準確率會停在一個平臺上大概 70% 上下。繼續(xù)加事實、繼續(xù)調(diào)檢索提升有限。真正把準確率從 70% 推到 90% 以上的是反饋循環(huán)。機制是這樣的agent 每次提交錯誤答案trace 里都帶著非常精確的信息——語料庫沒能區(qū)分的兩個實體以及 agent 實際選了哪個錯誤項。比如把 University of Aberdeen 錯提交成 University of Edinburgh把 9 提交成 7。這些失敗本身就是診斷信號。把這些失敗轉(zhuǎn)成新的 KI專門針對區(qū)分點。每條 disambiguation KI 的 title 里同時包含正確答案和錯誤答案并說明區(qū)別{ 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如果區(qū)分信息只埋在 description 里檢索階段可能召回不到disambiguation 就失效了。把這類 KI 寫回同一個檢索層之后重新跑評測效果很明顯96 題里 29 個失敗有 21 個被翻轉(zhuǎn)準確率從 69.8% 升到 91.7%輸入 token 又降了 12%超時從 37 降到 27。agent 不僅更準還更快了。這個循環(huán)要持續(xù)跑因為失敗模式會漂移。這一輪修好了 Hooker 和 Rauwolf 的混淆下一輪 agent 可能提交 Francisco Hernández而這個名字不在上一輪的 disambiguation 覆蓋范圍內(nèi)。所以反饋循環(huán)不是一次性優(yōu)化而是常態(tài)運行的過程觀察 agent 在哪里停滯、哪里延遲提交、哪里提交錯誤把這些信號反饋回抽取 prompt 和 disambiguation 構(gòu)建。落地時建議把反饋循環(huán)做成一個定時任務(wù)每天拉取 agent 的錯誤 trace自動生成候選 disambiguation KI過 guardrail 后寫入索引。抽取 prompt 的迭代可以按周做用一批失敗樣本重新調(diào)優(yōu)。最后給一個實用建議抽取階段和查詢階段用 TaoToken 同一個 Key但模型分開配。抽取用輕量模型批量跑查詢重寫用能力強的模型保證意圖理解。這樣成本和質(zhì)量能分開控制計費也集中在一個入口。接入文檔在這里配置細節(jié)可以對照著調(diào)# 接入文檔 # https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite整套方案跑下來核心不是某個單點配置而是三個環(huán)節(jié)咬合抽取要針對領(lǐng)域調(diào)優(yōu)檢索要支持混合匹配和聚合反饋循環(huán)要持續(xù)把失敗轉(zhuǎn)成新事實。缺任何一個預(yù)計算上下文都只能停在 demo 階段。