Unable to use ConversationBufferWindowMemory in async mode with FastAPI

[omarabdelrhmaan]

I'm developing a service using langchain and FastAPI and have the chat history managed with SQLChatMessageHistory in async mode. and when I pass the memory using ConversationBufferWindowMemory I receive the following error

ValueError: Attempting to use a sync method in when async mode is turned on. Please use the corresponding async method instead.

Code snippet

 async_chat_history = SQLChatMessageHistory(
            session_id, connection=engine, async_mode=True
        )

        # TODO: CHECK HOW TO CONVERT THIS TO ASYNC
        memory = ConversationBufferWindowMemory(
            memory_key="chat_history",
            chat_memory=async_chat_history,
            return_messages=True,
            k=5,
            # input_key="user_input",
            # output_key="output_key",
        )

Agent Initialization

agent = create_openai_functions_agent(
          llm=llm, tools=tools, prompt=prompt_template
      )

      agent_executor = AgentExecutor(
          agent=agent,
          tools=tools,
          memory=memory,
          verbose=True,
          max_iterations=6,
          handle_parsing_errors=True,
          return_intermediate_steps=True,
          # callbacks=[token_callback],
      )

[dosubot[bot]]

Hey there, @omarabdelrhmaan! I'm here to help you out with any bugs, questions, or contributions you might have. Let's dive into this together!

To resolve the issue with using ConversationBufferWindowMemory in async mode with FastAPI, you need to ensure that all methods used are asynchronous. Here's how you can adjust your setup:

1. Use Async Methods: Make sure to use the async methods provided by ConversationBufferWindowMemory. This includes methods like aload_memory_variables, asave_context, and aclear.

2. Initialize Memory in Async Mode: Ensure that ConversationBufferWindowMemory is properly set up to handle async operations. You might need to explicitly call async methods when interacting with the memory.

Here's an example of how you might adjust your code:

from langchain.memory.buffer import ConversationBufferWindowMemory
from langchain_community.chat_message_histories.sql import SQLChatMessageHistory
from sqlalchemy.ext.asyncio import create_async_engine

# Create an async engine
async_engine = create_async_engine("sqlite+aiosqlite:///memory.db")

# Initialize SQLChatMessageHistory in async mode
async_chat_history = SQLChatMessageHistory(
    session_id=session_id,
    connection=async_engine,
    async_mode=True
)

# Initialize ConversationBufferWindowMemory in async mode
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    chat_memory=async_chat_history,
    return_messages=True,
    k=5
)

# Ensure you use async methods when interacting with memory
async def process_conversation(inputs, outputs):
    # Load memory variables asynchronously
    memory_variables = await memory.aload_memory_variables(inputs)
    
    # Save context asynchronously
    await memory.asave_context(inputs, outputs)
    
    return memory_variables

Make sure to replace any synchronous calls with their asynchronous counterparts to avoid the ValueError you're encountering. This should help you integrate ConversationBufferWindowMemory with FastAPI in async mode ^[1][2].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu I'm not managing the conversation Memory manually. i just setup it. could you please elaborate your solution and how it could be integrated with ConversationBufferWindowMemory

[dosubot[bot]]

To integrate ConversationBufferWindowMemory in async mode without manually managing the conversation memory, you can use the async methods provided by the class. Here's a more detailed example:

1. Initialize the Memory: Set up ConversationBufferWindowMemory with SQLChatMessageHistory in async mode.

2. Use Async Methods: Utilize aload_memory_variables and asave_context to handle memory operations asynchronously.

Here's how you can do it:

from fastapi import FastAPI, Request
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories.sql import SQLChatMessageHistory
from sqlalchemy.ext.asyncio import create_async_engine

app = FastAPI()

# Create an async engine
async_engine = create_async_engine("sqlite+aiosqlite:///memory.db")

# Initialize SQLChatMessageHistory in async mode
async_chat_history = SQLChatMessageHistory(
    session_id="your_session_id",
    connection=async_engine,
    async_mode=True
)

# Initialize ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    chat_memory=async_chat_history,
    return_messages=True,
    k=5
)

@app.post("/chat")
async def chat(request: Request):
    data = await request.json()
    user_input = data.get("input")

    # Load memory variables asynchronously
    memory_variables = await memory.aload_memory_variables({"input": user_input})

    # Process the input and generate a response (mocked here)
    response = "AI response to: " + user_input

    # Save the context asynchronously
    await memory.asave_context({"input": user_input}, {"output": response})

    return {"response": response, "memory": memory_variables}

