3fi5 dZddlmZddlZddlZddlZddlmZddlm Z m Z m Z m Z ddl mZddlmZmZmZe r ddlmZdd lmZej.eZd Zd Zd Zd ZGddeZ ddZGddeZ y)zKafka-based chat message history by using confluent-kafka-python. confluent-kafka-python is under Apache 2.0 license. https://github.com/confluentinc/confluent-kafka-python ) annotationsN)Enum) TYPE_CHECKINGListOptionalSequence)BaseChatMessageHistory) BaseMessagemessage_to_dictmessages_from_dict)TopicPartition AdminClientzbootstrap.serversi $ceZdZdZdZdZdZy)ConsumeStartPositionzConsume start position for Kafka consumer to get chat history messages. LAST_CONSUMED: Continue from the last consumed offset. EARLIEST: Start consuming from the beginning. LATEST: Start consuming from the latest offset. rrN)__name__ __module__ __qualname____doc__ LAST_CONSUMEDEARLIESTLATESTn/var/www/auto_recruiter/arenv/lib/python3.12/site-packages/langchain_community/chat_message_histories/kafka.pyrrs MH Frrc Fddlm} |jj}||vr6t ||j }t jd|d|d|S ||||dt|i g} |j| } | jD]\} } | jt jd|d |S#t$r}t jd||d}~wwxYw#t$r"}t jd |d ||d}~wwxYw) zCreate topic if it doesn't exist, and return the number of partitions. If the topic already exists, we don't change the topic configuration. r)NewTopicTopic z already exists with z partitionszFailed to list topics: Nz retention.ms)num_partitionsreplication_factorconfigz createdzFailed to create topic : )confluent_kafka.adminr list_topicstopicslen partitionsloggerinfo Exceptionerrorstr create_topicsitemsresult) admin_client topic_namer# partitionttl_msr topic_metadatar"er(futures_fs rensure_topic_existsr<*s?/ %113::  '  !;!F!FGN KK $9.9IU " ! (  $1"CK0  F,,V4MMO DAq HHJ  fZL12 +  .qc23"  .zl"QC@As1AC 6AC5 C2C--C25 D >DD ceZdZdZeeef d dZ d ddZ ddZ d ddZ d ddZ d ddZ e ddZdd Zdd Zy )KafkaChatMessageHistoryaChat message history stored in Kafka. 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. c ddlm}ddlm}||_||_|t|i|_ t|j|||||_ |t|i|_ y#tt f$r t dwxYw)a Args: session_id: The ID for single chat session. It is used as Kafka topic name. bootstrap_servers: Comma-separated host/port pairs to establish connection to Kafka cluster https://kafka.apache.org/documentation.html#adminclientconfigs_bootstrap.servers ttl_ms: Time-to-live (milliseconds) for automatic expiration of entries. Default 7 days. -1 for no expiration. It translates to https://kafka.apache.org/documentation.html#topicconfigs_retention.ms replication_factor: The replication factor for the topic. Default 1. partition: The number of partitions for the topic. Default 3. r)Producerrz_Could not import confluent_kafka package. Please install it with `pip install confluent_kafka`.N) confluent_kafkar@r&r ImportErrorModuleNotFoundError session_idbootstrap_serversBOOTSTRAP_SERVERS_CONFIGr3r<r"producer)selfrDrEr6r#r5r@rs r__init__z KafkaChatMessageHistory.__init__s*  0 9%!2')ACT(UV1   z+=y& !":&  !% 3 34I J  1$"3!44RST%  LL.assign_beginnings*) ,+ ,  $ $%8 9rcD|D] }|_ |j|yr^r_)rbrcrdrYs r assign_latestz>KafkaChatMessageHistory.__read_messages..assign_latests*) &% &  $ $%8 9r) on_assignNg?)timeoutzConsumer error: zEmpty message valuez'Failed to consume messages from Kafka: )rbrZrczlist[TopicPartition]returnNone)rArXrYrZrFrErDrrr subscribetimer)pollr.r+rLrQappendrNloadsr-closer )rHconsume_start_posmax_message_count max_time_secrZconsumer_configrergrRconsumerstart_time_secrTr8rXrYs @@r__read_messagesz'KafkaChatMessageHistory.__read_messagess" KJ %d&<&<   $8$?$??"*   :' :>R :  : :' :>R :  : "O,  $8$A$AA""DOO#4@P"Q"&:&A&AA""DOO#4 "N""DOO#45!YY[N , n4|C$0S]FW5W"---4?==?.LL#3GMMO3D!EF==?*NN#89 7==? ;<%. NN !(++   LLB1#F GG  NN s%%E(G(( H1H  HHH%cF|jtj||S)avRetrieve messages from Kafka topic from the beginning. This method resets the consumer to the beginning and consumes messages. Args: max_message_count: Maximum number of messages to consume. max_time_sec: Time limit in seconds to consume messages. Returns: List of messages. rrrsrt)'_KafkaChatMessageHistory__read_messagesrrrHrsrts rmessages_from_beginningz/KafkaChatMessageHistory.messages_from_beginnings+##2;;/%$  rcF|jtj||S)aReset to the end offset. Try to consume messages if available. Args: max_message_count: Maximum number of messages to consume. max_time_sec: Time limit in seconds to consume messages. Returns: List of messages. rz)r{rrr|s rmessages_from_latestz,KafkaChatMessageHistory.messages_from_latest*s+##299/%$  rcF|jtj||S)aRetrieve messages from Kafka topic from the last consumed message. Please note this method is stateful. Internally, it uses Kafka consumer to consume messages, and maintains the commit offset. Args: max_message_count: Maximum number of messages to consume. max_time_sec: Time limit in seconds to consume messages. Returns: List of messages. rz)r{rrr|s rmessages_from_last_consumedz3KafkaChatMessageHistory.messages_from_last_consumed<s+##2@@/%$  rc"|jS)a Retrieve the messages for the session, from Kafka topic continuously from last consumed message. This method is stateful and maintains consumed(committed) offset based on consumer group. Alternatively, use messages_from_last_consumed() with specified parameters. Use messages_from_beginning() to read from the earliest message. Use messages_from_latest() to read from the latest message. )rrHs rrRz KafkaChatMessageHistory.messagesPs//11rcX |jj|jg}|jD]\}}|j t j d|jdy#t$r,}t jd|jd||d}~wwxYw)z3Clear the chat history by deleting the Kafka topic.r!z deletedzFailed to delete topic r%N) r3 delete_topicsrDr1r2r+r,r-r.)rHr9r:r;r8s rclearzKafkaChatMessageHistory.clear\s ''55t6GHG  1   KK& 1: ;  LL24??2C2aSI JG sA1A44 B)='B$$B)cy)zNRelease the resources. Nothing to be released at this moment. Nrrs rrqzKafkaChatMessageHistory.closegs rN) rDr/rEr/r6intr#rr5r)@)rRzSequence[BaseMessage]rSfloatrjrk)rrrrs Optional[int]rtOptional[float]rjList[BaseMessage])r)rsrrtrrjr)rjr)rjrk)rrrrDEFAULT_TTL_MSDEFAULT_REPLICATION_FACTORDEFAULT_PARTITIONrIrVr{r}rrpropertyrRrrqrrrr>r>Vs08|%"<* $P$P$P $P  $P  $PR(+' %  &L,/L,)L,& L,  L,^UX !. BQ  &UX !. BQ  &UX !. BQ  ( 2 2  rr>) r3rr4r/r#rr5rr6rrjr)!r __future__rrNloggingrmenumrtypingrrrrlangchain_core.chat_historyr langchain_core.messagesr r r rAr r&r getLoggerrr+rFrrrrr<r>rrrrs #  ::>TT.1   8 $. 4 )))) )  )  )XU 4U r