import gradio as gr import pandas as pd import asyncio import logging import sys import os import io from contextlib import asynccontextmanager # Change from absolute imports to relative imports from database import initialize_database, add_participant, get_participants_dataframe from matching_agent import create_matching_agent, run_matching from tinyagent.hooks.logging_manager import LoggingManager from tinyagent.hooks.logging_manager import LoggingManager import logging # --- Logging Setup --- log_manager = LoggingManager(default_level=logging.INFO) log_manager.set_levels({ 'tinyagent.hooks.gradio_callback': logging.DEBUG, 'tinyagent.tiny_agent': logging.DEBUG, 'tinyagent.mcp_client': logging.DEBUG, 'tinyagent.code_agent': logging.DEBUG, }) console_handler = logging.StreamHandler(sys.stdout) log_manager.configure_handler( console_handler, format_string='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.DEBUG ) # Gradio log handler log_stream = io.StringIO() gradio_handler = logging.StreamHandler(log_stream) log_manager.configure_handler( gradio_handler, format_string='%(asctime)s - %(levelname)s - %(name)s - %(message)s', level=logging.DEBUG # Capture detailed logs for the UI ) logger = log_manager.get_logger('app') # --- Initial Setup --- logger.info("Initializing database...") initialize_database() # Check for API key if not os.environ.get("OPENAI_API_KEY"): raise ValueError("The OPENAI_API_KEY environment variable is not set. Please set it before running the app.") # --- Session Management --- @asynccontextmanager async def agent_session(): """Create and manage a session-specific agent.""" session_agent = create_matching_agent(log_manager=log_manager) try: yield session_agent finally: await session_agent.close() logger.info("Session agent resources cleaned up") # --- Gradio UI Functions --- def register_participant(name, email, linkedin, background, goals): """Callback function to register a new participant.""" if not all([name, email]): return "Please provide at least a name and email.", get_participants_dataframe() participant_data = { "name": name, "email": email, "linkedin_profile": linkedin, "background": background, "goals": goals } try: add_participant(participant_data) feedback = f"✅ Success! Participant '{name}' registered." logger.info(f"Registered new participant: {email}") except Exception as e: feedback = f"❌ Error! Could not register participant. Reason: {e}" logger.error(f"Failed to register participant {email}: {e}") return feedback, get_participants_dataframe() def refresh_participants_list(): """Callback to reload the participant data from the database.""" return get_participants_dataframe() async def run_matching_process(organizer_criteria, progress=gr.Progress(track_tqdm=True)): """Async callback to run the team matching process, with live log streaming.""" # 1. Clear previous logs and show panel log_stream.seek(0) log_stream.truncate(0) progress(0, desc="Initializing...") yield [ gr.update(visible=True, open=True), # log_panel "Starting matching process...\n", # log_output "🚀 Setting up the matching process..." # matching_results_out ] # 2. Get participants and perform initial checks logger.info("Fetching participants...") progress(0, desc="Fetching participants...") participants_df = get_participants_dataframe() if len(participants_df) < 2: warning_msg = "Matching process aborted: not enough participants." logger.warning(warning_msg) yield [ gr.update(), # log_panel log_stream.getvalue(), # log_output "Cannot run matching with fewer than 2 participants." # matching_results_out ] return logger.info(f"Running matching for {len(participants_df)} participants.") progress(0.2, desc="🧠 Agent is thinking...") # 3. Create a session-specific agent and run matching async with agent_session() as session_agent: logger.info("Created session-specific agent") # Create a background task for the core matching logic match_task = asyncio.create_task( run_matching(session_agent, participants_df, organizer_criteria) ) # 4. Stream logs while the agent works while not match_task.done(): await asyncio.sleep(0.5) # Poll for new logs every 0.5s yield [ gr.update(), # log_panel log_stream.getvalue(), # log_output gr.update() # matching_results_out - No change to final output yet ] # 5. Process the final result from the agent try: final_report = await match_task logger.info("Matching process completed successfully.") progress(1.0, desc="✅ Done!") yield [ gr.update(), # log_panel log_stream.getvalue(), # log_output final_report # matching_results_out ] except Exception as e: logger.error(f"An error occurred during the matching process: {e}", exc_info=True) yield [ gr.update(), # log_panel log_stream.getvalue(), # log_output f"An error occurred: {e}" # matching_results_out ] # --- Gradio App Definition --- with gr.Blocks(theme=gr.themes.Default( font=[gr.themes.GoogleFont("Inter"), "Arial", "sans-serif"] ), title="HackBuddyAI") as app: gr.Markdown("# 🤖 HackBuddyAI") with gr.Tabs(): with gr.TabItem("👤 Participant Registration"): gr.Markdown("## Welcome, Participant!") gr.Markdown("Fill out the form below to register for the hackathon.") with gr.Row(): with gr.Column(): name_in = gr.Textbox(label="Full Name") email_in = gr.Textbox(label="Email Address") linkedin_in = gr.Textbox(label="LinkedIn Profile URL", placeholder="Optional") with gr.Column(): background_in = gr.Textbox(label="Your Background & Skills", lines=5, placeholder="e.g., Python developer with 3 years of experience, specializing in Django and REST APIs...") goals_in = gr.Textbox(label="Your Goals for this Hackathon", lines=5, placeholder="e.g., I want to learn about machine learning and work on a cool data visualization project...") submit_button = gr.Button("Register", variant="primary") registration_feedback = gr.Markdown() with gr.TabItem("👑 Organizer Dashboard"): gr.Markdown("## Welcome, Organizer!") gr.Markdown("Here you can view registered participants and run the AI-powered team matching process.") with gr.Accordion("View Registered Participants", open=False): refresh_button = gr.Button("🔄 Refresh List") participants_df_out = gr.DataFrame(value=get_participants_dataframe, interactive=False) gr.Markdown("### Run Matching") organizer_criteria_in = gr.Textbox( label="Matching Criteria", lines=4, value="Create teams of 3. Try to balance skills in each team (e.g., frontend, backend, data).", placeholder="Describe your ideal team composition..." ) run_button = gr.Button("🚀 Run AI Matching", variant="primary") gr.Markdown("### 🤝 Matched Teams") matching_results_out = gr.Markdown("Matching has not been run yet.") with gr.Accordion("Agent Logs", open=False, visible=False) as log_panel: log_output = gr.Code( label="Live Logs", lines=15, interactive=False, ) # Footer gr.Markdown( "
" "Built with ❤️ by TinyAgent" "
Start building your own AI agents with TinyAgent" "
" ) # --- Event Handlers --- submit_button.click( fn=register_participant, inputs=[name_in, email_in, linkedin_in, background_in, goals_in], outputs=[registration_feedback, participants_df_out] ) refresh_button.click( fn=refresh_participants_list, inputs=[], outputs=[participants_df_out] ) run_button.click( fn=run_matching_process, inputs=[organizer_criteria_in], outputs=[log_panel, log_output, matching_results_out] ) # --- Launching the App --- if __name__ == "__main__": try: logger.info("Launching Gradio app...") # queue() is important for handling multiple users and async calls app.queue().launch(share=False) except KeyboardInterrupt: logger.info("Gradio app shutting down.") finally: logger.info("Shutdown complete.")