#!/usr/bin/env python3 """ MIT License Copyright (c) 2025 Lin Yang, Yichen Huang Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import gradio as gr import os import sys import subprocess import tempfile import time from datetime import datetime import threading import queue import logging # Configure detailed logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) # Provider configuration with agent files and available models PROVIDERS = { 'Google Gemini': { 'agent_file': 'code/agent.py', 'api_key': 'GOOGLE_API_KEY', 'models': [ 'gemini-2.0-flash', # Higher free tier limits (1500 RPD) 'gemini-2.0-flash-lite', 'gemini-2.5-flash', # Good balance 'gemini-2.5-flash-lite', 'gemini-2.5-pro', # Best but limited (25 RPD) 'gemini-3-pro-preview' ], 'default_model': 'gemini-2.0-flash' # Default to higher limit model }, 'OpenAI': { 'agent_file': 'code/agent_oai.py', 'api_key': 'OPENAI_API_KEY', 'models': [ 'gpt-5-pro-2025-10-06', 'gpt-5.1-2025-11-13', 'gpt-5-nano-2025-08-07' ], 'default_model': 'gpt-5-pro-2025-10-06' }, 'XAI': { 'agent_file': 'code/agent_xai.py', 'api_key': 'XAI_API_KEY', 'models': [ 'grok-4-0709', 'grok-4-turbo' ], 'default_model': 'grok-4-0709' } } # Legacy mapping for backward compatibility AGENT_FILES = { 'Google Gemini': 'code/agent.py', 'OpenAI': 'code/agent_oai.py', 'XAI': 'code/agent_xai.py' } # Sample problems SAMPLE_PROBLEMS = { "-- Select a sample problem --": "", "IMO-style: Red and Blue Cards": """Let n be a positive integer. There are n red cards and n blue cards. Initially, every red card has the real number 0 written on it, and every blue card has the real number 1 written on it. An operation consists of selecting one red card and one blue card such that the number x on the red card is strictly less than the number y on the blue card, erasing both numbers, and replacing them both with the average (x+y)/2. Find the minimum positive integer n such that it is possible, using a finite sequence of these operations, to make the sum of the numbers on the n red cards strictly greater than 100.""", "Number Theory: Prime Sums": """Find all prime numbers p for which there exist non-negative integers x, y, and z such that p = x³ + y³ + z³.""", "Combinatorics: Permutation Divisibility": """Determine all positive integers n for which there exists a permutation (a₁, a₂, ..., aₙ) of (1, 2, ..., n) such that for each i ∈ {1, 2, ..., n}, the sum a₁ + a₂ + ... + aᵢ is divisible by i.""" } def check_api_keys(): """Check which API keys are available.""" logger.info("=" * 60) logger.info("Checking API keys availability...") available = [] for provider_name, config in PROVIDERS.items(): api_key = os.getenv(config['api_key']) if api_key: available.append(provider_name) logger.info("✓ %s found (length: %d)", config['api_key'], len(api_key)) else: logger.warning("✗ %s not found", config['api_key']) logger.info("Available providers: %s", available if available else "None") logger.info("=" * 60) return available def get_models_for_provider(provider): """Get available models for a provider.""" if provider in PROVIDERS: return PROVIDERS[provider]['models'] return [] def extract_clean_solution(raw_log_content): """ Extract clean solution text from raw log content. Only extracts content AFTER "Found a correct solution in run" marker. Args: raw_log_content: Raw content from the log file Returns: Clean solution text or empty string """ if not raw_log_content: return "" # First, check if solution was found solution_marker = "Found a correct solution in run" if solution_marker not in raw_log_content: # No solution found yet, return empty return "" # Find where the final solution starts (use rfind to get LAST occurrence) marker_index = raw_log_content.rfind(solution_marker) if marker_index == -1: return "" # Extract everything after this marker content_after_marker = raw_log_content[marker_index:] # Now clean up the extracted content lines = content_after_marker.split('\n') clean_lines = [] skip_patterns = [ '>>>>>>>', # Debug markers '[2025-', # Timestamps 'Logging to file:', 'Found a correct solution in run', # Skip the marker itself '{', # JSON lines '}', '"', # Quoted JSON strings 'null', ] in_solution = False for line in lines: # Skip empty lines at the start if not in_solution and not line.strip(): continue # Skip debug/system lines should_skip = False for pattern in skip_patterns: if pattern in line: should_skip = True break if should_skip: continue # Check if we've reached actual solution content if any(marker in line for marker in ['Summary', '## Summary', '### Summary', '**Summary**', 'Detailed Solution', 'Method Sketch']): in_solution = True if in_solution or line.strip().startswith(('*', '-', '1.', '2.', '3.', '$', '#')): clean_lines.append(line) result = '\n'.join(clean_lines).strip() # If we got nothing, return empty if not result or len(result) < 50: return "" return result def solve_problem(problem_text, provider, model, max_runs, num_agents, other_prompts, progress=gr.Progress()): """ Solve a math problem using the selected AI provider and model. Args: problem_text: The problem statement provider: AI provider to use model: Model to use for the provider max_runs: Maximum number of attempts per agent num_agents: Number of parallel agents other_prompts: Additional prompts (comma-separated) progress: Gradio progress indicator Yields: status, partial_solution, log_text, final_solution_text """ logger.info("=" * 60) logger.info("NEW SOLVE REQUEST") logger.info("=" * 60) logger.info("Provider: %s", provider) logger.info("Model: %s", model) logger.info("Max runs: %d", max_runs) logger.info("Num agents: %d", num_agents) logger.info("Other prompts: %s", other_prompts if other_prompts else "None") logger.info("Problem length: %d characters", len(problem_text) if problem_text else 0) # Validate inputs if not problem_text or not problem_text.strip(): logger.error("Validation failed: Empty problem text") yield "❌ Error", "", "Please enter a problem statement", "" return if provider not in PROVIDERS: logger.error("Validation failed: Invalid provider: %s", provider) yield "❌ Error", "", f"Invalid provider: {provider}", "" return # Validate model if model not in PROVIDERS[provider]['models']: logger.error("Validation failed: Invalid model %s for provider %s", model, provider) yield "❌ Error", "", f"Invalid model {model} for {provider}", "" return # Check API key logger.info("Checking API key availability for %s...", provider) available_providers = check_api_keys() if provider not in available_providers: logger.error("API key not found for %s", provider) logger.error("Available providers: %s", available_providers) yield "❌ Error", "", f"API key not found for {provider}. Please set the appropriate environment variable.", "" return logger.info("API key verified for %s", provider) # Create temporary file for the problem with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: f.write(problem_text.strip()) problem_file = f.name logger.info("Created temporary problem file: %s", problem_file) # Create log directory session_id = f"session_{int(time.time())}" log_dir = os.path.join('logs', session_id) os.makedirs(log_dir, exist_ok=True) logger.info("Created log directory: %s", log_dir) logger.info("Session ID: %s", session_id) try: agent_file = PROVIDERS[provider]['agent_file'] log_output = [] # Create environment with model specified env = os.environ.copy() env['MODEL_NAME'] = model logger.info("Set MODEL_NAME environment variable: %s", model) if num_agents == 1: # Single agent mode logger.info("Starting SINGLE AGENT mode") logger.info("Agent file: %s", agent_file) logger.info("Model: %s", model) progress(0, desc="Starting single agent...") yield f"🔄 Running {provider} ({model})...", "", "", "" log_file = os.path.join(log_dir, 'solution.log') solution_file = os.path.join(log_dir, 'solution_clean.txt') logger.info("Solution log file: %s", log_file) logger.info("Clean solution file: %s", solution_file) cmd = [ sys.executable, agent_file, problem_file, '--log', log_file, '--solution', solution_file, '--max_runs', str(max_runs) ] if other_prompts and other_prompts.strip(): cmd.extend(['--other_prompts', other_prompts.strip()]) logger.info("Command: %s", ' '.join(cmd)) logger.info("Environment: MODEL_NAME=%s", model) logger.info("Starting subprocess...") log_output.append(f"[{datetime.now().strftime('%H:%M:%S')}] Starting {provider} agent with model {model}...") log_output.append(f"Command: {' '.join(cmd)}\n") yield f"🔄 Running {provider} ({model})...", "", "\n".join(log_output), "" # Run the process try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, env=env ) logger.info("Subprocess started with PID: %d", process.pid) except Exception as e: logger.error("Failed to start subprocess: %s", str(e)) raise # Stream output logger.info("Reading subprocess output...") line_count = 0 last_update_time = time.time() for line in process.stdout: line = line.strip() if line: log_output.append(line) line_count += 1 # Log every 50th line to avoid spam if line_count % 50 == 0: logger.info("Processed %d lines from subprocess", line_count) # Update progress every 5 lines or every 2 seconds current_time = time.time() if line_count % 5 == 0 or (current_time - last_update_time) >= 2: progress(min(0.9, line_count / 100), desc=f"Processing... ({line_count} lines)") # Show simple progress message (don't show partial solution during processing) display_solution = f"⏳ Working on solution...\n\n🤖 {provider} ({model}) is processing your problem\n📊 Processed {line_count} lines\n\n💡 The final solution will appear below when complete." # Yield with progress message (no partial solution) yield f"🔄 Running {provider} ({model})...", display_solution, "\n".join(log_output[-100:]), "" last_update_time = current_time process.wait() return_code = process.returncode logger.info("Subprocess completed with return code: %d", return_code) logger.info("Total output lines: %d", line_count) progress(1.0, desc="Complete!") # Check for solution in the clean solution file logger.info("Checking for solution file...") solution_found = False solution_text = "" # Check if the clean solution file exists if os.path.exists(solution_file): try: with open(solution_file, 'r', encoding='utf-8') as f: solution_text = f.read().strip() if solution_text: solution_found = True logger.info("✓ SOLUTION FOUND! Size: %d bytes", len(solution_text)) else: logger.warning("Solution file exists but is empty") except Exception as e: logger.error("Error reading solution file: %s", str(e)) log_output.append(f"\n❌ Error reading solution file: {str(e)}") else: logger.warning("No solution file found at: %s", solution_file) if solution_found: logger.info("Returning SUCCESS result to user") log_output.append("\n" + "=" * 60) log_output.append("✅ SOLUTION FOUND!") log_output.append("=" * 60) # Clear partial solution, show only in final solution box yield "✅ Solution Found!", "✅ Complete! See 'Final Solution' section below.", "\n".join(log_output), solution_text else: logger.info("Returning NO SOLUTION result to user") log_output.append("\n" + "=" * 60) log_output.append("⚠️ No solution found") log_output.append("=" * 60) yield "⚠️ No Solution Found", "", "\n".join(log_output), "" else: # Parallel mode logger.info("Starting PARALLEL MODE with %d agents", num_agents) logger.info("Agent file: %s", agent_file) logger.info("Model: %s", model) progress(0, desc=f"Starting {num_agents} parallel agents...") yield f"🔄 Running {num_agents} parallel {provider} ({model}) agents...", "", "", "" cmd = [ sys.executable, 'code/run_parallel.py', problem_file, '-n', str(num_agents), '-d', log_dir, '-a', agent_file ] if other_prompts and other_prompts.strip(): cmd.extend(['-o', other_prompts.strip()]) logger.info("Command: %s", ' '.join(cmd)) logger.info("Environment: MODEL_NAME=%s", model) logger.info("Working directory: %s", os.path.dirname(os.path.abspath(__file__))) log_output.append(f"[{datetime.now().strftime('%H:%M:%S')}] Starting {num_agents} parallel {provider} agents with model {model}...") log_output.append(f"Command: {' '.join(cmd)}\n") yield f"🔄 Running {num_agents} agents ({model})...", "", "\n".join(log_output), "" # Run parallel process try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, cwd=os.path.dirname(os.path.abspath(__file__)), env=env ) logger.info("Parallel subprocess started with PID: %d", process.pid) except Exception as e: logger.error("Failed to start parallel subprocess: %s", str(e)) raise # Stream output logger.info("Reading parallel subprocess output...") line_count = 0 last_update_time = time.time() for line in process.stdout: line = line.strip() if line: log_output.append(line) line_count += 1 # Log every 50th line if line_count % 50 == 0: logger.info("Processed %d lines from parallel subprocess", line_count) # Update progress every 5 lines or every 2 seconds current_time = time.time() if line_count % 5 == 0 or (current_time - last_update_time) >= 2: progress(min(0.9, line_count / 200), desc=f"Processing... ({line_count} lines)") # Show simple progress message (don't show partial solution during processing) display_solution = f"⏳ Working on solution...\n\n🤖 {num_agents} × {provider} ({model}) agents are processing your problem\n📊 Processed {line_count} lines\n\n💡 The final solution will appear below when complete." # Yield with progress message (no partial solution) yield f"🔄 Running {num_agents} agents ({model})...", display_solution, "\n".join(log_output[-100:]), "" last_update_time = current_time process.wait() return_code = process.returncode logger.info("Parallel subprocess completed with return code: %d", return_code) logger.info("Total output lines: %d", line_count) progress(1.0, desc="Complete!") # Check for solution in clean solution files logger.info("Checking for solution files in %s...", log_dir) solution_found = False solution_text = "" solution_file_found = None # Look for any *_solution_clean.txt files try: solution_files = [f for f in os.listdir(log_dir) if f.endswith('_solution_clean.txt')] logger.info("Found %d solution files in %s", len(solution_files), log_dir) for solution_file_name in solution_files: solution_path = os.path.join(log_dir, solution_file_name) try: with open(solution_path, 'r', encoding='utf-8') as f: content = f.read().strip() if content: solution_found = True solution_text = content solution_file_found = solution_file_name logger.info("✓ SOLUTION FOUND in %s!", solution_file_name) break except Exception as e: logger.warning("Error reading %s: %s", solution_file_name, str(e)) pass except Exception as e: logger.error("Error listing solution files: %s", str(e)) if solution_found: logger.info("Returning SUCCESS result to user (parallel mode)") log_output.append("\n" + "=" * 60) log_output.append(f"✅ SOLUTION FOUND in {solution_file_found}!") log_output.append("=" * 60) # Clear partial solution, show only in final solution box yield "✅ Solution Found!", "✅ Complete! See 'Final Solution' section below.", "\n".join(log_output), solution_text else: logger.info("No solution found in any agent log file") log_output.append("\n" + "=" * 60) log_output.append("⚠️ No solution found by any agent") log_output.append("=" * 60) yield "⚠️ No Solution Found", "", "\n".join(log_output), "" except Exception as e: logger.exception("EXCEPTION occurred during solve_problem:") error_msg = f"❌ Error: {str(e)}" yield "❌ Error", "", error_msg, "" finally: # Clean up temporary problem file try: os.unlink(problem_file) logger.info("Cleaned up temporary problem file: %s", problem_file) except Exception as e: logger.warning("Failed to cleanup temporary file %s: %s", problem_file, str(e)) pass logger.info("=" * 60) logger.info("SOLVE REQUEST COMPLETED") logger.info("=" * 60) def load_sample_problem(sample_choice): """Load a sample problem from the dropdown.""" logger.info("Sample problem selected: %s", sample_choice) return SAMPLE_PROBLEMS.get(sample_choice, "") def create_interface(): """Create the Gradio interface.""" # Check available providers available_providers = check_api_keys() if not available_providers: # Create a warning interface if no API keys are set with gr.Blocks(title="IMO Math Problem Solver") as demo: gr.Markdown(""" # ⚠️ API Keys Required Please set at least one API key as an environment variable: - `GOOGLE_API_KEY` for Google Gemini - `OPENAI_API_KEY` for OpenAI GPT-5 - `XAI_API_KEY` for XAI Grok-4 Then restart the application. """) return demo # Create the main interface with gr.Blocks(title="IMO Math Problem Solver") as demo: gr.Markdown(""" # 🎓 IMO Math Problem Solver ### AI-powered solution for International Mathematical Olympiad problems Powered by Google Gemini, OpenAI, and XAI | MIT License © 2025 Lin Yang, Yichen Huang """) with gr.Row(): with gr.Column(scale=2): # Problem input section gr.Markdown("## 📝 Problem Statement") sample_dropdown = gr.Dropdown( choices=list(SAMPLE_PROBLEMS.keys()), label="📚 Or select a sample problem", value="-- Select a sample problem --" ) problem_input = gr.Textbox( label="Enter your mathematical problem", placeholder="Type or paste your problem here...\n\nExample: Find the minimum positive integer n such that...", lines=10, max_lines=20 ) # Link sample dropdown to problem input sample_dropdown.change( fn=load_sample_problem, inputs=[sample_dropdown], outputs=[problem_input] ) # Configuration section gr.Markdown("## ⚙️ Configuration") with gr.Row(): provider_select = gr.Dropdown( choices=available_providers, label="🤖 AI Provider", value=available_providers[0] if available_providers else None ) model_select = gr.Dropdown( choices=get_models_for_provider(available_providers[0]) if available_providers else [], label="🎯 Model", value=PROVIDERS[available_providers[0]]['default_model'] if available_providers else None ) # Update model dropdown when provider changes def update_models(provider): models = get_models_for_provider(provider) default = PROVIDERS[provider]['default_model'] if provider in PROVIDERS else None return gr.Dropdown(choices=models, value=default) provider_select.change( fn=update_models, inputs=[provider_select], outputs=[model_select] ) with gr.Row(): max_runs_input = gr.Slider( minimum=1, maximum=50, value=10, step=1, label="🔄 Max Attempts per Agent" ) num_agents_input = gr.Slider( minimum=1, maximum=20, value=1, step=1, label="👥 Number of Parallel Agents" ) other_prompts_input = gr.Textbox( label="💡 Additional Prompts (optional, comma-separated)", placeholder="e.g., focus_on_geometry, use_induction" ) solve_button = gr.Button("🚀 Solve Problem", variant="primary", size="lg") gr.Markdown(""" ### 💡 Tips: - Start with **1 agent** for testing - Use **5-10 parallel agents** for difficult problems - Increase **max attempts** for complex problems - Press **Ctrl+Enter** in the text box to solve """) with gr.Column(scale=3): # Results section gr.Markdown("## 📊 Results") status_output = gr.Textbox( label="Status", value="Ready to solve", interactive=False ) # Separate box for partial solution streaming (won't be covered by progress) partial_solution_output = gr.Textbox( label="🔄 Partial Solution (Live Streaming - Work in Progress)", lines=25, max_lines=35, interactive=False, value="", placeholder="Partial solution will stream here in real-time as it's being generated...\n\nThe final, complete solution will appear in the 'Final Solution' section below." ) with gr.Accordion("📜 Detailed Logs", open=False): log_output = gr.Textbox( label="Backend Processing Logs", lines=15, max_lines=25, interactive=False, placeholder="Detailed backend logs will appear here..." ) with gr.Accordion("✅ Final Solution (Click to View When Complete)", open=False): solution_output = gr.Textbox( label="Complete Verified Solution", lines=30, max_lines=100, interactive=False, placeholder="The final, verified solution will appear here once processing is complete...", autoscroll=False ) # Set up the solve button action solve_button.click( fn=solve_problem, inputs=[ problem_input, provider_select, model_select, max_runs_input, num_agents_input, other_prompts_input ], outputs=[status_output, partial_solution_output, log_output, solution_output] ) # Also allow Enter key in problem input (with Ctrl/Cmd modifier) problem_input.submit( fn=solve_problem, inputs=[ problem_input, provider_select, model_select, max_runs_input, num_agents_input, other_prompts_input ], outputs=[status_output, partial_solution_output, log_output, solution_output] ) # Solution output is always visible for streaming return demo if __name__ == "__main__": # Create logs directory if it doesn't exist os.makedirs('logs', exist_ok=True) logger.info("Created logs directory") print("=" * 60) print("IMO Math Problem Solver - Gradio Interface") print("=" * 60) logger.info("Starting IMO Math Problem Solver - Gradio Interface") logger.info("Python version: %s", sys.version) logger.info("Gradio version: %s", gr.__version__) logger.info("Working directory: %s", os.getcwd()) # Check for API keys logger.info("Performing initial API key check...") available = check_api_keys() if available: print(f"Available providers: {', '.join(available)}") logger.info("Available providers: %s", ', '.join(available)) else: print("⚠️ WARNING: No API keys found!") print("Please set at least one of these environment variables:") print(" - GOOGLE_API_KEY") print(" - OPENAI_API_KEY") print(" - XAI_API_KEY") logger.warning("No API keys found at startup!") print("=" * 60) # Create and launch the interface logger.info("Creating Gradio interface...") demo = create_interface() logger.info("Gradio interface created successfully") # Launch with public link for sharing # Set share=True to create a public link (72 hours) logger.info("Launching Gradio server...") logger.info("Server config: host=0.0.0.0, port=7860, share=False") demo.launch( server_name="0.0.0.0", server_port=7860, share=False, # Set to True to create a public shareable link show_error=True ) logger.info("Gradio server started successfully")