Elasticsearch Vector Database

The example uses OpenAI-backed embeddings or models. Set your key before running it:

export OPENAI_API_KEY="your-api-key"

Setup

uv pip install -U "elasticsearch[async]" pypdf openai sqlalchemy agno

Run Elasticsearch locally with Docker:

docker run -d \
  --name elasticsearch \
  -p 9200:9200 \
  -p 9300:9300 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "ES_JAVA_OPTS=-Xms1g -Xmx1g" \
  docker.elastic.co/elasticsearch/elasticsearch:9.1.0

The elasticsearch client's major version must match your cluster's. An 8.x cluster rejects a 9.x client.

Example

elasticsearch_db.py

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.elasticsearch import Elasticsearch

knowledge = Knowledge(
    name="Elasticsearch Recipe Knowledge Base",
    description="This is a knowledge base that uses Elasticsearch",
    vector_db=Elasticsearch(
        index_name="recipe",
    ),
)

knowledge.insert(
    name="Thai Recipes",
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    metadata={"doc_type": "recipe_book"},
)

agent = Agent(
    knowledge=knowledge,
    # Enable the agent to search the knowledge base
    search_knowledge=True,
    # A db is required for the agent to read its own chat history
    db=SqliteDb(db_file="tmp/elasticsearch.db"),
    # Enable the agent to read the chat history
    read_chat_history=True,
)
agent.print_response("How to make Thai curry?")

Async Support ⚡

Elasticsearch also supports asynchronous operations with ainsert() and aprint_response().

async_elasticsearch_db.py

import asyncio

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.elasticsearch import Elasticsearch

vector_db = Elasticsearch(
    index_name="recipe_async",
)

knowledge_base = Knowledge(
    vector_db=vector_db,
)

agent = Agent(knowledge=knowledge_base)


async def main():
    await knowledge_base.ainsert(
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
    )

    # Create and use the agent
    await agent.aprint_response("How to make Tom Kha Gai", markdown=True)

    # The async client holds an aiohttp session that Python will not close for you:
    # skip this and the script exits with "ResourceWarning: Unclosed connector" and a
    # leaked socket.
    await vector_db.async_close()


if __name__ == "__main__":
    asyncio.run(main())

Call async_close() when you finish to close the async client's connection.

Elasticsearch Params

Name	Type	Default	Description
`index_name`	`str`	-	Name of the Elasticsearch index.
`url`	`Union[str, List[str]]`	`"http://localhost:9200"`	Cluster URL. The scheme selects TLS, and credentials embedded in the URL are applied. Pass a list to spread requests across nodes. Ignored when `cloud_id` is set.
`dimension`	`Optional[int]`	`None`	Embedding dimensions. Defaults to the embedder's `dimensions`.
`embedder`	`Optional[Embedder]`	`OpenAIEmbedder()`	Embedder for documents and queries.
`distance`	`Distance`	`Distance.cosine`	Distance metric, mapped to an Elasticsearch similarity. Ignored when `similarity` is set.
`similarity`	`Optional[Similarity]`	`None`	Similarity for the `dense_vector` field: `cosine`, `l2_norm`, `dot_product` or `max_inner_product`. Takes precedence over `distance`.
`search_type`	`SearchType`	`SearchType.vector`	Default search type: `vector`, `keyword` or `hybrid`.
`hybrid_strategy`	`HybridStrategy`	`HybridStrategy.boost`	How hybrid search combines its vector and keyword results: `boost` or `rrf`.
`api_key`	`Optional[str]`	`None`	Elasticsearch API key.
`cloud_id`	`Optional[str]`	`None`	Elastic Cloud ID. Takes precedence over `url`.
`basic_auth`	`Optional[tuple]`	`None`	`(username, password)` for HTTP basic authentication.
`verify_certs`	`bool`	`True`	Verify the cluster's TLS certificate.
`ca_certs`	`Optional[str]`	`None`	Path to a CA bundle for verifying the cluster's certificate.
`timeout`	`int`	`30`	Request timeout in seconds.
`max_retries`	`int`	`10`	Maximum retry attempts.
`retry_on_timeout`	`bool`	`True`	Retry requests that time out.
`num_candidates`	`Optional[int]`	`None`	Neighbors each shard considers before picking the top results. Defaults to 10 times the limit, at least 50 and at most 10000.
`index_settings`	`Optional[Dict[str, Any]]`	`None`	Settings sent when the index is created. Defaults to `{"index": {"number_of_replicas": 0}}`.
`reranker`	`Optional[Reranker]`	`None`	Deprecated. Pass the reranker to `Knowledge` instead.
`id`	`Optional[str]`	`None`	Vector database ID. Derived from the URL or cloud ID and the index name when omitted.
`name`	`Optional[str]`	`None`	Name of the vector database.
`description`	`Optional[str]`	`None`	Description of the vector database.
`**kwargs`	`Any`	-	Extra arguments passed to the `elasticsearch.Elasticsearch` and `elasticsearch.AsyncElasticsearch` clients.

Setup

Example

Async Support ⚡

Elasticsearch Params

On this page