added apis for executives sops
This commit is contained in:
+212
-309
@@ -6,315 +6,6 @@ from typing import List, Dict, Optional
|
||||
from src.prompts.sops import *
|
||||
from src.models.response_schemas import *
|
||||
|
||||
|
||||
|
||||
|
||||
class SopGenerator:
|
||||
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]
|
||||
|
||||
def get_roles(self, docs) -> RolesResponse:
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": '''Suppose you are a role/position extractor from a company document.
|
||||
You extract the roles as a list, e.g., ["financial analyst", "data scientist", etc.].
|
||||
If no roles are found, return an empty list.''',
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
response_format=RolesResponse,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
return json.loads(response.choices[0].message.content)
|
||||
|
||||
def check_role_sop(self, roles: str, docs) -> SOPsFound:
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Your role is to check if the SOPs for the provided roles "{roles}" are found in the document.
|
||||
You are validating if the document can provide the SOPs.
|
||||
Return status=True with a proper message if found, and status=False with a proper message if not.
|
||||
Keep the message short, e.g., "SOPs found for the role: {roles}" or "SOPs not found for the role: {roles}".'''
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
response_format=SOPsFound,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
return json.loads(response.choices[0].message.content)
|
||||
|
||||
def generate_sops_from_doc(self, roles: List[str], docs) -> SOPsResponse:
|
||||
roles_sops_all = {}
|
||||
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
|
||||
for role in roles:
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Your job is to extract Standard Operating Procedure (SOP) questions specifically for the role of "{role}" from the provided text.
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the questions under three categories: "must," "shall," and "will."
|
||||
Direct Questions: The questions should be directly addressed to the person in the role. Do not reference the role itself in the question.
|
||||
Contextual Inference: If SOPs for the role are not explicitly stated, infer them from the context, but only if there is clear evidence within the text. Do not generate or assume SOPs that are not directly supported by the document.
|
||||
Empty Lists: If no SOPs are found for the role, return an empty list for each category.
|
||||
Format: The questions should be direct and concise, e.g., "Have you completed all the required reports?"
|
||||
Example:
|
||||
|
||||
Category: Must
|
||||
|
||||
Have you completed all the required reports?
|
||||
Category: Shall
|
||||
|
||||
Are you ensuring that all team members follow the safety protocols?
|
||||
Category: Will
|
||||
|
||||
Are you planning to review the budget next week?''',
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
response_format=RoleSOPs,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
role_sop = json.loads(response.choices[0].message.content)
|
||||
roles_sops_all[role] = role_sop
|
||||
|
||||
|
||||
return roles_sops_all
|
||||
|
||||
def generate_sops_from_info(self, roles: List[Dict[str, str]]):
|
||||
|
||||
roles_sops_all = {}
|
||||
|
||||
for role_info in roles:
|
||||
role_title = role_info.get("title", "Unknown Role")
|
||||
print(f"Role title : {role_title}")
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Your job is to generate Standard Operating Procedures (SOPs) for the role of "{role_title}" based on the following information provided:
|
||||
|
||||
Responsibilities: {role_info.get("responsibilities", "Not provided")}
|
||||
Objectives: {role_info.get("objectives", "Not provided")}
|
||||
Tools: {role_info.get("tools", "Not provided")}
|
||||
Challenges: {role_info.get("challenges", "Not provided")}
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the SOPs under three categories: "must," "shall," and "will."
|
||||
Direct Instructions: The SOPs should directly address the responsibilities, objectives, and challenges.
|
||||
Contextual Inference: If SOPs for the role are not explicitly stated, infer them from the context provided.
|
||||
Empty Lists: If no SOPs are generated, return an empty list for each category.
|
||||
Format: The SOPs should be direct and concise.
|
||||
''',
|
||||
}
|
||||
],
|
||||
response_format=RoleSOPs,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
role_sop = json.loads(response.choices[0].message.content)
|
||||
roles_sops_all[role_title] = role_sop
|
||||
|
||||
return roles_sops_all
|
||||
|
||||
|
||||
|
||||
|
||||
def generate_sops_by_role_and_area(self, role: str, areas: str) -> RoleSops:
|
||||
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Your job is to generate Standard Operating Procedures (SOPs) for the role of "{role}" with a focus on the areas "{areas}" based on the following instructions:
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the SOPs under the selected categories: a checkboxex of the three categories "must" , "shall" and "will"
|
||||
Direct Instructions: The SOPs should directly address responsibilities, objectives, and challenges related to the area of "{areas}" for the role of "{role}".
|
||||
Contextual Inference: If SOPs for the area are not explicitly stated, infer them from the role and area context provided.
|
||||
Empty Lists: If no SOPs are generated, return an empty list for each category.
|
||||
Format: The SOPs should be direct and concise.
|
||||
''',
|
||||
}
|
||||
],
|
||||
response_format=RoleSOPs,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
return json.loads(response.choices[0].message.content)
|
||||
|
||||
|
||||
def generate_executive_sops_from_questionnaire(self, data: dict) -> RoleSops:
|
||||
"""
|
||||
Generate SOPs based on the answers from an executive questionnaire.
|
||||
|
||||
:param data: A dictionary containing the vision, strategic goals, and department goals.
|
||||
:return: SOPs categorized by "must", "shall", and "will".
|
||||
"""
|
||||
vision_list = data.get("organization vision", [])
|
||||
strategic_goals = data.get("organization strategic goals", [])
|
||||
department_goals = data.get("department goals", [])
|
||||
|
||||
# Format vision and goals as text
|
||||
formatted_vision = "\n".join([f"- {vision}" for vision in vision_list])
|
||||
formatted_goals = "\n".join([f"- {goal}" for goal in strategic_goals])
|
||||
formatted_department_goals = "\n".join([
|
||||
f"{dept}: " + ", ".join([f"{goal}" for goal in goals])
|
||||
for dept_dict in department_goals
|
||||
for dept, goals in dept_dict.items()
|
||||
])
|
||||
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Generate Standard Operating Procedures (SOPs) for an executive role based on the following information:
|
||||
|
||||
Organizational Vision:
|
||||
{formatted_vision}
|
||||
|
||||
Organizational Strategic Goals:
|
||||
{formatted_goals}
|
||||
|
||||
Departmental Strategic Goals:
|
||||
{formatted_department_goals}
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the SOPs under three categories: "must," "shall," and "will."
|
||||
Direct Instructions: The SOPs should address leadership responsibilities for achieving the vision, strategic contribution, and goals outlined.
|
||||
Empty Lists: If no SOPs are generated, return an empty list for each category.
|
||||
Format: SOPs should be direct and concise.
|
||||
''',
|
||||
}
|
||||
],
|
||||
response_format=RoleSops,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
return json.loads(response.choices[0].message.content)
|
||||
|
||||
|
||||
|
||||
|
||||
def generate_executive_sops_from_doc(self,docs) -> SOPsResponse:
|
||||
|
||||
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''Your job is to extract Standard Operating Procedure (SOP) questions specifically for the role of "{role}" from the provided text.
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the questions under three categories: "must," "shall," and "will."
|
||||
Direct Questions: The questions should be directly addressed to the person in the role. Do not reference the role itself in the question.
|
||||
Contextual Inference: If SOPs for the role are not explicitly stated, infer them from the context, but only if there is clear evidence within the text. Do not generate or assume SOPs that are not directly supported by the document.
|
||||
Empty Lists: If no SOPs are found for the role, return an empty list for each category.
|
||||
Format: The questions should be direct and concise, e.g., "Have you completed all the required reports?"
|
||||
Example:
|
||||
|
||||
Category: Must
|
||||
|
||||
Have you completed all the required reports?
|
||||
Category: Shall
|
||||
|
||||
Are you ensuring that all team members follow the safety protocols?
|
||||
Category: Will
|
||||
|
||||
Are you planning to review the budget next week?''',
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
response_format=VisionMissionResponse,
|
||||
max_tokens=1024,
|
||||
temperature=0.1
|
||||
)
|
||||
return response
|
||||
|
||||
def extract_vision_and_mission(self, docs: str):
|
||||
"""
|
||||
Use LLM to extract Vision and Mission from the document text.
|
||||
|
||||
:param document_text: The text content of the document.
|
||||
:return: (vision_section, mission_section)
|
||||
"""
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": '''You are a helpful assistant that extracts specific sections from business documents.
|
||||
Your task is to extract the "Vision" and "Mission" sections (or "Goals" if "Mission" is not found).
|
||||
mission is basically same as goals just mission as goals if not found
|
||||
Provide the sections exactly as they appear in the document.'''
|
||||
},
|
||||
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
max_tokens=1024,
|
||||
temperature=0.1,
|
||||
response_format=VisionMissionResponse,
|
||||
)
|
||||
|
||||
# Parse the response from the LLM
|
||||
extracted_text = json.loads(response.choices[0].message.content)
|
||||
|
||||
# Assuming the response contains fields for 'vision' and 'mission' (or 'goals')
|
||||
vision_section = extracted_text["vision"]
|
||||
mission_section = extracted_text["mission"]
|
||||
|
||||
return vision_section, mission_section
|
||||
|
||||
|
||||
|
||||
|
||||
class SopGeneratorDocument:
|
||||
def __init__(self):
|
||||
self.api_key = os.getenv("OPENAI_API_KEY")
|
||||
@@ -362,6 +53,70 @@ class SopGeneratorDocument:
|
||||
except:
|
||||
return False
|
||||
|
||||
def extract_vision_mission(self, docs) -> VisionMissionResponse:
|
||||
"""
|
||||
Extracts Vision, Mission, and SOPs categorized into 'must,' 'shall,' and 'will' from the document.
|
||||
|
||||
:param docs: The document(s) from which to extract information.
|
||||
:return: VisionMissionResponse containing the vision, mission, and role-specific SOPs.
|
||||
"""
|
||||
|
||||
try:
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
prompt = get_vision_mission_extraction_from_doc()
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''{prompt}'''
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": text} for text in docs_text],
|
||||
}
|
||||
],
|
||||
response_format=VisionMissionResponse,
|
||||
max_tokens=4096,
|
||||
temperature=0.1
|
||||
)
|
||||
|
||||
# Parse the response from the LLM
|
||||
extracted_text = json.loads(response.choices[0].message.content)
|
||||
|
||||
return extracted_text
|
||||
|
||||
except:
|
||||
return False
|
||||
|
||||
def extract_departments_and_managers(self, docs):
|
||||
"""
|
||||
Extract departments and managerial roles from the document.
|
||||
|
||||
:param docs: List of document chunks
|
||||
:return: Dictionary containing departments and their managerial roles
|
||||
"""
|
||||
try:
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
prompt = get_departments_and_roles_extraction_prompt()
|
||||
|
||||
response = self.client.beta.chat.completions.parse(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{"role": "system", "content": prompt},
|
||||
{"role": "user", "content": [{"type": "text", "text": text} for text in docs_text]}
|
||||
],
|
||||
response_format=DepartmentsAndRolesResponse,
|
||||
max_tokens=4096,
|
||||
temperature=0.1
|
||||
)
|
||||
|
||||
|
||||
return json.loads(response.choices[0].message.content)
|
||||
except json.JSONDecodeError:
|
||||
return False
|
||||
|
||||
|
||||
|
||||
|
||||
class SopPersonalAssessment:
|
||||
@@ -462,3 +217,151 @@ class SopPersonalAssessment:
|
||||
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,executives:List) -> 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("goals", "No goals provided")
|
||||
|
||||
prompt = get_sop_executive_from_vision_goals(executives)
|
||||
|
||||
user_content = f'''
|
||||
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 = SopGeneratorDocument()
|
||||
departments_and_roles = sop_doc.extract_departments_and_managers(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['managerial_roles']:
|
||||
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['title']} 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['title'], 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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user