Setup:

Install confluent-kafka-python.

.. code-block:: bash

pip install confluent_kafka

Instantiate:

.. code-block:: python

from langchain_community.chat_message_histories import KafkaChatMessageHistory

history = KafkaChatMessageHistory( session_id="your_session_id", bootstrap_servers="host:port", )

Add and retrieve messages:

.. code-block:: python

Add messages

history.add_messages([message1, message2, message3, ...])

Retrieve messages

message_batch_0 = history.messages

retrieve messages after message_batch_0

message_batch_1 = history.messages

Reset to beginning and retrieve messages

messages_from_beginning = history.messages_from_beginning()

Retrieving messages is stateful. Internally, it uses Kafka consumer to read. The consumed offset is maintained persistently.

To retrieve messages, you can use the following methods:

• messages: continue consuming chat messages from last one.
• messages_from_beginning: reset the consumer to the beginning of the chat history and return messages. Optional parameters:
  1. max_message_count: maximum number of messages to return.
  2. max_time_sec: maximum time in seconds to wait for messages.
• messages_from_latest: reset to end of the chat history and try consuming messages. Optional parameters same as above.
• messages_from_last_consumed: continuing from the last consumed message, similar to messages. Optional parameters same as above.

max_message_count and max_time_sec are used to avoid blocking indefinitely when retrieving messages. As a result, the method to retrieve messages may not return all messages. Change max_message_count and max_time_sec to retrieve all history messages.