Chat Message History RabbitMQ Integration

[S-Makrod]

Checked

• I searched existing ideas and did not find a similar one
• I added a very descriptive title
• I've clearly described the feature request and motivation for it

Feature request

We would like to add a new third-party integration to LangChain by introducing a chat message history integration with RabbitMQ. The documentation for RabbitMQ can be found here: https://www.rabbitmq.com/. This feature would give users the ability to consume from and publish to RabbitMQ clusters.

The tool would take connection information provided by the user, such as the URL to the RabbitMQ server as well as the session name, connect to the RabbitMQ server, and provide an API to publish and consume to the server. This API would be inherited from the BaseChatMessageHistory class in LangChain.

Here is some sample usage to get a better idea of what we would like to add:

Sample Usage

Note this is a basic example taken from the KafkaChatMessageHistory documentation and adapted to RabbitMQ.

from langchain_community.chat_message_histories import RabbitMQChatMessageHistory

# queue name
chat_session_id = "chat-message-history-rabbitmq"
# RabbitMQ host
bootstrap_servers = "localhost:64797"

history = RabbitMQChatMessageHistory(
    chat_session_id,
    bootstrap_servers,
)

history.add_user_message("hi!")
history.add_ai_message("whats up?")

history.messages

Note: The current plan would create a dependency with the Python package Pika. This package is currently the recommended Python client by RabbitMQ, so RabbitMQChatMessageHistory would use the package internally to connect to RabitMQ servers.

Motivation

Looking at current third-party integrations, the only message broker that is integrated with chat message history is Kafka. There is definitely a gap here, as there are many popular message brokers like Kafka, RabbitMQ, Amazon SQS, and Google Cloud P/S.

By adding this integration, we would add a popular message broker to LangChain's chat message history suite, which would allow easier integrations for LangChaing users.

Overall, if we look at market data, RabbitMQ has over a quarter of the message broker market share and 16,000+ clients world wide. This makes it the second most popular message broker after Apache Kafka. It would be great to be able to add an integration for this, as it is an extremely popular service.

Proposal (If applicable)

We (a group of 4 from the University of Toronto) are planning to make an addition to the tools contained in libs/community/langchain_community/chat_message_histories and follow a similar structure to the existing chat message history tools. If this idea is approved we hope to submit a PR by early November.

Here is a extremely early stage outline of the changes we are anticipating to make. We would love to get feedback on our idea and any suggestions for the implementation.

Modifications to existing files

We plan to modify the following files:

1. libs/community/langchain_community/chat_message_history/__init__.py to register the tool
2. libs/community/tests/unit_tests/chat_message_histories/test_imports.py to test the tool

New files

We plan to create the following files:

1. libs/community/langchain_community/chat_message_history/rabbitmq.py where we will define the class RabbitMQChatMessageHistory which inherits from BaseChatMessageHistory

Some (Incomplete) Sample Code

class RabbitMQChatMessageHistory(BaseChatMessageHistory):
    def __init__(
        self,
        queue: str,
        bootstrap_servers: str,
    ):
        try:
            from pika import BlockingConnection, ConnectionParameters
        except (ImportError, ModuleNotFoundError):
            raise ImportError(
                "Could not import pika package. "
                "Please install it with `pip install pika`."
            )

        self.queue = queue
        self.bootstrap_servers = bootstrap_servers

        # Will add more complex connection if approved
        # this is just a basic example
        self.connection = BlockingConnection(ConnectionParameters(bootstrap_servers))
        self.channel = connection.channel()

        # can also have things like max messages to read, max time to wait, etc.
        # ...

    @property
    def messages(self) -> List[BaseMessage]:
        # abstracted message consuming logic which would be here
        return self._consume()

    def clear(self) -> None:
        try:
            channel.queue_delete(queue=self.queue)
        except Exception as e:
            print(f"Failed to delete queue {self.queue}: {e}")
            raise e

    def close(self) -> None:
        self.connection.close()