How to pass chat history to the arun method?

[SarahSohana27]

I have the following code where I want to pass the chat memory to a custom chat pipeline. But it's giving the error:
Output dict must have exactly one key.

below is the code? is it the correct way of passing the chat history to the arun method?
```
async def achat(
self, message: str, chat_history: Optional[List[ChatMessage]] = None
):
try:
# Retrieve chat history
chat_history = self._memory.get()
# Add the new message to the chat history
chat_history.append(ChatMessage(role="user", content=message))
# prepare inputs
chat_history_str = "\n".join([str(x) for x in chat_history])
# Process the message with the query pipeline
chat_response = await self.query_pipeline.arun(query=message, chat_history=chat_history_str)
# Add the response to the chat history
chat_history.append(ChatMessage(role="assistant", content=chat_response))
# Update the memory with the new chat history
self._memory.chat_store.set_messages(
self._memory.chat_store_key, chat_history
)
print('chat history', chat_history)
return chat_response
except Exception as e:
return "An error occurred while processing your query: {}".format(e)

[dosubot[bot]]

I found a similar discussion on how to add chat memory in a custom chat engine, which might be helpful for your situation <sup>[1]</sup>.

Regarding your code, the error "Output dict must have exactly one key" suggests that the arun method is expecting a specific structure for its input. The arun method processes keyword arguments (**kwargs) into a JSON payload, so it's important to ensure that the input format matches what the method expects.

In your code, you are passing chat_history as a string. Instead, you should ensure that the arun method is designed to handle this input format. If the method expects a dictionary with specific keys, you need to structure your input accordingly. Here's a suggestion:

1. Check the expected input format for the arun method. It might require a dictionary with specific keys rather than a single string.

2. Modify the input to match the expected format. For example, if arun expects a dictionary with a key like {"chat_history": chat_history_str}, ensure you pass it in that format.

Here's a possible modification:

async def achat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
):
    try:
        # Retrieve chat history
        chat_history = self._memory.get()
        # Add the new message to the chat history
        chat_history.append(ChatMessage(role="user", content=message))
        # Prepare inputs as a dictionary
        inputs = {"query": message, "chat_history": chat_history}
        # Process the message with the query pipeline
        chat_response = await self.query_pipeline.arun(**inputs)
        # Add the response to the chat history
        chat_history.append(ChatMessage(role="assistant", content=chat_response))
        # Update the memory with the new chat history
        self._memory.chat_store.set_messages(
            self._memory.chat_store_key, chat_history
        )
        print('chat history', chat_history)
        return chat_response
    except Exception as e:
        return "An error occurred while processing your query: {}".format(e)

Ensure that the arun method is designed to handle the inputs dictionary correctly. Adjust the keys in the inputs dictionary to match what arun expects.

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

The arun method in the query_pipeline is designed to accept keyword arguments (**kwargs) that are serialized into a JSON payload. The method expects the input to be structured in a way that it can process, typically as a dictionary with specific keys. In your CustomChatEngine class, you are correctly preparing the inputs as a dictionary with keys like query and chat_history.

Here's a refined version of your achat method, ensuring that the inputs are structured correctly for the arun method:

async def achat(
    self, message: str, chat_history: Optional[List[ChatMessage]] = None
):
    try:
        # Retrieve chat history
        chat_history = self._memory.get()
        # Add the new message to the chat history
        chat_history.append(ChatMessage(role="user", content=message))
        # Prepare inputs as a dictionary
        inputs = {"query": message, "chat_history": chat_history}
        # Process the message with the query pipeline
        chat_response = await self.query_pipeline.arun(**inputs)
        # Add the response to the chat history
        chat_history.append(ChatMessage(role="assistant", content=chat_response))
        # Update the memory with the new chat history
        self._memory.chat_store.set_messages(
            self._memory.chat_store_key, chat_history
        )
        print('chat history', chat_history)
        return chat_response
    except Exception as e:
        return "An error occurred while processing your query: {}".format(e)

Ensure that the arun method in your query_pipeline is designed to handle the inputs dictionary with the keys you are providing. If the method expects a different structure, adjust the keys in the inputs dictionary accordingly. The error "Output dict must have exactly one key" suggests that the method might be expecting a specific key or a single key-value pair in the output, so verify the expected output format as well <sup>[1][2]</sup>.

To continue talking to Dosu, mention @dosu.

[SarahSohana27]

@dosu self._memory.chat_store.set_messages() what is chat_store here? how do I access chat_store?

[dosubot[bot]]

chat_store is likely a component of your memory management system that handles storing and retrieving chat messages. It acts as a storage mechanism for maintaining the chat history. To access chat_store, you should ensure that your BaseMemory class or the specific memory implementation you are using has a chat_store attribute or method.

Here's how you might typically access it:

1. Check the Memory Class: Look into the implementation of your BaseMemory class to see if it has a chat_store attribute or method. This is where chat_store would be defined.

2. Initialization: Ensure that when you initialize your CustomChatEngine, the memory parameter is an instance of a class that includes chat_store.

3. Usage: In your code, self._memory.chat_store.set_messages() is used to update the stored chat history. Make sure that chat_store has methods like set_messages and get_messages to handle storing and retrieving messages.

If you don't have access to the implementation of BaseMemory, you might need to consult the documentation or codebase where BaseMemory is defined to understand how chat_store is structured and accessed.

To continue talking to Dosu, mention @dosu.

[SarahSohana27]

@dosu can you redirect me to the codebase of BaseChatEngine and BaseMemory?

[dosubot[bot]]

The BaseChatEngine class is defined in the file llama-index-legacy/llama_index/legacy/chat_engine/types.py <sup>[1]</sup>. The BaseMemory class is defined in the file llama-index-legacy/llama_index/legacy/memory/types.py <sup>[2]</sup>.

To continue talking to Dosu, mention @dosu.

[SarahSohana27]

@dosu can you fix this code to use the chat memory while running a query?

from llama_index.legacy.chat_engine.types import (
    AgentChatResponse,
    BaseChatEngine,
    StreamingAgentChatResponse,
    ChatMessage,
)
from typing import AsyncGenerator, Generator, List, Optional, Union
from llama_index.core.memory import BaseMemory, ChatMemoryBuffer
from llama_index.core.storage.chat_store import SimpleChatStore
import uuid



class CustomChatEngine(BaseChatEngine):
    def __init__(self, query_pipeline, *args, **kwargs):
        chat_store = SimpleChatStore()
        memory = ChatMemoryBuffer.from_defaults(
            token_limit=15000, chat_store=chat_store, chat_store_key=uuid.uuid4().hex
        )

        super().__init__(*args, **kwargs)
        self.query_pipeline = query_pipeline
        self._memory = memory
        print(query_pipeline, self._memory)

    def chat(self, message, chat_history):
        try:
            # Retrieve chat history
            chat_history = self._memory.get_messages()
           
            # Process the message with the query pipeline
            response = self.query_pipeline.run(query=message, chat_history= chat_history)

            print(response)

             # Add the new message to the chat history
            chat_history.append(ChatMessage(role="user", content=message))
            # Add the response to the chat history
            chat_history.append(ChatMessage(role="assistant", content=response))

            # Update the memory with the new chat history
            self._memory.set_messages(
                self._memory.chat_store_key, chat_history
            )

            return response
        except Exception as e:
            return "An error occurred while processing your query: {}".format(e)

    def reset(self) -> None:
        # Reset conversation state if applicable
        self._memory.set_messages(self._memory.chat_store_key, [])

    async def achat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ):
        try:
            # Retrieve chat history
            chat_history = self._memory.get_messages()
            print('chat history before', chat_history)
            # Prepare inputs as a dictionary
            inputs = {"query": message, "chat_history": chat_history}
            # Process the message with the query pipeline
            chat_response = await self.query_pipeline.arun(**inputs)

            # chat_response = await self.query_pipeline.arun(query=message, chat_history=chat_history_str)
            print('response', chat_response)

            # Add the new message to the chat history
            chat_history.append(ChatMessage(role="user", content=message))

            # Add the response to the chat history
            chat_history.append(ChatMessage(role="assistant", content=chat_response))
            
            # Update the memory with the new chat history
            self._memory.set_messages(
                self._memory.chat_store_key, chat_history
            )
            print('chat history', chat_history)
            return chat_response
        except Exception as e:
            return "An error occurred while processing your query: {}".format(e)

    def stream_chat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> StreamingAgentChatResponse:
        # Implement your streaming chat logic here
        return StreamingAgentChatResponse(response="Streaming response")

    async def astream_chat(
        self, message: str, chat_history: Optional[List[ChatMessage]] = None
    ) -> StreamingAgentChatResponse:
        # Implement your async streaming chat logic here
        return StreamingAgentChatResponse(response="Async streaming response")

    @property
    def chat_history(self) -> List[ChatMessage]:
        # Return chat history if applicable
        return []

[dosubot[bot]]

To fix the code and properly use chat memory while running a query, ensure that you are correctly retrieving and updating the chat history in both the chat and achat methods. Here are the necessary adjustments:

1. Retrieve Chat History: Use self._memory.get() to retrieve the chat history.

2. Update Chat Memory: After processing the message and generating a response, update the chat memory using self._memory.put() for synchronous methods and await self._memory.aput() for asynchronous methods.

Here's the corrected code:

class CustomChatEngine(BaseChatEngine):
    def __init__(self, query_pipeline, *args, **kwargs):
        chat_store = SimpleChatStore()
        memory = ChatMemoryBuffer.from_defaults(
            token_limit=15000, chat_store=chat_store, chat_store_key=uuid.uuid4().hex
        )

        super().__init__(*args, **kwargs)
        self.query_pipeline = query_pipeline
        self._memory = memory
        print(query_pipeline, self._memory)

    def chat(self, message, chat_history):
        try:
            # Retrieve chat history
            chat_history = self._memory.get()
           
            # Process the message with the query pipeline
            response = self.query_pipeline.run(query=message, chat_history=chat_history)

            print(response)

            # Add the new message to the chat history
            self._memory.put(ChatMessage(role="user", content=message))
            # Add the response to the chat history
            self._memory.put(ChatMessage(role="assistant", content=response))

            return response
        except Exception as e:
            return "An error occurred while processing your query: {}".format(e)

    def reset(self) -> None:
        # Reset conversation state if applicable
        self._memory.reset()

    async def achat(self, message: str, chat_history: Optional[List[ChatMessage]] = None):
        try:
            # Retrieve chat history
            chat_history = self._memory.get()
            print('chat history before', chat_history)
            # Prepare inputs as a dictionary
            inputs = {"query": message, "chat_history": chat_history}
            # Process the message with the query pipeline
            chat_response = await self.query_pipeline.arun(**inputs)

            print('response', chat_response)

            # Add the new message to the chat history
            await self._memory.aput(ChatMessage(role="user", content=message))
            # Add the response to the chat history
            await self._memory.aput(ChatMessage(role="assistant", content=chat_response))
            
            print('chat history', chat_history)
            return chat_response
        except Exception as e:
            return "An error occurred while processing your query: {}".format(e)

    def stream_chat(self, message: str, chat_history: Optional[List[ChatMessage]] = None) -> StreamingAgentChatResponse:
        # Implement your streaming chat logic here
        return StreamingAgentChatResponse(response="Streaming response")

    async def astream_chat(self, message: str, chat_history: Optional[List[ChatMessage]] = None) -> StreamingAgentChatResponse:
        # Implement your async streaming chat logic here
        return StreamingAgentChatResponse(response="Async streaming response")

    @property
    def chat_history(self) -> List[ChatMessage]:
        # Return chat history if applicable
        return []

This code ensures that the chat history is properly managed by retrieving it at the start of each method and updating it after processing the message and response <sup>[1][2]</sup>.

To continue talking to Dosu, mention @dosu.

[SarahSohana27]

@dosu why I am getting this error?
Output dict must have exactly one key.
after changing the code:

       ```

Retrieve chat history

        chat_history = self._memory.get()
        print('chat history before', chat_history)
        # Prepare inputs as a dictionary
        inputs = {"query": message, "chat_history": chat_history}
        # Process the message with the query pipeline
        chat_response = await self.query_pipeline.arun(**inputs)

        print('response', chat_response)

[dosubot[bot]]

The error "Output dict must have exactly one key" occurs because the get_output function expects the output_dict to contain exactly one key when src_key is None. This is enforced by the following code:

if src_key is None:
    # ensure that output_dict only has one key
    if len(output_dict) != 1:
        raise ValueError("Output dict must have exactly one key.")
    output = next(iter(output_dict.values()))

When you changed the code to use the arun method with inputs as a dictionary containing query and chat_history, it likely resulted in an output_dict with more than one key, which violates the condition checked in the get_output function. To resolve this, ensure that the output_dict passed to get_output contains exactly one key when src_key is None <sup>[1]</sup>.

To continue talking to Dosu, mention @dosu.