import os import json import re from openai import OpenAI from pydantic import BaseModel, Field from typing import List, Dict, Optional from src.prompts.sops import * from src.prompts.chatbot import * from src.models.sop_response_schemas import * from src.models.bot_response_schema import * from scripts.assessment_data import generate_summary_stats_v2 from dotenv import load_dotenv from scripts.statistics_data import (generate_summary_stats,create_json_body,fetch_data) load_dotenv() import random import json from datetime import datetime, timedelta import pandas as pd # Define possible areas areas = [ "Agile Methodologies", "API Development", "Code Review", "Collaboration with Cross-Functional Teams", "Continuous Integration/Continuous Deployment (CI/CD)", "Debugging Techniques", "Documentation", "Performance Optimization", "System Design", "Unit Testing", "Version Control" ] # Generate random dates def random_date(): start_date = datetime(2024, 1, 1) return (start_date + timedelta(days=random.randint(1, 300))).strftime('%Y-%m-%dT00:00:00.000Z') # Generate dummy assessments def generate_dummy_data(num_assessments=50, max_users_per_assessment=5): data = [] for i in range(1, num_assessments + 1): assessment_id = str(i) assessment_name = f"Assessment {i}" start_date = random_date() red_flags = random.randint(0, 5) open_items = random.randint(0, 50) completed_items = random.randint(0, 50) total_assigned_items = open_items + completed_items # Generate random users for each assessment user_details = [] num_users = random.randint(1, max_users_per_assessment) for _ in range(num_users): user_name = f"User_{random.randint(1, 100)}" user_total_items = random.randint(1, 50) user_completed_items = random.randint(0, user_total_items) area_list = random.sample(areas, random.randint(1, 5)) user_details.append({ "name": user_name, "total_assigned_items": user_total_items, "completed_items": user_completed_items, "area_list": area_list }) data.append({ "assessment_id": assessment_id, "red_flags": red_flags, "open_items": open_items, "completed_items": completed_items, "total_assigned_items": total_assigned_items, "assessment_name": assessment_name, "start_date": start_date, "user_details": user_details }) return {"error": False, "data": data} # Generate dummy data dummy_data = generate_dummy_data(num_assessments=100, max_users_per_assessment=5) #print(dummy_data) #SopGeneratorDocument class Chatbot: def __init__(self): self.api_key = os.getenv("OPENAI_API_KEY") self.client = OpenAI(api_key=self.api_key) self.model = "gpt-4o-mini" def clean_text(self, text): # Remove all surrogate characters return re.sub(r'[\uD800-\uDFFF]', '', text) def _extract_text_from_docs(self, docs): """Extract text content from document objects.""" print(docs) return [self.clean_text(doc.page_content) for doc in docs] # Existing methods... def validate_worker(self, question, docs) -> VisionMissionResponse: """ This method is responsible for validating the worker's response ("yes" or "no") using the provided document(s). The system generates a prompt, extracts document content, and validates the response using the GPT-4 model. :param question: The yes/no question asked to the worker. :param docs: A list of document(s) uploaded by the worker. :return: VisionMissionResponse containing the validated result or None if an error occurs. """ try: docs_text = self._extract_text_from_docs(docs) prompt = validate_worker_prompt() response = self.client.beta.chat.completions.parse( model=self.model, messages=[ { "role": "system", "content": f'''{prompt} The document is uploaded next.''' }, { "role": "user", "content": f"Question :{question}", }, { "role": "user", "content": [{"type": "text", "text": text} for text in docs_text], } ], response_format=ValidateWorker, max_tokens=1024, temperature=0.1 ) # Parse the response from the LLM extracted_text = json.loads(response.choices[0].message.content) return extracted_text except Exception as e: print(f"An error occurred: {e}") return None def suggest_more_areas(self, position, existing_areas) -> VisionMissionResponse: """ This method is responsible for suggesting more areas based on the worker's position and existing areas. The system generates a prompt, and then uses the GPT-4 model to return more areas based on the query. :param position: The worker's position. :param existing areas: The existing areas. :return: VisionMissionResponse containing the suggested areas or None if an error occurs. """ try: prompt = suggest_more_areas_prompt() response = self.client.beta.chat.completions.parse( model=self.model, messages=[ { "role": "system", "content": f'''{prompt} ''' }, { "role": "user", "content": f"position :{position}", }, { "role": "user", "content": f"existing areas :{existing_areas}", } ], response_format=Areas, max_tokens=1024, temperature=0.1 ) # Parse the response from the LLM extracted_text = json.loads(response.choices[0].message.content) return extracted_text except Exception as e: print(f"An error occurred: {e}") return None def predict_based_on_past_assessment(self, query, company_info, companyid) -> Result: """ This method generates predictions based on past assessment data of a company. It queries the backend for the company's assessment data, generates a prompt, and then uses the GPT-4 model to return predictions based on the query. :param query: The question or query asked by the user. :param company_info: General information about the company (name, size, departments, etc.). :param companyid: Unique identifier of the company to fetch its specific data. :return: Result containing the prediction result or None if an error occurs. """ try: # Define the path to the company's assessment data (stored as a CSV) json_body_area = create_json_body("problematic-areas", companyid) json_body_assessment = create_json_body("user-stats-by-assessment", companyid) # Fetching data problematic_areas_data = fetch_data(json_body_area) assessment_data = fetch_data(json_body_assessment) summary_stats = generate_summary_stats(assessment_data, problematic_areas_data) print(summary_stats) # Generate the prompt using the company info and the summary statistics prompt = predict_based_past_assessment_prompt( query=query, company_info=company_info, summary_stats=summary_stats ) # Interact with GPT-4 model to get a response response = self.client.beta.chat.completions.parse( model=self.model, messages=[ { "role": "system", "content": f"{prompt}" }, { "role": "user", "content": f"{query}", } ], response_format=Result, max_tokens=1024, temperature=0.1 ) # Extract and return the response from the GPT-4 model extracted_text = json.loads(response.choices[0].message.content) return extracted_text except Exception as e: print(f"An error occurred: {e}") return None def predict_next_n_assessment(self,companyid, N) -> AssessmentPredictionsResponse: """ This method generates predictions based on past assessment data of a company. It queries the backend for the company's assessment data, generates a prompt, and then uses the GPT-4 model to return predictions based on the query. :param query: The question or query asked by the user. :param company_info: General information about the company (name, size, departments, etc.). :param companyid: Unique identifier of the company to fetch its specific data. :param N: Number of assessments to predict. :return: Result containing the prediction result or None if an error occurs. """ try: json_body_area = create_json_body("problematic-areas", companyid) json_body_assessment = create_json_body("user-stats-by-assessment", companyid) # Fetching data problematic_areas_data = fetch_data(json_body_area) assessment_data = fetch_data(json_body_assessment) summary_stats = generate_summary_stats(assessment_data, problematic_areas_data) print(summary_stats) # Define the path to the company's assessment data (stored as a CSV) #data_path = os.path.join('data', 'raw', 'erp_company_assessment', f'{companyid}_raw_data.csv') # Generate summary statistics from the company's assessment data #summary_stats = generate_summary_stats_v2(file_path=data_path) # Generate the prompt using the company info and the summary statistics prompt = predict_next_n_assessments_prompt_v2() # Interact with GPT-4 model to get a response MODEL = "gpt-4o" response = self.client.beta.chat.completions.parse( model=MODEL, messages=[ { "role": "system", "content": f"{prompt}" }, { "role": "user", "content": f"Summary stattistics of company past assessment: {summary_stats}", }, { "role": "user", "content": f"Number of next assessments to predict: {N}", } ], #response_format=AssessmentPredictionsResponse, max_tokens=1024, temperature=0.1 ) # Extract the response from the GPT-4 model preds = json.loads(response.choices[0].message.content) # Initialize dictionary to store assessments with dynamic names #predictions = {} # Loop through the predicted assessments and rename them dynamically #for i in range(N): #assessment_key = f"assessment_{i + 1}" #predictions[assessment_key] = extracted_text["predictions"][i]['AssessmentN'] # Return the dynamically named assessments return preds except Exception as e: print(f"An error occurred: {e}") return None def recommend_assessment_frequencies(self,sops,options) -> AssessmentSuggestion: """ Process the SOPs data and return assessment frequencies. """ try: chunk_size = 1000 # Define your chunk size # Convert the 'sops' dictionary values into a single text string for chunking sops_text = '\n'.join([str(item) for sublist in sops.values() for item in sublist]) # Break the text into chunks of the defined size docs_text = [sops_text[i:i + chunk_size] for i in range(0, len(sops_text), chunk_size)] # Create a list of documents docs = [{"type": "text", "text": text} for text in docs_text] # Generate the prompt using the company info and the summary statistics prompt = recommend_assessment_frequency_prompt() # Update your prompt to handle managers and workers response = self.client.beta.chat.completions.parse( model=self.model, messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": f'provided sops {docs}'}, {"role": "user", "content": f'options: {options}'} ], response_format=AssessmentSuggestion, # Use the updated response schema max_tokens=4096, temperature=0.1 ) return json.loads(response.choices[0].message.content) except Exception as e: print(f"An error occurred: {e}") return None def predict_goal_achievement_probability(self, company_info, companyid) -> AchievementPrediction: """ This method generates predictions based on past assessment data of a company. It queries the backend for the company's assessment data, generates a prompt, and then uses the GPT-4 model to return predictions based on the query. :param company_info: General information about the company (name, size, departments, etc.). :param companyid: Unique identifier of the company to fetch its specific data. :return: Result containing the prediction result or None if an error occurs. """ try: # Define the path to the company's assessment data (stored as a CSV) #data_path = os.path.join('data', 'raw', 'erp_company_assessment', f'{companyid}_raw_data.csv') # Generate summary statistics from the company's assessment data #summary_stats = generate_summary_stats_v2(file_path=data_path) json_body_area = create_json_body("problematic-areas", companyid) json_body_assessment = create_json_body("user-stats-by-assessment", companyid) # Fetching data problematic_areas_data = fetch_data(json_body_area) assessment_data = fetch_data(json_body_assessment) summary_stats = generate_summary_stats(assessment_data, problematic_areas_data) # Generate the prompt using the company info and the summary statistics prompt = predict_goal_achievement_probability_prompt() # Interact with GPT-4 model to get a response response = self.client.beta.chat.completions.parse( model=self.model, messages=[ { "role": "system", "content": prompt }, { "role": "user", "content": f"company info: {company_info}", }, { "role": "user", "content": f"Summary stats of past assement: {summary_stats}", } ], response_format=AchievementPrediction, max_tokens=1024, temperature=0.1 ) # Extract the response from the GPT-4 model predictions = json.loads(response.choices[0].message.content) # Return the dynamically named assessments return predictions except Exception as e: print(f"An error occurred: {e}") return None