Spaces:
Runtime error
Runtime error
File size: 5,823 Bytes
1acafb2 | 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 | 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['CHAT_GPT_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):
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.api_key}'
}
data = {
'model': 'gpt-3.5-turbo-1106',
'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 ""}
"""
print(patientData)
return patientData
def processOutputResults(self, jsonOutput):
preprocessed_json_string = jsonOutput.replace('\n', '')
data = json.loads(preprocessed_json_string)
dataLength=len(data.keys())
result_list = []
for key, value in data.items():
result_list.append(key)
for variant, sentence in value.items():
result_list.append(sentence)
if dataLength < 5:
additional_keys_needed = 5 - dataLength
for key in range(additional_keys_needed):
result_list.append(f"Keyword {dataLength+key+1}") # Empty key
for _ in range(3):
result_list.append("") # Empty sentence
return result_list
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 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 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)
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)
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 getRephrasedSentences(self, sentences):
response = self.request_gpt_for_response(Prompts.REPHRASE_PROMPT.value,sentences)
return response |