This setup allows you to handle conversation memory automatically in async mode with FastAPI, using the provided async methods ^[1][2].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

how can this be integrated with agent_executor.ainvoke

here's my chatservice class

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}
        print(f"Agent Memory {agent_executor.memory}")
        memory = agent_executor.memory
        response = await agent_executor.ainvoke(
            {
                "input": query,
            }
        )
        return response

[dosubot[bot]]

To integrate ConversationBufferWindowMemory with agent_executor.ainvoke, ensure that your memory setup is asynchronous and compatible with the agent executor. Here's how you can adjust your ChatService class:

1. Ensure Asynchronous Memory: Make sure your ConversationBufferWindowMemory is initialized with async methods, as shown in previous examples.

2. Use ainvoke Properly: When calling ainvoke, ensure that the input format matches the expected structure.

Here's an example of how you might integrate it:

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}

        # Ensure memory is set up for async operations
        memory = agent_executor.memory

        # Use ainvoke with the correct input structure
        response = await agent_executor.ainvoke(
            {
                "question": query,  # Ensure this matches the expected input key
            }
        )
        return response

Make sure that the input key ("question") matches what ainvoke expects, as it requires a dictionary with at least a key "question" ^[1].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

below is my chat service class that handles the user query with ainvoke. how can the memory be managed async here

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}
        
        response = await agent_executor.ainvoke(
            {
                "input": query,
            }
        )
        return response

[dosubot[bot]]

To manage memory asynchronously in your ChatService class using ainvoke, ensure that the memory associated with the AgentExecutor is set up for async operations. Here's how you can adjust your class:

1. Ensure Asynchronous Memory Setup: Make sure the memory used by AgentExecutor is initialized with async capabilities, such as ConversationBufferWindowMemory with async methods.

2. Use Async Methods: When interacting with memory, use async methods like aload_memory_variables and asave_context.

Here's an example:

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}

        # Ensure memory is set up for async operations
        memory = agent_executor.memory

        # Load memory variables asynchronously if needed
        memory_variables = await memory.aload_memory_variables({"input": query})

        # Use ainvoke with the correct input structure
        response = await agent_executor.ainvoke(
            {
                "question": query,  # Ensure this matches the expected input key
                "chat_history": memory_variables.get("chat_history", [])
            }
        )

        # Optionally save context asynchronously
        await memory.asave_context({"input": query}, {"output": response})

        return response

Ensure that the input to ainvoke matches the expected structure, typically including "question" and "chat_history" keys ^[1].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

Still experiencing the same error

File "/lib/python3.8/site-packages/langchain_community/chat_message_histories/sql.py", line 340, in _make_sync_session
raise ValueError(
ValueError: Attempting to use a sync method in when async mode is turned on. Please use the corresponding async method instead.

[dosubot[bot]]

To resolve the ValueError you're encountering, ensure that all interactions with SQLChatMessageHistory and ConversationBufferWindowMemory are using async methods. Here's a checklist to help you troubleshoot:

1. Async Initialization: Confirm that SQLChatMessageHistory is initialized with async_mode=True.

2. Use Async Methods: Ensure that you are using async methods like aload_memory_variables, asave_context, and aclear for ConversationBufferWindowMemory.

3. AgentExecutor Setup: Verify that the AgentExecutor is correctly using the async memory. When calling ainvoke, ensure the input structure is correct and includes any necessary memory variables.

Here's a refined example:

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}

        # Ensure memory is set up for async operations
        memory = agent_executor.memory

        # Load memory variables asynchronously
        memory_variables = await memory.aload_memory_variables({"input": query})

        # Use ainvoke with the correct input structure
        response = await agent_executor.ainvoke(
            {
                "input": query,
                "chat_history": memory_variables.get("chat_history", [])
            }
        )

        # Save context asynchronously
        await memory.asave_context({"input": query}, {"output": response})

        return response

Double-check that all methods interacting with the memory are async, and ensure the input to ainvoke matches the expected format ^[1][2].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

Here's my full setup and still receiving the same error

engine = sessionmanager.get_engine()

       # Initialize memory with PostgreSQL persistence
       async_chat_history = SQLChatMessageHistory(
           session_id, connection=engine, async_mode=True
       )

       # TODO: CHECK HOW TO CONVERT THIS TO ASYNC
       memory = ConversationBufferWindowMemory(
           memory_key="chat_history",
           chat_memory=async_chat_history,
           return_messages=True,
           k=5,
           # input_key="user_input",
           # output_key="output_key",

       )
       
       agent = create_openai_functions_agent(
           llm=llm, tools=tools, prompt=prompt_template
       )

       agent_executor = AgentExecutor(
           agent=agent,
           tools=tools,
           memory=memory,
           verbose=True,
           max_iterations=6,
           handle_parsing_errors=True,
           return_intermediate_steps=True,
           # callbacks=[token_callback],
       )
       
       **Chat Service**

