""" app.py ------ Main application entry point for the modular AI agent system. Handles UI (Gradio), agent orchestration, knowledge loading, and chat interface logic. - Initializes logging and structlog for consistent, timestamped logs. - Loads and manages agent configurations and RAG retriever. - Provides chat, knowledge ingestion, and agent builder functions. - Integrates with OpenAI via llm_connector and supports streaming responses. - Designed for extensibility and future integration with Model Context Protocol (MCP) and Agent-to-Agent (A2A) standards. """ # --- Imports --- import gradio as gr import json import re import openai from openai import RateLimitError, APIError, APIConnectionError, OpenAI from typing import Dict, cast from core.utils.rag import KnowledgeLoader, SimpleRAGRetriever from core.utils.skills_registry import tool_registry, get_tool_by_name import asyncio import structlog import logging from structlog.stdlib import LoggerFactory, BoundLogger from core.utils.llm_connector import AgentLLMConnector import pandas as pd # ChatMessage not available in Gradio 4.20.0 - removed import # --- Per-User Session Management --- from user_session_manager import session_manager, get_username_from_request, SessionKeys from session_helpers import ( get_user_simple_chat_history, set_user_simple_chat_history, get_user_builder_chat_histories, set_user_builder_chat_histories, get_user_deployed_chat_histories, set_user_deployed_chat_histories, get_current_username, log_user_access ) from core.agents.agent_utils import linkify_citations, build_agent, load_prefilled, prepare_download, preload_demo_chat, _safe_title, extract_clinical_variables_from_history from config import agents_config, skills_library, prefilled_agents from core.ui.ui import show_landing, show_builder, show_chat, refresh_active_agents_widgets # Note: NOT importing build_ui from core.ui.ui - using local complex build_ui function instead import tools import os # Print API keys for debugging # Print all SERPER_* environment variables for robust debugging #for k, v in os.environ.items(): # if k.startswith("SERPER"): # print(f"{k} in Python: {v}") #print("OPENAI_API_KEY in Python:", os.getenv("OPENAI_API_KEY")) #print("SERPER_API_KEY in Python:", os.getenv("SERPER_API_KEY")) # Logging setup logging.basicConfig(filename="app.log", level=logging.INFO, format="%(message)s") structlog.configure(logger_factory=LoggerFactory()) logger: BoundLogger = structlog.get_logger() # Structlog config structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.dev.ConsoleRenderer() ], logger_factory=LoggerFactory(), wrapper_class=structlog.make_filtering_bound_logger(logging.INFO), cache_logger_on_first_use=True, ) # OpenAI API Key Check OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") if not OPENAI_API_KEY: raise RuntimeError("OPENAI_API_KEY environment variable is not set. Please set it before running the application.") OPENAI_API_KEY = cast(str, OPENAI_API_KEY) # initialize client once, pulling key from your environment client = OpenAI(api_key=OPENAI_API_KEY) def simple_chat_response(user_message, request: gr.Request): """ A bare-bones GPT-3.5-turbo chat using the v1.0+ SDK. Converts between Gradio 4.20.0 tuple format and OpenAI messages format. NOW WITH PER-USER SESSION ISOLATION - each user gets their own chat history. """ # Get user-specific history history = get_user_simple_chat_history(request) if history is None: history = [] log_user_access(request, "simple_chat_response") # Convert Gradio history tuples to OpenAI messages format messages = [] for user_msg, assistant_msg in history: if user_msg: messages.append({"role": "user", "content": user_msg}) if assistant_msg: messages.append({"role": "assistant", "content": assistant_msg}) # Add current user message messages.append({"role": "user", "content": user_message}) # Call OpenAI API completion = client.chat.completions.create( model="gpt-3.5-turbo", messages=messages, temperature=0.7, ) # Extract assistant reply reply_content = completion.choices[0].message.content reply = reply_content.strip() if reply_content else "" # Add new exchange to history in Gradio tuple format updated_history = history + [[user_message, reply]] # Save updated history for this user set_user_simple_chat_history(request, updated_history) return updated_history, "" # --- Chat orchestration logic is now in chat_orchestrator.py --- from core.agents.chat_orchestrator import simulate_agent_response_stream, build_log, _stream_to_agent, MAX_HISTORY, orchestrators def reset_chat(agent_json): """ Clears history, input, invocation log, and active_children state. Also clears orchestrator state to prevent response persistence. """ # Clear orchestrator state to prevent persistence across conversations from core.agents.chat_orchestrator import orchestrators orchestrators.clear() agent_data = json.loads(agent_json) name = agent_data.get("agent_name", "Agent") welcome = f"πŸ‘‹ Hello! I'm {name}. How can I assist you today?" # Use tuple format for Gradio chatbot: [["user_message", "assistant_message"]] chat_history = [["", welcome]] invocation_log = "" active_children= [] return chat_history, "", invocation_log, active_children def load_agent_to_builder(agent_name): if agent_name in agents_config: agent_data = json.loads(agents_config[agent_name]) return ( agent_data.get("agent_type", ""), agent_data.get("agent_name", ""), agent_data.get("agent_mission", ""), agent_data.get("skills", []) ) else: return None, "", "", [] def remove_selected_agent(agent_name: str): # 1) Remove from your in-memory store if agent_name in agents_config: del agents_config[agent_name] # 2) Re-render the list of active agents if agents_config: active_md = "### 🧠 Active Agents\n" + "\n".join(f"- {name}" for name in agents_config) else: active_md = "### 🧠 Active Agents\n_(None yet)_" # 3) Build the new dropdown state new_choices = list(agents_config.keys()) # gr.update to reset selection (value=None) and update choices dropdown_update = gr.update(choices=new_choices, value=None) # Return in the same order you wired the outputs return active_md, dropdown_update def update_skills(selected_type: str): """Callback to repopulate the skills checkbox based on chosen agent type.""" return gr.update(choices=skills_library.get(selected_type, [])) def handle_uploaded_files(files): """ Callback for the File component: spin up (or reuse) your RAG retriever and index any newly uploaded docs. """ global rag_retriever rag_retriever = SimpleRAGRetriever(openai_api_key=os.getenv("OPENAI_API_KEY")) if files: for f in files: rag_retriever.add_knowledge(f) # Return a tuple of (visible-update, message) to your upload_alert Markdown return gr.update(visible=True), "βœ… Files uploaded and indexed successfully!" def populate_from_preset(prefilled_name): if prefilled_name != "None": at, an, am, sk = load_prefilled(prefilled_name, prefilled_agents) return at, an, am, sk, True return None, "", "", [], False def on_agent_type_change(selected_type, was_prefilled): # always refresh the skill choices new_skills = skills_library.get(selected_type, []) skill_update = gr.update(choices=new_skills, value=[]) if was_prefilled: # consume the flag, but leave name & mission intact return skill_update, gr.update(), gr.update(), False # manual change: clear everything return skill_update, gr.update(value=""), gr.update(value=""), False def chat_selected_agent(agent_name): agent_json = agents_config.get(agent_name, "") if agent_json: # reset_chat returns 4 values: # (chat_history, cleared_input, invocation_log, active_children) chat_history, cleared_input, invocation_log, active_children = reset_chat(agent_json) return chat_history, cleared_input, invocation_log, active_children, agent_json # If no agent is selected, clear everything return [], "", "", [], "" def load_history(agent_name, histories): # if we’ve never chatted, start with a greeting if agent_name not in histories: histories[agent_name] = [ ["", f"πŸ‘‹ Hello! I'm {agent_name}. How can I help?"] ] return histories[agent_name] def convert_to_gradio_format(history_data): """ Convert any history format to Gradio-compatible tuple format. Handles: - Single dictionary: {'role': 'assistant', 'content': 'message'} - List of dictionaries: [{'role': 'user', 'content': '...'}, ...] - Already tuple format: [['user_msg', 'assistant_msg'], ...] """ if not history_data: return [] # Handle single dictionary case if isinstance(history_data, dict): role = history_data.get("role", "") content = history_data.get("content", "") if role == "user": return [[content, ""]] elif role == "assistant": return [["", content]] else: return [["", content]] # Default to assistant # Handle list case if isinstance(history_data, list): # Check if already in tuple format if history_data and isinstance(history_data[0], (list, tuple)): return history_data # Already in correct format # Convert from dictionary format if history_data and isinstance(history_data[0], dict): tuple_history = [] i = 0 while i < len(history_data): user_msg = "" assistant_msg = "" # Look for user message if i < len(history_data) and history_data[i].get("role") == "user": user_msg = history_data[i].get("content", "") i += 1 # Look for assistant message if i < len(history_data) and history_data[i].get("role") == "assistant": assistant_msg = history_data[i].get("content", "") i += 1 # Add the pair if we have at least one message if user_msg or assistant_msg: tuple_history.append([user_msg, assistant_msg]) else: i += 1 # Skip unknown message types return tuple_history # Fallback: return empty list return [] def chatpanel_handle(agent_name, user_text, request: gr.Request): """ Uses your simulate_agent_response_stream (tool-aware) in a blocking way, so that tool invocations actually happen. NOW WITH PER-USER SESSION ISOLATION - each user has their own chat histories. Returns (final_history, updated_histories, cleared_input). """ # Get user-specific histories histories = get_user_deployed_chat_histories(request) log_user_access(request, f"chatpanel_handle for agent {agent_name}") # 1) Look up the JSON you saved in agents_config agent_json = agents_config.get(agent_name) if not agent_json: return [], histories, "", "" # 2) Grab the prior history (or seed a greeting) history = histories.get(agent_name, []) if not history: history = [["", f"πŸ‘‹ Hello! I'm {agent_name}. How can I help?"]] # 3) Call your streaming function synchronously # simulate_agent_response_stream is async, so we need to run it in an event loop import asyncio async def run_stream(): async for updated_history, _, invocation_log, _, challenger_info in simulate_agent_response_stream( agent_json=agent_json, history=history, user_input=user_text, debug_flag=False, active_children=[] ): yield updated_history, invocation_log loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) gen = run_stream() final_history = history final_invocation_log = "" try: while True: updated_history, invocation_log = loop.run_until_complete(gen.__anext__()) # Convert any format to Gradio tuple format final_history = convert_to_gradio_format(updated_history) final_invocation_log = invocation_log except StopAsyncIteration: pass # 4) Save back and clear the input box histories[agent_name] = final_history # Save updated histories for this user set_user_deployed_chat_histories(request, histories) return final_history, histories, "", final_invocation_log def refresh_chat_dropdown(): return gr.update(choices=list(agents_config.keys()), value=None) def build_ui(): # --- App Layout --- with gr.Blocks() as app: # Separate chat histories for each panel simple_chat_history = gr.State([]) builder_chat_histories = gr.State({}) deployed_chat_histories = gr.State({}) # --- Custom CSS (Memory Optimized) --- app.css = """ /* Optimized CSS - reduced complexity for memory efficiency */ :root { --futuristic-bg: #fcfdff; --futuristic-panel: #e8ecf7; --futuristic-accent: #00ffe7; --futuristic-accent2: #7f5cff; --futuristic-card: #fafdff; --futuristic-border: #c7d0e6; --futuristic-text: #23263a; --futuristic-muted: #7a7e8c; } body, .gradio-container, .gr-block, .gr-app { background: var(--futuristic-bg) !important; color: var(--futuristic-text) !important; } #landing_card, #agent_form { opacity: 1; transition: opacity 0.3s ease; } .hidden { opacity: 0 !important; pointer-events: none; } #agent_form .gr-row { display: flex !important; align-items: stretch !important; background: #fff !important; border-radius: 16px; min-height: 600px; padding: 0 0.5em; } #start_button { background: linear-gradient(90deg, #7f5cff, #00ffe7); color: white !important; border: none; border-radius: 8px; font-weight: 700; font-size: 18px; padding: 12px 24px; } #to_chat_button { background: linear-gradient(90deg, #7f5cff, #a259ff); color: #fff !important; border: none; border-radius: 8px; font-weight: 700; font-size: 18px; padding: 12px 24px; } .left-panel, .right-panel { background: linear-gradient(135deg, #e8ecf7 60%, #fafdff 100%); padding: 15px; border-radius: 16px; border: 1px solid var(--futuristic-border); } .landing-card { background: linear-gradient(135deg, #23263a 60%, #1a1f2b 100%); border: 1.5px solid var(--futuristic-accent2); border-radius: 28px; padding: 32px 36px 32px 36px; margin-bottom: 40px; } .landing-title { text-align: center; font-size: 2.1rem; font-weight: 700; margin-bottom: 0.5em; } .landing-subtitle { text-align: center; font-size: 1.15rem; color: var(--futuristic-muted); margin-bottom: 1.2em; } .steps-box { background: var(--futuristic-card); border: 1px solid var(--futuristic-border); border-radius: 16px; max-width: 840px; margin: 0 auto 1.5em auto; padding: 18px 28px 18px 28px; text-align: left; } button.generate-btn { background: linear-gradient(90deg, var(--futuristic-accent2), var(--futuristic-accent)); color: #181c27 !important; border: none !important; font-weight: 600; } .agent-controls { display: inline-flex !important; flex-wrap: nowrap !important; align-items: center; gap: 0.5rem; } /* Simplified patient cards - removed complex animations */ .patient-card { background: linear-gradient(135deg, var(--futuristic-card) 60%, #ffffff 100%) !important; border: 2px solid var(--futuristic-border) !important; border-radius: 12px !important; padding: 1rem !important; cursor: pointer !important; transition: all 0.2s ease !important; margin: 0.5rem !important; } .patient-card:hover { transform: translateY(-2px) !important; border-color: var(--futuristic-accent2) !important; } """ # Helper functions for chat control state management def show_initial_instruction_state(): """Return initial state with instruction message and disabled controls""" instruction_chat = [["", "πŸ“‹ **Agent generated successfully!**\n\nTo start testing your agent:\n1. Select your agent from the dropdown menu above\n2. Click 'πŸ’¬ Chat with Selected Agent'\n3. Then you can type your questions in the chat box\n\n*Please select an agent from the dropdown to begin testing.*"]] return ( instruction_chat, # builder_chatbot gr.update(value="", interactive=False, placeholder="Please select an agent to start chatting..."), # chat_input gr.update(interactive=False), # builder_send_button gr.update(interactive=False), # reset_button "", # invocation_log [] # active_children ) def enable_chat_controls_with_agent(agent_name): """Enable chat controls and show proper agent greeting when agent is selected""" agent_json = agents_config.get(agent_name, "") if agent_json: # Get agent data for greeting agent_data = json.loads(agent_json) agent_display_name = agent_data.get("agent_name", agent_name) # Create greeting message greeting_chat = [["", f"πŸ‘‹ Hello! I'm {agent_display_name}. How can I assist you today?"]] return ( greeting_chat, # builder_chatbot with agent greeting gr.update(value="", interactive=True, placeholder="Type your question here…"), # chat_input enabled gr.update(interactive=True), # builder_send_button enabled gr.update(interactive=True), # reset_button enabled "", # clear invocation_log [], # clear active_children agent_json # agent_output ) else: # No valid agent selected, return disabled state return show_initial_instruction_state() + ("",) # Add empty agent_output # Update instruction when agent is selected from dropdown def update_instruction_on_dropdown_change(agent_name): """Update instruction message when agent is selected from dropdown""" if agent_name: instruction_msg = f"πŸ“‹ **Agent '{agent_name}' selected!**\n\nTo start testing this agent:\nβ€’ Click 'πŸ’¬ Chat with Selected Agent' button above\nβ€’ Then you can type your questions in the chat box\n\n*Click the chat button to begin testing.*" else: instruction_msg = "πŸ“‹ **Welcome to the Agent Builder!**\n\nTo start testing your agents:\n1. Generate an agent using the form on the left\n2. Select your agent from the dropdown menu above\n3. Click 'πŸ’¬ Chat with Selected Agent'\n4. Then you can type your questions in the chat box\n\n*Please create and select an agent to begin testing.*" return [["", instruction_msg]] # 1) HEADER & LANDING CARD with gr.Group(elem_id="landing_card", elem_classes="landing-card", visible=True) as landing_panel: gr.Markdown("
🦠 Infectious Diseases Agent Builder
", elem_id=None, elem_classes=None) gr.Markdown("
Build your own ID-focused chat agent in 5 easy steps β€” no coding required.
", elem_id=None, elem_classes=None) gr.HTML("""
""") start_button = gr.Button( "πŸš€ Get Started", elem_id="start_button", elem_classes="animated-button" ) gr.HTML("
") # Only the simple GPT-3.5 Chatbot (no active agents or builder UI) gr.Markdown("### πŸ’¬ Try A Simple Chatbot Before You Build Your ID Agents") simple_chatbot = gr.Chatbot(label="GPT-3.5 Chat") simple_input = gr.Textbox( placeholder="Ask anything…", show_label=False, lines=2, max_lines=4, ) simple_send = gr.Button("Send") simple_reset = gr.Button("Reset") # 2) AGENT FORM (HIDDEN UNTIL CLICK) prefill_flag = gr.State(False) with gr.Group(elem_id="agent_form", visible=False) as agent_form: # Move Back to Home button to the very top back_button = gr.Button("πŸ”™ Back to Home", elem_id="back_button") # Steps box at the top of the builder panel for user guidance gr.HTML("""
""") gr.Markdown("### πŸŽ›οΈ Infectious Diseases Agent Builder") with gr.Row(): # Left panel with gr.Column(scale=3, elem_classes="left-panel"): prefilled = gr.Dropdown(choices=["None"] + list(prefilled_agents.keys()), label="Start with a prefilled agent?") with gr.Accordion("πŸ› οΈ Basic Settings", open=True): agent_type = gr.Radio( choices=[ "πŸ›‘οΈ Antimicrobial Stewardship", "🦠 Infection Prevention and Control", "πŸ”¬ Research Assistant", "πŸ₯ Clinical Assistant", "πŸ“š Education Assistant", "🎼 Orchestrator", ], label="Select Agent Type", elem_id="select_agent_type_radio" ) agent_name = gr.Textbox(label="Agent Name", placeholder="e.g., SmartSteward", max_lines=1) agent_mission = gr.Textbox(label="Agent Mission", placeholder="Describe what your agent should do…", lines=4) skills = gr.CheckboxGroup(choices=[], label="Select Skills") with gr.Accordion("βš™οΈ Advanced Options", open=False): link1, link2, link3, link4 = [ gr.Textbox(label=f"Trusted Source Link {i} (optional)") for i in range(1,5) ] web_access_toggle = gr.Checkbox(label="Allow Internet Search 🌐", value=True, interactive=True) allow_fallback_toggle = gr.Checkbox(label="Allow Fallback to LLM General Knowledge πŸ€–", value=True) challenger_toggle = gr.Checkbox(label="Enable Adversarial AI Validation (Challenger)", value=False, info="If enabled, agent replies will be critiqued by an adversarial LLM before being shown to the user.") # --- Auto-toggle logic for web_access_toggle --- def update_web_access_toggle(l1, l2, l3, l4): links = [l1, l2, l3, l4] any_links = any(l.strip() for l in links if l) if any_links: # If any trusted link is present, force checked and disable return gr.update(value=True, interactive=False) else: # If all empty, allow user to toggle return gr.update(interactive=True) # Wire up the logic: any change to link1-4 updates web_access_toggle for link in [link1, link2, link3, link4]: link.change( fn=update_web_access_toggle, inputs=[link1, link2, link3, link4], outputs=[web_access_toggle] ) generate_button = gr.Button("✨ Generate Agent Config", elem_classes="generate-btn") with gr.Accordion("πŸ“¦ Generated Agent Config", open=False): agent_loader = gr.Markdown("") agent_output = gr.Code(label="Configuration (JSON)", language="json") download_button = gr.DownloadButton(label="Download Config") # Move Upload Knowledge Files section below agent config with gr.Accordion("πŸ“š Upload Knowledge Files (Global)", open=False): uploaded_files = gr.File(label="Upload Knowledge Files", file_count="multiple") upload_alert = gr.Markdown("", visible=False) # Right panel with gr.Column(scale=9, elem_classes="right-panel"): builder_active_agents = gr.Markdown("### 🧠 Active Agents\n_(None yet)_") # dropdown + action buttons inline with gr.Row(elem_classes="agent-controls"): with gr.Column(scale=3, elem_classes="agent-dropdown-column"): agent_remove_dropdown = gr.Dropdown( label="Select an agent", choices=[], elem_classes="agent-dropdown" ) with gr.Column(scale=1, elem_classes="agent-button-group"): chat_agent_button = gr.Button( "πŸ’¬ Chat with Selected Agent", elem_classes="action-btn" ) edit_agent_button = gr.Button( "πŸ›  Edit Selected Agent", elem_classes="action-btn" ) remove_agent_button = gr.Button( "❌ Remove Selected Agent", elem_classes="action-btn" ) show_debug = gr.Checkbox(label="πŸ”Ž Show tool reasoning", value=False) # Only one chatbot in builder panel builder_chatbot = gr.Chatbot( label="πŸ’¬ Live Conversation with Your ID Agent", value=[["", "πŸ“‹ **Welcome to the Agent Builder!**\n\nTo start testing your agents:\n1. Generate an agent using the form on the left\n2. Select your agent from the dropdown menu above\n3. Click 'πŸ’¬ Chat with Selected Agent'\n4. Then you can type your questions in the chat box\n\n*Please create and select an agent to begin testing.*"]] ) chat_input = gr.Textbox( placeholder="Please select an agent to start chatting...", show_label=False, lines=3, max_lines=5, interactive=False ) active_children = gr.State([]) # will hold a list of JSON-configs # --- Builder panel clinical variable fields (hidden, but needed for wiring) --- builder_deescalation_culture = gr.Textbox(visible=False) builder_deescalation_meds = gr.Textbox(visible=False) builder_stewardship_site = gr.Textbox(visible=False) builder_stewardship_biofilm = gr.Textbox(visible=False) builder_stewardship_response = gr.Textbox(visible=False) builder_stewardship_crcl = gr.Textbox(visible=False) builder_stewardship_severity = gr.Textbox(visible=False) builder_stewardship_allergies = gr.Textbox(visible=False) builder_empiric_age = gr.Textbox(visible=False) builder_empiric_allergies = gr.Textbox(visible=False) builder_empiric_labs = gr.Textbox(visible=False) builder_empiric_culture = gr.Textbox(visible=False) builder_empiric_meds = gr.Textbox(visible=False) builder_empiric_site = gr.Textbox(visible=False) builder_empiric_biofilm = gr.Textbox(visible=False) builder_empiric_response = gr.Textbox(visible=False) builder_empiric_crcl = gr.Textbox(visible=False) builder_empiric_severity = gr.Textbox(visible=False) # Add the builder send button under the chatbox builder_send_button = gr.Button("Send", elem_id="builder_send_button", interactive=False) reset_button = gr.Button("πŸ”„ Reset Chat", interactive=False) invocation_log = gr.Markdown( value="", label="πŸ” Tool Invocation Log", visible=True ) gr.Markdown("---\nBuilt with ❀️ for ID Week 2025 β€” Empowering Infectious Diseases Innovation") # Move the Chat With the Agents You Deployed button to the bottom, below the disclaimer to_chat_button = gr.Button("πŸ—¨οΈ Chat With the Agents You Deployed", elem_id="to_chat_button") # --- Builder panel send button logic --- def builderpanel_handle_with_dynamic_vars( agent_name, user_text, histories, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity ): agent_json = agents_config.get(agent_name) if agent_json: agent_data = json.loads(agent_json) skills = agent_data.get("skills", []) history = histories.get(agent_name, []) # --- Trusted links wiring --- trusted_links = [] for k in ["trusted_links", "trusted_links_1", "trusted_links_2", "trusted_links_3", "trusted_links_4"]: # Support both list and individual keys if isinstance(agent_data.get(k), list): trusted_links.extend([l for l in agent_data[k] if l]) elif isinstance(agent_data.get(k), str) and agent_data[k]: trusted_links.append(agent_data[k]) # Also check for link1-link4 keys (legacy) for k in ["link1", "link2", "link3", "link4"]: if agent_data.get(k): trusted_links.append(agent_data[k]) trusted_links = [l for l in trusted_links if l] # Do not prepend trusted links to every user message; just keep them available for tools # if trusted_links: # links_str = ", ".join(trusted_links) # user_text = f"Trusted sources for this agent: {links_str}\n\n" + user_text # --- End trusted links wiring --- # Deescalation tool if "recommend_deescalation" in skills: var_names = ["culture", "meds", "site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection", "known_allergies"] user_vars = { "culture": deescalation_culture, "meds": deescalation_meds, "site_of_infection": stewardship_site, "risk_of_biofilm": stewardship_biofilm, "current_response": stewardship_response, "creatinine_clearance": stewardship_crcl, "severity_of_infection": stewardship_severity, "known_allergies": stewardship_allergies } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" # Only prepend if at least one field is non-empty if any(user_vars[k] for k in var_names): user_text = f"[DEESCALATION_TOOL_INPUT] {json.dumps(user_vars)}\n" + user_text elif "alert_prolonged_antibiotic_use" in skills: var_names = ["site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection", "known_allergies"] user_vars = { "site_of_infection": stewardship_site, "risk_of_biofilm": stewardship_biofilm, "current_response": stewardship_response, "creatinine_clearance": stewardship_crcl, "severity_of_infection": stewardship_severity, "known_allergies": stewardship_allergies } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" if any(user_vars[k] for k in var_names): user_text = f"[ALERT_PROLONGED_ABX_INPUT] {json.dumps(user_vars)}\n" + user_text elif "recommend_empiric_therapy" in skills: # Remove 'known_allergies' as a separate required field (it's covered by 'allergies') var_names = [ "age", "allergies", "labs", "culture", "meds", "site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection" ] user_vars = { "age": empiric_age, "allergies": empiric_allergies, "labs": empiric_labs, "culture": empiric_culture, "meds": empiric_meds, "site_of_infection": empiric_site, "risk_of_biofilm": empiric_biofilm, "current_response": empiric_response, "creatinine_clearance": empiric_crcl, "severity_of_infection": empiric_severity } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" # If any required field is missing, prompt = ... missing = [k.replace('_', ' ').capitalize() for k in var_names if not user_vars[k].strip()] if missing: prompt = f"Please provide the following required information for empiric therapy: {', '.join(missing)}." # Show this as an assistant message and do not call the tool history.append(["", prompt]) return history, histories, "" # All required fields present, prepend tool input user_text = f"[EMPIRIC_THERAPY_INPUT] {json.dumps(user_vars)}\n" + user_text # Use the same chat handling logic, but ensure the builder_chatbot is updated and history is preserved # Call chatpanel_handle, but get extra challenger info from simulate_agent_response_stream import asyncio from core.agents.chat_orchestrator import simulate_agent_response_stream agent_json_val = agents_config.get(agent_name) history_val = histories.get(agent_name, []) result = None async def run_stream(): gen = simulate_agent_response_stream( agent_json=agent_json_val, history=history_val, user_input=user_text, debug_flag=False, active_children=[] ) last_result = None async for result in gen: last_result = result return last_result loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(run_stream()) if result is not None and len(result) == 5: final_history, _, invocation_log, _, challenger_info = result # Ensure format conversion for builder panel final_history = convert_to_gradio_format(final_history) else: final_history, updated_histories, cleared_input, invocation_log = chatpanel_handle(agent_name, user_text, histories) challenger_info = None return final_history, updated_histories, cleared_input, invocation_log, "" except Exception: # fallback to old behavior if error final_history, updated_histories, cleared_input, invocation_log = chatpanel_handle(agent_name, user_text, histories) challenger_info = None return final_history, updated_histories, cleared_input, invocation_log, "" # Update histories updated_histories = histories.copy() updated_histories[agent_name] = final_history # Prepare challenger markdown (debug log for builder panel) challenger_md = "" if isinstance(challenger_info, dict): orig = challenger_info.get("original_reply", "") crit = challenger_info.get("challenger_critique", "") final = challenger_info.get("final_reply", "") # Ensure critique is never None or empty in the UI if not crit or str(crit).strip().lower() == "none": crit = "OK" # Only show the final (challenger-approved) answer in the chatbox if final and final_history and isinstance(final_history, list) and len(final_history) > 0: # final_history might come in dictionary format from simulate_agent_response_stream # Convert to tuple format if needed last_message = final_history[-1] if isinstance(last_message, dict) and "content" in last_message: # Convert from dict format to tuple format if last_message.get("role") == "assistant": final_history[-1] = ["", final] elif last_message.get("role") == "user": # Keep user message, but this shouldn't happen for final challenger message final_history[-1] = [last_message.get("content", ""), ""] elif isinstance(last_message, (list, tuple)) and len(last_message) >= 2: # Already in tuple format final_history[-1] = [last_message[0], final] # In the builder panel log, show the critique, but in the chatbox, only show the final answer # If the challenger changed the answer, only show the suggested revision as the final answer, not the critique text if final != orig and crit != "OK": challenger_md = f"**Original Agent Answer:**\n\n{orig}\n\n**Challenger Critique:**\n\n{crit}\n\n**Final Answer Shown to User:**\n\n{final}" else: challenger_md = f"**Original Agent Answer:**\n\n{orig}\n\n**Final Answer Shown to User:**\n\n{final}" return final_history, updated_histories, "", invocation_log, challenger_md # Add a Markdown for challenger debug info under the invocation log, only in the builder panel challenger_debug_md = gr.Markdown("", visible=True) # Place the Markdown visually under the invocation log in the builder panel only builder_send_button.click( fn=builderpanel_handle_with_dynamic_vars, inputs=[ agent_remove_dropdown, chat_input, builder_chat_histories, builder_deescalation_culture, builder_deescalation_meds, builder_stewardship_site, builder_stewardship_biofilm, builder_stewardship_response, builder_stewardship_crcl, builder_stewardship_severity, builder_stewardship_allergies, builder_empiric_age, builder_empiric_allergies, builder_empiric_labs, builder_empiric_culture, builder_empiric_meds, builder_empiric_site, builder_empiric_biofilm, builder_empiric_response, builder_empiric_crcl, builder_empiric_severity ], outputs=[builder_chatbot, builder_chat_histories, chat_input, invocation_log, challenger_debug_md] ) # Hide the challenger debug markdown in the deployed chat and simple chat panels # (No code needed, as those panels do not use this Markdown block) # (3) Agent-Chat panel, hidden by default with gr.Group(elem_id="agent_chat", visible=False) as chat_panel: gr.Markdown("### πŸ—¨οΈ Chat with Your ID Agents") chat_active_agents = gr.Markdown("### 🧠 Active Agents\n_(None yet)_") # Dropdown to pick which agent to chat with agent_picker = gr.Dropdown( label="Select Agent", choices=list(agents_config.keys()), interactive=True ) # The ChatGPT-style history box chat_view = gr.Chatbot(label="Conversation") # --- Stewardship Tool Clinical Variables Section (Deescalation & Alert Prolonged Abx Use) --- with gr.Accordion("Stewardship Clinical Variables", open=False, visible=False) as stewardship_vars_section: deescalation_culture = gr.Textbox(label="Culture & Sensitivity Results", lines=2, visible=False) deescalation_meds = gr.Textbox(label="Current Antibiotic Regimen", lines=2, visible=False) stewardship_site = gr.Textbox(label="Site of Infection", lines=1) stewardship_biofilm = gr.Textbox(label="Risk or Presence of Biofilm", lines=1) stewardship_response = gr.Textbox(label="Current Response to Antibiotics", lines=1) stewardship_crcl = gr.Textbox(label="Creatinine Clearance", lines=1) stewardship_severity = gr.Textbox(label="Severity of Infection", lines=1) stewardship_allergies = gr.Textbox(label="Known Drug Allergies", lines=1) # --- Empiric Therapy Tool Clinical Variables Section --- with gr.Accordion("Empiric Therapy Clinical Variables", open=False, visible=False) as empiric_vars_section: empiric_age = gr.Textbox(label="Age", lines=1) empiric_allergies = gr.Textbox(label="Allergies", lines=1) empiric_labs = gr.Textbox(label="Recent Labs", lines=2) empiric_culture = gr.Textbox(label="Culture & Sensitivity Results", lines=2) empiric_meds = gr.Textbox(label="Current Antibiotic Regimen", lines=2) empiric_site = gr.Textbox(label="Site of Infection", lines=1) empiric_biofilm = gr.Textbox(label="Risk or Presence of Biofilm", lines=1) empiric_response = gr.Textbox(label="Current Response to Antibiotics", lines=1) empiric_crcl = gr.Textbox(label="Creatinine Clearance", lines=1) empiric_severity = gr.Textbox(label="Severity of Infection", lines=1) empiric_known_allergies = gr.Textbox(label="Known Drug Allergies", lines=1) # --- Infection Prevention and Control Clinical Variables Section --- with gr.Accordion("IPC Clinical Variables", open=False, visible=False) as ipc_vars_section: ipc_facility_name = gr.Textbox(label="Facility Name", lines=1) ipc_location = gr.Textbox(label="Location/Unit", lines=1) ipc_infection_type = gr.Textbox(label="Type of Infection (HAI, SSI, CLABSI, etc.)", lines=1) ipc_onset_date = gr.Textbox(label="Infection Onset Date", lines=1) ipc_device_days = gr.Textbox(label="Device Days (Central Line, Ventilator, etc.)", lines=1) ipc_pathogen = gr.Textbox(label="Pathogen Identified", lines=1) ipc_resistance_pattern = gr.Textbox(label="Resistance Pattern (MRSA, CRE, etc.)", lines=1) ipc_isolation_status = gr.Textbox(label="Current Isolation Precautions", lines=1) ipc_compliance_issues = gr.Textbox(label="Compliance/Breach Issues", lines=2) # --- Clinical Assistant Clinical Variables Section --- with gr.Accordion("Clinical Assessment Variables", open=False, visible=False) as clinical_vars_section: clinical_chief_complaint = gr.Textbox(label="Chief Complaint", lines=2) clinical_history_present = gr.Textbox(label="History of Present Illness", lines=3) clinical_past_medical = gr.Textbox(label="Past Medical History", lines=2) clinical_medications = gr.Textbox(label="Current Medications", lines=2) clinical_allergies = gr.Textbox(label="Allergies", lines=1) clinical_social_history = gr.Textbox(label="Social History (Travel, Exposures)", lines=2) clinical_vital_signs = gr.Textbox(label="Vital Signs", lines=1) clinical_physical_exam = gr.Textbox(label="Physical Examination Findings", lines=3) clinical_lab_results = gr.Textbox(label="Laboratory Results", lines=2) clinical_imaging = gr.Textbox(label="Imaging Results", lines=2) # --- Orchestrator Coordination Variables Section --- with gr.Accordion("Multi-Agent Coordination Variables", open=False, visible=False) as orchestrator_vars_section: # Stewardship Variables (8 fields) orchestrator_culture = gr.Textbox(label="Culture Results", lines=1) orchestrator_meds = gr.Textbox(label="Current Medications", lines=1) orchestrator_site = gr.Textbox(label="Site of Infection", lines=1) orchestrator_biofilm = gr.Textbox(label="Risk of Biofilm", lines=1) orchestrator_response = gr.Textbox(label="Current Response", lines=1) orchestrator_crcl = gr.Textbox(label="Creatinine Clearance", lines=1) orchestrator_severity = gr.Textbox(label="Severity of Infection", lines=1) orchestrator_allergies = gr.Textbox(label="Known Allergies", lines=1) # IPC Variables (9 fields) orchestrator_facility_name = gr.Textbox(label="Facility Name", lines=1) orchestrator_location = gr.Textbox(label="Location/Unit", lines=1) orchestrator_infection_type = gr.Textbox(label="Type of Infection (HAI, SSI, CLABSI, etc.)", lines=1) orchestrator_onset_date = gr.Textbox(label="Infection Onset Date", lines=1) orchestrator_device_days = gr.Textbox(label="Device Days (Central Line, Ventilator, etc.)", lines=1) orchestrator_pathogen = gr.Textbox(label="Pathogen Identified", lines=1) orchestrator_resistance_pattern = gr.Textbox(label="Resistance Pattern (MRSA, CRE, etc.)", lines=1) orchestrator_isolation_status = gr.Textbox(label="Current Isolation Precautions", lines=1) orchestrator_compliance_issues = gr.Textbox(label="Compliance/Breach Issues", lines=2) # Clinical Assistant Variables (10 fields) orchestrator_chief_complaint = gr.Textbox(label="Chief Complaint", lines=2) orchestrator_history_present = gr.Textbox(label="History of Present Illness", lines=3) orchestrator_past_medical = gr.Textbox(label="Past Medical History", lines=2) orchestrator_medications = gr.Textbox(label="Current Medications", lines=2) orchestrator_patient_allergies = gr.Textbox(label="Patient Allergies", lines=1) orchestrator_social_history = gr.Textbox(label="Social History (Travel, Exposures)", lines=2) orchestrator_vital_signs = gr.Textbox(label="Vital Signs", lines=1) orchestrator_physical_exam = gr.Textbox(label="Physical Examination Findings", lines=3) orchestrator_lab_results = gr.Textbox(label="Laboratory Results", lines=2) orchestrator_imaging = gr.Textbox(label="Imaging Results", lines=2) # Add chat input box to chat panel chat_panel_input = gr.Textbox( placeholder="Type your question here…", show_label=False, lines=3, max_lines=5 ) # Only show chat_send, chat_reset, and chat_back in the chat panel with gr.Row(elem_classes="chat-panel-buttons"): chat_send = gr.Button("Send", elem_classes="chat-only-btn") chat_reset = gr.Button("πŸ”„ Reset Chat", elem_classes="chat-only-btn") chat_back = gr.Button("πŸ”™ Back to Builder", elem_id="chat_back", elem_classes="chat-only-btn") # Patient cards section - positioned at bottom of chat panel with gr.Group(elem_classes="patient-cards-section"): gr.Markdown("### 🎯 Select Context-Aware Chat Scenario (Optional)", elem_classes="patient-cards-header") # First row - 3 cards with gr.Row(elem_classes="patient-cards-row"): # Patient Card 1: Stewardship Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_1 = gr.Button( "πŸ“‹ Patient A: ICU Sepsis\n\nπŸ›‘οΈ SmartSteward Case\n\n68F, ICU day 5, on vancomycin + piperacillin-tazobactam for sepsis. Blood cultures positive for MSSA. Patient improving, normal renal function.", elem_classes="patient-card-btn" ) # Patient Card 2: IPC Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_2 = gr.Button( "🦠 Patient B: CLABSI Investigation\n\n🦠 InfectoGuard Case\n\n45M, ICU patient with central line x6 days. Developed fever, positive blood cultures for MRSA. Potential healthcare-associated infection.", elem_classes="patient-card-btn" ) # Patient Card 3: Research Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_3 = gr.Button( "πŸ”¬ Research Query\n\nπŸ”¬ ResearchRanger Case\n\nLiterature search needed for novel carbapenem-resistant Enterobacterales treatment options and resistance mechanisms.", elem_classes="patient-card-btn" ) # Second row - 3 cards with gr.Row(elem_classes="patient-cards-row"): # Patient Card 4: Clinical Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_4 = gr.Button( "πŸ₯ Patient C: Complex Diagnosis\n\nπŸ₯ ClinicoPilot Case\n\n32M with fever, rash, and joint pain after recent travel to Southeast Asia. Multiple differential diagnoses to consider.", elem_classes="patient-card-btn" ) # Patient Card 5: Education Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_5 = gr.Button( "πŸ“š Education Request\n\nπŸ“š EduMedCoach Case\n\nMedical student requesting board exam questions and educational materials on antimicrobial resistance mechanisms.", elem_classes="patient-card-btn" ) # Patient Card 6: Orchestrator Case with gr.Column(elem_classes="patient-card", scale=1): patient_card_6 = gr.Button( "🎼 Complex Multi-Agent Case\n\n🎼 ID Maestro Case\n\n75M with multiple ID issues: MDRO pneumonia, C. diff colitis, and suspected endocarditis requiring comprehensive analysis.", elem_classes="patient-card-btn" ) # Store patient data in hidden state patient_data = gr.State({}) # Define patient case data patient_cases = { "patient_1": { "name": "Patient A", "age": "68", "summary": "68-year-old female in ICU, day 5 of admission for sepsis", "current_meds": "vancomycin 1g q12h, piperacillin-tazobactam 4.5g q6h", "culture_results": "Blood cultures (day 3): methicillin-sensitive Staphylococcus aureus (MSSA), sensitive to cefazolin, nafcillin, clindamycin", "site_of_infection": "bloodstream", "biofilm_risk": "central venous catheter present", "response": "clinically improving, fever resolved, WBC trending down", "creatinine_clearance": "75 mL/min (normal)", "severity": "severe sepsis, now stable", "allergies": "NKDA", "agent_focus": "πŸ›‘οΈ Antimicrobial Stewardship", "context": "This patient is a perfect candidate for antibiotic deescalation given the MSSA blood culture results and clinical improvement. Current broad-spectrum therapy can likely be narrowed." }, "patient_2": { "name": "Patient B", "age": "45", "summary": "45-year-old male, ICU patient with central line-associated bloodstream infection", "diagnosis": "Central line-associated bloodstream infection (CLABSI)", "central_line_days": "6 days", "culture_results": "Blood cultures positive for methicillin-resistant Staphylococcus aureus (MRSA)", "symptoms": "fever (38.8Β°C), chills, no other obvious source", "location": "Methodist Hospital, Dallas, Texas", "agent_focus": "🦠 Infection Prevention and Control", "context": "This case requires evaluation for NHSN CLABSI criteria, appropriate isolation precautions for MRSA, and reporting requirements for healthcare-associated infections in Texas." }, "patient_3": { "name": "Research Query", "topic": "Carbapenem-resistant Enterobacterales (CRE) treatment", "research_focus": "Novel treatment options for CRE infections", "specific_interests": "resistance mechanisms, combination therapies, newer antibiotics", "urgency": "clinical decision support needed", "agent_focus": "πŸ”¬ Research Assistant", "context": "Literature search and evidence synthesis needed for treatment of carbapenem-resistant Enterobacterales infections, including mechanism-based approaches and newest therapeutic options." }, "patient_4": { "name": "Patient C", "age": "32", "summary": "32-year-old male with fever, rash, and arthralgia after travel", "travel_history": "Recent travel to Southeast Asia (Thailand, Vietnam) 3 weeks ago", "symptoms": "fever (39.1Β°C), maculopapular rash on trunk and extremities, polyarthralgia", "duration": "symptoms for 5 days", "differential": "considering dengue fever, chikungunya, Zika virus, typhus, malaria", "agent_focus": "πŸ₯ Clinical Assistant", "context": "Complex infectious disease case requiring systematic evaluation of travel-related illnesses and patient education about diagnostic workup and treatment options." }, "patient_5": { "name": "Education Request", "level": "Medical student, 3rd year", "topic": "Antimicrobial resistance mechanisms", "request": "Board exam questions and educational materials", "focus_areas": "beta-lactamase types, carbapenemases, ESBL, AmpC", "format_needed": "multiple choice questions, flashcards, presentation slides", "agent_focus": "πŸ“š Education Assistant", "context": "Educational content creation for antimicrobial resistance mechanisms, suitable for medical student board exam preparation with varying difficulty levels." }, "patient_6": { "name": "Patient D", "age": "75", "summary": "75-year-old male with multiple infectious complications", "problem_1": "Ventilator-associated pneumonia with XDR Pseudomonas aeruginosa", "problem_2": "Clostridioides difficile colitis (severe, recurrent)", "problem_3": "Suspected infective endocarditis (blood cultures pending)", "comorbidities": "diabetes, chronic kidney disease (CrCl 30 mL/min), heart failure", "current_status": "ICU day 12, on multiple antibiotics, clinically complex", "agent_focus": "🎼 Orchestrator", "context": "Complex multi-system infectious disease case requiring coordination between stewardship, infection control, and clinical decision-making across multiple agents and specialties." } } # Patient card click handlers def load_patient_1(): case = patient_cases["patient_1"] context_msg = f"""**Patient Case Loaded: {case['name']}** **Clinical Summary:** {case['summary']} - **Age:** {case['age']} years old - **Current Antibiotics:** {case['current_meds']} - **Culture Results:** {case['culture_results']} - **Site of Infection:** {case['site_of_infection']} - **Current Response:** {case['response']} - **Renal Function:** {case['creatinine_clearance']} - **Allergies:** {case['allergies']} **Agent Focus:** {case['agent_focus']} *How can I help with this stewardship case?*""" # Auto-populate stewardship clinical variables return ( [["", context_msg]], case, case.get('culture_results', ''), # deescalation_culture case.get('current_meds', ''), # deescalation_meds case.get('site_of_infection', ''), # stewardship_site case.get('biofilm_risk', ''), # stewardship_biofilm case.get('response', ''), # stewardship_response case.get('creatinine_clearance', ''), # stewardship_crcl case.get('severity', ''), # stewardship_severity case.get('allergies', ''), # stewardship_allergies case.get('age', ''), # empiric_age case.get('allergies', ''), # empiric_allergies '', # empiric_labs (not in patient data) case.get('culture_results', ''), # empiric_culture case.get('current_meds', ''), # empiric_meds case.get('site_of_infection', ''), # empiric_site case.get('biofilm_risk', ''), # empiric_biofilm case.get('response', ''), # empiric_response case.get('creatinine_clearance', ''), # empiric_crcl case.get('severity', ''), # empiric_severity '', '', '', '', '', '', '', '', '', # ipc fields (9 fields) '', '', '', '', '', '', '', '', '', '', # clinical fields (10 fields) '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '' # orchestrator fields (27 fields) ) def load_patient_2(): case = patient_cases["patient_2"] context_msg = f"""**Patient Case Loaded: {case['name']}** **Clinical Summary:** {case['summary']} - **Age:** {case['age']} years old - **Diagnosis:** {case['diagnosis']} - **Central Line Duration:** {case['central_line_days']} - **Culture Results:** {case['culture_results']} - **Symptoms:** {case['symptoms']} - **Location:** {case['location']} **Agent Focus:** {case['agent_focus']} *How can I help with this infection prevention case?*""" # Auto-populate IPC clinical variables for Patient B return ( [["", context_msg]], case, '', '', '', '', '', '', '', '', # stewardship fields (8 fields) '', '', '', '', '', '', '', '', '', '', # empiric fields (10 fields) '', # ipc_facility_name (blank for user input) '', # ipc_location (blank for user input) case.get('diagnosis', ''), # ipc_infection_type 'admission + ' + case.get('central_line_days', ''), # ipc_onset_date case.get('central_line_days', ''), # ipc_device_days 'MRSA' if 'MRSA' in case.get('culture_results', '') else case.get('culture_results', ''), # ipc_pathogen 'MRSA' if 'MRSA' in case.get('culture_results', '') else 'pending resistance testing', # ipc_resistance_pattern 'Contact precautions for MRSA', # ipc_isolation_status 'Review central line maintenance', # ipc_compliance_issues '', '', '', '', '', '', '', '', '', '', # clinical fields (10 fields) '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '' # orchestrator fields (27 fields) ) def load_patient_3(): case = patient_cases["patient_3"] context_msg = f"""**Research Query Loaded: {case['name']}** **Research Focus:** {case['research_focus']} - **Topic:** {case['topic']} - **Specific Interests:** {case['specific_interests']} - **Urgency:** {case['urgency']} **Agent Focus:** {case['agent_focus']} *How can I help with your research needs?*""" # Return empty clinical variables for research case return ( [["", context_msg]], case, '', '', '', '', '', '', '', '', # stewardship fields (8 fields) '', '', '', '', '', '', '', '', '', '', # empiric fields (10 fields) '', '', '', '', '', '', '', '', '', # ipc fields (9 fields) '', '', '', '', '', '', '', '', '', '', # clinical fields (10 fields) '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '' # orchestrator fields (27 fields) ) def load_patient_4(): case = patient_cases["patient_4"] context_msg = f"""**Patient Case Loaded: {case['name']}** **Clinical Summary:** {case['summary']} - **Age:** {case['age']} years old - **Travel History:** {case['travel_history']} - **Symptoms:** {case['symptoms']} - **Duration:** {case['duration']} - **Differential Diagnosis:** {case['differential']} **Agent Focus:** {case['agent_focus']} *How can I help with this clinical case?*""" # Return clinical assistant-focused variables for Patient E case return ( [["", context_msg]], case, '', '', '', '', '', '', '', '', # stewardship fields (8 fields) '', '', '', '', '', '', '', '', '', '', # empiric fields (10 fields) '', '', '', '', '', '', '', '', '', # ipc fields (9 fields) case['symptoms'], # chief_complaint f"Patient with travel history to {case['travel_history']} presenting with {case['symptoms']} for {case['duration']}", # history_present 'Travel medicine history as noted', # past_medical_history '', # current_medications '', # allergies f"Recent travel to {case['travel_history']}", # social_history '', # vital_signs '', # physical_exam '', # lab_results '', # imaging_results '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '' # orchestrator fields (27 fields) ) def load_patient_5(): case = patient_cases["patient_5"] context_msg = f"""**Education Request Loaded: {case['name']}** **Educational Details:** - **Student Level:** {case['level']} - **Topic:** {case['topic']} - **Request:** {case['request']} - **Focus Areas:** {case['focus_areas']} - **Formats Needed:** {case['format_needed']} **Agent Focus:** {case['agent_focus']} *How can I help with your educational materials?*""" # Return empty clinical variables for education case return ( [["", context_msg]], case, '', '', '', '', '', '', '', '', # stewardship fields (8 fields) '', '', '', '', '', '', '', '', '', '', # empiric fields (10 fields) '', '', '', '', '', '', '', '', '', # ipc fields (9 fields) '', '', '', '', '', '', '', '', '', '', # clinical fields (10 fields) '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '' # orchestrator fields (27 fields) ) def load_patient_6(): case = patient_cases["patient_6"] context_msg = f"""**Complex Case Loaded: {case['name']}** **Clinical Summary:** {case['summary']} - **Age:** {case['age']} years old - **Problem 1:** {case['problem_1']} - **Problem 2:** {case['problem_2']} """ # Auto-populate clinical variables for complex orchestrator case return ( [["", context_msg]], case, '', # deescalation_culture (no specific culture data) 'multiple antibiotics', # deescalation_meds (from current_status) 'multiple sites (respiratory, GI)', # stewardship_site (inferred from problems) 'biofilm risk from devices', # stewardship_biofilm (inferred) 'complex, multiple infections', # stewardship_response case.get('comorbidities', '').split(',')[1].strip() if 'CrCl' in case.get('comorbidities', '') else '30 mL/min', # stewardship_crcl 'severe, multiple complications', # stewardship_severity 'review allergies needed', # stewardship_allergies case.get('age', ''), # empiric_age 'review allergies needed', # empiric_allergies '', # empiric_labs '', # empiric_culture 'multiple antibiotics', # empiric_meds 'multiple sites (respiratory, GI)', # empiric_site 'biofilm risk from devices', # empiric_biofilm 'complex, multiple infections', # empiric_response '30 mL/min', # empiric_crcl 'severe, multiple complications', # empiric_severity '', '', '', '', '', '', '', '', '', # ipc fields (9 fields) 'Multiple acute medical problems', # chief_complaint f"Complex patient with {case['problem_1']}, {case['problem_2']}, and {case['problem_3']}", # history_present case['comorbidities'], # past_medical_history 'Multiple antibiotics, supportive care', # current_medications 'Review needed', # allergies 'ICU setting, complex care', # social_history 'Multiple abnormalities expected', # vital_signs 'Complex findings across systems', # physical_exam 'Multiple abnormal values', # lab_results 'Multiple studies indicated', # imaging_results # Auto-populate orchestrator variables from case data '', # orchestrator_culture 'multiple antibiotics', # orchestrator_meds 'multiple sites (respiratory, GI)', # orchestrator_site 'biofilm risk from devices', # orchestrator_biofilm 'complex, multiple infections', # orchestrator_response '30 mL/min', # orchestrator_crcl 'severe, multiple complications', # orchestrator_severity 'review allergies needed', # orchestrator_allergies '', # orchestrator_facility_name 'ICU', # orchestrator_location 'multiple HAIs', # orchestrator_infection_type 'admission', # orchestrator_onset_date 'multiple devices', # orchestrator_device_days 'multiple pathogens', # orchestrator_pathogen 'review resistance patterns', # orchestrator_resistance_pattern 'contact precautions', # orchestrator_isolation_status 'complex care coordination', # orchestrator_compliance_issues 'Multiple acute medical problems', # orchestrator_chief_complaint f"Complex patient with {case['problem_1']}, {case['problem_2']}, and {case['problem_3']}", # orchestrator_history_present case['comorbidities'], # orchestrator_past_medical 'Multiple antibiotics, supportive care', # orchestrator_medications 'Review needed', # orchestrator_patient_allergies 'ICU setting, complex care', # orchestrator_social_history 'Multiple abnormalities expected', # orchestrator_vital_signs 'Complex findings across systems', # orchestrator_physical_exam 'Multiple abnormal values', # orchestrator_lab_results 'Multiple studies indicated' # orchestrator_imaging ) # Store patient data in hidden state patient_data = gr.State({}) # Define patient case data patient_cases = { "patient_1": { "name": "Patient A", "age": "68", "summary": "68-year-old female in ICU, day 5 of admission for sepsis", "current_meds": "vancomycin 1g q12h, piperacillin-tazobactam 4.5g q6h", "culture_results": "Blood cultures (day 3): methicillin-sensitive Staphylococcus aureus (MSSA), sensitive to cefazolin, nafcillin, clindamycin", "site_of_infection": "bloodstream", "biofilm_risk": "central venous catheter present", "response": "clinically improving, fever resolved, WBC trending down", "creatinine_clearance": "75 mL/min (normal)", "severity": "severe sepsis, now stable", "allergies": "NKDA", "agent_focus": "πŸ›‘οΈ Antimicrobial Stewardship", "context": "This patient is a perfect candidate for antibiotic deescalation given the MSSA blood culture results and clinical improvement. Current broad-spectrum therapy can likely be narrowed." }, "patient_2": { "name": "Patient B", "age": "45", "summary": "45-year-old male, ICU patient with central line-associated bloodstream infection", "diagnosis": "Central line-associated bloodstream infection (CLABSI)", "central_line_days": "6 days", "culture_results": "Blood cultures positive for methicillin-resistant Staphylococcus aureus (MRSA)", "symptoms": "fever (38.8Β°C), chills, no other obvious source", "location": "Methodist Hospital, Dallas, Texas", "agent_focus": "🦠 Infection Prevention and Control", "context": "This case requires evaluation for NHSN CLABSI criteria, appropriate isolation precautions for MRSA, and reporting requirements for healthcare-associated infections in Texas." }, "patient_3": { "name": "Research Query", "topic": "Carbapenem-resistant Enterobacterales (CRE) treatment", "research_focus": "Novel treatment options for CRE infections", "specific_interests": "resistance mechanisms, combination therapies, newer antibiotics", "urgency": "clinical decision support needed", "agent_focus": "πŸ”¬ Research Assistant", "context": "Literature search and evidence synthesis needed for treatment of carbapenem-resistant Enterobacterales infections, including mechanism-based approaches and newest therapeutic options." }, "patient_4": { "name": "Patient C", "age": "32", "summary": "32-year-old male with fever, rash, and arthralgia after travel", "travel_history": "Recent travel to Southeast Asia (Thailand, Vietnam) 3 weeks ago", "symptoms": "fever (39.1Β°C), maculopapular rash on trunk and extremities, polyarthralgia", "duration": "symptoms for 5 days", "differential": "considering dengue fever, chikungunya, Zika virus, typhus, malaria", "agent_focus": "πŸ₯ Clinical Assistant", "context": "Complex infectious disease case requiring systematic evaluation of travel-related illnesses and patient education about diagnostic workup and treatment options." }, "patient_5": { "name": "Education Request", "level": "Medical student, 3rd year", "topic": "Antimicrobial resistance mechanisms", "request": "Board exam questions and educational materials", "focus_areas": "beta-lactamase types, carbapenemases, ESBL, AmpC", "format_needed": "multiple choice questions, flashcards, presentation slides", "agent_focus": "πŸ“š Education Assistant", "context": "Educational content creation for antimicrobial resistance mechanisms, suitable for medical student board exam preparation with varying difficulty levels." }, "patient_6": { "name": "Patient D", "age": "75", "summary": "75-year-old male with multiple infectious complications", "problem_1": "Ventilator-associated pneumonia with XDR Pseudomonas aeruginosa", "problem_2": "Clostridioides difficile colitis (severe, recurrent)", "problem_3": "Suspected infective endocarditis (blood cultures pending)", "comorbidities": "diabetes, chronic kidney disease (CrCl 30 mL/min), heart failure", "current_status": "ICU day 12, on multiple antibiotics, clinically complex", "agent_focus": "🎼 Orchestrator", "context": "Complex multi-system infectious disease case requiring coordination between stewardship, infection control, and clinical decision-making across multiple agents and specialties." } } # Connect patient card click handlers patient_card_1.click( fn=load_patient_1, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) patient_card_2.click( fn=load_patient_2, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) patient_card_3.click( fn=load_patient_3, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) patient_card_4.click( fn=load_patient_4, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) patient_card_5.click( fn=load_patient_5, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) patient_card_6.click( fn=load_patient_6, inputs=[], outputs=[ chat_view, patient_data, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) # --- Show/hide stewardship fields based on agent selection --- def update_dynamic_vars_visibility(agent_name): agent_json = agents_config.get(agent_name) if agent_json: agent_data = json.loads(agent_json) skills = agent_data.get("skills", []) agent_type = agent_data.get("agent_type", "") # Stewardship tools if "recommend_deescalation" in skills or "alert_prolonged_antibiotic_use" in skills: show_culture = "recommend_deescalation" in skills show_meds = "recommend_deescalation" in skills return ( gr.update(visible=True), # stewardship_vars_section gr.update(visible=show_culture), # deescalation_culture gr.update(visible=show_meds), # deescalation_meds gr.update(visible=False), # empiric_vars_section gr.update(visible=False), # ipc_vars_section gr.update(visible=False), # clinical_vars_section gr.update(visible=False) # orchestrator_vars_section ) # Empiric therapy tool elif "recommend_empiric_therapy" in skills: return ( gr.update(visible=False), # stewardship_vars_section gr.update(visible=False), # deescalation_culture gr.update(visible=False), # deescalation_meds gr.update(visible=True), # empiric_vars_section gr.update(visible=False), # ipc_vars_section gr.update(visible=False), # clinical_vars_section gr.update(visible=False) # orchestrator_vars_section ) # IPC tools elif "IPC_reporting" in skills or "NHSN_criteria_evaluator" in skills or "recommend_isolation_precautions" in skills: return ( gr.update(visible=False), # stewardship_vars_section gr.update(visible=False), # deescalation_culture gr.update(visible=False), # deescalation_meds gr.update(visible=False), # empiric_vars_section gr.update(visible=True), # ipc_vars_section gr.update(visible=False), # clinical_vars_section gr.update(visible=False) # orchestrator_vars_section ) # Clinical Assistant tools elif "retrieve_guidelines" in skills or "explain_in_layman_language" in skills or "history_taking" in skills: return ( gr.update(visible=False), # stewardship_vars_section gr.update(visible=False), # deescalation_culture gr.update(visible=False), # deescalation_meds gr.update(visible=False), # empiric_vars_section gr.update(visible=False), # ipc_vars_section gr.update(visible=True), # clinical_vars_section gr.update(visible=False) # orchestrator_vars_section ) # Orchestrator (check by agent type since it has no specific skills) elif "🎼 Orchestrator" in agent_type: return ( gr.update(visible=False), # stewardship_vars_section gr.update(visible=False), # deescalation_culture gr.update(visible=False), # deescalation_meds gr.update(visible=False), # empiric_vars_section gr.update(visible=False), # ipc_vars_section gr.update(visible=False), # clinical_vars_section gr.update(visible=True) # orchestrator_vars_section ) # Hide all return ( gr.update(visible=False), # stewardship_vars_section gr.update(visible=False), # deescalation_culture gr.update(visible=False), # deescalation_meds gr.update(visible=False), # empiric_vars_section gr.update(visible=False), # ipc_vars_section gr.update(visible=False), # clinical_vars_section gr.update(visible=False) # orchestrator_vars_section ) agent_picker.change( fn=update_dynamic_vars_visibility, inputs=[agent_picker], outputs=[ stewardship_vars_section, deescalation_culture, deescalation_meds, empiric_vars_section, ipc_vars_section, clinical_vars_section, orchestrator_vars_section ] ) # Client-side script gr.HTML(""" """) # --- Interactions --- # Simple GPT-3.5 Chat callbacks (no skills, no internet) def simple_send_handler(user_message, history): updated_history, cleared = simple_chat_response(user_message, history) return updated_history, cleared, updated_history simple_send.click( simple_send_handler, inputs=[simple_input, simple_chat_history], outputs=[simple_chatbot, simple_input, simple_chat_history], ) simple_input.submit( simple_send_handler, inputs=[simple_input, simple_chat_history], outputs=[simple_chatbot, simple_input, simple_chat_history], ) def simple_reset_handler(): return [], "", [] simple_reset.click( simple_reset_handler, inputs=[], outputs=[simple_chatbot, simple_input, simple_chat_history], ) start_button.click( fn=lambda: (gr.update(visible=False), gr.update(visible=True), gr.update(visible=False), ""), inputs=[], outputs=[landing_panel, agent_form, chat_panel, challenger_debug_md], ) back_button.click( fn=show_landing, inputs=[], outputs=[landing_panel, agent_form, chat_panel], ) to_chat_button.click( fn=lambda: (gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), ""), outputs=[landing_panel, agent_form, chat_panel, challenger_debug_md] ).then( refresh_active_agents_widgets, inputs=[], outputs=[chat_active_agents, agent_picker] ) # Only wire up chat_back for the agent_chat panel (third page) chat_back.click( fn=lambda: (gr.update(visible=False), gr.update(visible=True), gr.update(visible=False), ""), outputs=[landing_panel, agent_form, chat_panel, challenger_debug_md] ) # when you pick a new agent, reload its history into chat_view agent_picker.change( fn=load_history, inputs=[agent_picker, deployed_chat_histories], outputs=[chat_view] ) # when you click Send, append & re-render def chatpanel_handle_with_dynamic_vars( agent_name, user_text, histories, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity ): agent_json = agents_config.get(agent_name) # Prevent tool invocation on empty/whitespace input (e.g., initial greeting or agent selection) if user_text is None or not str(user_text).strip(): # Just return the current history, do not invoke any tool return histories.get(agent_name, []), histories, user_text or "" # --- IPC dynamic requirements integration --- ipc_tool_triggered = False ipc_jurisdiction = None ipc_info = None if agent_json: agent_data = json.loads(agent_json) skills = agent_data.get("skills", []) history = histories.get(agent_name, []) # --- Trusted links wiring --- trusted_links = [] for k in ["trusted_links", "trusted_links_1", "trusted_links_2", "trusted_links_3", "trusted_links_4"]: if isinstance(agent_data.get(k), list): trusted_links.extend([l for l in agent_data[k] if l]) elif isinstance(agent_data.get(k), str) and agent_data[k]: trusted_links.append(agent_data[k]) for k in ["link1", "link2", "link3", "link4"]: if agent_data.get(k): trusted_links.append(agent_data[k]) trusted_links = [l for l in trusted_links if l] # --- End trusted links wiring --- # IPC tool dynamic requirements fetch if "ipc_reporting" in skills: # TODO: Fix IPC requirements fetch # fetch_ipc_requirements = tools.fetch_ipc_requirements # Always extract the latest jurisdiction from the most recent user message us_states = [ "Alabama", "Alaska", "Arizona", "Arkansas", "California", "Colorado", "Connecticut", "Delaware", "Florida", "Georgia", "Hawaii", "Idaho", "Illinois", "Indiana", "Iowa", "Kansas", "Kentucky", "Louisiana", "Maine", "Maryland", "Massachusetts", "Michigan", "Minnesota", "Mississippi", "Missouri", "Montana", "Nebraska", "Nevada", "New Hampshire", "New Jersey", "New Mexico", "New York", "North Carolina", "North Dakota", "Ohio", "Oklahoma", "Oregon", "Pennsylvania", "Rhode Island", "South Carolina", "South Dakota", "Tennessee", "Texas", "Utah", "Vermont", "Virginia", "Washington", "West Virginia", "Wisconsin", "Wyoming" ] # Try to find a state name (case-insensitive) in the latest user message state_match = None for state in us_states: if re.search(rf"\\b{state}\\b", user_text, re.IGNORECASE): state_match = state break match = re.search(r"\b(CDC|WHO|United States|US|World Health Organization|global)\b", user_text, re.IGNORECASE) if state_match: ipc_jurisdiction = state_match elif match: ipc_jurisdiction = match.group(1) else: # If not in this message, look back in history for last mentioned jurisdiction ipc_jurisdiction = None for item in reversed(history): if isinstance(item, (list, tuple)) and len(item) >= 1 and item[0]: # item[0] is the user message in tuple format user_content = item[0] for state in us_states: if re.search(rf"\\b{state}\\b", user_content, re.IGNORECASE): ipc_jurisdiction = state break if not ipc_jurisdiction: match2 = re.search(r"\b(CDC|WHO|United States|US|World Health Organization|global)\b", user_content, re.IGNORECASE) if match2: ipc_jurisdiction = match2.group(1) if ipc_jurisdiction: break if not ipc_jurisdiction: ipc_jurisdiction = "CDC" # Default # Fetch requirements for the extracted jurisdiction # ipc_info = fetch_ipc_requirements(ipc_jurisdiction) ipc_info = None # TODO: Fix IPC requirements fetch ipc_tool_triggered = True # Prepend IPC requirements info to user_text for the agent if ipc_info: req_fields_list = ipc_info.get("fields", []) req_fields = ", ".join(req_fields_list) summary = ipc_info.get("summary", "") source_url = ipc_info.get("source_url") warning = ipc_info.get("warning") else: req_fields_list = [] req_fields = "" summary = "" source_url = "" warning = "" if ipc_info: # Only process if we have valid info # --- Prevent repeated tool invocation logic --- # Gather current field values from user_text/history (simple: just use user_text for now) current_submission = { "jurisdiction": ipc_jurisdiction, "fields": req_fields_list, "summary": summary, "user_text": user_text.strip() } # Find last IPC_SUBMISSION in history # Note: In tuple format, system messages are not stored, so this will be None last_ipc_submission = None # Legacy code for dictionary format - skipped in tuple format # for msg in reversed(history): # if msg["role"] == "system" and msg["content"].startswith("[IPC_SUBMISSION]"): # try: # last_ipc_submission = json.loads(msg["content"][len("[IPC_SUBMISSION]"):].strip()) # except Exception: # last_ipc_submission = None # break # Only proceed if something changed if last_ipc_submission == current_submission: # Requirements and user input unchanged, do not re-invoke return history, histories, "" # --- Dynamic required fields logic --- # Only prompt for missing fields if the live/static requirements specify them missing_fields = [] if req_fields_list: # Try to extract field values from user_text (very basic: look for each field name in user_text, case-insensitive) for field in req_fields_list: if not re.search(rf"\b{re.escape(field)}\b", user_text, re.IGNORECASE): missing_fields.append(field) if missing_fields: prompt = f"Please provide the following required information for IPC reporting: {', '.join(missing_fields)}." history.append(["", prompt]) return history, histories, "" # If no required fields or all are present, proceed user_text = f"[IPC_REQUIREMENTS] Jurisdiction: {ipc_jurisdiction}\nRequired fields: {req_fields}\nSummary: {summary}\n" + user_text # Always add a visible assistant message to the chat so the user sees which requirements are being used visible_msg = f"πŸ“‹ Reporting requirements for {ipc_jurisdiction}:
" if req_fields: visible_msg += f"Required fields: {req_fields}. " if summary: visible_msg += f"Summary: {summary} " if source_url: visible_msg += f'
Source: {source_url}' if warning: visible_msg += f'
Warning: {warning}' try: with open("debug_log.txt", "a", encoding="utf-8") as f: f.write(f"[DEBUG] IPC visible_msg: {visible_msg}\n") except Exception as e: pass # Add the requirements message if it's new or the jurisdiction changed if not (history and len(history) > 0 and len(history[-1]) > 1 and ipc_jurisdiction in history[-1][1]): history.append(["", visible_msg]) # Store this submission in history as a hidden system message (keep as internal format) # Note: This system message won't be displayed in the chat UI # Deescalation tool if "recommend_deescalation" in skills: var_names = ["culture", "meds", "site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection", "known_allergies"] user_vars = { "culture": deescalation_culture, "meds": deescalation_meds, "site_of_infection": stewardship_site, "risk_of_biofilm": stewardship_biofilm, "current_response": stewardship_response, "creatinine_clearance": stewardship_crcl, "severity_of_infection": stewardship_severity, "known_allergies": stewardship_allergies } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" # Only prepend if at least one field is non-empty if any(user_vars[k] for k in var_names): user_text = f"[DEESCALATION_TOOL_INPUT] {json.dumps(user_vars)}\n" + user_text # Alert prolonged antibiotic use tool elif "alert_prolonged_antibiotic_use" in skills: var_names = ["site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection", "known_allergies"] user_vars = { "site_of_infection": stewardship_site, "risk_of_biofilm": stewardship_biofilm, "current_response": stewardship_response, "creatinine_clearance": stewardship_crcl, "severity_of_infection": stewardship_severity, "known_allergies": stewardship_allergies } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" if any(user_vars[k] for k in var_names): user_text = f"[ALERT_PROLONGED_ABX_INPUT] {json.dumps(user_vars)}\n" + user_text # Empiric therapy tool elif "recommend_empiric_therapy" in skills: var_names = [ "age", "allergies", "labs", "culture", "meds", "site_of_infection", "risk_of_biofilm", "current_response", "creatinine_clearance", "severity_of_infection" ] user_vars = { "age": empiric_age, "allergies": empiric_allergies, "labs": empiric_labs, "culture": empiric_culture, "meds": empiric_meds, "site_of_infection": empiric_site, "risk_of_biofilm": empiric_biofilm, "current_response": empiric_response, "creatinine_clearance": empiric_crcl, "severity_of_infection": empiric_severity } extracted = extract_clinical_variables_from_history(history, var_names) for k in var_names: if not user_vars[k]: user_vars[k] = extracted.get(k) or "" # If any required field is missing, prompt = ... missing = [k.replace('_', ' ').capitalize() for k in var_names if not user_vars[k].strip()] if missing: prompt = f"Please provide the following required information for empiric therapy: {', '.join(missing)}." # Show this as an assistant message and do not call the tool history.append(["", prompt]) return history, histories, "" # All required fields present, prepend tool input if any(user_vars[k] for k in var_names): user_text = f"[EMPIRIC_THERAPY_INPUT] {json.dumps(user_vars)}\n" + user_text # Use simulate_agent_response_stream for all agents to ensure challenger logic is applied import asyncio from core.agents.chat_orchestrator import simulate_agent_response_stream agent_json_val = agents_config.get(agent_name) history_val = histories.get(agent_name, []) result = None # Prevent repeated tool invocation: if the last assistant message is a tool request for the same required fields, do not re-invoke if history_val and len(history_val) > 0: # Check the last tuple - assistant message is the second element last_tuple = history_val[-1] if len(last_tuple) >= 2 and last_tuple[1]: # Check if assistant message exists and is not empty last_content = last_tuple[1] if "required fields" in last_content.lower() and "ipc_reporting" in last_content.lower(): # Don't re-invoke, just return return history_val, histories, "" async def run_stream(): final_history = history_val final_invocation_log = "" challenger_info = None gen = simulate_agent_response_stream( agent_json=agent_json_val, history=history_val, user_input=user_text, debug_flag=False, active_children=[] ) async for updated_history, _, invocation_log, _, challenger_info in gen: final_history = convert_to_gradio_format(updated_history) final_invocation_log = invocation_log return final_history, final_invocation_log, challenger_info loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(run_stream()) if result is not None and len(result) == 3: final_history, invocation_log, challenger_info = result else: final_history, updated_histories, cleared_input, invocation_log = chatpanel_handle(agent_name, user_text, histories) challenger_info = None return final_history, updated_histories, cleared_input, invocation_log except Exception: # fallback to old behavior if error final_history, updated_histories, cleared_input, invocation_log = chatpanel_handle(agent_name, user_text, histories) challenger_info = None return final_history, updated_histories, cleared_input, invocation_log # Update histories updated_histories = histories.copy() updated_histories[agent_name] = final_history # Prepare challenger markdown (debug log for builder panel) challenger_md = "" challenger_enabled = False try: if agent_json_val: agent_data = json.loads(agent_json_val) challenger_enabled = agent_data.get("challenger_enabled", False) except Exception: pass if isinstance(challenger_info, dict) and challenger_enabled: orig = challenger_info.get("original_reply", "") crit = challenger_info.get("challenger_critique", "") final = challenger_info.get("final_reply", "") challenger_md = f"**Original Agent Answer:**\n\n{orig}\n\n**Challenger Critique:**\n\n{crit}\n\n**Final Answer Shown to User:**\n\n{final}" # Only show the final (challenger-approved) answer in the chatbox if final and final_history and isinstance(final_history, list) and len(final_history) > 0: # final_history might come in dictionary format from simulate_agent_response_stream # Convert to tuple format if needed last_message = final_history[-1] if isinstance(last_message, dict) and "content" in last_message: # Convert from dict format to tuple format if last_message.get("role") == "assistant": final_history[-1] = ["", final] elif last_message.get("role") == "user": # Keep user message, but this shouldn't happen for final challenger message final_history[-1] = [last_message.get("content", ""), ""] elif isinstance(last_message, (list, tuple)) and len(last_message) >= 2: # Already in tuple format final_history[-1] = [last_message[0], final] # If challenger is not enabled, do not show the markdown at all elif not challenger_enabled: challenger_md = "" return final_history, updated_histories, "", invocation_log, challenger_md chat_send.click( fn=chatpanel_handle_with_dynamic_vars, inputs=[ agent_picker, chat_panel_input, deployed_chat_histories, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity ], outputs=[chat_view, deployed_chat_histories, chat_panel_input] ) # --- Reset button for deployed agent chat panel --- def reset_and_clear_deployed_history(agent_name, histories): # Clear orchestrator state to prevent persistence across conversations from core.agents.chat_orchestrator import orchestrators orchestrators.clear() if not agent_name: return ( [], histories, "", "", "", "", "", "", "", "", "", # stewardship fields cleared (8 fields) "", "", "", "", "", "", "", "", "", "", # empiric fields cleared (10 fields) "", "", "", "", "", "", "", "", "", # ipc fields cleared (9 fields) "", "", "", "", "", "", "", "", "", "", # clinical fields cleared (10 fields) "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "" # orchestrator fields cleared (27 fields) ) # Remove the agent's chat history and re-seed with greeting if agent_name in histories: del histories[agent_name] chat_history = [["", f"πŸ‘‹ Hello! I'm {agent_name}. How can I help?"]] histories[agent_name] = chat_history return ( chat_history, histories, "", "", "", "", "", "", "", "", "", # stewardship fields cleared (8 fields) "", "", "", "", "", "", "", "", "", "", # empiric fields cleared (10 fields) "", "", "", "", "", "", "", "", "", # ipc fields cleared (9 fields) "", "", "", "", "", "", "", "", "", "", # clinical fields cleared (10 fields) "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "" # orchestrator fields cleared (27 fields) ) chat_reset.click( reset_and_clear_deployed_history, inputs=[agent_picker, deployed_chat_histories], outputs=[ chat_view, deployed_chat_histories, chat_panel_input, deescalation_culture, deescalation_meds, stewardship_site, stewardship_biofilm, stewardship_response, stewardship_crcl, stewardship_severity, stewardship_allergies, empiric_age, empiric_allergies, empiric_labs, empiric_culture, empiric_meds, empiric_site, empiric_biofilm, empiric_response, empiric_crcl, empiric_severity, ipc_facility_name, ipc_location, ipc_infection_type, ipc_onset_date, ipc_device_days, ipc_pathogen, ipc_resistance_pattern, ipc_isolation_status, ipc_compliance_issues, clinical_chief_complaint, clinical_history_present, clinical_past_medical, clinical_medications, clinical_allergies, clinical_social_history, clinical_vital_signs, clinical_physical_exam, clinical_lab_results, clinical_imaging, orchestrator_culture, orchestrator_meds, orchestrator_site, orchestrator_biofilm, orchestrator_response, orchestrator_crcl, orchestrator_severity, orchestrator_allergies, orchestrator_facility_name, orchestrator_location, orchestrator_infection_type, orchestrator_onset_date, orchestrator_device_days, orchestrator_pathogen, orchestrator_resistance_pattern, orchestrator_isolation_status, orchestrator_compliance_issues, orchestrator_chief_complaint, orchestrator_history_present, orchestrator_past_medical, orchestrator_medications, orchestrator_patient_allergies, orchestrator_social_history, orchestrator_vital_signs, orchestrator_physical_exam, orchestrator_lab_results, orchestrator_imaging ] ) agent_type.change( fn=on_agent_type_change, inputs=[agent_type, prefill_flag], outputs=[skills, agent_name, agent_mission, prefill_flag] ) prefilled.change( fn=populate_from_preset, inputs=[prefilled], outputs=[agent_type, agent_name, agent_mission, skills, prefill_flag] ) uploaded_files.upload(fn=handle_uploaded_files, inputs=[uploaded_files], outputs=[upload_alert, upload_alert]) def handle_generate(agent_type, agent_name, agent_mission, selected_skills, web_access, allow_fallback, uploaded_files, link1, link2, link3, link4, challenger_toggle): # Accept challenger_toggle as an argument agent_json = build_agent(agent_type, agent_name, agent_mission, selected_skills, web_access, allow_fallback, uploaded_files, link1, link2, link3, link4) # Add challenger_enabled to the agent config JSON agent_data = json.loads(agent_json) agent_data["challenger_enabled"] = challenger_toggle agent_json = json.dumps(agent_data) agents_config[agent_name] = agent_json return agent_json generate_button.click( lambda: gr.update(visible=True, value="⏳ Generating your agent..."), inputs=[], outputs=[agent_loader] ).then( # 1) add / save the agent handle_generate, inputs=[agent_type, agent_name, agent_mission, skills, web_access_toggle, allow_fallback_toggle, uploaded_files, link1, link2, link3, link4, challenger_toggle], outputs=[agent_output] # <- only the JSON ).then( # 2) show initial instruction instead of preload demo chat lambda: show_initial_instruction_state(), inputs=[], outputs=[builder_chatbot, chat_input, builder_send_button, reset_button, invocation_log, active_children] ).then( # 2.5) auto-load agent fields into builder panel lambda agent_json: ( json.loads(agent_json).get("agent_type", ""), json.loads(agent_json).get("agent_name", ""), json.loads(agent_json).get("agent_mission", ""), json.loads(agent_json).get("skills", []) ), inputs=[agent_output], outputs=[agent_type, agent_name, agent_mission, skills] ).then( # 3) refresh markdown & dropdown *atomically* refresh_active_agents_widgets, inputs=[], outputs=[builder_active_agents, agent_remove_dropdown] ).then( # 4) done spinner lambda: gr.update(visible=True, value="βœ… Agent generated successfully!"), inputs=[], outputs=[agent_loader] ).then( refresh_chat_dropdown, inputs=[], outputs=[agent_picker] ) edit_agent_button.click( load_agent_to_builder, inputs=[agent_remove_dropdown], outputs=[agent_type, agent_name, agent_mission, skills] ) chat_agent_button.click( fn=enable_chat_controls_with_agent, inputs=[agent_remove_dropdown], outputs=[builder_chatbot, chat_input, builder_send_button, reset_button, invocation_log, active_children, agent_output] ) remove_agent_button.click( remove_selected_agent, inputs=[agent_remove_dropdown], outputs=[builder_active_agents, agent_remove_dropdown] ).then( refresh_chat_dropdown, inputs=[], outputs=[agent_picker] ) download_button.click( prepare_download, inputs=[agent_output], outputs=[download_button] ) # Only keep reset for builder panel, and chat_send for chat panel def reset_and_clear_builder_history(agent_json, histories): # Clear orchestrator state to prevent persistence across conversations from core.agents.chat_orchestrator import orchestrators orchestrators.clear() if not agent_json or agent_json.strip() == "": # No agent selected, return disabled state with instruction instruction_state = show_initial_instruction_state() return instruction_state[0], instruction_state[1], instruction_state[5], histories # chatbot, input, active_children, histories # Valid agent selected, show agent greeting with enabled controls agent_data = json.loads(agent_json) name = agent_data.get("agent_name", "Agent") welcome = f"πŸ‘‹ Hello! I'm {name}. How can I assist you today?" chat_history = [["", welcome]] if name in histories: del histories[name] return chat_history, gr.update(value="", interactive=True, placeholder="Type your question here…"), [], histories reset_button.click( reset_and_clear_builder_history, inputs=[agent_output, builder_chat_histories], outputs=[builder_chatbot, chat_input, active_children, builder_chat_histories] ) return app if __name__ == "__main__": try: print("πŸš€ Launching ID Agents with Beta Testing Authentication...") # Authentication credentials (Gradio built-in auth) raw_auth = os.getenv("AUTH_CREDENTIALS", "") auth_credentials = [tuple(pair.split(":")) for pair in raw_auth.split(",") if ":" in pair] auth_message = """ **ID Agents Beta Testing Access** Welcome to the ID Agents beta testing environment! Please use your assigned credentials to access the application. """ # Check if running on Hugging Face Spaces (with fixed config) try: from hf_config import configure_hf_environment, get_hf_launch_config if configure_hf_environment(): # Use FIXED HF Spaces configuration with authentication launch_config = get_hf_launch_config() launch_config["auth"] = auth_credentials launch_config["auth_message"] = auth_message print("πŸ” Authentication enabled for HF Spaces deployment (FIXED)") print("πŸš€ Launching Complex ID Agents on Hugging Face Spaces...") else: # Local development with authentication for testing launch_config = { "auth": auth_credentials, "auth_message": auth_message, "share": False, "server_name": "127.0.0.1", "server_port": 7860, "show_error": True } print("πŸ” Authentication enabled for local testing") print("πŸ”§ Launching Complex ID Agents locally...") except ImportError: # Fallback configuration with authentication launch_config = { "auth": auth_credentials, "auth_message": auth_message, "share": False, "server_name": "127.0.0.1", "server_port": 7860, "show_error": True } print("πŸ” Authentication enabled with fallback configuration") print("πŸ”§ Launching Complex ID Agents with default configuration...") print("πŸ“‹ Available beta test accounts:") for username, password in auth_credentials: print(f" β€’ {username} / {password}") build_ui().launch(**launch_config) except Exception as e: logger.error(f"Failed to launch ID Agents: {e}") print(f"❌ Failed to launch ID Agents: {e}") print("πŸ’‘ Check your API keys and environment configuration") import traceback traceback.print_exc()