作者来自 Elastic Kathleen DeRusso, Matt Nowzari, Apostolos Matsagkas, Peter Pisljar将 AI agent 上下文存储在 AI Index 中使用更少的 tokens 为更智能的 agent 提供支持。包含使用 ES|QL 和 Kibana Workflows 的分步演练。Agent 在回答任何问题之前都会消耗 tokens 来探索你的数据包括检查 mappings、采样文档、探查应该使用哪个 index。Elasticsearch AI Indices 让你可以预先计算这些工作一次并将其存储为 Knowledge IndicatorKI一种结构化、可搜索的记录agent 可以直接检索而不必每次都从头重新发现这些信息。本演练将向你展示如何构建完整的 pipeline创建 AI Index使用 Kibana Workflow 生成用于路由的 KI并通过可移植的 ES|QL skill 将它们连接到任意 agent harness。如果你希望在阅读本文示例的同时端到端地运行这些内容我们还提供了一个 notebook。这是博客系列的第 1 部分将通过 KI 和 AI indices 对上下文进行管理提供技术演练。虽然 AI indices 将包含在未来的 Stack 版本中但目前我们建议使用 Serverless。工作原理AI Index、Kibana Workflows 和 query-ki skill在本演练中构建上下文包含三个部分一个AI IndexKI 存储在其中。它是一个常规的 Elasticsearch index 或 data stream通过特定的命名约定触发 component templates从而自动配置正确的 mappings。Kibana Workflows它们从你的数据源读取数据运行 LLM 将内容结构化为 KI并将这些 KI 写入 AI Index。query-kiskill一个使用 ES|QL 直接从 AI Index 查询 KI 的 skillchat agent 可以将其作为工具调用。前置条件本教程假设你已经具备一个 Elasticsearch Serverless 项目。如果你还没有可以注册试用版。一个用于访问 Elasticsearch 项目的 API key。创建用于 agent 路由的示例 indices首先我们需要一些数据源。数据源可以是已经存在于 Elasticsearch indices 中的数据也可以是通过 connectors 或 ES|QL data sources 访问的外部数据。在本文中我们将创建一些包含示例数据的 indices。首先我们使用三个数据集作为示例BEIR/fiqa金融、beir-nfcorpus生物医学 / 营养和 beir-scifact科学事实核查。每个 index 都包含其自身的_meta.description。以下是我们为这些 indices 定义的 mappings{ beir-fiqa: { mappings: { _meta: { description: FiQA: financial question answering corpus from StackExchange Finance community posts and web crawls. Covers investments, banking, taxes, and market analysis. BM25-only index. }, properties: { text: { type: text, meta: { description: Full document body text. } }, title: { type: text, meta: { description: Document or article title. } } } } } } { beir-nfcorpus: { mappings: { _meta: { description: NFCorpus: biomedical information retrieval corpus from NutritionFacts.org. Contains nutrition science and medical research documents on diet, disease, and health interventions. BM25-only index. }, properties: { text: { type: text, meta: { description: Full document body text. } }, title: { type: text, meta: { description: Document or article title. } } } } } } { beir-scifact: { mappings: { _meta: { description: SciFact: scientific fact-checking corpus of biomedical research abstracts used to verify factual claims in peer-reviewed literature. BM25-only index. }, properties: { text: { type: text, meta: { description: Full document body text. } }, title: { type: text, meta: { description: Document or article title. } } } } } }然后使用上面的便捷脚本通过 _bulk API 向每个 index 中加载一些文档。现在假设一个 agent 面对一个问题以及我们刚刚创建的这些 indices。agent 在开始时完全不知道哪个 index 与问题相关。如果没有预先计算好的上下文它要么执行探索性查询mappings、测试搜索来确定应该使用哪个数据源要么搜索全部三个 indices并希望合并后的结果中能找到有用的信息。无论采用哪种方式都会消耗 tokens如果把这种低效累积到 agent 执行的每一次查询中成本就会越来越高。创建你的 AI Index在生成任何 KI 之前你需要一个用于存储它们的 index。我们称之为AI Index。命名约定会触发自动配置。任何名称以ai-index-idx-开头的 index 都是常规 index以ai-index-ds-开头的则是 data stream。对于 observability 用例、时间序列数据以及对数据新鲜度要求较高的场景你应该选择 data stream。相反对于会长期存在的静态数据数据新鲜度并不是特别重要并且可能偶尔需要按需更新的场景标准 index 是不错的选择。AI indices 必须使用这种命名约定。当 Elasticsearch 看到ai-index-前缀时会自动应用 component templates从而配置正确的 mappings 和 settings。创建 AI Index 只需要一次调用PUT ai-index-idx-my-corpus要准确查看应用了哪些 component templates可以检查 mappingsGET ai-index-idx-my-corpus/_mapping响应会显示每个 AI Index 开箱即用就具有的字段{ ai-index-idx-my-corpus: { mappings: { properties: { timestamp: { type: date }, attributes: { type: flattened }, content: { type: text, fields: { semantic: { type: semantic_text, inference_id: .jina-embeddings-v5-text-small } } }, description: { type: text, fields: { semantic: { type: semantic_text, inference_id: .jina-embeddings-v5-text-small } } }, references: { properties: { uri: { type: keyword } } }, tags: { type: keyword }, title: { type: text, fields: { semantic: { type: semantic_text, inference_id: .jina-embeddings-v5-text-small } } }, type: { type: keyword } } } } }title、description和content都是text字段并带有一个类型为 semantic_text 的.semantic子字段支持混合检索。Data stream indicesai-index-ds-*还默认具有 90 天的数据保留策略。本文使用标准 indexai-index-idx-*。将 Index 元数据作为 Knowledge Indicator这个示例的目标用例是展示query-index-metadata-kiskill 如何将 agent 路由到正确的 Elasticsearch index即使 index 或字段名称比较模糊。这样可以减少因选择错误的 index或者基于不完整的 schema 探索来构造查询而导致的错误。由于我们正在为自己的 indices 创建 KI因此可以给 LLM 一个良好的起点使用人工编写的_meta.description内容为 index mappings 添加注释。这样workflow 就能利用更多上下文生成更好的 KI。为了解决这个问题我们将手动创建一个 Kibana Workflow对每个 index 进行分析并将用于路由的 KI 写入 AI Index。该 workflow 串联了四个步骤步骤类型功能get_mappingelasticsearch.request读取 mapping包括_meta.description和每个字段的描述。sample_docselasticsearch.search获取一些真实文档使 profile 能够反映实际的数据值结构。profile_indexai.agent将 index profile 生成为结构化输出。sink_index_kielasticsearch.bulk将 profile 作为 KI 写入 AI Index。将以下 YAML 粘贴到 Workflows 编辑器中version: 1 name: beir-index-profile-ki description: Profile an index into an index-selection Knowledge Indicator. enabled: true tags: - context-management - index-selection triggers: - type: manual consts: indices: - beir-fiqa - beir-nfcorpus - beir-scifact steps: - name: loop_indices type: foreach foreach: {{ consts.indices | json }} iteration-on-failure: continue: true steps: - name: get_mapping type: elasticsearch.request with: method: GET path: /{{ foreach.item }}/_mapping - name: sample_docs type: elasticsearch.search with: index: {{ foreach.item }} size: 3 query: match_all: {} - name: profile_index type: ai.agent timeout: 120s with: message: You are a data steward building an INDEX PROFILE for an enterprise data catalog. Downstream, an AI agent uses these profiles to decide WHICH Elasticsearch index to query for a given user question -- this is an index-SELECTION aid, not a place to answer the question itself. You are given (a) the index name, (b) its Elasticsearch mapping including human-written descriptions in _meta.description and each fields meta.description, and (c) a few sample documents. Produce a faithful, decision-useful profile. Rules: - Ground everything in the provided mapping samples. Never invent fields, values, or purpose. If unknown, use an empty string/array. - Optimize for routing: make it obvious what kinds of questions this index can authoritatively answer, and what it canNOT. - Prefer concrete field names and real example values from the samples over vague phrasing. - For joins, surface shared keys (e.g. *_id fields) that link this index to sibling indices, since cross-index questions hinge on them. Index name: {{ foreach.item }} Elasticsearch mapping (JSON): {{ steps.get_mapping.output | json }} Sample documents (JSON): {{ steps.sample_docs.output.hits.hits | map: _source | json }} schema: type: object properties: display_name: type: string description: A concise human-readable name for what this index represents ( 8 words). purpose: type: string description: 2-4 sentences describing what this index stores and its role. PRIMARY semantic surface for matching a question to this index. answers_questions: type: array items: type: string description: 3-7 representative natural-language questions this index can authoritatively answer. does_not_contain: type: array items: type: string description: 1-4 things a searcher might wrongly expect here but that live elsewhere, to prevent mis-routing. key_fields: type: array items: type: string description: 3-10 of the most query-relevant fields as field_name - what it is. when_to_use: type: string description: A single crisp routing heuristic - when should an agent pick THIS index? ( 30 words). example_esql: type: string description: One realistic, runnable ES|QL query against this index answering one of answers_questions. required: - display_name - purpose - answers_questions - key_fields - when_to_use - name: sink_index_ki type: elasticsearch.request with: method: PUT path: /ai-index-idx-my-corpus/_doc/{{ foreach.item | url_encode }} body: timestamp: {{ now | date: %Y-%m-%dT%H:%M:%S.%LZ }} type: index_metadata_entry title: {{ steps.profile_index.output.structured_output.display_name | default: foreach.item }} tags: - index-profile - {{ foreach.item }} attributes: display_name: {{ steps.profile_index.output.structured_output.display_name }} purpose: {{ steps.profile_index.output.structured_output.purpose }} when_to_use: {{ steps.profile_index.output.structured_output.when_to_use }} answers_questions: {{ steps.profile_index.output.structured_output.answers_questions | json }} does_not_contain: {{ steps.profile_index.output.structured_output.does_not_contain | json }} key_fields: {{ steps.profile_index.output.structured_output.key_fields | json }} example_esql: {{ steps.profile_index.output.structured_output.example_esql }} source_index: {{ foreach.item }} content: SOURCE / PROVENANCE This is an INDEX PROFILE for routing/index-selection. Backing Elasticsearch index: {{ foreach.item }} Inspect it directly with ES|QL: FROM {{ foreach.item }} | LIMIT 10 WHAT THIS INDEX IS {{ steps.profile_index.output.structured_output.purpose }} Questions this index can answer: {{ steps.profile_index.output.structured_output.answers_questions | join: | }} When to use this index: {{ steps.profile_index.output.structured_output.when_to_use }} Example query: {{ steps.profile_index.output.structured_output.example_esql }} description: Index profile: {{ steps.profile_index.output.structured_output.display_name }}. Does NOT contain: {{ steps.profile_index.output.structured_output.does_not_contain | join: ; }}. Key fields: {{ steps.profile_index.output.structured_output.key_fields | join: ; }}.让我们逐步了解这个 workflow 的工作方式。我们通过foreach循环遍历三个指定的 indices。对于每一个 indexget_mapping获取 Elasticsearch index mappings包括我们之前添加的_meta.description注释。sample_docs获取 3 个真实文档。与单独依赖 schema 相比具体示例能够为 LLM 提供更好的信息。profile_index使用 index 名称、mappings 和示例文档调用ai.agent。LLM 返回结构化输出其中描述了该 index 的用途、关键字段以及一个展示如何使用该 index 的示例 ES|QL 查询。sink_index_ki将结果作为index_metadata_entry类型的 KI 写入 AI Index并以 index 名称作为 key从而确保重复运行时具有幂等性。需要注意以下几点这个 workflow 将一组特定的 indices 硬编码了。在实际应用中你可以从 index pattern 或动态数据源中获取这个列表。foreach循环还会按顺序执行各个迭代这对于本指南来说没问题但在生产环境中会比较慢因为每次迭代都需要调用一次 LLM。对于大规模场景可以使用 workflow.executeAsync 或原生并行支持。cheat sheet 中介绍了这两种方式的使用技巧。在profile_index步骤中agent prompt 是关键所在。它决定了 KI 的准确性和实用性。如果你不需要加载其他工具使用 ai.prompt 可以提高 workflow 的效率同时降低成本。成本可以通过多种方式进行控制。更丰富的 prompt 和结构化输出通常会带来更高的 token 使用量当然你选择的模型也会显著影响总成本。Elastic Inference ServiceEIS可以作为一个很好的测试环境你可以针对profile_index的ai.agent步骤测试不同的模型从而比较不同模型在生成 KI 时的表现。查询你的 AI Index 以验证 Knowledge Indicatorsbeir-index-profile-kiworkflow 运行后使用以下 ES|QL 查询直接在 Discover 标签页中查询 AI Index以确认写入了哪些内容FROM ai-index-idx-my-corpus | WHERE type index_metadata_entry | KEEP title, content, description, attributes, tags | LIMIT 10这将得到以下输出构建一个可移植的 skill 来检索 AI agent 上下文检索是 AI Index 中的关键组件。KI 是 AI Index 中的一个文档而查找一个 KI 只需要执行一次 ES|QL 查询。我们将这个查询封装成一个小型、可移植的 skill这样任何 agent 都可以调用它而不受其运行的 harness 影响。我们将这个 skill 编写为一个SKILL.md包含 name 和 description 的 YAML header后面跟着 markdown instructions。这与许多 harness 使用的 Agent Skills 格式相同包括 Claude Code、LangChain 的 Deep Agents 以及其他 harness它们都可以直接加载这种格式。harness 会预先读取 header 内容只有当问题与 description 匹配时才会加载完整的 instructions。这个 skill 对 harness 唯一的要求就是提供一种针对 Elasticsearch 运行 ES|QL 的方式。以下是一个query-index-metadata-kiskill 示例--- name: query-index-metadata-ki description: - Retrieve Knowledge Indicators (pre-computed context) from the Elasticsearch AI Index before answering. Use it to find which index to search (routing profiles). Trigger on any question that depends on choosing a data source. allowed-tools: esql_query --- # Retrieving Knowledge Indicators Knowledge Indicators (KIs) live in Elasticsearch indices named ai-index-*. Retrieve them by calling the esql_query tool with the query below. Substitute the users question for query, and index_metadata_entry as the ki_type for routing profiles. esql FROM ai-index-idx-* METADATA _id, _index, _score | WHERE type ki_type | FORK (WHERE MATCH(content, query) OR MATCH(description, query) | SORT _score DESC | LIMIT 20) (WHERE MATCH(content.semantic, query) OR MATCH(description.semantic, query) | SORT _score DESC | LIMIT 20) | FUSE | SORT _score DESC | KEEP title, content, description, tags | LIMIT 5 Ground your answer in what the query returns, and cite the KI titles you used. If nothing relevant comes back, say so rather than guessing.让我们拆解一下这个 skill 的工作方式我们将index-metadata-entry定义为一种 KI 类型 / 用例。我们在 AI indices 上执行混合 ES|QL 搜索并使用合适的type进行过滤同时使用 RRF 作为融合结果的默认方法。当确定哪些 indices 与查询相关时KI 结果将直接为 agent 的回答提供依据。由于这个 skill 只是 instructions 加上一个查询因此无论你的 agent 在哪里运行它都可以随之使用。你可以将同一个文件用于 Kibana Workflow agent、Claude Code、LangChain Deep Agents 或任何其他 harness而无需修改其中任何一行。将你的 AI Index 连接到 agent harness我们希望展示如何使用 AI indices通过任意 harness 查询你的数据。在这些示例中我们将使用 LangChain Deep Agents 和一个兼容 OpenAI 的 key但也可以轻松替换为其他 agent harness包括 Elastic Agent Builder。首先让我们创建一个基线看看 agent 在不使用 KI 的情况下会有怎样的表现# Example question: Is there scientific evidence that vitamin D supplementation prevents cancer? import os import sys import time from elasticsearch import Elasticsearch from langchain_core.messages import AIMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from deepagents import create_deep_agent if len(sys.argv) 2: sys.exit(fUsage: python {sys.argv[0]} your question) es Elasticsearch(os.environ[ES_URL], api_keyos.environ[ES_API_KEY]) tool def esql_query(query: str) - list[dict] | str: Execute an ES|QL query against Elasticsearch and return the matching rows. Args: query: A complete ES|QL query string, e.g. FROM beir-fiqa | LIMIT 5. Full-text search syntax: WHERE MATCH(field, value) — not field MATCH value. try: resp es.esql.query(queryquery, formatjson) cols [c[name] for c in resp[columns]] return [dict(zip(cols, row)) for row in resp[values]] except Exception as e: return fES|QL error: {e} tool def get_mapping(index: str) - dict: Return the field mapping for an Elasticsearch index or pattern. return es.indices.get_mapping(indexindex).body baseline_agent create_deep_agent( modelChatOpenAI( # any OpenAI-compatible endpoint; configure via LLM_* env vars base_urlos.environ.get(LLM_BASE_URL, https://openrouter.ai/api/v1), modelos.environ.get(LLM_MODEL, anthropic/claude-sonnet-4.5), api_keyos.environ[LLM_API_KEY], ), tools[esql_query, get_mapping], system_prompt( You are a research assistant with access to three Elasticsearch indices: beir-fiqa, beir-nfcorpus, and beir-scifact. You do NOT know which index is relevant for a given question. Use get_mapping to inspect an indexs description and fields, then query the most relevant one with esql_query. Ground your answer strictly in what the queries return. ), ) start time.perf_counter() result baseline_agent.invoke( { messages: [ { role: user, content: sys.argv[1], } ] } ) latency time.perf_counter() - start print(\n--- Tool calls ---) for m in result[messages]: if isinstance(m, AIMessage) and m.tool_calls: for tc in m.tool_calls: print(f [{tc[name]}] {str(tc[args])[:120]}) total sum( len(m.tool_calls) for m in result[messages] if isinstance(m, AIMessage) and m.tool_calls ) print(fTotal: {total}\n) print(--- Usage ---) input_tokens sum( (m.usage_metadata or {}).get(input_tokens, 0) for m in result[messages] if isinstance(m, AIMessage) and m.usage_metadata ) output_tokens sum( (m.usage_metadata or {}).get(output_tokens, 0) for m in result[messages] if isinstance(m, AIMessage) and m.usage_metadata ) print(fTokens: {input_tokens output_tokens} (input {input_tokens}, output {output_tokens})) print(fLatency: {latency:.2f}s\n) print(--- Answer ---) print(result[messages][-1].content)以下是一个修改后的示例可以运行相同的 agent但现在具备搜索 AI indices 并返回 KI 的能力# Example question: Is there scientific evidence that vitamin D supplementation prevents cancer? import os import sys import time from elasticsearch import Elasticsearch from langchain_core.messages import AIMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from deepagents import create_deep_agent from deepagents.backends.filesystem import FilesystemBackend if len(sys.argv) 2: sys.exit(fUsage: python {sys.argv[0]} your question) es Elasticsearch(os.environ[ES_URL], api_keyos.environ[ES_API_KEY]) tool def esql_query(query: str) - list[dict] | str: Execute an ES|QL query against Elasticsearch and return the matching rows. Args: query: A complete ES|QL query string, e.g. FROM beir-fiqa | LIMIT 5. Full-text search syntax: WHERE MATCH(field, value) — not field MATCH value. try: resp es.esql.query(queryquery, formatjson) cols [c[name] for c in resp[columns]] return [dict(zip(cols, row)) for row in resp[values]] except Exception as e: return fES|QL error: {e} backend FilesystemBackend(root_dir., virtual_modeFalse) agent create_deep_agent( modelChatOpenAI( # any OpenAI-compatible endpoint; configure via LLM_* env vars base_urlos.environ.get(LLM_BASE_URL, https://openrouter.ai/api/v1), modelos.environ.get(LLM_MODEL, anthropic/claude-sonnet-4.5), api_keyos.environ[LLM_API_KEY], ), tools[esql_query], skills[skills], backendbackend, system_prompt( You are a research assistant with access to several Elasticsearch indices. You do NOT know which index is relevant for a given question. Before searching, always use the query-ki skill with type index_metadata_entry to retrieve the routing profile for the right index, then query that index directly. Ground your answer strictly in what the queries return and cite the KI you used for routing. ), ) start time.perf_counter() result agent.invoke( { messages: [ { role: user, content: sys.argv[1], } ] } ) latency time.perf_counter() - start print(\n--- Tool calls ---) for m in result[messages]: if isinstance(m, AIMessage) and m.tool_calls: for tc in m.tool_calls: print(f [{tc[name]}] {str(tc[args])[:120]}) total sum( len(m.tool_calls) for m in result[messages] if isinstance(m, AIMessage) and m.tool_calls ) print(fTotal: {total}\n) print(--- Usage ---) input_tokens sum( (m.usage_metadata or {}).get(input_tokens, 0) for m in result[messages] if isinstance(m, AIMessage) and m.usage_metadata ) output_tokens sum( (m.usage_metadata or {}).get(output_tokens, 0) for m in result[messages] if isinstance(m, AIMessage) and m.usage_metadata ) print(fTokens: {input_tokens output_tokens} (input {input_tokens}, output {output_tokens})) print(fLatency: {latency:.2f}s\n) print(--- Answer ---) print(result[messages][-1].content)这个 agent 将始终查询 KI indices 来获取答案。Knowledge Indicators 能减少多少 agent token 使用量由于我们使用的是 agents这些脚本的结果具有非确定性。不过当我使用查询Is there scientific evidence that vitamin D supplementation prevents cancer?运行这些结果时两个 agents 得出了相同的结论但它们采用了不同的路径来得到这个结论指标基线无 AI Index使用 AI Index总工具调用次数128read_file调用次数02get_mapping调用次数30esql_query调用次数96查询的 indices 总数2在beir-scifact和beir-nfcorpus之间反复查询1beir-nfcorpus消耗的 tokens167,76392,711延迟39.58 秒36.15 秒答案有依据、正确有依据、正确KI 得出的答案既有依据又正确但一个有趣的数据点是使用 KI 时整体工具使用量和 token 使用量都更少延迟大致相当。以下是两条路径的并排对比在 Serverless 中运行完整的 AI Index pipeline本演练深入介绍了自行构建 AI indices 和 KI 的方式。在生产环境中你不会手动编写这些 workflowssetup agent 会生成它们而反馈循环则会根据 agent 自身的 traces 不断优化 KI。但其核心组件正是你刚刚使用的这些通过 workflow 提取 KI将它们存储在 AI Index 中然后通过 skill 检索它们。管理上下文对于构建相关且高效的 agentic search 系统至关重要而 AI indices 提供了一种利用 Elastic stack 的完整能力来管理这些上下文的方式。在 Serverless 中试用一下并在我们的 Discuss forums 或 Community Slack 中的#stack-kibanachannel 告诉我们你的想法原文Elasticsearch AI Indices: building context for agents | Elasticsearch Labs