smolagents-wip/agent_test_client.py

#!/usr/bin/env python3
"""
SmolAgent Test Client

A Gradio-based test client for the BasicSmolAgent that:
1. Fetches random questions from the evaluation API
2. Executes the agent with detailed tracking
3. Displays comprehensive execution information
4. Supports custom question testing
5. Tests against evaluation questions from questions_evaluated.py

Usage: python agent_test_client.py
"""

import gradio as gr
import requests
from agent import BasicSmolAgent
import traceback
import time
from contextlib import redirect_stdout, redirect_stderr
import io
import pandas as pd
from questions_evaluated import questions
import json
import sys
from typing import Optional, Dict, Any, List

# Configuration
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"

class AgentExecutionTracker:
    """Tracks and logs agent execution with detailed information"""
    
    def __init__(self):
        self.reset()
    
    def reset(self):
        """Reset tracking for new execution"""
        self.logs = []
        self.start_time = ""
        self.end_time = ""
        self.question = ""
        self.agent_response = ""
        self.final_answer = ""
        self.captured_stdout = ""
        self.captured_stderr = ""
    
    def log(self, level, message):
        """Add a log entry with timestamp"""
        timestamp = time.strftime("%H:%M:%S")
        self.logs.append(f"[{timestamp}] {level}: {message}")
    
    def get_formatted_log(self):
        """Get comprehensive formatted execution log"""
        lines = [
            "🤖 AGENT EXECUTION LOG",
            "=" * 60,
            f"📝 Question: {self.question}",
            f"⏰ Started: {self.start_time}",
            f"⏱️ Ended: {self.end_time}",
            "",
            "📋 EXECUTION STEPS:",
            "-" * 40
        ]
        
        # Add all log entries
        for log_entry in self.logs:
            lines.append(log_entry)
        
        # Add captured outputs if any
        if self.captured_stdout.strip():
            lines.extend([
                "",
                "📤 CAPTURED STDOUT:",
                "-" * 30,
                self.captured_stdout,
                "-" * 30
            ])
        
        if self.captured_stderr.strip():
            lines.extend([
                "",
                "⚠️ CAPTURED STDERR:",
                "-" * 30,
                self.captured_stderr,
                "-" * 30
            ])
        
        # Add final results
        lines.extend([
            "",
            "🎯 RESULTS:",
            "-" * 20,
            f"Agent Response Length: {len(self.agent_response)} characters",
        ])
        
        if self.final_answer:
            lines.append(f"Final Answer: {self.final_answer}")
        
        return "\n".join(lines)

