| import os |
| import json |
| import requests |
| import gradio as gr |
| from typing import TypedDict, Annotated, List |
| from langchain_groq import ChatGroq |
| from langchain_core.messages import HumanMessage |
| from langgraph.graph import StateGraph, END |
| from langgraph.graph.message import add_messages |
|
|
| def create_agent(api_key: str): |
| llm = ChatGroq( |
| model="llama-3.3-70b-versatile", |
| temperature=0.1, |
| api_key=api_key |
| ) |
|
|
| class AgentState(TypedDict): |
| messages: Annotated[list, add_messages] |
| patient_input: str |
| search_query: str |
| pubmed_results: List[dict] |
| clinical_recommendation: dict |
| conversation_history: List[dict] |
|
|
| def search_pubmed(query: str, max_results: int = 5) -> List[dict]: |
| search_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi" |
| search_params = {"db": "pubmed", "term": query, "retmax": max_results, "retmode": "json", "sort": "relevance"} |
| search_response = requests.get(search_url, params=search_params) |
| search_data = search_response.json() |
| article_ids = search_data["esearchresult"]["idlist"] |
|
|
| if not article_ids: |
| return [{"title": "No results found", "abstract": "No relevant articles found.", "authors": "", "year": "", "url": ""}] |
|
|
| fetch_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi" |
| fetch_params = {"db": "pubmed", "id": ",".join(article_ids), "retmode": "xml", "rettype": "abstract"} |
| fetch_response = requests.get(fetch_url, params=fetch_params) |
|
|
| from bs4 import BeautifulSoup |
| soup = BeautifulSoup(fetch_response.text, "xml") |
| articles = [] |
|
|
| for article in soup.find_all("PubmedArticle"): |
| try: |
| title = article.find("ArticleTitle") |
| title = title.text if title else "No title" |
| abstract = article.find("AbstractText") |
| abstract = abstract.text if abstract else "No abstract available" |
| year = article.find("PubDate") |
| year = year.find("Year").text if year and year.find("Year") else "Unknown" |
| authors = article.find_all("LastName") |
| authors = ", ".join([a.text for a in authors[:3]]) + " et al." if authors else "Unknown" |
| pmid = article.find("PMID") |
| pmid = pmid.text if pmid else "" |
| articles.append({ |
| "title": title, |
| "abstract": abstract[:500] + "..." if len(abstract) > 500 else abstract, |
| "authors": authors, "year": year, "pmid": pmid, |
| "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" |
| }) |
| except: |
| continue |
|
|
| return articles if articles else [{"title": "Parse error", "abstract": "Could not parse results.", "authors": "", "year": "", "url": ""}] |
|
|
| def extract_search_query(state: AgentState) -> AgentState: |
| prompt = f"""You are a medical AI assistant. Generate the best PubMed search query for these symptoms. |
| Patient input: {state['patient_input']} |
| Return ONLY the search query (max 8 words), nothing else.""" |
| response = llm.invoke([HumanMessage(content=prompt)]) |
| return {**state, "search_query": response.content.strip()} |
|
|
| def search_medical_literature(state: AgentState) -> AgentState: |
| results = search_pubmed(state["search_query"], max_results=5) |
| return {**state, "pubmed_results": results} |
|
|
| def generate_recommendation(state: AgentState) -> AgentState: |
| articles_text = "" |
| for i, article in enumerate(state["pubmed_results"]): |
| articles_text += f"\nArticle {i+1}:\nTitle: {article['title']}\nAuthors: {article['authors']} ({article['year']})\nAbstract: {article['abstract']}\nURL: {article['url']}\n" |
|
|
| prompt = f"""You are an expert clinical decision support AI. |
| Patient symptoms: {state['patient_input']} |
| Relevant Medical Literature: {articles_text} |
| |
| Generate a structured response in this EXACT JSON format: |
| {{ |
| "possible_conditions": ["condition1", "condition2", "condition3"], |
| "recommended_tests": ["test1", "test2", "test3"], |
| "treatment_considerations": ["consideration1", "consideration2"], |
| "urgency_level": "Low/Medium/High/Emergency", |
| "reasoning": "Brief explanation", |
| "important_disclaimer": "This is AI-generated information for educational purposes only. Always consult a qualified healthcare professional.", |
| "sources": ["Article title - Authors (Year)"] |
| }} |
| Return ONLY the JSON object.""" |
|
|
| response = llm.invoke([HumanMessage(content=prompt)]) |
| try: |
| clean = response.content.strip() |
| if "```json" in clean: |
| clean = clean.split("```json")[1].split("```")[0].strip() |
| elif "```" in clean: |
| clean = clean.split("```")[1].split("```")[0].strip() |
| recommendation = json.loads(clean) |
| except: |
| recommendation = { |
| "possible_conditions": ["Unable to parse"], |
| "recommended_tests": [], |
| "treatment_considerations": [], |
| "urgency_level": "Unknown", |
| "reasoning": response.content, |
| "important_disclaimer": "Always consult a qualified healthcare professional.", |
| "sources": [] |
| } |
| return {**state, "clinical_recommendation": recommendation} |
|
|
| def format_response(state: AgentState) -> AgentState: |
| history = state.get("conversation_history", []) |
| history.append({"patient_input": state["patient_input"], "recommendation": state["clinical_recommendation"]}) |
| return {**state, "conversation_history": history} |
|
|
| graph = StateGraph(AgentState) |
| graph.add_node("extract_query", extract_search_query) |
| graph.add_node("search_pubmed", search_medical_literature) |
| graph.add_node("generate_recommendation", generate_recommendation) |
| graph.add_node("format_response", format_response) |
| graph.set_entry_point("extract_query") |
| graph.add_edge("extract_query", "search_pubmed") |
| graph.add_edge("search_pubmed", "generate_recommendation") |
| graph.add_edge("generate_recommendation", "format_response") |
| graph.add_edge("format_response", END) |
|
|
| return graph.compile(), llm |
|
|
| def run_agent(agent, patient_input: str, history: list) -> tuple: |
| initial_state = { |
| "messages": [], |
| "patient_input": patient_input, |
| "search_query": "", |
| "pubmed_results": [], |
| "clinical_recommendation": {}, |
| "conversation_history": history |
| } |
| result = agent.invoke(initial_state) |
| rec = result["clinical_recommendation"] |
|
|
| output = f"""🚨 URGENCY LEVEL: {rec.get('urgency_level', 'Unknown')} |
| |
| 🔬 POSSIBLE CONDITIONS: |
| {chr(10).join([f"• {c}" for c in rec.get('possible_conditions', [])])} |
| |
| 🧪 RECOMMENDED TESTS: |
| {chr(10).join([f"• {t}" for t in rec.get('recommended_tests', [])])} |
| |
| 💊 TREATMENT CONSIDERATIONS: |
| {chr(10).join([f"• {t}" for t in rec.get('treatment_considerations', [])])} |
| |
| 🧠 CLINICAL REASONING: |
| {rec.get('reasoning', '')} |
| |
| 📚 SOURCES FROM PUBMED: |
| {chr(10).join([f"[{i+1}] {s}" for i, s in enumerate(rec.get('sources', []))])} |
| |
| ⚠️ DISCLAIMER: {rec.get('important_disclaimer', '')}""" |
|
|
| return output, result["conversation_history"] |
|
|
| with gr.Blocks(title="Clinical Decision Support Agent") as demo: |
| gr.Markdown("# 🏥 Clinical Decision Support Agent") |
| gr.Markdown("Powered by LangGraph + LLaMA 3.3 70B + Real PubMed Literature") |
| gr.Markdown("⚠️ **For educational purposes only. Always consult a qualified healthcare professional.**") |
|
|
| with gr.Row(): |
| api_key_input = gr.Textbox( |
| label="🔑 Enter your Groq API Key", |
| placeholder="gsk_xxxxxxxxxxxx", |
| type="password" |
| ) |
|
|
| history_state = gr.State([]) |
| agent_state = gr.State(None) |
|
|
| def initialize_agent(api_key): |
| if not api_key.strip(): |
| return None, "❌ Please enter a valid Groq API key" |
| try: |
| agent, _ = create_agent(api_key) |
| return agent, "✅ Agent initialized successfully!" |
| except Exception as e: |
| return None, f"❌ Error: {str(e)}" |
|
|
| init_btn = gr.Button("🚀 Initialize Agent", variant="primary") |
| init_status = gr.Textbox(label="Status", interactive=False) |
|
|
| init_btn.click( |
| fn=initialize_agent, |
| inputs=[api_key_input], |
| outputs=[agent_state, init_status] |
| ) |
|
|
| with gr.Row(): |
| with gr.Column(): |
| symptom_input = gr.Textbox( |
| label="Describe Patient Symptoms", |
| placeholder="Example: I have fever of 39°C for 3 days, cough with yellow sputum, chest pain...", |
| lines=4 |
| ) |
| submit_btn = gr.Button("🔍 Analyze Symptoms", variant="primary") |
| clear_btn = gr.Button("🗑️ Clear Conversation") |
|
|
| with gr.Column(): |
| output_text = gr.Textbox( |
| label="Clinical Recommendation", |
| lines=20, |
| interactive=False |
| ) |
|
|
| def analyze(agent, symptoms, history): |
| if agent is None: |
| return "❌ Please initialize the agent first with your Groq API key.", history |
| if not symptoms.strip(): |
| return "Please describe your symptoms.", history |
| output, new_history = run_agent(agent, symptoms, history) |
| return output, new_history |
|
|
| submit_btn.click( |
| fn=analyze, |
| inputs=[agent_state, symptom_input, history_state], |
| outputs=[output_text, history_state] |
| ) |
|
|
| clear_btn.click( |
| fn=lambda: ([], ""), |
| outputs=[history_state, output_text] |
| ) |
|
|
| demo.launch() |
