Merge pull request #12 from YakobusIP/main

Move production ready deployment from main to production

public-prediction/get_gpt_answer.py

@@ -0,0 +1,17 @@
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage, SystemMessage
+
+
+class GetGPTAnswer:
+    def __init__(self):
+        self.llm_gpt4o = ChatOpenAI(model="gpt-4o")
+
+    def generate_gpt4o_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_gpt4o.invoke(messages)
+        return gpt4_answer.content

public-prediction/kafka_consumer.py

@@ -0,0 +1,90 @@
+import json
+import os
+import requests
+from kafka import KafkaConsumer
+from get_gpt_answer import GetGPTAnswer
+from typing import List
+from concurrent.futures import ThreadPoolExecutor
+from predict_custom_model import predict_custom_trained_model
+from google.protobuf.json_format import MessageToDict
+
+
+def get_gpt_responses(data: dict[str, any], gpt_helper: GetGPTAnswer):
+    data["gpt4o_answer"] = gpt_helper.generate_gpt4o_answer(data["question"])
+    return data
+
+
+def process_batch(batch: List[dict[str, any]], batch_size: int, gpt_helper: GetGPTAnswer):
+    with ThreadPoolExecutor(max_workers=batch_size) as executor:
+        futures = [executor.submit(
+            get_gpt_responses, data, gpt_helper) for data in batch]
+        results = [future.result() for future in futures]
+
+    predictions = predict_custom_trained_model(
+        instances=results, project=os.environ.get("PROJECT_ID"), endpoint_id=os.environ.get("ENDPOINT_ID"))
+
+    results = []
+    for prediction in predictions:
+        result_dict = {}
+        for key, value in prediction._pb.items():
+            # Ensure that 'value' is a protobuf message
+            if hasattr(value, 'DESCRIPTOR'):
+                result_dict[key] = MessageToDict(value)
+            else:
+                print(f"Item {key} is not a convertible protobuf message.")
+        results.append(result_dict)
+
+    return results
+
+
+def send_results_back(full_results: dict[str, any], job_application_id: str):
+    print(f"Sending results back with job_app_id {job_application_id}")
+    url = "https://ta-2-sistem-cerdas-be-vi2jkj4riq-et.a.run.app/api/anti-cheat/result"
+    headers = {
+        "Content-Type": "application/json",
+        "x-api-key": os.environ.get("X-API-KEY")
+    }
+
+    body = {
+        "job_application_id": job_application_id,
+        "evaluations": full_results
+    }
+
+    response = requests.patch(url, json=body, headers=headers)
+    print(f"Data sent with status code {response.status_code}")
+    print(response.content)
+
+
+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="ai-detector",
+    )
+
+    print("Successfully connected to Kafka at", os.environ.get("KAFKA_IP"))
+
+    BATCH_SIZE = 5
+    gpt_helper = GetGPTAnswer()
+
+    for message in consumer:
+        try:
+            incoming_message = json.loads(message.value.decode("utf-8"))
+            full_batch = incoming_message["data"]
+        except json.JSONDecodeError:
+            print("Failed to decode JSON from message:", message.value)
+            print("Continuing...")
+            continue
+
+        print("Parsing successful. Processing job_app_id {0}".format(
+            incoming_message['job_application_id']))
+
+        full_results = []
+        for i in range(0, len(full_batch), BATCH_SIZE):
+            batch = full_batch[i:i+BATCH_SIZE]
+            batch_results = process_batch(batch, BATCH_SIZE, gpt_helper)
+            full_results.extend(batch_results)
+
+        send_results_back(full_results, incoming_message["job_application_id"])

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/predict_custom_model.py

@@ -0,0 +1,45 @@
+from typing import Dict, List, Union
+from google.cloud import aiplatform
+from google.protobuf import json_format
+from google.protobuf.struct_pb2 import Value
+from google.oauth2 import service_account
+
+
+def predict_custom_trained_model(
+    project: str,
+    endpoint_id: str,
+    instances: Union[Dict, List[Dict]],
+    location: str = "us-central1",
+    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
+):
+    """
+    `instances` can be either single instance of type dict or a list
+    of instances.
+    """
+    # The AI Platform services require regional API endpoints.
+    client_options = {"api_endpoint": api_endpoint}
+
+    credentials = service_account.Credentials.from_service_account_file(
+        "steady-climate-416810-ea1536e1868c.json")
+    # Initialize client that will be used to create and send requests.
+    # This client only needs to be created once, and can be reused for multiple requests.
+    client = aiplatform.gapic.PredictionServiceClient(
+        credentials=credentials,
+        client_options=client_options)
+    # The format of each instance should conform to the deployed model's prediction input schema.
+    instances = instances if isinstance(instances, list) else [instances]
+    instances = [
+        json_format.ParseDict(instance_dict, Value()) for instance_dict in instances
+    ]
+    parameters_dict = {}
+    parameters = json_format.ParseDict(parameters_dict, Value())
+    endpoint = client.endpoint_path(
+        project=project, location=location, endpoint=endpoint_id
+    )
+    response = client.predict(
+        endpoint=endpoint, instances=instances, parameters=parameters
+    )
+    # The predictions are a google.protobuf.Value representation of the model's predictions.
+    predictions = response.predictions
+
+    return predictions

public-prediction/requirements.txt

@@ -0,0 +1,7 @@
+kafka-python
+langchain
+openai
+langchain-openai
+python-dotenv
+google-cloud-aiplatform
+requests