llm_res.py

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(
        """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 ], for each of the properties.

Raw data (in JSON format):
{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)