class SmolAgentTester:
    """Main tester class that handles agent execution and API calls"""
    
    def __init__(self):
        self.agent = None
        self.tracker = AgentExecutionTracker()
        self.api_url = DEFAULT_API_URL
    
    def _initialize_agent(self):
        """Initialize the BasicSmolAgent if not already done"""
        if self.agent is None:
            try:
                self.tracker.log("INIT", "Initializing BasicSmolAgent...")
                self.agent = BasicSmolAgent()
                self.tracker.log("INIT", "✅ BasicSmolAgent initialized successfully")
                return True
            except Exception as e:
                self.tracker.log("ERROR", f"Failed to initialize agent: {str(e)}")
                return False
        return True
    
    def fetch_random_question(self):
        """Fetch a random question from the evaluation API"""
        try:
            self.tracker.log("API", "Fetching random question from evaluation API...")
            
            response = requests.get(f"{self.api_url}/random-question", timeout=15)
            response.raise_for_status()
            
            question_data = response.json()
            task_id = question_data.get("task_id", "Unknown")
            question_text = question_data.get("question", "No question available")
            
            self.tracker.log("API", f"✅ Successfully fetched question (Task ID: {task_id})")
            
            return question_data
            
        except requests.exceptions.Timeout:
            self.tracker.log("ERROR", "Request timeout - API may be slow or unavailable")
            return None
        except requests.exceptions.ConnectionError:
            self.tracker.log("ERROR", "Connection error - Check internet connection")
            return None
        except requests.exceptions.HTTPError as e:
            self.tracker.log("ERROR", f"HTTP error {e.response.status_code} - API may be unavailable")
            return None
        except Exception as e:
            self.tracker.log("ERROR", f"Unexpected error fetching question: {str(e)}")
            return None
    
    def execute_agent(self, question):
        """Execute the agent with comprehensive tracking"""
        # Reset tracker for new execution
        self.tracker.reset()
        self.tracker.question = question
        self.tracker.start_time = time.strftime("%H:%M:%S")
        
        try:
            # Initialize agent if needed
            if not self._initialize_agent():
                self.tracker.end_time = time.strftime("%H:%M:%S")
                return "Failed to initialize agent"
            
            self.tracker.log("EXEC", "Starting agent execution...")
            self.tracker.log("QUESTION", f"Processing: {question[:100]}{'...' if len(question) > 100 else ''}")
            
            # Capture stdout and stderr during execution
            stdout_buffer = io.StringIO()
            stderr_buffer = io.StringIO()
            
            with redirect_stdout(stdout_buffer), redirect_stderr(stderr_buffer):
                result = self.agent(question)
            
            # Store captured outputs
            self.tracker.captured_stdout = stdout_buffer.getvalue()
            self.tracker.captured_stderr = stderr_buffer.getvalue()
            
            self.tracker.log("EXEC", "✅ Agent execution completed successfully")
            
            # FIXED: Handle non-string results from agent
            original_type = type(result).__name__
            if isinstance(result, str):
                result_str = result
                self.tracker.log("RESPONSE", f"Agent returned string ({len(result_str)} characters)")
            else:
                result_str = str(result)
                self.tracker.log("RESPONSE", f"Agent returned {original_type}: {result}")
                self.tracker.log("RESPONSE", f"Converted to string ({len(result_str)} characters)")

            # Extract final answer if present in string version
            if "FINAL ANSWER:" in result_str:
                final_answer = result_str.split("FINAL ANSWER:")[-1].strip()
                self.tracker.final_answer = final_answer
                self.tracker.log("ANSWER", f"Extracted final answer: {final_answer[:50]}{'...' if len(final_answer) > 50 else ''}")
            else:
                # If no "FINAL ANSWER:" format and original was not a string, use the converted string
                if not isinstance(result, str):
                    self.tracker.final_answer = result_str
                    self.tracker.log("ANSWER", f"No FINAL ANSWER format, using converted {original_type}: {result_str}")
                else:
                    self.tracker.final_answer = "No explicit final answer found"
                    self.tracker.log("ANSWER", "No explicit FINAL ANSWER format detected")

            self.tracker.agent_response = result_str
            self.tracker.end_time = time.strftime("%H:%M:%S")
            
            return result_str
            
        except Exception as e:
            error_msg = f"Agent execution failed: {str(e)}"
            self.tracker.log("ERROR", error_msg)
            self.tracker.log("ERROR", f"Traceback: {traceback.format_exc()}")
            self.tracker.end_time = time.strftime("%H:%M:%S")
            return f"ERROR: {error_msg}"

# Global tester instance
tester = SmolAgentTester()

def test_random_question():
    """Handle random question testing"""
    try:
        # Fetch random question
        question_data = tester.fetch_random_question()
        
        if not question_data:
            return (
                "❌ Failed to fetch random question from API", 
                "Please check your internet connection and try again.\nThe evaluation API might be temporarily unavailable.",
                "No response available"
            )
        
        question_text = question_data.get("question", "No question text available")
        task_id = question_data.get("task_id", "Unknown")
        
        # Execute agent
        agent_response = tester.execute_agent(question_text)
        
        # Format outputs
        question_info = f"📋 Task ID: {task_id}\n\n📝 Question:\n{question_text}"
        execution_log = tester.tracker.get_formatted_log()
        
        result_summary = f"🤖 Agent Response:\n{agent_response}\n\n"
        if tester.tracker.final_answer:
            result_summary += f"🎯 Final Answer: {tester.tracker.final_answer}"
        
        return question_info, execution_log, result_summary
        
    except Exception as e:
        error_msg = f"Unexpected error in random question test: {str(e)}\n{traceback.format_exc()}"
        return f"❌ Error: {error_msg}", "", ""

