303 lines
11 KiB
Python
303 lines
11 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
|
|
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"
|
|
|
|
# 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=2048,
|
|
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: List[dict]) -> RoleSops:
|
|
|
|
try:
|
|
sops_by_role = []
|
|
for role_info in roles:
|
|
role = role_info['role']
|
|
sop_types = role_info['sop_types'] # List of SOP types: ["will", "shall", "must"]
|
|
areas = role_info['areas'] # List of areas: ["communication", "development", etc.]
|
|
|
|
|
|
prompt = get_sop_personalassessment_from_area_role(role,areas,sop_types)
|
|
response = self.client.beta.chat.completions.parse(
|
|
model=self.model,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": f'''{prompt}
|
|
''',
|
|
}
|
|
],
|
|
response_format=RoleSops,
|
|
max_tokens=1024,
|
|
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
|
|
sops_by_role.append(extracted_text)
|
|
|
|
|
|
|
|
return sops_by_role
|
|
|
|
except:
|
|
return False
|
|
|
|
def generate_roles_from_questionnaire(self, questionnaire_data: List[dict]) -> 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}''',
|
|
}
|
|
],
|
|
response_format=Roles_response,
|
|
max_tokens=1024,
|
|
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=2048,
|
|
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=1024,
|
|
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], managers: List[str], departments: List[str]):
|
|
try:
|
|
prompt = get_sop_executive_from_questionnaire()
|
|
|
|
# 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 SOPs based on this questionnaire:\n{user_content}\n\nExecutives to consider: {', '.join(executives)}\nManagers to consider: {', '.join(managers)}\nDepartments to consider: {', '.join(departments)}"}
|
|
],
|
|
response_format={"type": "json_object"},
|
|
max_tokens=4096,
|
|
temperature=0.1
|
|
)
|
|
|
|
sops_data = json.loads(response.choices[0].message.content)
|
|
|
|
# Process executive SOPs
|
|
executive_sops = {}
|
|
for executive in executives:
|
|
if executive in sops_data['executives']:
|
|
executive_sops[executive] = Categories(**sops_data['executives'][executive])
|
|
else:
|
|
executive_sops[executive] = Categories()
|
|
|
|
# Process department manager SOPs
|
|
departments_with_sops = []
|
|
for dept_name in departments:
|
|
dept_data = next((d for d in sops_data['departments'] if d['name'].lower() == dept_name.lower()), None)
|
|
if dept_data:
|
|
managers_with_sops = [
|
|
ManagerWithSOPs(
|
|
title=manager,
|
|
sops=ManagerSOPs(**dept_data['managers'])
|
|
)
|
|
for manager in managers
|
|
|
|
]
|
|
if managers_with_sops:
|
|
departments_with_sops.append(DepartmentManagerSOPs(
|
|
name=dept_name,
|
|
managers=managers_with_sops
|
|
))
|
|
|
|
return {
|
|
"executive_sops": executive_sops,
|
|
"department_sops": ExecutiveManagerSOPsResponse(departments=departments_with_sops)
|
|
}
|
|
|
|
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"
|
|
|
|
|
|
|
|
|
|
|