# %% # imports import os import json from dotenv import load_dotenv from openai import OpenAI import gradio as gr import requests import urllib.parse from datetime import datetime # %% # Initialization load_dotenv(override=True) openai_api_key = os.getenv('OPENAI_API_KEY') if openai_api_key: print(f"OpenAI API Key exists and begins {openai_api_key[:8]}") else: print("OpenAI API Key not set") MODEL = "gpt-4o-mini" openai = OpenAI() # %% system_message = ( "You are a clinical trials assistant that uses the ClinicalTrials.gov API to answer questions. " "If the user asks for study data, you must always call the `search_studies` tool. " "Only use your own knowledge for greetings or general questions." "Always return detailed and structured responses." ) # %% def search_studies(query): import urllib.parse import requests print("\n==============================") print(f"[DEBUG] Received query: {query} (type: {type(query)})") print("==============================") if isinstance(query, dict): filters = [] query_parts = [] if "query" in query: query_string = urllib.parse.quote_plus(query["query"]) query_parts.append(f"query.cond={query_string}") if "phase" in query: filters.append(f"AREA[Phase]{query['phase']}") if "country" in query: filters.append(f"AREA[LocationCountry]{query['country']}") if "study_type" in query: filters.append(f"AREA[StudyType]{query['study_type']}") if "sex" in query: filters.append(f"AREA[Sex]{query['sex']}") if "age_group" in query: filters.append(f"AREA[StdAge]{query['age_group']}") if "status" in query: filters.append(f"AREA[OverallStatus]{query['status']}") if "sampling_method" in query: filters.append(f"AREA[SamplingMethod]{query['sampling_method']}") if "ipd_sharing" in query: filters.append(f"AREA[IPDSharing]{query['ipd_sharing']}") if "start_date_from" in query: filters.append(f"AREA[StartDate]RANGE[{query['start_date_from']},MAX]") if "start_date_to" in query: filters.append(f"AREA[StartDate]RANGE[MIN,{query['start_date_to']}]") if "completion_date_from" in query: filters.append(f"AREA[CompletionDate]RANGE[{query['completion_date_from']},MAX]") if "completion_date_to" in query: filters.append(f"AREA[CompletionDate]RANGE[MIN,{query['completion_date_to']}]") def normalize_sponsor(s): for suffix in ["Inc.", "Inc", "Ltd.", "Ltd", "GmbH", "LLC", "Corp.", "Corp"]: s = s.replace(suffix, "") return s.strip() if "sponsor" in query: sponsor_name = normalize_sponsor(query["sponsor"]) sponsor_string = urllib.parse.quote_plus(sponsor_name) query_parts.append(f"query.spons={sponsor_string}") page_size = query.get("max_results", 3) if ("city" in query or "facility" in query) and "max_results" not in query: page_size = 30 # ensure enough studies for post-filtering filter_advanced = " AND ".join(filters) if filter_advanced: filter_advanced = f"({filter_advanced})" encoded_filter = urllib.parse.quote(filter_advanced, safe="[]*") url = ( f"https://clinicaltrials.gov/api/v2/studies?" f"{'&'.join(query_parts)}" f"{f'&filter.advanced={encoded_filter}' if filter_advanced else ''}" f"&pageSize={page_size}" ) else: encoded_query = urllib.parse.quote_plus(query) url = f"https://clinicaltrials.gov/api/v2/studies?query.cond={encoded_query}&pageSize=3" print("Requesting:", url) try: response = requests.get(url) print("Status Code:", response.status_code) if response.status_code == 400 and any("query.cond=" in part for part in query_parts): print("[โš ๏ธ Fallback] Retrying with query.text instead of query.cond...") query_parts = [p.replace("query.cond=", "query.text=") for p in query_parts] url = ( f"https://clinicaltrials.gov/api/v2/studies?" f"{'&'.join(query_parts)}" f"{f'&filter.advanced={encoded_filter}' if filter_advanced else ''}" f"&pageSize={page_size}" ) print("Retrying:", url) response = requests.get(url) print("Retry Status Code:", response.status_code) if response.status_code == 200: data = response.json() trials = data.get("studies", []) if not trials: return "No studies found." def matches(study, key, value): section = study.get("protocolSection", {}) if key == "sponsor": return value.lower() in section.get("sponsorCollaboratorsModule", {}).get("leadSponsor", {}).get("name", "").lower() elif key == "intervention": interventions = section.get("armsInterventionsModule", {}).get("interventions", []) return any(value.lower() in i.get("name", "").lower() for i in interventions) elif key == "city": locs = section.get("contactsLocationsModule", {}).get("locations", []) return any(value.lower() in loc.get("city", "").lower() for loc in locs) elif key == "facility": locs = section.get("contactsLocationsModule", {}).get("locations", []) return any(value.lower() in loc.get("facility", "").lower() for loc in locs) return True for key in ["sponsor", "intervention", "city", "facility"]: if key in query: trials = [s for s in trials if matches(s, key, query[key])] if not trials: return "No studies found after applying filters." result = [] for study in trials: ps = study.get("protocolSection", {}) id_module = ps.get("identificationModule", {}) design_module = ps.get("designModule", {}) status_module = ps.get("statusModule", {}) elig_module = ps.get("eligibilityModule", {}) ipd_module = ps.get("ipdSharingStatementModule", {}) desc_module = ps.get("descriptionModule", {}) contact_module = ps.get("contactsLocationsModule", {}) sponsor_module = ps.get("sponsorCollaboratorsModule", {}) outcomes_module = ps.get("outcomesModule", {}) arms_module = ps.get("armsInterventionsModule", {}) nct_id = id_module.get("nctId", "N/A") title = id_module.get("briefTitle", "No Title") official_title = id_module.get("officialTitle", "N/A") phases = design_module.get("phases", []) study_type = design_module.get("studyType", "N/A") status = status_module.get("overallStatus", "N/A") start_date = status_module.get("startDateStruct", {}).get("date", "N/A") completion_date = status_module.get("completionDateStruct", {}).get("date", "N/A") sex = elig_module.get("sex", "N/A") std_ages = elig_module.get("stdAges", []) sampling_method = elig_module.get("samplingMethod", "N/A") criteria = elig_module.get("eligibilityCriteria", "N/A") locations = contact_module.get("locations", []) countries = sorted({loc.get("country") for loc in locations if loc.get("country")}) # Format location info (first 2 entries, truncate the rest) location_text_lines = [] for loc in locations: parts = [loc.get("facility"), loc.get("city"), loc.get("state"), loc.get("country")] clean = [p for p in parts if p] if clean: location_text_lines.append(", ".join(clean)) if location_text_lines: if len(location_text_lines) > 3: display_lines = location_text_lines[:2] location_text = "\n".join(f"- {line}" for line in display_lines) location_text += f"\n...and {len(location_text_lines)-2} more site(s)" else: location_text = "\n".join(f"- {line}" for line in location_text_lines) else: location_text = "N/A" description = desc_module.get("detailedDescription", "N/A") interventions = arms_module.get("interventions", []) intervention_names = [iv.get("name", "") for iv in interventions if iv.get("name")] intervention_text = ", ".join(intervention_names) if intervention_names else "N/A" sponsor = sponsor_module.get("leadSponsor", {}).get("name", "N/A") collaborators = sponsor_module.get("collaborators", []) collaborator_names = [c.get("name", "") for c in collaborators] result.append( f"### ๐Ÿงช {title}\n\n" f"**NCT ID:** `{nct_id}`\n" f"๐Ÿ”— [View on ClinicalTrials.gov](https://clinicaltrials.gov/study/{nct_id})\n\n" f"**Start Date:** {start_date}\n" f"**Completion Date:** {completion_date}\n\n" f"**Official Title:** {official_title}\n" f"**Type:** {study_type.title()}\n" f"**Phase:** {', '.join(phases) if phases else 'Not reported'}\n" f"**Status:** {status}\n" f"**Countries:** {', '.join(countries) if countries else 'N/A'}\n" f"**Locations:**\n{location_text}\n" f"**Interventions:** {intervention_text}\n" f"**Sponsor:** {sponsor}\n" f"**Collaborators:** {', '.join(collaborator_names) if collaborator_names else 'None'}\n" ) return "\n\n---\n\n".join(result).strip() return f"API returned error: {response.status_code}" except Exception as e: print("Exception occurred:", e) return "Error fetching study data." # %% # There's a particular dictionary structure that's required to describe our function: search_function = { "name": "search_studies", "description": "Search for clinical trials with strict filtering on all key metadata fields such as condition, country, phase, study type, sex, age group, sampling method, sponsor, intervention, locations, start dates, completion dates, etc.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Condition or keyword to search for. (e.g., 'lung cancer', 'IBD')" }, "phase": { "type": "string", "description": "Clinical trial phase. (e.g., 'Phase 1', 'Phase 2', 'Phase 3')" }, "status": { "type": "string", "description": "Recruitment status. (e.g., 'RECRUITING', 'COMPLETED')" }, "country": { "type": "string", "description": "Country where the trial is conducted. (e.g., 'Italy')" }, "study_type": { "type": "string", "description": "Type of study. (e.g., 'INTERVENTIONAL', 'OBSERVATIONAL')" }, "sex": { "type": "string", "description": "Sex eligibility. (e.g., 'Male', 'Female', 'All')" }, "age_group": { "type": "string", "description": "Standard age group. (e.g., 'CHILD', 'ADULT', 'OLDER_ADULT')" }, "sampling_method": { "type": "string", "description": "Participant sampling method. (e.g., 'PROBABILITY_SAMPLE', 'NON_PROBABILITY_SAMPLE')" }, "intervention": { "type": "string", "description": "Intervention or treatment keyword. (e.g., 'aspirin', 'TAE')" }, "sponsor": { "type": "string", "description": "Name of the lead sponsor or organization. (e.g., 'Pfizer', 'NIH')" }, "ipd_sharing": { "type": "string", "description": "Will individual participant data (IPD) be shared? (e.g., 'YES', 'NO', 'UND')" }, "city": { "type": "string", "description": "City where the trial site is located. (e.g., 'Chicago')" }, "facility": { "type": "string", "description": "Facility or hospital name where trial is conducted. (e.g., 'Mayo Clinic')" }, "start_date_from": { "type": "string", "description": "Earliest start date allowed (format: YYYY-MM or YYYY-MM-DD)" }, "start_date_to": { "type": "string", "description": "Latest start date allowed" }, "completion_date_from": { "type": "string", "description": "Earliest completion date allowed" }, "completion_date_to": { "type": "string", "description": "Latest completion date allowed" }, "max_results": { "type": "integer", "description": "Maximum number of studies to return" } }, "required": ["query"], "additionalProperties": False } } # %% # And this is included in a list of tools: tools = [{"type": "function", "function": search_function}] # %% def chat(message, history): messages = [{"role": "system", "content": system_message}] + history + [{"role": "user", "content": message}] # ๐Ÿ”„ First attempt: try to stream the LLM output response_stream = openai.chat.completions.create( model=MODEL, messages=messages, tools=tools, tool_choice="auto", stream=True ) full_response = "" tool_call_detected = False for chunk in response_stream: choice = chunk.choices[0] delta = choice.delta # ๐Ÿง  Detect tool call request during stream if hasattr(delta, "tool_calls") and delta.tool_calls: tool_call_detected = True break # Exit streaming โ€” can't continue past tool call if delta.content: full_response += delta.content yield full_response # Live stream to user # ๐Ÿงฐ Tool call fallback (non-streamed) if tool_call_detected: fallback = openai.chat.completions.create( model=MODEL, messages=messages, tools=tools, tool_choice="auto" # No stream here, required to get tool_calls ) message = fallback.choices[0].message print("Finish reason:", fallback.choices[0].finish_reason) print("Tool calls:", message.tool_calls if hasattr(message, 'tool_calls') else None) # ๐Ÿ”ง Call the tool(s) tool_responses = handle_tool_call(message) # Add the assistant tool call message and all corresponding tool responses messages.append(message) messages.extend(tool_responses) # ๐Ÿง  Now ask GPT to summarize the tool result(s) final_response_stream = openai.chat.completions.create( model=MODEL, messages=messages, stream=True ) final_output = "" for chunk in final_response_stream: delta = chunk.choices[0].delta if delta.content: final_output += delta.content yield final_output # Stream final GPT summary # ๐Ÿงฏ Final fallback if nothing streamed elif not full_response: fallback = openai.chat.completions.create( model=MODEL, messages=messages, tools=tools, tool_choice="auto" ) yield fallback.choices[0].message.content # %% def handle_tool_call(message): import json tool_responses = [] for tool_call in message.tool_calls: arguments = json.loads(tool_call.function.arguments) result = search_studies(arguments) tool_responses.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result if isinstance(result, str) else json.dumps(result) }) return tool_responses # %% example_prompts = [ "Show me trials in Taiwan studying Vedolizumab.", "List studies for Crohn's disease that started after 2015.", "Give me 5 completed trials on lung cancer in Japan.", "Provide latest Perennial Allergic Rhinitis study from Eli Lily", "Find interventional Phase 3 studies for breast cancer in France.", "List observational studies in Thailand with female participants over 65.", "Show me trials in United States, Houston, at MD Anderson.", "Show studies that started after 2022 for Asthma and are still ongoing.", "Show me studies that use HS135 in Canada.", "Show me trials that share individual participant data (IPD) in the USA.", "Get trial details for NCT06083857.", "Find completed prostate cancer trials for adults in Germany.", "List Phase 4 trials with probability sampling in South Korea." ] description = """\
Ask about any of the following criteria to find clinical trials:

๐Ÿฉบ Medical Conditions
e.g., 'lung cancer', 'prostate cancer'		 ๐Ÿ”ข NCT ID
e.g., 'NCT01234567'		 ๐Ÿงช Trial Phase
e.g., 'Phase 1', 'Phase 2', 'Phase 3', 'Phase 4'
๐Ÿงซ Study Type
e.g., 'INTERVENTIONAL', 'OBSERVATIONAL'		 ๐Ÿ“‹ Status
e.g., 'RECRUITING', 'COMPLETED', 'TERMINATED'		 ๐Ÿ’Š Interventions
e.g., 'aspirin', 'HS135', 'Vedolizumab'
๐Ÿข Sponsor
e.g., 'Pfizer', 'NIH', 'Amgen'		 ๐Ÿšป Sex
e.g., 'Male', 'Female', 'All'		 ๐Ÿง Age Group
e.g., 'CHILD', 'ADULT', 'OLDER_ADULT'
๐ŸŽฏ Sampling Method
e.g., 'PROBABILITY_SAMPLE', 'NON_PROBABILITY_SAMPLE'		 ๐Ÿ“ค IPD Sharing
e.g., 'YES', 'NO', 'UNDECIDED' (Individual Participant Data)		 ๐ŸŒ Location
e.g., country, city, or facility
๐Ÿ“… Start/Completion Date
e.g., '2020', '2020-05', or '2020-05-20'

๐Ÿ’ฌ Ask your question naturally, but include keywords like condition, location, phase, sex, or sponsor for best results.
You can combine filters (e.g., 'recruiting lung cancer trials in Canada by Pfizer') for more precise answers.

๐Ÿ’ก Try one of the examples below to get started!
""" gr.ChatInterface( fn=chat, # Your actual function type="messages", title="ClinicalTrials.gov Agent", description=description, chatbot=gr.Chatbot(label="CTGagent", type="messages", height=1000), examples=example_prompts ).launch(app_kwargs={"title": "CTGagent"}) # %%