def test_custom_question(custom_question):
    """Handle custom question testing"""
    if not custom_question.strip():
        return "❌ Please enter a question to test", "", ""
    
    try:
        # Execute agent with custom question
        agent_response = tester.execute_agent(custom_question.strip())
        
        # Format outputs
        question_info = f"📝 Custom Question:\n{custom_question.strip()}"
        execution_log = tester.tracker.get_formatted_log()
        
        result_summary = f"🤖 Agent Response:\n{agent_response}\n\n"
        if tester.tracker.final_answer:
            result_summary += f"🎯 Final Answer: {tester.tracker.final_answer}"
        
        return question_info, execution_log, result_summary
        
    except Exception as e:
        error_msg = f"Unexpected error in custom question test: {str(e)}\n{traceback.format_exc()}"
        return f"❌ Error: {error_msg}", "", ""

def get_evaluation_questions():
    """Get list of evaluation questions for dropdown"""
    question_choices = []
    for i, q in enumerate(questions):
        task_id = q.get("task_id", "Unknown")
        question_text = q.get("question", "No question")
        level = q.get("Level", "Unknown")
        
        # Truncate long questions for display
        display_text = question_text[:100] + "..." if len(question_text) > 100 else question_text
        label = f"[Level {level}] {task_id[:8]}... - {display_text}"
        
        question_choices.append((label, i))
    
    return question_choices

def test_evaluation_question(question_index):
    """Handle evaluation question testing"""
    if question_index is None:
        return "❌ Please select a question to test", "", "", ""
    
    try:
        selected_question = questions[question_index]
        question_text = selected_question.get("question", "No question text")
        task_id = selected_question.get("task_id", "Unknown")
        level = selected_question.get("Level", "Unknown")
        file_name = selected_question.get("file_name", "")
        
        # Execute agent
        agent_response = tester.execute_agent(question_text)
        
        # Format outputs
        question_info = f"📋 Task ID: {task_id}\n📊 Level: {level}\n📎 File: {file_name if file_name else 'None'}\n\n📝 Question:\n{question_text}"
        execution_log = tester.tracker.get_formatted_log()
        
        result_summary = f"🤖 Agent Response:\n{agent_response}\n\n"
        if tester.tracker.final_answer:
            result_summary += f"🎯 Final Answer: {tester.tracker.final_answer}"
        
        # Get the correct answer
        correct_answer = get_correct_answer(task_id)
        if correct_answer:
            correct_answer_display = f"✅ **Correct Answer:**\n{correct_answer}\n\n📋 **Task ID:** {task_id}\n📊 **Level:** {level}"
        else:
            correct_answer_display = f"❓ **Correct Answer:**\nNot found in metadata\n\n📋 **Task ID:** {task_id}\n📊 **Level:** {level}"
        
        return question_info, execution_log, result_summary, correct_answer_display
        
    except Exception as e:
        error_msg = f"Unexpected error in evaluation question test: {str(e)}\n{traceback.format_exc()}"
        return f"❌ Error: {error_msg}", "", "", ""

