Spaces:
Sleeping
Sleeping
File size: 5,288 Bytes
9270f6b 3237695 9270f6b 3237695 9270f6b 3237695 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 | 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)
variants = [value for value in data.values()]
# Fetch first three items from the array or pad with empty strings if not enough items are available
paddedVariants = variants[:3] + [''] * (3 - len(variants))
return paddedVariants
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) |