import ast import json import os from typing import Any, Dict, List import langchain import openai import pandas as pd import requests from dotenv import load_dotenv from langchain import OpenAI from langchain.chains.combine_documents.stuff import StuffDocumentsChain from langchain.chains.qa_with_sources.loading import load_qa_with_sources_chain from langchain.document_loaders import UnstructuredURLLoader from langchain.embeddings import OpenAIEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores import FAISS from langchain_community.document_loaders import JSONLoader from langchain_community.document_loaders.csv_loader import CSVLoader from langchain_core.prompts import ChatPromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI from langchain.chains.llm import LLMChain from langchain_core.prompts import PromptTemplate load_dotenv() # getting the json files def get_clinical_record_info(clinical_record_id: str) -> Dict[str, Any]: # Request: # curl -X GET "https://clinicaltrials.gov/api/v2/studies/NCT00841061" \ # -H "accept: text/csv" request_url = f"https://clinicaltrials.gov/api/v2/studies/{clinical_record_id}" response = requests.get(request_url, headers={"accept": "application/json"}) return response.json() def get_clinical_records_by_ids(clinical_record_ids: List[str]) -> List[Dict[str, Any]]: clinical_records = [] for clinical_record_id in clinical_record_ids: clinical_record_info = get_clinical_record_info(clinical_record_id) clinical_records.append(clinical_record_info) return clinical_records def process_json_data_for_llm(data): # Define the fields you want to keep fields_to_keep = [ "class_of_organization", "title", "overallStatus", "descriptionModule", "conditions", "interventions", "outcomesModule", "eligibilityModule", ] # Iterate through the dictionary and keep only the desired fields filtered_data = [] for item in data: try: organization_name = item["protocolSection"]["identificationModule"][ "organization" ]["fullName"] except: organization_name = "" try: project_title = item["protocolSection"]["identificationModule"][ "officialTitle" ] except: project_title = "" try: status = item["protocolSection"]["statusModule"]["overallStatus"] except: status = "" try: briefDescription = item["protocolSection"]["descriptionModule"][ "briefSummary" ] except: briefDescription = "" try: detailedDescription = item["protocolSection"]["descriptionModule"][ "detailedDescription" ] except: detailedDescription = "" try: conditions = item["protocolSection"]["conditionsModule"]["conditions"] except: conditions = [] try: keywords = item["protocolSection"]["conditionsModule"]["keywords"] except: keywords = [] try: interventions = item["protocolSection"]["armsInterventionsModule"][ "interventions" ] except: interventions = [] try: primary_outcomes = item["protocolSection"]["outcomesModule"][ "primaryOutcomes" ] except: primary_outcomes = [] try: secondary_outcomes = item["protocolSection"]["outcomesModule"][ "secondaryOutcomes" ] except: secondary_outcomes = [] try: eligibility = item["protocolSection"]["eligibilityModule"] except: eligibility = {} filtered_item = { "organization_name": organization_name, "project_title": project_title, "status": status, "briefDescription": briefDescription, "detailedDescription": detailedDescription, "keywords": keywords, "interventions": interventions, "primary_outcomes": primary_outcomes, "secondary_outcomes": secondary_outcomes, "eligibility": eligibility, } filtered_data.append(filtered_item) # for ele in filtered_data: # print(ele) def get_short_summary_out_of_json_files(data_json): prompt_template = """ You are an expert clinician working on the analysis of reports of clinical trials. # Task You will be given a set of descriptions of clinical trials. Your job is to come up with a short summary (100-200 words) of the descriptions of the clinical trials. Your users are clinical researchers who are experts in medicine, so you should be technical and specific, including scientific terms. Always be faithful to the original information written in the reports. To write your summary, you will need to read the following examples, labeled as "Report 1", "Report 2", and so on. Your answer should be a single paragraph (100-200 words) that summarizes the general content of all the reports. {text} General summary:""" prompt = PromptTemplate.from_template(prompt_template) llm = ChatOpenAI( temperature=0.4, model_name="gpt-4-turbo", api_key=os.environ["OPENAI_API_KEY"] ) llm_chain = LLMChain(llm=llm, prompt=prompt) # Define StuffDocumentsChain stuff_chain = StuffDocumentsChain( llm_chain=llm_chain, document_variable_name="text" ) descriptions = [ ( x["detailedDescription"] if "detailedDescription" in x and len(x["detailedDescription"]) > 0 else x["briefSummary"] ) for x in data_json if "detailedDescription" in x or "briefSummary" in x ] combined_descriptions = "" for i, description in enumerate(descriptions): combined_descriptions += f"Report {i+1}:\n{description}\n" print(f"Combined descriptions: {combined_descriptions}") result = stuff_chain.run(combined_descriptions) print(f"Result: {result}") return result def taggingTemplate(): class Classification(BaseModel): description: str = Field( description="text description grouping all the clinical trials using briefDescription and detailedDescription keys" ) project_title: list = Field( description="Extract the project title of all the clinical trials" ) status: list = Field( description="Extract the status of all the clinical trials" ) # keywords: list = Field( # description="Extract the most relevant keywords regrouping all the clinical trials" # ) interventions: list = Field( description="describe the interventions for each clinical trial using title, name and description" ) primary_outcomes: list = Field( description="get the primary outcomes of each clinical trial" ) # secondary_outcomes: list= Field(description= "get the secondary outcomes of each clinical trial") # eligibility: list = Field( # description="get the eligibilityCriteria grouping all the clinical trials" # ) # healthy_volunteers: list= Field(description= "determine whether the clinical trial requires healthy volunteers") # minimum_age: list = Field( # description="get the minimum age from each experiment" # ) # maximum_age: list = Field( # description="get the maximum age from each experiment" # ) # gender: list = Field(description="get the gender from each experiment") def get_dict(self): return { "summary": self.description, "project_title": self.project_title, "status": self.status, "keywords": self.keywords, "interventions": self.interventions, "primary_outcomes": self.primary_outcomes, # "secondary_outcomes": self.secondary_outcomes, "eligibility": self.eligibility, # "healthy_volunteers": self.healthy_volunteers, "minimum_age": self.minimum_age, "maximum_age": self.maximum_age, "gender": self.gender, } # LLM llm = ChatOpenAI( temperature=0.6, model="gpt-4", openai_api_key=os.environ["OPENAI_API_KEY"], ).with_structured_output(Classification) stuff_chain = StuffDocumentsChain(llm_chain=llm, document_variable_name="text") # tagging_chain = prompt_template | llm # return tagging_chain # clinical_record_info = get_clinical_records_by_ids(['NCT00841061', 'NCT03035123', 'NCT02272751', 'NCT03035123', 'NCT03055377']) # print(clinical_record_info) # with open('data.json', 'w') as f: # json.dump(clinical_record_info, f, indent=4) # tagging_chain = llm_config() def process_dictionaty_with_llm_to_generate_response(json_contents): processed_data = process_json_data_for_llm(json_contents) # res = tagging_chain.invoke({"input": processed_data}) # return res