def test_all_evaluation_questions():
    """Run all evaluation questions and return results"""
    try:
        results = []
        total_questions = len(questions)
        
        progress_info = f"🔄 Running {total_questions} evaluation questions...\n\n"
        
        for i, question_data in enumerate(questions):
            question_text = question_data.get("question", "No question text")
            task_id = question_data.get("task_id", "Unknown")
            level = question_data.get("Level", "Unknown")
            
            progress_info += f"Processing question {i+1}/{total_questions}: {task_id[:8]}...\n"
            
            try:
                # Execute agent
                agent_response = tester.execute_agent(question_text)
                
                # Extract final answer
                final_answer = tester.tracker.final_answer if tester.tracker.final_answer else "No answer extracted"
                
                # Get correct answer
                correct_answer = get_correct_answer(task_id)
                correct_answer_display = correct_answer if correct_answer else "Not found"
                
                results.append({
                    "Task ID": task_id,
                    "Level": level,
                    "Question": question_text[:100] + "..." if len(question_text) > 100 else question_text,
                    "Agent Answer": final_answer,
                    "Correct Answer": correct_answer_display,
                    "Response Length": len(agent_response),
                    "Status": "Success"
                })
                
            except Exception as e:
                # Get correct answer even if agent failed
                correct_answer = get_correct_answer(task_id)
                correct_answer_display = correct_answer if correct_answer else "Not found"
                
                results.append({
                    "Task ID": task_id,
                    "Level": level,
                    "Question": question_text[:100] + "..." if len(question_text) > 100 else question_text,
                    "Agent Answer": f"ERROR: {str(e)}",
                    "Correct Answer": correct_answer_display,
                    "Response Length": 0,
                    "Status": "Failed"
                })
        
        # Create DataFrame for results
        results_df = pd.DataFrame(results)
        
        # Summary statistics
        success_count = len([r for r in results if r["Status"] == "Success"])
        failure_count = total_questions - success_count
        
        summary = f"""
        ✅ EVALUATION COMPLETE
        
        📊 Summary:
        - Total Questions: {total_questions}
        - Successful: {success_count}
        - Failed: {failure_count}
        - Success Rate: {(success_count/total_questions)*100:.1f}%
        """
        
        return summary, results_df, "All evaluation questions processed!"
        
    except Exception as e:
        error_msg = f"Error running all evaluation questions: {str(e)}\n{traceback.format_exc()}"
        return f"❌ Error: {error_msg}", pd.DataFrame(), ""

