added manager and execuytive generator

This commit is contained in:
OwusuBlessing
2024-09-09 14:03:14 +01:00
parent 13ddfa3e10
commit 0d8ad2381b
7 changed files with 506 additions and 211 deletions
Binary file not shown.
+71 -7
View File
@@ -1,9 +1,8 @@
import os
from flask import Blueprint, request, jsonify, current_app
from werkzeug.utils import secure_filename
from src.services.sop_generator import (SopGeneratorDocument,
SopPersonalAssessment,SopGeneratorExecutive)
from src.services.sop_generator import (SopPersonalAssessment,SopGeneratorExecutive)
from src.services.document_parser import DocumentParser
from src.utils.utils import delete_all_files_in_directory
from src.utils.document_loader import load_document
@@ -103,7 +102,7 @@ def generate_questions_from_sop():
return jsonify({"error": "Document cannot extract SOPs", "message": status_check["message"]}), 400
# Generate SOPs based on the roles provided
sop_generator = SopGeneratorDocument()
sop_generator = DocumentParser()
sops = sop_generator.generate_sops_from_doc(docs)
# Cleanup: Delete all files in the upload directory after processing
@@ -147,7 +146,7 @@ def generate_sops():
docs = load_document(file_path)
# Generate SOPs based on the roles provided
sop_generator = SopGeneratorDocument()
sop_generator = DocumentParser()
sops = sop_generator.extract_sops_from_doc(docs)
# Cleanup: Delete all files in the upload directory after processing
delete_all_files_in_directory(upload_folder)
@@ -277,7 +276,7 @@ def generate_executive_sops_from_doc():
# Use the utility function to generate docs from the file
docs = load_document(file_path)
sop_doc = SopGeneratorDocument()
sop_doc = DocumentParser()
vision_mission = sop_doc.extract_vision_mission(docs)
if not vision_mission:
@@ -380,4 +379,69 @@ def generate_sops_from_questionnaire():
return jsonify({"sops": serializable_result, "message": "SOPs successfully generated from questionnaire."}), 200
except Exception as e:
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the request: {str(e)}"}), 500
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the request: {str(e)}"}), 500
@sops_bp.route('/executive/get_roles_doc', methods=['POST'])
def generate_sops_from_questionnaire():
try:
# Retrieve form data
reference_roles = request.get_json().get('reference_roles') # List of reference roles in JSON format
document = request.files.get('document') # The uploaded document
if not reference_roles or not document:
return jsonify({"error": "Missing data", "message": "Reference roles or document not provided."}), 400
# Use extractor to extract roles from the document
extractor = DocumentParser()
extracted_data = extractor.extract_departments_and_managers_workers([document])
if not extracted_data:
return jsonify({"error": "Extraction error", "message": "No roles were extracted from the document."}), 400
# Extract all managers with their name, title (position), and classification (role: PRP or SRP)
extracted_managers = []
for department in extracted_data['departments']:
extracted_managers.extend([{
'name': manager['name'],
'position': manager.get('position', 'Unknown Position'), # Assuming title is the position
'role': manager.get('classification', 'Unknown Role') # PRP or SRP classification
} for manager in department['managers']])
# Prepare assigned, unassigned, and unavailable managers
assigned_managers = [manager for manager in extracted_managers if manager['name'] in reference_roles]
unassigned_managers = [{'name': role, 'position': 'Reference Role', 'role': 'N/A'} for role in reference_roles if role not in [manager['name'] for manager in extracted_managers]]
unavailable_managers = [manager for manager in extracted_managers if manager['name'] not in reference_roles]
# Return the results with detailed manager information
return jsonify({
"assigned_roles": assigned_managers,
"unassigned_roles": unassigned_managers,
"unavailable_roles": unavailable_managers
}), 200
except Exception as e:
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the request: {str(e)}"}), 500
assigned_roles = [role for role in reference_roles if role in extracted_roles]
unassigned_roles = [role for role in reference_roles if role not in extracted_roles]
unavailable_roles = [role for role in extracted_roles if role not in reference_roles]
# Return the results
return jsonify({
"assigned_roles": assigned_roles,
"unassigned_roles": unassigned_roles,
"unavailable_roles": unavailable_roles
}), 200
except Exception as e:
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the request: {str(e)}"}), 500
+18 -6
View File
@@ -35,17 +35,26 @@ class Categories(BaseModel):
class ExecutivesSops(BaseModel):
executive_sops: List[RoleSops]
class ManagerialRole(BaseModel):
title: str
responsibilities: List[str]
class Manager(BaseModel):
name: str
position: str
role: str
responsibilities: List[str] = Field(default_factory=list)
class Worker(BaseModel):
name: str
position: str
responsibilities: List[str] = Field(default_factory=list)
class Department(BaseModel):
name: str
managerial_roles: List[ManagerialRole]
managers: List[Manager] = Field(default_factory=list) # Updated to managers
workers: List[Worker] = Field(default_factory=list) # New field for workers
class DepartmentsAndRolesResponse(BaseModel):
class DepartmentsAndWorkersResponse(BaseModel):
departments: List[Department]
class ManagerSOPs(BaseModel):
must: List[str] = Field(default_factory=list)
shall: List[str] = Field(default_factory=list)
@@ -60,4 +69,7 @@ class DepartmentManagerSOPs(BaseModel):
managers: List[ManagerWithSOPs]
class ExecutiveManagerSOPsResponse(BaseModel):
departments: List[DepartmentManagerSOPs]
departments: List[DepartmentManagerSOPs]
+56 -9
View File
@@ -93,31 +93,45 @@ def get_vision_mission_extraction_from_doc():
def get_departments_and_roles_extraction_prompt():
def get_departments_managers_workers_extraction_prompt():
return """
Extract departments and their senior managerial roles from the document.
Include only managerial positions (e.g., Department Head, Director, Manager).
For each role, list 2-3 key responsibilities.
Do not add any departments or roles that are not explicitly mentioned in the document.
Extract departments, their managers, and workers from the document.
For each department, include the managers (e.g., Department Head, Director, Manager) and their key responsibilities.
Additionally, for each department, extract the workers and their positions, and list 1-2 key responsibilities for each worker.
Do not add any departments, managers, or workers that are not explicitly mentioned in the document.
Managers: Include the managers (e.g., Department Head, Manager), their role , and key responsibilities.
- **PRP (Primary Responsible Person)**: A manager who has primary responsibility for decision-making and overseeing operations.
- **SRP (Secondary Responsible Person)**: A manager who supports the PRP, often assisting with tasks and providing backup in decision-making.
Format as JSON:
{
"departments": [
{
"name": "Department Name",
"managerial_roles": [
"managers": [
{
"title": "Managerial Role Title",
"name": "Manager Name",
"position": "manager Position",
"role": "PRP or SRP", # The classification field either PRP or SRP
"responsibilities": ["Key Responsibility 1", "Key Responsibility 2"]
}
],
"workers": [
{
"name": "Worker Name",
"position": "Worker Position",
"responsibilities": ["Key Responsibility 1", "Key Responsibility 2"]
}
]
}
]
}
If no departments or roles are found in the document, return an empty list for departments.
If no departments, managers, or workers are found in the document, return an empty list for departments.
"""
def get_sop_for_department_managers():
return '''Generate Standard Operating Procedures (SOPs) for the specified managerial role in the given department.
@@ -186,4 +200,37 @@ def get_sop_executive_from_questionnaire():
]
}
Ensure that each specified department has its own set of SOPs.
'''
'''
def generate_llm_comparison_prompt(reference_roles, extracted_managers):
reference_roles_str = ', '.join(reference_roles)
extracted_managers_str = '\n'.join([f"- {manager['name']} (Position: {manager['position']}, Role: {manager['role']})" for manager in extracted_managers])
prompt = f"""
You are tasked with comparing a list of reference roles with the extracted roles from a document.
Reference roles:
[{reference_roles_str}]
Extracted roles:
{extracted_managers_str}
Please classify the roles into the following categories:
1. **Assigned Roles**: Roles that are found in both the reference list and the extracted list.
2. **Unassigned Roles**: Roles that are found in the reference list but not in the extracted list.
3. **Unavailable Roles**: Roles that are found in the extracted list but not in the reference list.
Return the result in the following JSON format:
{
"assigned_roles": [
{"name": "Role Name", "position": "Role Position", "role": "PRP or SRP"}
],
"unassigned_roles": [
{"name": "Role Name", "position": "Reference Role"}
],
"unavailable_roles": [
{"name": "Role Name", "position": "Role Position", "role": "PRP or SRP"}
]
}
"""
return prompt
+197
View File
@@ -0,0 +1,197 @@
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.response_schemas import *
from dotenv import load_dotenv
load_dotenv()
#SopGeneratorDocument
class DocumentParser:
def __init__(self):
self.api_key = os.getenv("OPENAI_API_KEY")
self.client = OpenAI(api_key=self.api_key)
self.model = "gpt-4o-2024-08-06"
def _extract_text_from_docs(self, docs):
"""Extract text content from document objects."""
return [doc.page_content for doc in docs]
# Existing methods...
def extract_sops_from_doc(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_sop_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=SOPsResponse,
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_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'''
def extract_departments_and_managers_workers(self, docs):
"""
Extract departments, managers, and workers from the document.
:param docs: List of document chunks
:return: Dictionary containing departments, their managers, and workers.
"""
try:
docs_text = self._extract_text_from_docs(docs)
prompt = get_departments_managers_workers_extraction_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": [{"type": "text", "text": text} for text in docs_text]}
],
response_format=DepartmentsAndWorkersResponse, # Use the updated response schema
max_tokens=4096,
temperature=0.1
)
return json.loads(response.choices[0].message.content)
except json.JSONDecodeError:
return False
def generate_sops_(self, docs, reference_roles):
try:
# First, extract departments and managers from the document
sop_doc = DocumentParser()
departments_and_roles = sop_doc.extract_departments_and_managers(docs)
# Prepare extracted roles (only managers)
extracted_managers = []
for department in departments_and_roles['departments']:
extracted_managers.extend([
{
'name': manager['name'],
'position': manager.get('title', 'Unknown Position'),
'role': manager.get('classification', 'Unknown Role') # PRP or SRP classification
}
for manager in department['managerial_roles']
])
# Generate prompt for the LLM to compare reference roles with extracted roles
prompt = generate_llm_comparison_prompt(reference_roles, extracted_managers)
# Send prompt to the LLM for comparison
response = self.client.beta.chat.completions.parse(
model=self.model,
messages=[
{"role": "system", "content": "You are a role comparison assistant."},
{"role": "user", "content": prompt}
],
max_tokens=1024,
temperature=0.1
)
# Parse LLM response (assuming it returns a structured JSON with assigned, unassigned, and unavailable roles)
comparison_result = json.loads(response.choices[0].message.content)
# Return the result as a JSON response
return jsonify(comparison_result), 200
except Exception as e:
return jsonify({"error": "Processing error", "message": f"An error occurred: {str(e)}"}), 500
+155 -113
View File
@@ -5,117 +5,9 @@ from pydantic import BaseModel, Field
from typing import List, Dict, Optional
from src.prompts.sops import *
from src.models.response_schemas import *
class SopGeneratorDocument:
def __init__(self):
self.api_key = os.getenv("OPENAI_API_KEY")
self.client = OpenAI(api_key=self.api_key)
self.model = "gpt-4o-2024-08-06"
def _extract_text_from_docs(self, docs):
"""Extract text content from document objects."""
return [doc.page_content for doc in docs]
# Existing methods...
def extract_sops_from_doc(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_sop_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=SOPsResponse,
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_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
from src.services.document_parser import DocumentParser
from dotenv import load_dotenv
load_dotenv()
@@ -271,7 +163,155 @@ class SopGeneratorExecutive:
def generate_sops_for_department_managers(self, docs):
try:
# First, extract departments and managers
sop_doc = SopGeneratorDocument()
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"
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 = DocumentParser()
departments_and_roles = sop_doc.extract_departments_and_managers(docs)
if not departments_and_roles or not departments_and_roles.get('departments'):
@@ -363,5 +403,7 @@ class SopGeneratorExecutive:
except Exception as e:
print(f"Error in generate_sops_from_questionnaire: {str(e)}")
return False
+9 -76
View File
@@ -1,81 +1,14 @@
from src.services.sop_generator import SopGeneratorDocument, SopGeneratorExecutive
from src.services.sop_generator import SopGeneratorExecutive
from src.utils.document_loader import load_document
file_path = "/root/ds_erp_ai/data/raw/document.doc"
from src.services.document_parser import DocumentParser
from src.services.sop_generator import SopGeneratorExecutive
file_path = r"C:\Users\User\Desktop\Blessing_AI\MKD\test_erp_ai\erp_ai\test\erp_ai\data\raw\document.doc"
docs = load_document(file_path)
sop_doc = SopGeneratorDocument()
sop_executive = SopGeneratorExecutive()
if __name__ == "__main__":
# Test the generate_sops_from_questionnaire function
questionnaire_data = {
"organizational_vision": {
"question_1_answer": "Our vision is to lead the market in innovative product solutions.",
"question_2_answer": "We see our organization contributing by expanding into new regions and enhancing service quality.",
"question_3_answer": "The key elements of our vision focus on scalability, customer satisfaction, and technological advancement."
},
"organizational_strategic_goals": {
"question_1_answer": "The strategic direction is to increase market share and improve operational efficiency.",
"question_2_answer": "We aim to achieve a 20% reduction in operating costs by streamlining internal processes.",
"question_3_answer": "The process aligns with our goal to reduce costs while maintaining high product quality."
},
"departmental_strategic_goals": {
"sales_department_answer": "Increase sales by 15% in the next fiscal year through better lead generation and customer retention.",
"finance_department_answer": "Ensure a balanced budget by optimizing resource allocation and reducing overhead costs."
}
}
executives = ["CEO", "COO"]
managers = ["Sales Manager", "Finance Manager"]
SOP = DocumentParser()
so = SopGeneratorExecutive()
info = SOP.extract_departments_and_managers_workers(docs)
print(info)
sops_from_questionnaire = sop_executive.generate_sops_from_questionnaire(questionnaire_data, executives, managers)
if sops_from_questionnaire:
print("Generated SOPs from Questionnaire:")
# Print Executive SOPs
print("\nExecutive SOPs:")
for executive, sops in sops_from_questionnaire["executive_sops"].items():
print(f" {executive}:")
print(" Must SOPs:")
for sop in sops.must:
print(f" - {sop}")
print(" Shall SOPs:")
for sop in sops.shall:
print(f" - {sop}")
print(" Will SOPs:")
for sop in sops.will:
print(f" - {sop}")
# Print Department Manager SOPs
print("\nDepartment Manager SOPs:")
for department in sops_from_questionnaire["department_sops"].departments:
print(f" Department: {department.name}")
for manager in department.managers:
print(f" Manager: {manager.title}")
print(" Must SOPs:")
for sop in manager.sops.must:
print(f" - {sop}")
print(" Shall SOPs:")
for sop in manager.sops.shall:
print(f" - {sop}")
print(" Will SOPs:")
for sop in manager.sops.will:
print(f" - {sop}")
else:
print("Failed to generate SOPs from questionnaire.")
# You can keep the previous tests if you want
departments_and_roles = sop_doc.extract_departments_and_managers(docs)
if departments_and_roles:
print("\nExtracted Departments and Roles:")
for department in departments_and_roles.get('departments', []):
print(f"\nDepartment: {department['name']}")
for role in department.get('managerial_roles', []):
print(f" Role: {role['title']}")
print(f" Responsibilities: {', '.join(role['responsibilities'])}")
else:
print("Failed to extract departments and roles.")
v_ms = sop_doc.extract_vision_mission(docs)
print(f"\nVision and Mission: {v_ms}")