from langchain_core.messages import AIMessage, HumanMessage

from services.agent_service import AgentService

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}
        # Ensure memory is set up for async operations
        memory = agent_executor.memory

        print(f"Agent Memory {agent_executor.memory}")
        # Load memory variables asynchronously if needed
        memory_variables = await memory.aload_memory_variables({"input": query})
        # memory_variables = None
        print(f"Memory Variables: {memory_variables}")

        print(
            f"Memory Variables Chat history: {memory_variables.get('chat_history', [])}"
        )

        response = await agent_executor.ainvoke(
            {"input": query, "chat_history": memory_variables.get("chat_history", [])}
        )

        await memory.asave_context({"input": query}, {"output": response})
        return response

[dosubot[bot]]

To resolve the ValueError, ensure that all components are correctly set up for async operations. Here's a checklist and example adjustments:

1. Async Engine: Make sure you're using an async engine, such as create_async_engine from sqlalchemy.ext.asyncio.

2. Async Methods: Ensure all methods interacting with SQLChatMessageHistory and ConversationBufferWindowMemory are async.

3. Correct Initialization: Double-check that SQLChatMessageHistory is initialized with async_mode=True.

Here's a refined setup:

from sqlalchemy.ext.asyncio import create_async_engine
from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories.sql import SQLChatMessageHistory

# Create an async engine
async_engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")

# Initialize SQLChatMessageHistory in async mode
async_chat_history = SQLChatMessageHistory(
    session_id=session_id,
    connection=async_engine,
    async_mode=True
)

# Initialize ConversationBufferWindowMemory
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    chat_memory=async_chat_history,
    return_messages=True,
    k=5
)

class ChatService:
    async def chat(identifier: str, user_id: str, query: str):
        agent_executor = AgentService.get_agent(identifier)

        if not agent_executor:
            return {"error": f"No Agent found for identifier {identifier}"}

        # Ensure memory is set up for async operations
        memory = agent_executor.memory

        # Load memory variables asynchronously
        memory_variables = await memory.aload_memory_variables({"input": query})

        # Use ainvoke with the correct input structure
        response = await agent_executor.ainvoke(
            {
                "input": query,
                "chat_history": memory_variables.get("chat_history", [])
            }
        )

        # Save context asynchronously
        await memory.asave_context({"input": query}, {"output": response})

        return response

Ensure that your database connection string is correct for async operations, and all async methods are used consistently ^[1][2].

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

I've followed the same instructions you provided but still receiving the same error, any other thoughts to solve this or debug it?

[dosubot[bot]]

Double-check the following to resolve the ValueError:

1. Async Engine: Ensure you're using create_async_engine from sqlalchemy.ext.asyncio for your database connection.

2. Async Methods: Verify that all interactions with SQLChatMessageHistory and ConversationBufferWindowMemory use async methods like aget_messages, aadd_message, and aclear.

3. Correct Initialization: Confirm that SQLChatMessageHistory is initialized with async_mode=True.

4. Session Management: Ensure that your session management is using async sessions. Avoid any sync session calls.

5. Debugging: Use set_debug(True) to get detailed logs and identify where the sync method might be called.

Here's a quick checklist:

• Use create_async_engine for the database.
• Ensure all methods are async.
• Check for any accidental sync method calls.

If these steps don't resolve the issue, consider adding logging to trace where the sync method is being invoked.

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

@dosu

Where should set_debug(True) be placed?

[dosubot[bot]]

Place set_debug(True) at the beginning of your script, right after your imports. This will enable detailed logging throughout your application, helping you trace where the sync method might be called. Here's an example:

from sqlalchemy.ext.asyncio import create_async_engine
from langchain import set_debug

# Enable detailed logging
set_debug(True)

# Your existing setup
async_engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")

# Continue with your setup...

This should help you identify any sync method calls causing the error.

To continue talking to Dosu, mention @dosu.

[omarabdelrhmaan]

Hello,

In case someone was looking for a solution, I had to modify the langchain package, especially the SQLChatMessageHistory, ConversationBufferWindowMemory, and memory classes to process the async connections correctly. If you need the full implementation DM me and I'll be happy to help.

In the next days, I'll work on creating a pull request with these changes in the main repo of langchain.