Add async methods to BaseChatMessageHistory and BaseMemory

[cbornet]

Description:

Adds:

• async methods to BaseChatMessageHistory
• async methods to ChatMessageHistory
• async methods to BaseMemory
• async methods to BaseChatMemory
• async methods to ConversationBufferMemory
• tests of ConversationBufferMemory's async methods

Twitter handle: cbornet_

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
langchain	⬜️ Ignored (Inspect)	Visit Preview		Feb 1, 2024 10:42am

[eyurtsev]

Comandeering to resolve conflicts

[cbornet]

So IIUC, we only add aad_messages. Is that correct ?

[cbornet]

Comandeering to resolve conflicts

@eyurtsev done.

[eyurtsev]

Would you mind changing this into aget_messages()? So it's obvious it's a function call.

We can keep messages for the sync variant as a property for now -- though I think we may want to later on change that into a get_messages() for most implementations this involves IO and should not be assumed to be a cheap operation (I'm used to a convention that a @Property is cheap)

[cbornet]

Yes. That's especially problematic in this part

    def __str__(self) -> str:
        """Return a string representation of the chat history."""
        return get_buffer_string(self.messages)

If you call this in async mode, it might block the event loop.

[eyurtsev]

Would you mind adding in the following unit test:

async def test_async_interface() -> None:
    """Test async interface for BaseChatMessageHistory."""
    store: List[BaseMessage] = []

    class BulkAddHistory(BaseChatMessageHistory):
        def __init__(self, *, store: List[BaseMessage]) -> None:
            self.store = store

        @property
        def messages(self) -> List[BaseMessage]:
            return list(self.store)

        def add_messages(self, message: Sequence[BaseMessage]) -> None:
            """Add a message to the store."""
            self.store.extend(message)

        def clear(self) -> None:
            """Clear the store."""
            self.store = []

    chat_history = BulkAddHistory(store=store)
    await chat_history.aadd_messages(
        [HumanMessage(content="Hello"), HumanMessage(content="World")]
    )
    assert await chat_history.aget_messages() == [
        HumanMessage(content="Hello"),
        HumanMessage(content="World"),
    ]
    await chat_history.aadd_messages([HumanMessage(content="!")])
    assert await chat_history.aget_messages() == [
        HumanMessage(content="Hello"),
        HumanMessage(content="World"),
        HumanMessage(content="!"),
    ]
    await chat_history.aclear()
    assert await chat_history.aget_messages() == []

I'd like us to get close to 90% test coverage on all code in core. This unit test covers the async variant.

[cbornet]

Sure

[cbornet]

done

[eyurtsev]

Updated doc-string (feel free to incorporate or else I'll follow up with a PR)

class BaseChatMessageHistory(ABC):
    """Abstract base class for storing chat message history.

    Implementations guidelines:

    Implementations are expected to over-ride all or some of the following methods:

    * add_messages: sync variant for bulk addition of messages
    * aadd_messages: async variant for bulk addition of messages
    * messages: sync variant for getting messages
    * amessages: async variant for getting messages
    * clear: sync variant for clearing messages
    * aclear: async variant for clearing messages

    add_messages contains a default implementation that calls add_message
    for each message in the sequence. This is provided for backwards compatibility
    with existing implementations which only had add_message.

    Async variants all have default implementations that call the sync variants.
    Implementers can choose to over-ride the async implementations to provide
    truly async implementations.

    Usage guidelines:

    When used for updating history, users should favor usage of `add_messages`
    over `add_message` or other variants like `add_user_message` and `add_ai_message`
    to avoid unnecessary round-trips to the underlying persistence layer.

    Example: Shows a default implementation.

        .. code-block:: python

            class FileChatMessageHistory(BaseChatMessageHistory):
                storage_path:  str
                session_id: str

               @property
               def messages(self):
                   with open(os.path.join(storage_path, session_id), 'r:utf-8') as f:
                       messages = json.loads(f.read())
                    return messages_from_dict(messages)

               def add_messages(self, messages: Sequence[BaseMessage]) -> None:
                   all_messages = list(self.messages) # Existing messages
                   all_messages.extend(messages) # Add new messages

                   serialized = [message_to_dict(message) for message in all_messages]
                   # Can be further optimized by only writing new messages
                   # using append mode.
                   with open(os.path.join(storage_path, session_id), 'w') as f:
                       json.dump(f, messages)

               def clear(self):
                   with open(os.path.join(storage_path, session_id), 'w') as f:
                       f.write("[]")
    """

    messages: List[BaseMessage]
    """A property or attribute that returns a list of messages.
    
    In general, getting the messages may involve IO to the underlying
    persistence layer, so this operation is expected to incur some
    latency.
    """

    async def aget_messages(self) -> List[BaseMessage]:
        """Async version of getting messages.

        Can over-ride this method to provide an efficient async implementation.

        In general, fetching messages may involve IO to the underlying
        persistence layer.
        """
        return await run_in_executor(None, lambda: self.messages)

[eyurtsev]

@cbornet I suspect that we'll be deprecating these abstractions eventually. They're over-complicated for what they do. Regardless, I think OK adding async methods.

[cbornet]

If we 're not sure these abstractions will be kept, I think it's better to not add the async version for now. It's harder to remove things than to add them later.
I'll update.

[cbornet]

Well, actually those are used internally by aload_memory_variables. Should they be made private with _ ?
Since it's an implementation and not an interface, it's maybe not so bad to keep them.

[eyurtsev]

For public/private, let's match the syntax of any existing sync variants?

---

Regarding the abstraction, I was referring to BaseChatMemory and implementations.

If I understand correctly, their primary role is to generate a prompt for an LLM / chat model.

We definitely want the functionality, but need to figure out how to simplify the API and make it more intuitive. One possibility is to expose more standalone functions that can operate on messages (e.g., for history trimming based on number of messages, trimming based on # of tokens, summarization etc.)

[eyurtsev]

We for sure want to merge the improvements for BaseChatMessageHistory -- that abstraction we definitely want/need!

[eyurtsev]

Good to go. Waiting for confirmation from @cbornet on whether to keep the async methods in the Memory classes or avoid adding them for now.

[cbornet]

I think we're good to go!