404 lines
14 KiB
Python
404 lines
14 KiB
Python
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.models.sop_response_schemas import *
|
|
from src.services.sop_document_parser import DocumentParser
|
|
from dotenv import load_dotenv
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from typing import List, Dict
|
|
import json
|
|
from openai import OpenAI
|
|
import os
|
|
import time
|
|
from collections import deque
|
|
load_dotenv()
|
|
|
|
|
|
|
|
class SopPersonalAssessment:
|
|
def __init__(self):
|
|
self.api_key = os.getenv("OPENAI_API_KEY")
|
|
self.client = OpenAI(api_key=self.api_key)
|
|
self.model = "gpt-4o-mini"
|
|
self.rpm_limit = 3 # Requests per minute limit
|
|
self.batch_size = 3 # Process 3 roles at a time
|
|
self.retry_delays = [4, 8, 16, 32]
|
|
|
|
# Existing methods...
|
|
|
|
def extract_sops_from_questionnaire(self, questionnaire_data: dict) -> VisionMissionResponse:
|
|
"""
|
|
Extracts SOPs categorized into 'must,' 'shall,' and 'will' based on free-text questionnaire information.
|
|
|
|
:param questionnaire_data: A dictionary containing plain text responses to the questionnaire.
|
|
:return: SOPsResponse containing the vision, mission, and role-specific SOPs
|
|
"""
|
|
|
|
try:
|
|
# Extract the plain text answers from the questionnaire
|
|
vision = questionnaire_data.get("vision", "No vision provided")
|
|
roles = questionnaire_data.get("roles", "No roles provided") # This will be a plain text string
|
|
responsibilities = questionnaire_data.get("responsibilities", "No responsibilities provided") # Plain text string
|
|
project_details = questionnaire_data.get("project_details", "No additional project details provided")
|
|
|
|
# Get the appropriate prompt for the questionnaire scenario
|
|
prompt = get_sop_personalassessment_from_questionnaire()
|
|
|
|
# Combine the plain text data into a string to pass to the LLM
|
|
user_content = f'''
|
|
Vision: {vision}
|
|
Roles: {roles}
|
|
Responsibilities: {responsibilities}
|
|
Project Details: {project_details}
|
|
'''
|
|
|
|
# Send the combined plain text content to the LLM
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": f'''{prompt}'''
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": user_content,
|
|
}
|
|
],
|
|
response_format=SOPsResponse,
|
|
max_tokens=16000,
|
|
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"Error occurred: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_sops_by_role_and_area(self, roles,qna) -> RoleSops:
|
|
|
|
try:
|
|
|
|
|
|
MODEL = "gpt-4o-mini"
|
|
prompt = get_sop_personalassessment_from_area_rolev2()
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=MODEL,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": f'''{prompt}
|
|
''',
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": f'''povided Roles Data {roles}
|
|
''',
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": f'''Extra question and answers for more context {qna}
|
|
''',
|
|
}
|
|
],
|
|
response_format=RoleSopssLists,
|
|
max_tokens=6000,
|
|
temperature=0.1
|
|
)
|
|
extracted_text = json.loads(response.choices[0].message.content)
|
|
# You can customize this to generate SOPs based on the role, SOP types, and areas
|
|
|
|
return extracted_text["sops"]
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_roles_from_questionnaire(self, questionnaire_data: List[dict],role_slug:str) -> Roles_response:
|
|
|
|
try:
|
|
# List of areas: ["communication", "development", etc.]
|
|
|
|
prompt = get_roles_extraction_from_questionnaire()
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": f'''{prompt}''',
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": f'''Questionairre data : {questionnaire_data}''',
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": f'''Role slug to consider : {role_slug}''',
|
|
}
|
|
],
|
|
response_format=Roles_response,
|
|
max_tokens=4096,
|
|
temperature=0.1
|
|
)
|
|
extracted_text = json.loads(response.choices[0].message.content)
|
|
# You can customize this to generate roles based on the questionnaire data
|
|
|
|
return extracted_text
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {str(e)}")
|
|
return False
|
|
|
|
|
|
|
|
class SopGeneratorExecutive:
|
|
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_sops_from_executive_vision_goals_doc(self, data: dict,executive:str) -> SOPsResponse:
|
|
"""
|
|
Extracts SOPs categorized into 'must,' 'shall,' and 'will' based on executive vision and goals.
|
|
|
|
:param data: A dictionary containing vision and goals.
|
|
:return: SOPsResponse containing the SOPs for executives
|
|
"""
|
|
try:
|
|
|
|
vision = data.get("vision", "No vision provided")
|
|
goals = data.get("mission", "No goals provided")
|
|
|
|
prompt = get_sop_executive_from_vision_goals(executive)
|
|
|
|
user_content = f'''
|
|
Find the mission and vision provided below
|
|
Vision: {vision}
|
|
Goals: {goals}
|
|
|
|
'''
|
|
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": f'''{prompt}'''
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": user_content,
|
|
}
|
|
],
|
|
response_format=Categories,
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
|
|
extracted_text = json.loads(response.choices[0].message.content)
|
|
return extracted_text
|
|
|
|
except Exception as e:
|
|
print(f"Error occurred: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_sops_for_department_managers(self, docs):
|
|
try:
|
|
# First, extract departments and managers
|
|
sop_doc = DocumentParser()
|
|
departments_and_roles = sop_doc.extract_departments_and_managers_workers(docs)
|
|
|
|
if not departments_and_roles or not departments_and_roles.get('departments'):
|
|
return False
|
|
|
|
departments_with_sops = []
|
|
|
|
for department in departments_and_roles['departments']:
|
|
managers_with_sops = []
|
|
for role in department['managers']:
|
|
prompt = get_sop_for_department_managers()
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": prompt},
|
|
{"role": "user", "content": f"Generate SOPs for {role['position']} in {department['name']} department."}
|
|
],
|
|
response_format=ManagerSOPs,
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
manager_sops = json.loads(response.choices[0].message.content)
|
|
managers_with_sops.append(ManagerWithSOPs(title=role['position'], sops=manager_sops))
|
|
|
|
departments_with_sops.append(DepartmentManagerSOPs(
|
|
name=department['name'],
|
|
managers=managers_with_sops
|
|
))
|
|
|
|
return ExecutiveManagerSOPsResponse(departments=departments_with_sops)
|
|
|
|
except Exception as e:
|
|
print(f"Error in generate_sops_for_department_managers: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_sops_from_questionnaire(self, questionnaire_data: dict, executives: List[str]):
|
|
try:
|
|
print("Generating SOPs from questionnaire...")
|
|
prompt = get_sop_executive_from_questionnaire()
|
|
print(f"Prompt: {prompt}")
|
|
|
|
# Prepare the questionnaire data for the prompt
|
|
user_content = json.dumps(questionnaire_data, indent=2)
|
|
print(f"User content: {user_content}")
|
|
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": prompt},
|
|
{"role": "user", "content": f"Generate SOPs based on this questionnaire response:\n{user_content}\n\nExecutives to consider: {', '.join(executives)}"}
|
|
],
|
|
response_format={"type": "json_object"},
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
|
|
print("Response received from API.")
|
|
sops_data = json.loads(response.choices[0].message.content)
|
|
print(f"SOPs data: {sops_data}")
|
|
|
|
# Process executive SOPs
|
|
|
|
return {
|
|
"response": sops_data
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error in generate_sops_from_questionnaire: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_vision_mission_from_questionnaire(self, questionnaire_data: dict):
|
|
"""
|
|
Generate vision and mission based on the questionnaire data provided in the request body.
|
|
The request body is expected to contain plain-text information for vision and mission."""
|
|
try:
|
|
print("Generating vsion and mission from questionnaire...")
|
|
prompt = get_vision_mission_extraction_from_questionnaire_executive()
|
|
print(f"Prompt: {prompt}")
|
|
|
|
# Prepare the questionnaire data for the prompt
|
|
user_content = json.dumps(questionnaire_data, indent=2)
|
|
print(f"User content: {user_content}")
|
|
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": prompt},
|
|
{"role": "user", "content": f"questionnaire response:\n{user_content}"}
|
|
],
|
|
response_format=VisionMissionResponse2,
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
|
|
print("Response received from API.")
|
|
response = json.loads(response.choices[0].message.content)
|
|
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
print(f"Error in generate_sops_from_questionnaire: {str(e)}")
|
|
return False
|
|
|
|
|
|
def generate_dept_goals_from_questionnaire(self, questionnaire_data):
|
|
try:
|
|
print("Generating SOPs from questionnaire...")
|
|
prompt = get_dept_vision_mission_extraction_from_questionaiire()
|
|
print(f"Prompt: {prompt}")
|
|
|
|
# Prepare the questionnaire data for the prompt
|
|
user_content = json.dumps(questionnaire_data, indent=2)
|
|
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": prompt},
|
|
{"role": "user", "content": f"Generate based on this questionnaire response:\n{user_content}\n"}
|
|
],
|
|
response_format=DeptGoalsVisssion,
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
|
|
print("Response received from API.")
|
|
sops_data = json.loads(response.choices[0].message.content)
|
|
|
|
|
|
# Process executive SOPs
|
|
|
|
return sops_data
|
|
|
|
except Exception as e:
|
|
print(f"Error in generate_sops_from_questionnaire: {str(e)}")
|
|
return False
|
|
|
|
def generate_sops_from_questionnaire2(self, questionnaire_data: dict):
|
|
try:
|
|
print("Generating SOPs from questionnaire...")
|
|
prompt = get_sop_executive_from_questionnaire2()
|
|
print(f"Prompt: {prompt}")
|
|
|
|
# Prepare the questionnaire data for the prompt
|
|
user_content = json.dumps(questionnaire_data, indent=2)
|
|
print(f"User content: {user_content}")
|
|
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{"role": "system", "content": prompt},
|
|
{"role": "user", "content": f"Generate SOPs based on this questionnaire response:\n{user_content}\n"}
|
|
],
|
|
response_format={"type": "json_object"},
|
|
max_tokens=16000,
|
|
temperature=0.1
|
|
)
|
|
|
|
print("Response received from API.")
|
|
sops_data = json.loads(response.choices[0].message.content)
|
|
print(f"SOPs data: {sops_data}")
|
|
|
|
# Process executive SOPs
|
|
|
|
return {
|
|
"response": sops_data
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error in generate_sops_from_questionnaire: {str(e)}")
|
|
return False
|
|
|
|
|
|
class SopGeneratorManager:
|
|
def __init__(self):
|
|
self.api_key = os.getenv("OPENAI_API_KEY")
|
|
self.client = OpenAI(api_key=self.api_key)
|
|
self.model = "gpt-4o-mini"
|
|
|
|
|
|
|
|
|
|
|