def create_interface():
    """Create the main Gradio interface"""
    
    # Custom CSS for better styling
    css = """
    .gradio-container {
        max-width: 1200px !important;
    }
    .tab-nav {
        font-size: 16px !important;
    }
    """
    
    with gr.Blocks(
        title="SmolAgent Test Client", 
        css=css,
        theme=gr.themes.Base(
            primary_hue="blue",
            secondary_hue="gray"
        )
    ) as interface:
        
        # Header
        gr.Markdown("""
        # 🧪 SmolAgent Test Client
        
        **Interactive testing environment for the BasicSmolAgent**
        
        This tool allows you to thoroughly test the agent's capabilities with detailed execution tracking.
        You can fetch random questions from the evaluation API, test with custom questions, or run specific evaluation questions.
        """)
        
        # Main tabs
        with gr.Tabs():
            # Random Question Tab
            with gr.TabItem("🎲 Random Question Test", elem_id="random-tab"):
                gr.Markdown("### Fetch and test a random question from the evaluation API")
                gr.Markdown("Click the button below to fetch a random question and run the agent on it.")
                
                random_btn = gr.Button(
                    "🎲 Fetch Random Question & Execute Agent", 
                    variant="primary", 
                    size="lg",
                    scale=1
                )
                
                # Output sections
                with gr.Row():
                    with gr.Column(scale=1):
                        question_display = gr.Textbox(
                            label="📋 Question Information",
                            lines=6,
                            max_lines=10,
                            interactive=False,
                            show_copy_button=True
                        )
                    
                    with gr.Column(scale=1):
                        result_display = gr.Textbox(
                            label="🎯 Agent Response & Final Answer",
                            lines=6,
                            max_lines=10,
                            interactive=False,
                            show_copy_button=True
                        )
                
                execution_log_display = gr.Textbox(
                    label="🔍 Detailed Execution Log",
                    lines=20,
                    max_lines=30,
                    interactive=False,
                    show_copy_button=True,
                    placeholder="Execution log will appear here after running the agent..."
                )
                
                # Wire up the random question functionality
                random_btn.click(
                    fn=test_random_question,
                    inputs=[],
                    outputs=[question_display, execution_log_display, result_display]
                )
            
            # Custom Question Tab
            with gr.TabItem("✏️ Custom Question Test", elem_id="custom-tab"):
                gr.Markdown("### Test the agent with your own custom question")
                gr.Markdown("Enter any question you'd like to test the agent with.")
                
                custom_input = gr.Textbox(
                    label="📝 Your Question",
                    lines=3,
                    max_lines=5,
                    placeholder="Enter your question here...\n\nExample: What is the square root of 144?",
                    show_copy_button=True
                )
                
                custom_btn = gr.Button(
                    "🚀 Execute Agent on Custom Question", 
                    variant="secondary", 
                    size="lg"
                )
                
                # Output sections for custom questions
                with gr.Row():
                    with gr.Column(scale=1):
                        custom_question_display = gr.Textbox(
                            label="📋 Question Information",
                            lines=4,
                            max_lines=8,
                            interactive=False,
                            show_copy_button=True
                        )
                    
                    with gr.Column(scale=1):
                        custom_result_display = gr.Textbox(
                            label="🎯 Agent Response & Final Answer",
                            lines=4,
                            max_lines=8,
                            interactive=False,
                            show_copy_button=True
                        )
                
                custom_execution_log_display = gr.Textbox(
                    label="🔍 Detailed Execution Log",
                    lines=20,
                    max_lines=30,
                    interactive=False,
                    show_copy_button=True,
                    placeholder="Execution log will appear here after running the agent..."
                )
                
                # Wire up the custom question functionality
                custom_btn.click(
                    fn=test_custom_question,
                    inputs=[custom_input],
                    outputs=[custom_question_display, custom_execution_log_display, custom_result_display]
                )
            
            # Evaluation Questions Tab
            with gr.TabItem("📊 Evaluation Questions", elem_id="eval-tab"):
                gr.Markdown("### Test with specific evaluation questions")
                gr.Markdown(f"Select from {len(questions)} evaluation questions or run all of them.")
                
                with gr.Row():
                    with gr.Column(scale=2):
                        question_dropdown = gr.Dropdown(
                            choices=get_evaluation_questions(),
                            label="📝 Select Evaluation Question",
                            value=None
                        )
                    
                    with gr.Column(scale=1):
                        eval_single_btn = gr.Button(
                            "🚀 Run Selected Question", 
                            variant="secondary", 
                            size="lg"
                        )
                
                eval_all_btn = gr.Button(
                    "🔄 Run ALL Evaluation Questions", 
                    variant="primary", 
                    size="lg"
                )
                
                gr.Markdown("⚠️ **Warning**: Running all questions may take a long time!")
                
                # Single question results
                with gr.Row():
                    with gr.Column(scale=1):
                        eval_question_display = gr.Textbox(
                            label="📋 Question Information",
                            lines=6,
                            max_lines=10,
                            interactive=False,
                            show_copy_button=True
                        )
                    
                    with gr.Column(scale=1):
                        eval_result_display = gr.Textbox(
                            label="🎯 Agent Response & Final Answer",
                            lines=6,
                            max_lines=10,
                            interactive=False,
                            show_copy_button=True
                        )
                    
                    with gr.Column(scale=1):
                        eval_correct_answer_display = gr.Textbox(
                            label="✅ Correct Answer",
                            lines=6,
                            max_lines=10,
                            interactive=False,
                            show_copy_button=True,
                            placeholder="Correct answer will appear here..."
                        )
                
                eval_execution_log_display = gr.Textbox(
                    label="🔍 Detailed Execution Log",
                    lines=15,
                    max_lines=25,
                    interactive=False,
                    show_copy_button=True,
                    placeholder="Execution log will appear here after running a question..."
                )
                
                # All questions results
                gr.Markdown("### 📊 Batch Results")
                
                batch_summary_display = gr.Textbox(
                    label="📈 Batch Summary",
                    lines=8,
                    interactive=False,
                    show_copy_button=True,
                    placeholder="Summary will appear here after running all questions..."
                )
                
                batch_results_display = gr.DataFrame(
                    label="📋 Detailed Results Table",
                    headers=["Task ID", "Level", "Question", "Agent Answer", "Correct Answer", "Response Length", "Status"],
                    datatype=["str", "str", "str", "str", "str", "number", "str"],
                    interactive=False,
                    wrap=True
                )
                
                batch_status_display = gr.Textbox(
                    label="🔄 Status",
                    lines=2,
                    interactive=False,
                    placeholder="Status updates will appear here..."
                )
                
                # Wire up evaluation question functionality
                eval_single_btn.click(
                    fn=test_evaluation_question,
                    inputs=[question_dropdown],
                    outputs=[eval_question_display, eval_execution_log_display, eval_result_display, eval_correct_answer_display]
                )
                
                eval_all_btn.click(
                    fn=test_all_evaluation_questions,
                    inputs=[],
                    outputs=[batch_summary_display, batch_results_display, batch_status_display]
                )
        
        # Footer information
        gr.Markdown("---")
        gr.Markdown("""
        ### 📚 Features & Information
        
        **🔍 Execution Tracking:**
        - Comprehensive step-by-step logging with timestamps
        - Capture of stdout/stderr during agent execution
        - Detailed error reporting and stack traces
        - Performance timing information
        
        **🎯 Response Analysis:**
        - Full agent response display
        - Automatic final answer extraction
        - Response length and format analysis
        
        **⚡ Testing Capabilities:**
        - Random questions from the evaluation API endpoint
        - Custom question testing with any input
        - Individual evaluation question testing
        - Batch processing of all evaluation questions
        - Copy-friendly logs for external analysis
        - Real-time execution monitoring
        
        **🔧 Technical Details:**
        - Uses the existing BasicSmolAgent from agent.py
        - Connects to: `https://agents-course-unit4-scoring.hf.space/random-question`
        - Processes questions from questions_evaluated.py
        - Captures all agent tool usage and reasoning steps
        - Provides detailed execution diagnostics
        """)
        
        gr.Markdown("""
        ### 🚀 Quick Start Guide
        
        1. **Random Questions**: Click "Fetch Random Question & Execute Agent" to test with API questions
        2. **Custom Questions**: Enter your own question and click "Execute Agent on Custom Question"
        3. **Evaluation Questions**: Select a specific evaluation question or run all of them
        4. **Review Results**: Check execution logs for detailed insights into agent processing
        5. **Batch Analysis**: Use the "Run ALL" feature to get comprehensive performance metrics
        """)
    
    return interface

