ElasticsearchChatMessageHistory returns only 10 message(5 AI, 5 User) and that too oldest first

[sanjeet-toosi]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from langchain.memory import ElasticsearchChatMessageHistory

# i have added the credentials correctly 
chat_history = ElasticsearchChatMessageHistory(
                es_url = "",
                es_user ="",
                es_password = "",
                session_id = "1234",
                index = ""
            )

print("Number of messages in chat history {}".format(len(chat_history.messages)))

Description

I have 16 messages corresponding to session_id "1234" but it is returning only 10 and order of message is oldest first.

i want to get latest k messages.

System Info

langchain==0.2.6
langchain-cli==0.0.25
langchain-community==0.0.38
langchain-core==0.2.43
langchain-elasticsearch==0.2.2
langchain-openai==0.1.14
langchain-text-splitters==0.2.2

Platform: Mac

Python version: 3.12.3

[dnaby]

Hi @sanjeet-toosi !

I had a similar issue when working with ElasticsearchChatMessageHistory and managed to resolve it by extending the class to allow fetching more messages and ensuring flexibility in the number of messages retrieved.

The Problem

The issue with the 10-result limit in Elasticsearch is related to its default pagination mechanism. By default, Elasticsearch restricts the number of documents returned in a single query to 10, which is defined by the size parameter.

The Solution

To address these limitations, I extended the ElasticsearchChatMessageHistory class. This allowed me to customize the query and set a higher limit for the number of messages fetched. Here’s the code:

import logging
from typing import List
from langchain_core.messages import BaseMessage, messages_from_dict

# Define logger
logger = logging.getLogger(__name__)

class ExtendedElasticsearchChatMessageHistory(ElasticsearchChatMessageHistory):
    def __init__(self, index: str, session_id: str, fetch_size: int = 10000, **kwargs):
        """
        Args:
            index (str): Elasticsearch index name.
            session_id (str): Session ID for the chat history.
            fetch_size (int): Maximum number of messages to fetch.
            kwargs: Additional arguments for the parent class.
        """
        super().__init__(index=index, session_id=session_id, **kwargs)
        self.fetch_size = fetch_size

    @property
    def messages(self) -> List[BaseMessage]:
        """Retrieve all messages from Elasticsearch with a custom fetch size."""
        try:
            from elasticsearch import ApiError

            # Fetch up to `fetch_size` messages for the session
            result = self.client.search(
                index=self.index,
                query={"term": {"session_id": self.session_id}},
                sort="created_at:asc",  # Fetch in chronological order
                size=self.fetch_size,  # Custom fetch size
                from_=0
            )
        except ApiError as err:
            logger.error(f"Could not retrieve messages from Elasticsearch: {err}")
            raise err

        if result and len(result["hits"]["hits"]) > 0:
            items = [
                json.loads(document["_source"]["history"])
                for document in result["hits"]["hits"]
            ]
        else:
            items = []

        return messages_from_dict(items)

Fetching the Latest k Messages

To retrieve only the latest k messages, I wrote a utility function and leveraged this custom class.

def get_session_history(session_id: str) -> ExtendedElasticsearchChatMessageHistory:
    if session_id not in store:
        store[session_id] = ExtendedElasticsearchChatMessageHistory(
            es_url=os.environ.get('ES_URL'),
            es_user=os.environ.get('ES_USERNAME'),
            es_password=os.environ.get('ES_PASSWORD'),
            index=os.environ.get('ES_CHAT_HISTORY_INDEX'),
            session_id=session_id,
            fetch_size=10000
        )
    
    stored_messages = store[session_id].messages

    # Trim to the last `k` messages
    if len(stored_messages) >= last_k_messages:
        store[session_id].clear()
        store[session_id].add_messages(stored_messages[-last_k_messages:])
    
    chat_history = store[session_id]

    print(f"\nLength of chat history: {len(store[session_id].messages)}\n")
    
    return chat_history

You can refer to the LangChain documentation here to see how to use get_session_history in RunnableWithMessageHistory: LangChain Message History.

Hope this helps! 😊