updated
This commit is contained in:
+145
-48
@@ -7,6 +7,13 @@ 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()
|
||||
|
||||
|
||||
@@ -16,6 +23,9 @@ class SopPersonalAssessment:
|
||||
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...
|
||||
|
||||
@@ -72,42 +82,47 @@ class SopPersonalAssessment:
|
||||
print(f"Error occurred: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def generate_sops_by_role_and_area(self, roles: List[dict]) -> RoleSops:
|
||||
|
||||
def generate_sops_by_role_and_area(self, roles,qna) -> 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)
|
||||
|
||||
|
||||
MODEL = "gpt-4o-mini"
|
||||
prompt = get_sop_personalassessment_from_area_rolev2()
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
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=RoleSops,
|
||||
max_tokens=16000,
|
||||
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
|
||||
sops_by_role.append(extracted_text)
|
||||
|
||||
|
||||
|
||||
return sops_by_role
|
||||
|
||||
return extracted_text["sops"]
|
||||
|
||||
except:
|
||||
except Exception as e:
|
||||
print(f"Error occurred: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def generate_roles_from_questionnaire(self, questionnaire_data: List[dict]) -> Roles_response:
|
||||
|
||||
try:
|
||||
@@ -165,6 +180,7 @@ class SopGeneratorExecutive:
|
||||
Find the mission and vision provided below
|
||||
Vision: {vision}
|
||||
Goals: {goals}
|
||||
|
||||
'''
|
||||
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
@@ -232,56 +248,35 @@ class SopGeneratorExecutive:
|
||||
return False
|
||||
|
||||
|
||||
def generate_sops_from_questionnaire(self, questionnaire_data: dict, executives: List[str], managers: List[str], departments: List[str]):
|
||||
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:\n{user_content}\n\nExecutives to consider: {', '.join(executives)}\nManagers to consider: {', '.join(managers)}\nDepartments to consider: {', '.join(departments)}"}
|
||||
{"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
|
||||
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)
|
||||
"response": sops_data
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
@@ -289,6 +284,108 @@ class SopGeneratorExecutive:
|
||||
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):
|
||||
|
||||
Reference in New Issue
Block a user