def main():
    """Main function to launch the test client"""
    print("🚀 Starting SmolAgent Test Client...")
    print("📡 API Endpoint:", DEFAULT_API_URL)
    print("🤖 Agent Type: BasicSmolAgent")
    print(f"📊 Evaluation Questions: {len(questions)} loaded")
    print("-" * 50)
    
    # Create and launch interface
    interface = create_interface()
    interface.launch(
        debug=True,
        share=False,
        show_error=True,
        server_name="0.0.0.0",  # Allow external connections
        server_port=7860
    )

def load_metadata() -> Dict[str, str]:
    """Load metadata from metadata.jsonl and return a mapping of task_id to final answer"""
    metadata = {}
    try:
        with open('metadata.jsonl', 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        data = json.loads(line)
                        task_id = data.get('task_id')
                        final_answer = data.get('Final answer')
                        if task_id and final_answer is not None:
                            metadata[task_id] = str(final_answer)
                    except json.JSONDecodeError as e:
                        print(f"Warning: Could not parse JSON line: {line[:100]}...")
                        continue
    except FileNotFoundError:
        print("Warning: metadata.jsonl file not found")
    except Exception as e:
        print(f"Warning: Error loading metadata: {e}")
    
    return metadata

def get_correct_answer(task_id: str) -> Optional[str]:
    """Get the correct answer for a given task_id"""
    metadata = load_metadata()
    return metadata.get(task_id)

if __name__ == "__main__":
    main()