Spaces:
Sleeping
Sleeping
File size: 7,150 Bytes
9270f6b 2492d3e 9270f6b dea5118 9270f6b 2492d3e 9270f6b 2492d3e 9270f6b dea5118 9270f6b 2492d3e 9270f6b 2492d3e 9270f6b 2492d3e 9270f6b 2492d3e 9270f6b dea5118 9270f6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | import os
import requests
from enum import Enum
from dotenv import load_dotenv
from prompts import Prompts
import json
# Load environment variables from .env file
load_dotenv()
# Access environment variables
API_KEY = os.environ['openAI_key']
class Stage(Enum):
INITIAL_ASSESSMENT = "Initial Assessment"
FOLLOWUP_ASSESSMENT = "Follow Up Assessment"
EVALUATION = "Evaluation"
DETAILED_EVALUATION = "Detailed Evaluation"
PROGRESS_NOTE = "Progress Note"
RECOMMENDATION="Recommendations"
DISCHARGE_SUMMARY = "Discharge Summary"
SHORT_TERM_GOALS = "Short Term Goals"
LONG_TERM_GOALS = "Long Term Goals"
class GPT_Service:
def __init__(self):
self.api_key = API_KEY
self.base_url = 'https://api.openai.com/v1/chat/completions'
def request_gpt_for_response(self, system_prompt, userPrompt):
print(f"System Prompt: {system_prompt}\n user Prompt: {userPrompt}\n")
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.api_key}'
}
data = {
'model': 'gpt-4-turbo',
'temperature': 0.3,
'messages': [
{
'role': 'system',
'content': system_prompt
},
{
'role': 'user',
'content': userPrompt
}
]
}
try:
response = requests.post(
self.base_url,
json=data,
headers=headers,
)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
except requests.exceptions.RequestException as e:
raise ValueError(f"Error: {e}")
def formulateUserPrompt(self, assessmentPhase, keywords, medicalHistory):
patientData= f"""
Keywords: {keywords}.
{f"Patient has the following medical history: {medicalHistory}" if medicalHistory is not None and medicalHistory != ""
else ""}
"""
return patientData
def formulateUserPromptForDetailAssessment(self, rawPatientInformation, medicalHistory):
patientData= f"""
Sentences: {rawPatientInformation}.
{f"Patient has the following medical history: {medicalHistory}" if medicalHistory is not None and medicalHistory != ""
else ""}
"""
return patientData
def processOutputResults(self, jsonOutput):
preprocessed_json_string = jsonOutput.replace('\n', '')
data = json.loads(preprocessed_json_string)
result_list = []
for key, value in data.items():
for variant, sentence in value.items():
result_list.append(sentence)
return result_list
def getInitialAssessment(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.INITIAL_ASSESSMENT.value,keywords,medicalHistory)
response= self.request_gpt_for_response(Prompts.INITIAL_ASSESSMENT_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getShortTermGoals(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.SHORT_TERM_GOALS.value,keywords,medicalHistory)
response = self.request_gpt_for_response(Prompts.SHORT_TERM_GOALS_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getLongTermGoals(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.LONG_TERM_GOALS.value,keywords,medicalHistory)
response = self.request_gpt_for_response(Prompts.LONG_TERM_GOALS_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getRecommendation(self, keywords, medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.RECOMMENDATION.value,keywords,medicalHistory)
response= self.request_gpt_for_response(Prompts.RECOMMENDATION_PROMPT.value,userPrompt)
return self.processOutputResults(response)
# Detail Patient Evaluation Prompt
def getDetailInitialAssessment(self,medicalHistory,sentences):
rawPatientAssessment = " ".join(sentences)
userPrompt= self.formulateUserPromptForDetailAssessment(rawPatientAssessment,medicalHistory)
response= self.request_gpt_for_response(Prompts.DETAIL_PATIENT_ASSESSMENT_INITIAL_ASSESSMENT_GOALS.value,userPrompt)
return response
def getDetailShortTermGoals(self,medicalHistory,sentences):
rawPatientAssessment = " ".join(sentences)
userPrompt= self.formulateUserPromptForDetailAssessment(rawPatientAssessment,medicalHistory)
response= self.request_gpt_for_response(Prompts.DETAIL_PATIENT_ASSESSMENT_SHORT_TERM_GOALS.value,userPrompt)
return response
def getDetailLongTermGoals(self,medicalHistory,sentences):
rawPatientAssessment = " ".join(sentences)
userPrompt= self.formulateUserPromptForDetailAssessment(rawPatientAssessment,medicalHistory)
response= self.request_gpt_for_response(Prompts.DETAIL_PATIENT_ASSESSMENT_LONG_TERM_GOALS.value,userPrompt)
return response
def getDetailRecommendation(self, medicalHistory,sentences):
rawPatientAssessment = " ".join(sentences)
userPrompt= self.formulateUserPromptForDetailAssessment(rawPatientAssessment,medicalHistory)
response= self.request_gpt_for_response(Prompts.DETAIL_PATIENT_ASSESSMENT_RECOMMENDATION_GOALS.value,userPrompt)
return response
# Extra helper functions
def getFollowUpAssessment(self, keywords, medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.FOLLOWUP_ASSESSMENT.value,keywords,medicalHistory)
response= self.request_gpt_for_response(Prompts.FOLLOWUP_ASSESSMENT_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getEvaluation(self, keywords, medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.EVALUATION.value,keywords,medicalHistory)
response= self.request_gpt_for_response(Prompts.EVALUATION_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getDetailedEvaluation(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.DETAILED_EVALUATION.value,keywords,medicalHistory)
response= self.request_gpt_for_response(Prompts.DETAILED_EVALUATION_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getProgressNote(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.PROGRESS_NOTE.value,keywords,medicalHistory)
response = self.request_gpt_for_response(Prompts.PROGRESS_NOTE_PROMPT.value,userPrompt)
return self.processOutputResults(response)
def getDischargeSummary(self, keywords,medicalHistory):
userPrompt= self.formulateUserPrompt(Stage.DISCHARGE_SUMMARY.value,keywords,medicalHistory)
response = self.request_gpt_for_response(Prompts.DISCHARGE_SUMMARY_PROMPT.value,userPrompt)
return self.processOutputResults(response) |