Using async ChatMessageHistory with langchain-core.

[pprados]

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

import asyncio
import uuid
from pprint import pprint

import psycopg
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

from langchain_postgres import PostgresChatMessageHistory

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model

table_name = "chat_history"

async_connection = None


async def init_async_connection():
    global async_connection
    async_connection = await psycopg.AsyncConnection.connect(
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)


def aget_session_history(session_id: str) -> BaseChatMessageHistory:
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        async_connection=async_connection
    )


awith_message_history = RunnableWithMessageHistory(
    runnable,
    aget_session_history,
    input_messages_key="input",
    history_messages_key="history",
)


async def amain():
    await init_async_connection()  # Glups ! It's not a global initialization
    result = await awith_message_history.ainvoke(
        {"ability": "math", "input": "What does cosine mean?"},
        config={"configurable": {"session_id": str(uuid.uuid4())}},
    )
    pprint(result)


asyncio.run(amain())

Description

It's impossible to use and async ChatMessageHistory with langchain-core.

The ChatMessageHistory class is synchronous and doesn't have an async counterpart.
This is a problem because the RunnableWithMessageHistory class requires a ChatMessageHistory object to be passed to it. This means that it's impossible to use an async ChatMessageHistory with langchain-core.

I can't find any example of how to use it. I will try to create an example of how to use PostgresChatMessageHistory with async mode.

There are many problems:

• Bug in _exit_history()
• Bugs in PostgresChatMessageHistory and sync usage
• Bugs in PostgresChatMessageHistory and async usage

Bug in _exit_history()

In RunnableWithMessageHistory, the _exit_history() is called because the chain has | runnable.with_listeners(on_end=self._exit_history). This method is not async and it will raise an error. This method call add_messages() and not await aadd_messages().

import asyncio
import uuid
from pprint import pprint

import psycopg
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

from langchain_postgres import PostgresChatMessageHistory

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model

table_name = "chat_history"

async_connection = None


async def init_async_connection():
    global async_connection
    async_connection = await psycopg.AsyncConnection.connect(
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)


def aget_session_history(session_id: str) -> BaseChatMessageHistory:
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        async_connection=async_connection
    )


awith_message_history = RunnableWithMessageHistory(
    runnable,
    aget_session_history,
    input_messages_key="input",
    history_messages_key="history",
)


async def amain():
    await init_async_connection()  # Glups ! It's not a global initialization
    result = await awith_message_history.ainvoke(
        {"ability": "math", "input": "What does cosine mean?"},
        config={"configurable": {"session_id": str(uuid.uuid4())}},
    )
    pprint(result)


asyncio.run(amain())

Result

Error in RootListenersTracer.on_chain_end callback: ValueError('Please initialize the PostgresChatMessageHistory with a sync connection or use the aadd_messages method instead.')
AIMessage(content='Cosine is a trigonometric function that represents the ratio of the adjacent side to the hypotenuse in a right triangle.', response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 33, 'total_tokens': 59}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-a00fc81a-1844-4a47-98fa-7a30d6e51228-0')

Bugs in PostgresChatMessageHistory and sync usage

In PostgresChatMessageHistory, the design is problematic.
Langchain, with LCEL, is declarative programming. You have to declare a chain in global variables, then invoke them when necessary. This is how langserv is able to publish interfaces with add_route().

For optimization reasons, PostgresChatMessageHistory wishes to recycle connections. The class provides a constructor which accepts a sync_connection parameter. However, it is not possible to have a global connection, in order to reuse it when implementing get_session_history().

sync_connection = psycopg.connect(  # ERROR: A connection is not reentrant!
    user="postgres",
    password="password_postgres",
    host="localhost",
    port=5432)


def get_session_history(session_id: str) -> BaseChatMessageHistory:
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        sync_connection=sync_connection
    )

A connection is not reentrant! You can't use the same connection in multiple threads. But, the design of langchain-postgres is to have a global connection. This is a problem.

The alternative is to create a new connection each time you need to access the database.

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    sync_connection = psycopg.connect(  # ERROR: A connection is not reentrant!
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        sync_connection=sync_connection
    )

Then, why accept only a connection and not an engine? The engine is a connection pool.

Bugs in PostgresChatMessageHistory and async usage

If we ignore the problem mentioned above with _exit_history(), there are even more difficulties. It's not easy to initialize a global async connection. Because it's must be initialized in an async function.

async_connection = None

async def init_async_connection(): # Where call this function?
    global async_connection
    async_connection = await psycopg.AsyncConnection.connect(
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)

And, it's not possible to call init_async_connection() in get_session_history(). get_session_history() is not async. It's a problem.

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    async_connection = await psycopg.AsyncConnection.connect( # ERROR: 'await' outside async function
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        async_connection=async_connection
    )

It is therefore currently impossible to implement session history correctly in asynchronous mode.
Either you use a global connection, but that's not possible, or you open the connection in ̀get_session_history()`, but that's impossible.

The only solution is to completely break the use of LCEL, by building the chain just after the connection is opened. It's still very strange. To publish it with langserv, you need to use a RunnableLambda.

