feat: kafka preparation and event loop

public-prediction/get_gpt_answer.py

@@ -0,0 +1,28 @@
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage, SystemMessage
+
+
+class GetGPTAnswer:
+    def __init__(self):
+        self.llm_gpt35 = ChatOpenAI(model="gpt-3.5-turbo")
+        self.llm_gpt4 = ChatOpenAI(model="gpt-4-turbo")
+
+    def generate_gpt35_answer(self, question: str):
+        messages = [
+            SystemMessage(
+                content="Please answer the following question based solely on your internal knowledge, without external references. Assume you are the human."),
+            HumanMessage(question)
+        ]
+
+        gpt35_answer = self.llm_gpt35.invoke(messages)
+        return gpt35_answer.content
+
+    def generate_gpt4_answer(self, question: str):
+        messages = [
+            SystemMessage(
+                content="Please answer the following question based solely on your internal knowledge, without external references. Assume you are the human."),
+            HumanMessage(question)
+        ]
+
+        gpt4_answer = self.llm_gpt4.invoke(messages)
+        return gpt4_answer.content

public-prediction/kafka_consumer.py

@@ -0,0 +1,44 @@
+import json
+import os
+from kafka import KafkaConsumer
+from get_gpt_answer import GetGPTAnswer
+from typing import List
+from concurrent.futures import ThreadPoolExecutor
+
+
+def get_gpt_responses(data: dict[str, any], gpt_helper: GetGPTAnswer):
+    # data["gpt35_answer"] = gpt_helper.generate_gpt35_answer(data["question"])
+    # data["gpt4_answer"] = gpt_helper.generate_gpt4_answer(data["question"])
+    data["gpt35_answer"] = "This is gpt35 answer"
+    data["gpt4_answer"] = "This is gpt4 answer"
+    return data
+
+
+def process_batch(batch: List[dict[str, any]], batch_size: int):
+    with ThreadPoolExecutor(max_workers=batch_size) as executor:
+        gpt_helper = GetGPTAnswer()
+        futures = [executor.submit(
+            get_gpt_responses, data, gpt_helper) for data in batch]
+        results = [future.result() for future in futures]
+
+    print("Batch ready with gpt responses", results)
+
+
+def consume_messages():
+    consumer = KafkaConsumer(
+        "ai-detector",
+        bootstrap_servers=[os.environ.get("KAFKA_IP")],
+        auto_offset_reset='earliest',
+        client_id="ai-detector-1",
+        group_id=None,
+        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
+    )
+
+    BATCH_SIZE = 5
+
+    for message in consumer:
+        full_batch = message.value
+
+        for i in range(0, len(full_batch), BATCH_SIZE):
+            batch = full_batch[i:i+BATCH_SIZE]
+            process_batch(batch, BATCH_SIZE)

public-prediction/main.py

@@ -0,0 +1,6 @@
+from kafka_consumer import consume_messages
+from dotenv import load_dotenv
+
+if __name__ == "__main__":
+    load_dotenv()
+    consume_messages()

public-prediction/requirements.txt

@@ -0,0 +1,5 @@
+kafka-python
+langchain
+openai
+langchain-openai
+python-dotenv