Skip to main content
Open In ColabOpen on GitHub

Kafka

Kafka 是一个用于发布和订阅记录流的分布式消息系统。 该演示展示了如何使用 KafkaChatMessageHistory 从 Kafka 集群存储和检索聊天消息。

运行演示需要一个正在运行的Kafka集群。你可以按照此 说明 在本地创建一个Kafka集群。

from langchain_community.chat_message_histories import KafkaChatMessageHistory

chat_session_id = "chat-message-history-kafka"
bootstrap_servers = "localhost:64797"  # host:port. `localhost:Plaintext Ports` if setup Kafka cluster locally
history = KafkaChatMessageHistory(
    chat_session_id,
    bootstrap_servers,
)

构建 KafkaChatMessageHistory 的可选参数：

  • ttl_ms：聊天消息的生存时间，单位为毫秒。
  • partition：用于存储聊天消息的主题分区数量。
  • replication_factor：用于存储聊天消息的主题的复制因子。

KafkaChatMessageHistory 内部使用 Kafka 消费者来读取聊天消息，并且能够持久化地标记已消费的位置。它具有以下方法来获取聊天消息：

  • messages：从上一条继续消费聊天消息。
  • messages_from_beginning：将消费者重置到历史记录的开头并消费消息。可选参数：
    1. max_message_count：最多读取消息的数量。
    2. max_time_sec：读取消息的最长时间（秒）。
  • messages_from_latest：将消费者重置到聊天历史的末尾并尝试消费消息。可选参数与上述相同。
  • messages_from_last_consumed：从最后已消费的消息继续返回消息，类似于 messages，但带有可选参数。

max_message_count 和 max_time_sec 用于在检索消息时避免无限期阻塞。 因此，使用 messages 及其他方法检索消息可能无法返回聊天历史中的所有消息。您需要指定 max_message_count 和 max_time_sec 才能在单个批次中检索全部聊天历史。

添加消息并检索。

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

history.messages
[HumanMessage(content='hi!'), AIMessage(content='whats up?')]

再次调用 messages 会返回一个空列表，因为消费者已到达聊天历史的末尾。

history.messages
[]

添加新消息并继续使用。

history.add_user_message("hi again!")
history.add_ai_message("whats up again?")
history.messages
[HumanMessage(content='hi again!'), AIMessage(content='whats up again?')]

重置消费者并从开头开始读取：

history.messages_from_beginning()
[HumanMessage(content='hi again!'),
 AIMessage(content='whats up again?'),
 HumanMessage(content='hi!'),
 AIMessage(content='whats up?')]

将消费者设置到聊天历史的末尾，添加几条新消息，然后进行消费：

history.messages_from_latest()
history.add_user_message("HI!")
history.add_ai_message("WHATS UP?")
history.messages
[HumanMessage(content='HI!'), AIMessage(content='WHATS UP?')]