import ast import json import os import statistics from collections import Counter from typing import Any, Dict, List import langchain import openai import pandas as pd import regex as re import requests from dotenv import load_dotenv from langchain import OpenAI from langchain.chains.combine_documents.stuff import StuffDocumentsChain from langchain.chains.llm import LLMChain 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, PromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field from langchain_openai import ChatOpenAI 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) # return filtered_data # # for ele in filtered_data: # # print(ele) def process_dictionaty_with_llm_to_generate_response(json_data): # processed_data = process_json_data_for_llm(json_data) # res = tagging_chain.invoke({"input": processed_data}) # return res # Iterate through the dictionary and keep only the desired fields filtered_data = [] for item in json_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) return filtered_data def get_short_summary_out_of_json_files(data_json): prompt_template = """You are an expert on clinicial trials and their analysis of their reports. # Task You will be given a text of descriptions of multiple clinical trials realed to similar diseases. Your job is to come up with a short and detailed summary of the descriptions of the clinical trials. Your users are clinical researchers, so you should be technical and specific, including scientific terms in the summary. {text}""" 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. Format your answer in Markdown format, **highlighting** the most important concepts, and _italicizing_ the technical concepts extracted from the reports. Be very specific about the details of the clinical trials. {text} General summary:""" prompt = PromptTemplate.from_template(prompt_template) llm = ChatOpenAI( temperature=0.5, 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_summarization: {result}") return result def analyze_data(data): print(f"Data: {data}") # Extract minimum and maximum ages: Turn ['18 Years', '20 Years'] into [18, 20] min_ages = [ int(re.search(r"\d+", age).group()) for age in data["minimum_age"] if age ] max_ages = [ int(re.search(r"\d+", age).group()) for age in data["maximum_age"] if age ] # primary_timeframe= [int(age.split()[0]) for age in data['[primary_outcome]'] if age] # Calculate average minimum and maximum ages avg_min_age = statistics.mean(min_ages) if min_ages else None avg_max_age = statistics.mean(max_ages) if max_ages else None # Find most common gender gender_counter = Counter(data["gender"]) most_common_gender = gender_counter.most_common(1)[0][0] # Flatten keywords list and find common keywords # keywords = [keyword for sublist in data["keywords"] for keyword in sublist] # common_keywords = [word for word, count in Counter(keywords).most_common()] return { "avg_min_age": avg_min_age, "avg_max_age": avg_max_age, "most_common_gender": most_common_gender, } def tagging_insights_from_json(data_json): processed_json = process_dictionaty_with_llm_to_generate_response(data_json) tagging_prompt = ChatPromptTemplate.from_template( """ You are an expert on clinicial trials and analysis of their reports. Extract the desired information from the following JSON data. Only extract the properties mentioned in the 'Classification' function. Output a list of the extracted properties, starting with [ and ending with ]. JSON data: {input} """ ) 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 titles 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 for each 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 timeframe of each clinical trial" # ) # secondary_outcomes: list= Field(description= "get the secondary outcomes of each clinical trial") # eligibility: list = Field( # description="get the timeframe of each clinical trial" # ) # 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 { # "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-turbo", openai_api_key=os.environ["OPENAI_API_KEY"], ).with_structured_output(Classification) # stuff_chain = StuffDocumentsChain(llm_chain=llm, document_variable_name="text") tagging_chain = tagging_prompt | llm res = tagging_chain.invoke({"input": processed_json}) unprocessed_results_dict = res.get_dict() results_dict = analyze_data(unprocessed_results_dict) # stats_dict= {'Average Minimum age': avg_min_age, # 'Average Maximum age': avg_max_age, # 'Most common gender undergoing the trials': most_common_gender, # 'common keywords found in the trials': common_keywords} print(f"Result_tagging: {results_dict}") return results_dict # 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 = tagging_insights_from_json(json_data)