Building context in Elasticsearch: how AI Indices power smarter agents using fewer tokens

{
  "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."
          }
        }
      }
    }
  }
}

PUT ai-index-idx-my-corpus

GET ai-index-idx-my-corpus/_mapping

{
  "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"
        }
      }
    }
  }
}

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
            field's `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: "; " }}.

FROM ai-index-idx-my-corpus
    | WHERE type == "index_metadata_entry"
    | KEEP title, content, description, attributes, tags
    | LIMIT 10

---
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 user's 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.

# 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(f'Usage: python {sys.argv[0]} "your question"')

es = Elasticsearch(os.environ["ES_URL"], api_key=os.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(query=query, format="json")
        cols = [c["name"] for c in resp["columns"]]
        return [dict(zip(cols, row)) for row in resp["values"]]
    except Exception as e:
        return f"ES|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(index=index).body


baseline_agent = create_deep_agent(
    model=ChatOpenAI(  # any OpenAI-compatible endpoint; configure via LLM_* env vars
        base_url=os.environ.get("LLM_BASE_URL", "https://openrouter.ai/api/v1"),
        model=os.environ.get("LLM_MODEL", "anthropic/claude-sonnet-4.5"),
        api_key=os.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 index's 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(f"Total: {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(f"Tokens: {input_tokens + output_tokens} (input {input_tokens}, output {output_tokens})")
print(f"Latency: {latency:.2f}s\n")

print("--- Answer ---")
print(result["messages"][-1].content)

# 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(f'Usage: python {sys.argv[0]} "your question"')

es = Elasticsearch(os.environ["ES_URL"], api_key=os.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(query=query, format="json")
        cols = [c["name"] for c in resp["columns"]]
        return [dict(zip(cols, row)) for row in resp["values"]]
    except Exception as e:
        return f"ES|QL error: {e}"


backend = FilesystemBackend(root_dir=".", virtual_mode=False)

agent = create_deep_agent(
    model=ChatOpenAI(  # any OpenAI-compatible endpoint; configure via LLM_* env vars
        base_url=os.environ.get("LLM_BASE_URL", "https://openrouter.ai/api/v1"),
        model=os.environ.get("LLM_MODEL", "anthropic/claude-sonnet-4.5"),
        api_key=os.environ["LLM_API_KEY"],
    ),
    tools=[esql_query],
    skills=["skills"],
    backend=backend,
    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(f"Total: {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(f"Tokens: {input_tokens + output_tokens} (input {input_tokens}, output {output_tokens})")
print(f"Latency: {latency:.2f}s\n")

print("--- Answer ---")
print(result["messages"][-1].content)