AsyncElasticsearchChatMessageHistory

Setup:

Install langchain_elasticsearch and start Elasticsearch locally using the start-local script.

pip install -qU langchain_elasticsearch
curl -fsSL https://elastic.co/start-local | sh

This will create an elastic-start-local folder. To start Elasticsearch and Kibana:

cd elastic-start-local
./start.sh

Elasticsearch will be available at http://localhost:9200. The password for the elastic user and API key are stored in the .env file in the elastic-start-local folder.

Initialize the ElasticsearchChatMessageHistory instance.

Instantiate:

from langchain_elasticsearch import ElasticsearchChatMessageHistory

history = ElasticsearchChatMessageHistory(
    index="chat-history",
    session_id="user-123",
    es_url="http://localhost:9200"
)

Instantiate with API key (URL):

from langchain_elasticsearch import ElasticsearchChatMessageHistory

history = ElasticsearchChatMessageHistory(
    index="chat-history",
    session_id="user-123",
    es_url="http://localhost:9200",
    es_api_key="your-api-key"
)

Instantiate with username/password (URL):

from langchain_elasticsearch import ElasticsearchChatMessageHistory

history = ElasticsearchChatMessageHistory(
    index="chat-history",
    session_id="user-123",
    es_url="http://localhost:9200",
    es_user="elastic",
    es_password="password"
)

If you want to use a cloud hosted Elasticsearch instance, you can pass in the es_cloud_id argument instead of the es_url argument.

Instantiate from cloud (with API key):

from langchain_elasticsearch import ElasticsearchChatMessageHistory

history = ElasticsearchChatMessageHistory(
    index="chat-history",
    session_id="user-123",
    es_cloud_id="<cloud_id>",
    es_api_key="your-api-key"
)

You can also connect to an existing Elasticsearch instance by passing in a pre-existing Elasticsearch connection via the client argument.

Instantiate from existing connection:

from langchain_elasticsearch import ElasticsearchChatMessageHistory
from elasticsearch import Elasticsearch

client = Elasticsearch("http://localhost:9200")
history = ElasticsearchChatMessageHistory(
    index="chat-history",
    session_id="user-123",
    client=client
)

Add messages:

from langchain_core.messages import HumanMessage, AIMessage

history.add_message(HumanMessage(content="Hello!"))
history.add_message(AIMessage(content="Hi there! How can I help?"))

Get messages:

messages = history.messages
for msg in messages:
    print(f"{msg.type}: {msg.content}")

Clear history:

history.clear()

For synchronous applications, use the ElasticsearchChatMessageHistory class. For asynchronous applications, use the AsyncElasticsearchChatMessageHistory class.