async def async_lambda_history(input:Dict[str,Any],config:Dict[str,Any]):
    async_connection = await psycopg.AsyncConnection.connect(
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)

    def _get_session_history(session_id: str) -> BaseChatMessageHistory:
        return PostgresChatMessageHistory(
            table_name,
            session_id,
            async_connection=async_connection
        )

    awith_message_history = RunnableWithMessageHistory(
        runnable,
        _get_session_history,
        input_messages_key="input",
        history_messages_key="history",
    )
    result = await awith_message_history.ainvoke(
        input,
        config=config,
    )
    pprint(result)

def nope():
    pass

lambda_chain=RunnableLambda(func=nope,afunc=async_lambda_history)

async def lambda_amain():
    result = await lambda_chain.ainvoke(
        {"ability": "math", "input": "What does cosine mean?"},
        config={"configurable": {"session_id": str(uuid.uuid4())}},
    )
    pprint(result)

asyncio.run(lambda_amain())

It's a very strange way to use langchain.

But a good use of langchain in a website consists precisely in using only asynchronous approaches. This must include history management.

What is the best solution?

System Info

langchain==0.1.20
langchain-community==0.0.38
langchain-core==0.2.1
langchain-openai==0.1.7
langchain_postgres==0.0.6
langchain-rag==0.1.46
langchain-text-splitters==0.0.1

Ubuntu

[dosubot[bot]]

To use an asynchronous ChatMessageHistory with langchain-core, you need to override the necessary methods to provide truly asynchronous behavior. Here is an example implementation using InMemoryChatMessageHistory:

from typing import List, Sequence
from langchain_core.messages import BaseMessage
from langchain_core.runnables import run_in_executor
from langchain_core.pydantic_v1 import BaseModel, Field

class InMemoryChatMessageHistory(BaseChatMessageHistory, BaseModel):
    messages: List[BaseMessage] = Field(default_factory=list)

    async def aget_messages(self) -> List[BaseMessage]:
        return self.messages

    def add_message(self, message: BaseMessage) -> None:
        self.messages.append(message)

    async def aadd_messages(self, messages: Sequence[BaseMessage]) -> None:
        await run_in_executor(None, self.add_messages, messages)

    def clear(self) -> None:
        self.messages = []

    async def aclear(self) -> None:
        await run_in_executor(None, self.clear)

To manage the message history more effectively, you can use the RunnableWithMessageHistory class. Here is an example of how to use it with an in-memory ChatMessageHistory:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

with_message_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

with_message_history.invoke(
    {"ability": "math", "input": "What does cosine mean?"},
    config={"configurable": {"session_id": "abc123"}},
)

Addressing Your Specific Issues

1. Bug in _exit_history():

   • The _exit_history() method in RunnableWithMessageHistory should be updated to handle asynchronous operations. You can override this method to use await aadd_messages().

2. Bugs in PostgresChatMessageHistory and sync usage:

   • Ensure that the PostgresChatMessageHistory class is properly initialized with an asynchronous connection. You may need to modify the class to support asynchronous operations fully.

3. Bugs in PostgresChatMessageHistory and async usage:

   • Initialize the async connection within an async context and pass it to the PostgresChatMessageHistory instance.

Here is an updated version of your example code with these considerations:

import asyncio
import uuid
from pprint import pprint

import psycopg
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

from langchain_postgres import PostgresChatMessageHistory

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're an assistant who's good at {ability}. Respond in 20 words or fewer",
        ),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
runnable = prompt | model

table_name = "chat_history"

async_connection = None

async def init_async_connection():
    global async_connection
    async_connection = await psycopg.AsyncConnection.connect(
        user="postgres",
        password="password_postgres",
        host="localhost",
        port=5432)

async def aget_session_history(session_id: str) -> BaseChatMessageHistory:
    return PostgresChatMessageHistory(
        table_name,
        session_id,
        async_connection=async_connection
    )

awith_message_history = RunnableWithMessageHistory(
    runnable,
    aget_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

async def amain():
    await init_async_connection()
    result = await awith_message_history.ainvoke(
        {"ability": "math", "input": "What does cosine mean?"},
        config={"configurable": {"session_id": str(uuid.uuid4())}},
    )
    pprint(result)

asyncio.run(amain())

This implementation ensures that the asynchronous methods (aget_messages, aadd_messages, and aclear) are properly managed, providing a way to handle session history in an asynchronous mode.

Details

Sources

• create_retrieval_chain does not store RedisChatMessageHistory
• libs/core/langchain_core/chat_history.py
• docs/docs/how_to/message_history.ipynb
• libs/community/langchain_community/tools/sql_database/tool.py
• libs/langchain/langchain/indexes/_sql_record_manager.py
• libs/core/tests/unit_tests/runnables/test_history.py
• libs/langchain/langchain/utilities/init.py
• docs/docs/how_to/callbacks_async.ipynb
• libs/core/tests/unit_tests/chat_history/test_chat_history.py
• libs/community/langchain_community/chat_message_histories/postgres.py
• libs/core/langchain_core/runnables/history.py
• docs/docs/how_to/qa_chat_history_how_to.ipynb

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[hamdan-27]

Hi @pprados , hope you are well. I am having the same issue with my project. Have you been able to find a solution yet? Thank you

[pprados]

Use langchain-postgres

[hamdan-27]

I am using langchain_postgres, but don't have an idea how to initialise the Async connection to be used with PosgresChatMessageHistory. Could you show me a snippet if possible? Thanks

[pprados]

Read this blog post ;-)

[hamdan-27]

Thanks! Saw your PR for the async support on SQLChatMessageHistory which suits my use case best. You're a legend!