Add Support for Asynchronous Connection Pooling in PostgresChatMessageHistory

[shamspias]

Currently, the PostgresChatMessageHistory class in the langchain-postgres package does not support asynchronous connection pooling using psycopg_pool.AsyncConnectionPool. This limitation leads to inefficiencies and potential resource exhaustion in high-load asynchronous applications, as each database operation may open a new connection instead of reusing existing ones.

When attempting to use PostgresChatMessageHistory with an AsyncConnectionPool, we encounter errors due to missing parameters and unexpected arguments. Specifically, the __init__ method does not accept a conn_pool parameter, and the parameter order causes issues with positional-only parameters.

What We Need:

• Add support for asynchronous connection pooling by modifying the PostgresChatMessageHistory class to accept an AsyncConnectionPool instance.
• Adjust the __init__ method to include a conn_pool parameter and modify the parameter order to avoid positional-only parameter issues.
• Update asynchronous methods (aget_messages, aadd_messages, etc.) to utilize the connection pool when provided.
• Ensure backward compatibility by maintaining existing functionality for sync_connection and async_connection.
• Add unit tests to verify the new functionality with the connection pool.
• Update documentation and README to reflect the changes and provide examples of using the class with an AsyncConnectionPool.

Code to Reproduce Error:

Attempting to use PostgresChatMessageHistory with an AsyncConnectionPool leads to errors.

import uuid
import asyncio
from langchain_postgres.chat_message_histories import PostgresChatMessageHistory
from psycopg_pool import AsyncConnectionPool
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

# Initialize the connection pool
pool = AsyncConnectionPool(conninfo="postgresql://user:password@host:port/dbname")

async def main():
    table_name = "chat_history"
    session_id = str(uuid.uuid4())

    # Attempt to create PostgresChatMessageHistory with conn_pool
    chat_history = PostgresChatMessageHistory(
        table_name=table_name,
        session_id=session_id,
        conn_pool=pool,
    )

    # Add messages to the chat history
    await chat_history.aadd_messages([
        SystemMessage(content="System message"),
        AIMessage(content="AI response"),
        HumanMessage(content="Human message"),
    ])

    # Retrieve messages from the chat history
    messages = await chat_history.aget_messages()
    print(messages)

# Run the async main function
asyncio.run(main())

Errors Encountered:

TypeError: __init__() got an unexpected keyword argument 'conn_pool'
TypeError: Parameter 'table_name' unfilled
TypeError: Parameter 'session_id' unfilled

Explanation:

• Unexpected Keyword Argument 'conn_pool': The PostgresChatMessageHistory class does not currently accept a conn_pool parameter, leading to this error.
• Parameter Unfilled Errors: The __init__ method uses positional-only parameters for table_name and session_id, causing issues when they are passed as keyword arguments.

Request:

Please update the PostgresChatMessageHistory class to support asynchronous connection pooling and adjust the parameter handling to resolve these errors. This enhancement will improve efficiency and resource management in asynchronous applications using the langchain-postgres package.

Related Issues

• How to use PostgresChatMessageHistory with async connections when using RunnableWithMessageHistory? #122