erp-ai-latest/src/services/chatbot.py

import os
import json
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 _extract_text_from_docs(self, docs):
        """Extract text content from document objects."""
        return [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