sop for execuitve generator added
This commit is contained in:
+119
-2
@@ -59,7 +59,7 @@ def get_roles():
|
||||
|
||||
|
||||
|
||||
@sops_bp.route('/generate_sops', methods=['POST'])
|
||||
@sops_bp.route('/generate_sops_from_doc', methods=['POST'])
|
||||
def generate_sops():
|
||||
# Check if the POST request has the file part
|
||||
if 'document' not in request.files:
|
||||
@@ -101,7 +101,7 @@ def generate_sops():
|
||||
return jsonify({"error": "Document cannot extract SOPs", "message": status_check["message"]}), 400
|
||||
|
||||
# Generate SOPs based on the roles provided
|
||||
sops = sop_generator.generate_sops(roles, docs)
|
||||
sops = sop_generator.generate_sops_from_doc(roles, docs)
|
||||
|
||||
# Cleanup: Delete all files in the upload directory after processing
|
||||
delete_all_files_in_directory(upload_folder)
|
||||
@@ -114,3 +114,120 @@ def generate_sops():
|
||||
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the document: {str(e)}"}), 500
|
||||
|
||||
return jsonify({"error": "File type not allowed", "message": "The uploaded file type is not allowed. Please upload a PDF, DOC, or DOCX file."}), 400
|
||||
|
||||
|
||||
|
||||
@sops_bp.route('/generate_sops_from_info', methods=['POST'])
|
||||
def generate_sops_from_info():
|
||||
"""
|
||||
Generate SOPs based on role information provided in the request body.
|
||||
"""
|
||||
try:
|
||||
# Get role information from the request body
|
||||
roles_info = request.json.get('roles_info')
|
||||
if not roles_info:
|
||||
return jsonify({"error": "No role information provided", "message": "Please provide role information in the 'roles_info' field."}), 400
|
||||
|
||||
# Generate SOPs based on the provided role information
|
||||
sops_response = sop_generator.generate_sops_from_info(roles=roles_info)
|
||||
|
||||
return jsonify({"sops": sops_response, "message": "SOPs successfully generated based on the provided role information."}), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": "Processing error", "message": f"An error occurred while generating SOPs: {str(e)}"}), 500
|
||||
|
||||
|
||||
@sops_bp.route('/generate_sops_by_role_and_area', methods=['POST'])
|
||||
def generate_sops_by_role_and_area():
|
||||
try:
|
||||
# Get role and area from the request body
|
||||
role = request.json.get('role')
|
||||
area = request.json.get('area')
|
||||
|
||||
if not role or not area:
|
||||
return jsonify({"error": "Missing parameters", "message": "Both 'role' and 'area' fields are required."}), 400
|
||||
|
||||
# Generate SOPs based on the provided role and area
|
||||
sops_response = sop_generator.generate_sops_by_role_and_area(role=role, area=area)
|
||||
|
||||
return jsonify({"sops": sops_response, "message": f"SOPs successfully generated for role '{role}' in area '{area}'."}), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": "Processing error", "message": f"An error occurred while generating SOPs: {str(e)}"}), 500
|
||||
|
||||
|
||||
|
||||
@sops_bp.route('/executive/generate_sops_from_questionnaire', methods=['POST'])
|
||||
def generate_executive_sops_from_questionnaire():
|
||||
try:
|
||||
# Get data from the request body
|
||||
data = request.json
|
||||
|
||||
# Generate SOPs based on the questionnaire answers
|
||||
sops_response = sop_generator.generate_executive_sops_from_questionnaire(data)
|
||||
|
||||
return jsonify({"sops": sops_response, "message": "SOPs successfully generated from the questionnaire."}), 200
|
||||
|
||||
except Exception as e:
|
||||
return jsonify({"error": "Processing error", "message": f"An error occurred while generating SOPs: {str(e)}"}), 500
|
||||
|
||||
|
||||
|
||||
|
||||
@sops_bp.route('/executive/generate_sops_from_doc', methods=['POST'])
|
||||
def generate_executive_sops_from_doc():
|
||||
"""
|
||||
Generate SOPs for executives based on a document containing vision and mission.
|
||||
"""
|
||||
# Check if the POST request has the file part
|
||||
if 'document' not in request.files:
|
||||
return jsonify({"error": "No file part", "message": "Please upload a file with the key 'document'."}), 400
|
||||
|
||||
file = request.files['document']
|
||||
|
||||
# If the user does not select a file, the browser may also submit an empty part without filename
|
||||
if file.filename == '':
|
||||
return jsonify({"error": "No selected file", "message": "A file was not selected for upload. Please select a valid file."}), 400
|
||||
|
||||
if file and allowed_file(file.filename):
|
||||
filename = secure_filename(file.filename)
|
||||
upload_folder = current_app.config['UPLOAD_FOLDER']
|
||||
file_path = os.path.join(upload_folder, filename)
|
||||
|
||||
# Save the file to the upload folder
|
||||
file.save(file_path)
|
||||
|
||||
try:
|
||||
# Use the utility function to generate docs from the file
|
||||
docs = load_document(file_path)
|
||||
|
||||
|
||||
# Use LLM to extract Vision and Mission sections from the document
|
||||
vision_section, mission_section = sop_generator.extract_vision_and_mission(docs)
|
||||
|
||||
if not vision_section or not mission_section:
|
||||
# Cleanup: Delete all files in the upload directory if parsing fails
|
||||
delete_all_files_in_directory(upload_folder)
|
||||
return jsonify({"error": "Missing Vision and Mission", "message": "The document does not contain or properly define the company's vision and mission."}), 400
|
||||
|
||||
# Organize extracted data
|
||||
extracted_data = {
|
||||
"role": "Executive",
|
||||
"organization vision": [vision_section],
|
||||
"organization strategic goals": [mission_section]
|
||||
}
|
||||
|
||||
# Generate SOPs based on the extracted vision and goals
|
||||
sops_response = sop_generator.generate_executive_sops_from_questionnaire(extracted_data)
|
||||
|
||||
# Cleanup: Delete all files in the upload directory after processing
|
||||
delete_all_files_in_directory(upload_folder)
|
||||
|
||||
return jsonify({"sops": sops_response, "message": "SOPs successfully generated from the document."}), 200
|
||||
|
||||
except Exception as e:
|
||||
# Cleanup: Delete all files in the upload directory if an error occurs
|
||||
delete_all_files_in_directory(upload_folder)
|
||||
return jsonify({"error": "Processing error", "message": f"An error occurred while processing the document: {str(e)}"}), 500
|
||||
|
||||
return jsonify({"error": "File type not allowed", "message": "The uploaded file type is not allowed. Please upload a PDF, DOC, or DOCX file."}), 400
|
||||
|
||||
@@ -22,6 +22,12 @@ class RolesResponse(BaseModel):
|
||||
class SOPsResponse(BaseModel):
|
||||
roles_sops: Dict[str, SOPs] = Field(default_factory=dict)
|
||||
|
||||
class VisionMissionResponse(BaseModel):
|
||||
vision: Optional[str]
|
||||
mission: Optional[str]
|
||||
message: str
|
||||
|
||||
|
||||
class SopGenerator:
|
||||
def __init__(self):
|
||||
self.api_key = os.getenv("OPENAI_API_KEY")
|
||||
@@ -77,7 +83,7 @@ class SopGenerator:
|
||||
)
|
||||
return json.loads(response.choices[0].message.content)
|
||||
|
||||
def generate_sops(self, roles: List[str], docs) -> SOPsResponse:
|
||||
def generate_sops_from_doc(self, roles: List[str], docs) -> SOPsResponse:
|
||||
roles_sops_all = {}
|
||||
|
||||
docs_text = self._extract_text_from_docs(docs)
|
||||
@@ -88,14 +94,26 @@ class SopGenerator:
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": f'''You are a Standard Operating Procedure (SOP) extractor.
|
||||
Your task is to find SOPs for the role "{role}" in the provided text.
|
||||
SOPs should be categorized under "must", "shall", and "will".
|
||||
If the SOPs for the role are not explicitly stated, you are required to infer them from the context provided in the document,
|
||||
but only if there is clear evidence within the text.
|
||||
Do not generate or assume SOPs that are not directly supported by the document.
|
||||
Your extraction should strictly adhere to the content of the document, ensuring that no information is fabricated or inferred beyond what is present.
|
||||
If no SOPs are found for the role, return an empty list for each category.''',
|
||||
"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",
|
||||
@@ -108,6 +126,211 @@ class SopGenerator:
|
||||
)
|
||||
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, area: 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 area "{area}" based on the following instructions:
|
||||
|
||||
Instructions:
|
||||
|
||||
Categorization: Organize the SOPs under three categories: "must," "shall," and "will."
|
||||
Direct Instructions: The SOPs should directly address responsibilities, objectives, and challenges related to the area of "{area}" 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)
|
||||
|
||||
|
||||
print(F"extracted text:{extracted_text}")
|
||||
# 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
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ def delete_file(file_path):
|
||||
print(f"Error deleting file {file_path}: {e}")
|
||||
|
||||
|
||||
import os
|
||||
|
||||
def delete_all_files_in_directory(directory_path):
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user