Spaces:
Runtime error
Runtime error
from fastapi import FastAPI, File, UploadFile | |
from fastapi.responses import StreamingResponse | |
from fastapi.responses import FileResponse, HTMLResponse | |
import os | |
import io | |
from fastapi.middleware.cors import CORSMiddleware | |
import google.generativeai as genai | |
import torch | |
from auto_gptq import AutoGPTQForCausalLM | |
from langchain import HuggingFacePipeline, PromptTemplate | |
from langchain.chains import RetrievalQA | |
from langchain.document_loaders import PyPDFDirectoryLoader | |
from langchain.embeddings import HuggingFaceInstructEmbeddings | |
from langchain.text_splitter import RecursiveCharacterTextSplitter | |
from langchain.vectorstores import Chroma | |
from langchain.vectorstores import FAISS | |
from transformers import AutoTokenizer, TextStreamer, pipeline | |
"""Patient Response : hi doctor \n\n"What brings you here today? Please provide me with some background information about your health so far." \nPatient Response : im arafath i have fever""" | |
DEVICE = "cuda:0" if torch.cuda.is_available() else "cpu" | |
GOOGLE_API_KEY = os.environ['GOOGLE_API_KEY'] | |
genai.configure(api_key=GOOGLE_API_KEY) | |
gemini_model = genai.GenerativeModel('gemini-pro') | |
try: from pip._internal.operations import freeze | |
except ImportError: # pip < 10.0 | |
from pip.operations import freeze | |
pkgs = freeze.freeze() | |
for pkg in pkgs: print(pkg) | |
embeddings = HuggingFaceInstructEmbeddings( | |
model_name="hkunlp/instructor-large", model_kwargs={"device": DEVICE} | |
) | |
new_db = FAISS.load_local("faiss_index", embeddings) | |
model_name_or_path = "TheBloke/Llama-2-13B-chat-GPTQ" | |
model_basename = "model" | |
tokenizer = AutoTokenizer.from_pretrained(model_name_or_path, use_fast=True) | |
model = AutoGPTQForCausalLM.from_quantized( | |
model_name_or_path, | |
revision="gptq-4bit-128g-actorder_True", | |
model_basename=model_basename, | |
use_safetensors=True, | |
trust_remote_code=True, | |
device=DEVICE, | |
inject_fused_attention=False, | |
quantize_config=None, | |
) | |
#default promts it will work when we don't set the our custom system propts | |
DEFAULT_SYSTEM_PROMPT = """ | |
You are a helpful, respectful and honest assistant. give answer for any questions. | |
""".strip() | |
def generate_prompt(prompt: str, system_prompt: str = DEFAULT_SYSTEM_PROMPT) -> str: | |
return f""" | |
[INST] <<SYS>> | |
{system_prompt} | |
<</SYS>> | |
{prompt} [/INST] | |
""".strip() | |
# setting the RAG pipeline | |
streamer = TextStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True) | |
text_pipeline = pipeline( | |
"text-generation", | |
model=model, | |
tokenizer=tokenizer, | |
max_new_tokens=4096, | |
temperature=2, | |
top_p=0.95, | |
repetition_penalty=1.15, | |
streamer=streamer, | |
) | |
global llm,llm2,llm3 | |
llm = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
llm2 = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
llm3 = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
# when the user query is not related to trained PDF data model will give the response from own knowledge | |
SYSTEM_PROMPT = "give answer from external data's. don't use the provided context" | |
template = generate_prompt( | |
""" | |
{context} | |
Question: {question} | |
""", | |
system_prompt=SYSTEM_PROMPT, | |
) | |
prompt = PromptTemplate(template=template, input_variables=["context", "question"]) | |
global qa_chain,qa_chain_a,qa_chain_v | |
qa_chain_v = RetrievalQA.from_chain_type( | |
llm=llm3, | |
chain_type="stuff", | |
retriever=store.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
qa_chain = RetrievalQA.from_chain_type( | |
llm=llm, | |
chain_type="stuff", | |
retriever=new_db.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
qa_chain_a = RetrievalQA.from_chain_type( | |
llm=llm2, | |
chain_type="stuff", | |
retriever=new_db.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
report_prompt_template = """ | |
this is report format | |
Patient Name: [Insert name here]<br> | |
Age: [Insert age here]<br> | |
sex: [Insert here]<br> | |
Chief Complaint: [insert here]<br> | |
History of Present Illness:[insert here]<br> | |
Past Medical History: [insert here]<br> | |
Medication List: [insert here]<br> | |
Social History: [insert here]<br> | |
Family History: [insert here]<br> | |
Review of Systems: [insert here]<br> | |
ICD Code: [insert here] | |
convert this bellow details into above format don't add any other details .don't use the provided pdfs data's.\n\n""" | |
# 4. prompt sets for ask some defined questions and its will guide the model correct way | |
final_question ={ | |
8:"Do you have a history of medical conditions, such as allergies, chronic illnesses, or previous surgeries? If so, please provide details.", | |
9:"What medications are you currently taking, including supplements and vitamins?", | |
10:"Can you please Describe Family medical history (particularly close relatives): Does anyone in your immediate family suffer from similar symptoms or health issues?", | |
11:"Can you please Describe Social history: Marital status, occupation, living arrangements, education level, and support system.", | |
12:"Could you describe your symptoms, and have you noticed any changes or discomfort related to your respiratory, cardiovascular, gastrointestinal, or other body systems?" | |
} | |
# 1 . basic first prompt for handled the llama in correct like a family physician | |
sys = "You are a general family physician.\n\n" | |
# 5 . prommpts for get the diagnosis with ICD code based on the conversation, its will handle unrelated questions also(not related to diagnosis) | |
end_sys_prompts = "\n\ngive correct treatment and most related diagnosis with ICD code don't ask any questions. if question is not related to provided data don't give answer from this provided data's" | |
def refresh_model(): | |
global llm,llm2,llm3 | |
llm = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
llm2 = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
llm3 = HuggingFacePipeline(pipeline=text_pipeline, model_kwargs={"temperature": 2}) | |
global qa_chain,qa_chain_a,qa_chain_v | |
qa_chain_v = RetrievalQA.from_chain_type( | |
llm=llm3, | |
chain_type="stuff", | |
retriever=store.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
qa_chain = RetrievalQA.from_chain_type( | |
llm=llm, | |
chain_type="stuff", | |
retriever=new_db.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
qa_chain_a = RetrievalQA.from_chain_type( | |
llm=llm2, | |
chain_type="stuff", | |
retriever=new_db.as_retriever(search_kwargs={"k": 2}), | |
return_source_documents=True, | |
chain_type_kwargs={"prompt": prompt}, | |
) | |
print("model refreshed") | |
app = FastAPI() | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
async def llm_response(chain,id,mode): | |
id = int(id) | |
global qa_chain,qa_chain_a,qa_chain_v | |
refresh_model() | |
def QA(): | |
print("\nopen QA mode running ========================================\n") | |
try: | |
print("\n google gemini===================\n") | |
gemini_chat = gemini_model.start_chat(history=[]) | |
gemini_response = gemini_chat.send_message('give next small response for laste patient response like a doctor. '+str(chain)) | |
return gemini_response.text | |
except: | |
print("\n llmmaa ===================\n") | |
result_ex = qa_chain(sys+chain+"""\n\n\nalways give small and single response based on the patient | |
response. don't ask any question give simple response""") | |
if "Patient:" in str(result_ex['result']) or "Patient response:" in str(result_ex['result']) or "Patient Response" in str(result_ex['result']): | |
return str((str(result_ex['result']).split("\n\n")[-1]).split(":")[-1]) | |
else: | |
try: | |
return str(result_ex['result']).split("\n\n")[1] | |
except: | |
return str(result_ex['result']).split("\n\n") | |
if str(mode)=="dirrect_QA" and id==3: | |
diagnosis_and_treatment = qa_chain_v(sys+chain+end_sys_prompts) | |
diagnosis_and_treatment = str(diagnosis_and_treatment['result']) | |
print(diagnosis_and_treatment) | |
print("dirrect answer") | |
return diagnosis_and_treatment | |
if str(mode)=="dirrect_QA" and id>3: | |
return QA() | |
if id<13: | |
if id>=8: | |
return final_question[id] | |
else: | |
if id<5: | |
# 2 . prompmt control the natural way on question asking based on patient response,symptomps type | |
question = qa_chain(sys+chain +"""\n\nask single small queston to get details based on the patient response,and don't ask | |
same question again, and don't provide treatment and diagnosis ask next small and short question , | |
always don't ask same question again and again , always only ask next single small question""") | |
else: | |
# 3. prompt will guide the model to ask yes or no questions based on patient response,symptomps type | |
question = qa_chain(sys+chain +"""\n\nask single small queston to get details based on the patient response,and don't ask | |
same question again, and don't provide treatment and diagnosis ask next small and short question with yes or no format , | |
always don't ask same question again and again , always only ask next single small question""") | |
try: | |
if "Patient:" in str(question['result']) or "Patient response:" in str(question['result']) or "Patient Response" in str(question['result']): | |
return str((str(question['result']).split("\n\n")[-1]).split(":")[-1]) | |
else: | |
return str(question['result']).split("\n\n")[1] | |
except: | |
if "Patient:" in str(question['result']) or "Patient response:" in str(question['result']) or "Patient Response" in str(question['result']): | |
return str(question['result']).split(":")[-1] | |
else: | |
return str(question['result']) | |
if id==13: | |
diagnosis_and_treatment = qa_chain_v(sys+chain+end_sys_prompts) | |
diagnosis_and_treatment = str(diagnosis_and_treatment['result']) | |
print(mode,diagnosis_and_treatment) | |
report = qa_chain_a(report_prompt_template+sys+chain+"\n\ntreatment & diagnosis with ICD code below\n"+diagnosis_and_treatment) | |
print(str(report['result'])) | |
print("h&P") | |
return str(report['result']) | |
return QA() | |