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

[pprados]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).

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()
    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())

Error Message and Stack Trace (if applicable)

Error in RootListenersTracer.on_chain_end callback: ValueError('Please initialize the PostgresChatMessageHistory with a sync connection or use the aadd_messages method instead.')

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.

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

[pprados]

The solution is to use

SQLChatMessageHistory( 
                session_id=session_id,
                connection_string=history_url, # Open a new connection each time
            )
``̀

[eyurtsev]

@pprados

The solution is to use

SQLChatMessageHistory( 
                session_id=session_id,
                connection_string=history_url, # Open a new connection each time
            )
``̀

I assume this is a work-around around one of the issues? I think you identified a number of separate issues that need to be addressed? Is that correct?

[pprados]

Yes. This is a work-around around one of the issues in PostgresChatMessageHistory() (see here.
For the problem with _exit_history(), is here, for each async chat message history implementation, with a check of usage.

[pprados]

@eyurtsev

I'm trying to re-implement the SQLChatMessageHistory class, to really use the asynchronous approach, and there's the same problem. The add_message() method is invoked by RunnableWithMessageHistory._exit_history().

An implementation workaround is to use this:

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # The methode RunnableWithMessageHistory._exit_history() call
        #  add_message method by mistake and not aadd_message.
        # See https://github.com/langchain-ai/langchain/discussions/22022
        if self.async_mode:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(self.aadd_message(smessages))
        else:
            with self.session_maker() as session:
                for message in messages:
                    session.add(self.converter.to_sql_model(message, self.session_id))
                session.commit()

I'll propose a PR for this, improving the interface for accepting an (Async)Engine, in addition to the connection string. That way, it will be possible to use the connection pool.

I'm going to use an approach similar to the PR I'm proposing to finally make langchain use resilient.

@baskaryan I'd like you to take the time to validate this PR, which prohibits me from publishing an article explaining how to make langchain use resilient.
This PR is just the beginning of the story
Without it, it's simply not possible in the current state of implementation. There are situations where a crash at the wrong time completely breaks applications (During the import of documents into the vector database, during the saving of the history which is done message by message, and not the entire conversation in a single transaction, etc.).

[waadarsh]

is this resolved?

[pprados]

My PR fixed this bug.

[alew3]

await init_async_connection()

@pprados do you have sample code of how to use it correctly with async with your fix?