Update app.py

app.py

@@ -37,6 +37,526 @@ from langchain_community.embeddings import HuggingFaceEmbeddings
 from langchain_community.tools import DuckDuckGoSearchRun
 from langchain_core.documents import Document
 
+
+# =============================================================================
+# CONFIGURATION
+# =============================================================================
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+MAX_TURNS = 25  # Increased for planning/reflection
+MAX_MESSAGE_LENGTH = 8000
+REFLECT_EVERY_N_TURNS = 5
+
+# =============================================================================
+# GLOBAL RAG COMPONENTS
+# =============================================================================
+global_embeddings = None
+global_text_splitter = None
+
+def initialize_rag_components():
+    """Initialize RAG components globally."""
+    global global_embeddings, global_text_splitter
+    
+    if global_embeddings is None:
+        print("Initializing RAG embeddings...")
+        try:
+            global_embeddings = HuggingFaceEmbeddings(
+                model_name="sentence-transformers/all-MiniLM-L6-v2",
+                model_kwargs={'device': 'cpu'}
+            )
+            print("✅ Embeddings initialized.")
+        except Exception as e:
+            print(f"⚠️ Failed to initialize embeddings: {e}")
+            return False
+    
+    if global_text_splitter is None:
+        print("Initializing text splitter...")
+        global_text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200,
+            length_function=len,
+            separators=["\n\n", "\n", ". ", " ", ""]
+        )
+        print("✅ Text splitter initialized.")
+    
+    return True
+
+# =============================================================================
+# ASR INITIALIZATION
+# =============================================================================
+asr_pipeline = None
+try:
+    print("Loading ASR (Whisper) pipeline globally...")
+    device = 0 if torch.cuda.is_available() else -1
+    device_name = "cuda:0" if device == 0 else "cpu"
+    print(f"Attempting to use device: {device_name} for ASR.")
+    asr_pipeline = pipeline(
+        "automatic-speech-recognition",
+        model="openai/whisper-base",
+        torch_dtype=torch.float16 if device == 0 else torch.float32,
+        device=device
+    )
+    print("✅ ASR (Whisper) pipeline loaded successfully.")
+except Exception as e:
+    print(f"⚠️ Warning: Could not load ASR pipeline globally. Error: {e}")
+    asr_pipeline = None
+
+# =============================================================================
+# UTILITY FUNCTIONS
+# =============================================================================
+def remove_fences_simple(text):
+    """Remove code fences from text."""
+    original_text = text
+    text = text.strip()
+    if text.startswith("```") and text.endswith("```"):
+        text = text[3:-3].strip()
+        if '\n' in text:
+            first_line, rest = text.split('\n', 1)
+            if first_line.strip().replace('_','').isalnum() and len(first_line.strip()) < 15:
+                text = rest.strip()
+        return text
+    return original_text
+
+def truncate_if_needed(content: str, max_length: int = MAX_MESSAGE_LENGTH) -> str:
+    """Truncate content if it exceeds max length."""
+    if len(content) > max_length:
+        return content[:max_length] + f"\n...[truncated, {len(content)} total chars]"
+    return content
+
+def find_file(path: str) -> Optional[Path]:
+    """Find a file by trying multiple path variations."""
+    script_dir = Path.cwd()
+    safe_path = Path(path).as_posix()
+    
+    paths_to_try = [
+        script_dir / safe_path,
+        Path(safe_path),
+        script_dir / Path(path).name
+    ]
+    
+    for attempt_path in paths_to_try:
+        if attempt_path.exists():
+            return attempt_path
+    
+    return None
+
+# =============================================================================
+# PLANNING & REFLECTION TOOLS
+# =============================================================================
+
+class ThinkInput(BaseModel):
+    reasoning: str = Field(description="Your step-by-step reasoning for a logic puzzle (keep under 200 chars)")
+
+@tool(args_schema=ThinkInput)
+def think_through_logic(reasoning: str) -> str:
+    """
+    Use this to work through logic puzzles, riddles, or reasoning problems.
+    
+    Call this when:
+    - The question is a riddle or brain teaser
+    - You need to reason through a logical problem
+    - No external information is needed, just thinking
+    
+    After thinking through the logic, use calculator if math is involved,
+    then validate_answer and final_answer_tool.
+    
+    NOTE: Keep reasoning summary brief (under 200 chars).
+    """
+    print(f"🧠 Thinking through logic: {reasoning[:100]}...")
+    
+    return f"""✅ Logic reasoning recorded: {reasoning}
+
+Now:
+1. If there's any math to calculate, use calculator()
+2. Once you have the answer, call validate_answer()
+3. Then call final_answer_tool() with just the answer"""
+
+
+class PlanInput(BaseModel):
+    question: str = Field(description="Brief summary of the task (keep under 100 chars)")
+
+@tool(args_schema=PlanInput)
+def create_plan(question: str) -> str:
+    """
+    Creates a step-by-step plan for answering a question.
+    CRITICAL: Call this FIRST for any multi-step or complex question.
+    
+    This helps you think through:
+    1. What information do you need?
+    2. In what order should you gather it?
+    3. What tools will you use?
+    
+    After calling this, execute the plan step-by-step.
+    
+    NOTE: Keep the question summary brief (under 100 chars) to avoid errors.
+    """
+    print(f"📋 Planning phase initiated for: {question[:100]}...")
+    
+    return f"""✅ Plan Created. Now execute these steps methodically:
+
+PLANNING FRAMEWORK:
+1. GOAL: What exact answer format is needed?
+2. REQUIREMENTS: What data/information is required?
+3. STRATEGY: What's the most efficient path?
+4. EXECUTION: List concrete actions in order
+
+Now proceed with Step 1 of your plan."""
+
+
+class ReflectInput(BaseModel):
+    current_situation: str = Field(description="What you've tried so far (keep brief, under 100 chars)")
+
+@tool(args_schema=ReflectInput)
+def reflect_on_progress(current_situation: str) -> str:
+    """
+    Reflects on your progress and suggests what to do next.
+    
+    Call this when:
+    - You feel stuck or uncertain
+    - Tools keep failing
+    - You're not making progress
+    - You've taken 5+ steps without getting closer to the answer
+    
+    This helps you step back and reconsider your approach.
+    
+    NOTE: Keep the situation summary brief (under 100 chars).
+    """
+    print(f"🤔 Reflection initiated: {current_situation[:100]}...")
+    
+    return f"""🔍 REFLECTION ANALYSIS:
+
+Current situation: {current_situation}
+
+CRITICAL QUESTIONS TO ASK YOURSELF:
+1. Have I gathered the information I actually need?
+2. Am I using the right tools for this task?
+3. Am I going in circles (repeating similar actions)?
+4. Should I try a completely different approach?
+5. Do I have enough information to answer now?
+
+NEXT STEPS:
+- If stuck: Try a different tool or search query
+- If missing info: Identify exactly what's missing
+- If have info: Proceed to final_answer_tool
+- If uncertain: Break problem into smaller pieces
+
+Take a different approach now."""
+
+
+class ValidateInput(BaseModel):
+    proposed_answer: str = Field(description="The answer you plan to submit")
+    original_question: str = Field(description="The original question")
+
+@tool(args_schema=ValidateInput)
+def validate_answer(proposed_answer: str, original_question: str) -> str:
+    """
+    Validates your proposed answer before submission.
+    CRITICAL: ALWAYS call this before final_answer_tool.
+    
+    Checks:
+    - Does the answer match what was asked?
+    - Is it in the correct format?
+    - Are there any obvious issues?
+    
+    If validation passes, then call final_answer_tool.
+    If validation fails, gather more information or correct the format.
+    """
+    print(f"✓ Validating answer: '{proposed_answer[:50]}...'")
+    
+    issues = []
+    warnings = []
+    
+    # Check for conversational fluff
+    fluff_phrases = ["the answer is", "based on", "according to", "i found that", "here is", "final answer"]
+    if any(phrase in proposed_answer.lower() for phrase in fluff_phrases):
+        issues.append("❌ Remove conversational text. Provide ONLY the answer.")
+    
+    # Check for number format if question asks for numbers
+    number_keywords = ["how many", "what number", "count", "total", "sum"]
+    if any(kw in original_question.lower() for kw in number_keywords):
+        if not any(char.isdigit() for char in proposed_answer):
+            warnings.append("⚠️ Question seems to ask for a number, but answer contains no digits.")
+    
+    # Check for list format
+    if "list" in original_question.lower() and "," not in proposed_answer:
+        warnings.append("⚠️ Question asks for a list, consider comma-separated format.")
+    
+    # Check for yes/no questions
+    if original_question.lower().strip().startswith(("is ", "are ", "was ", "were ", "do ", "does ", "did ", "can ", "will ")):
+        if proposed_answer.lower() not in ["yes", "no", "true", "false"]:
+            warnings.append("⚠️ This looks like a yes/no question. Consider simple yes/no answer.")
+    
+    # Check for code fences or markdown
+    if "```" in proposed_answer:
+        issues.append("❌ Remove code fences (```) from the answer.")
+    
+    # Check length
+    if len(proposed_answer) > 500:
+        warnings.append("⚠️ Answer is quite long. Are you sure this is just the answer and not an explanation?")
+    
+    if issues:
+        return "🚫 VALIDATION FAILED:\n" + "\n".join(issues) + "\n\nFix these issues before calling final_answer_tool."
+    
+    if warnings:
+        return "⚠️ VALIDATION WARNINGS:\n" + "\n".join(warnings) + "\n\nConsider these points, but you may proceed if confident."
+    
+    return "✅ VALIDATION PASSED: Answer looks good! Proceed with final_answer_tool now."
+
+
+# =============================================================================
+# CORE TOOLS
+# =============================================================================
+
+class SearchInput(BaseModel):
+    query: str = Field(description="The search query.")
+
+@tool(args_schema=SearchInput)
+def search_tool(query: str) -> str:
+    """
+    Searches the web using DuckDuckGo.
+    Use for: recent information, facts, general web searches.
+    
+    Tips:
+    - Keep queries concise and specific
+    - Include year for time-sensitive queries (e.g., "GDP Brazil 2016")
+    - Try different phrasings if first search doesn't help
+    """
+    if not isinstance(query, str) or not query.strip():
+        return "Error: Invalid input. 'query' must be a non-empty string."
+    
+    print(f"🔍 Searching: {query}")
+    try:
+        search = DuckDuckGoSearchRun()
+        result = search.run(query)
+        if len(result) > MAX_MESSAGE_LENGTH:
+            result = result[:MAX_MESSAGE_LENGTH] + f"\n...[truncated, {len(result)} total chars]"
+        return result
+    except Exception as e:
+        return f"Error running search for '{query}': {str(e)}"
+
+
+class CalcInput(BaseModel):
+    expression: str = Field(description="Mathematical expression to evaluate (e.g., '2 + 2', 'sqrt(16)', '45 * 1.2')")
+
+@tool(args_schema=CalcInput)
+def calculator(expression: str) -> str:
+    """
+    Evaluates mathematical expressions.
+    Use this for ANY calculations instead of code_interpreter.
+    
+    Supports: +, -, *, /, **, sqrt, sin, cos, tan, log, exp, pi, e, abs, round
+    
+    Examples:
+    - calculator("127 * 83")
+    - calculator("sqrt(144)")
+    - calculator("(45 + 23) / 2")
+    """
+    if not isinstance(expression, str) or not expression.strip():
+        return "Error: Invalid expression."
+    
+    print(f"🧮 Calculating: {expression}")
+    
+    try:
+        # Create safe namespace with math functions
+        import math
+        safe_dict = {
+            'sqrt': math.sqrt, 'sin': math.sin, 'cos': math.cos, 'tan': math.tan,
+            'log': math.log, 'log10': math.log10, 'exp': math.exp,
+            'pi': math.pi, 'e': math.e, 'abs': abs, 'round': round,
+            'pow': pow, 'sum': sum, 'min': min, 'max': max
+        }
+        
+        result = eval(expression, {"__builtins__": {}}, safe_dict)
+        return f"{result}"
+    except Exception as e:
+        return f"Error evaluating '{expression}': {str(e)}\nMake sure to use proper syntax (e.g., sqrt(16), not sqrt 16)"
+
+
+class CodeInput(BaseModel):
+    code: str = Field(description="Python code to execute. MUST include print() for output.")
+
+@tool(args_schema=CodeInput)
+def code_interpreter(code: str) -> str:
+    """
+    Executes Python code for complex data processing.
+    
+    WHEN TO USE:
+    - Data analysis (CSV, Excel files)
+    - Complex calculations with loops/conditionals
+    - String manipulation
+    - Date/time calculations
+    
+    WHEN NOT TO USE:
+    - Simple math (use calculator instead)
+    - Web searches (use search_tool)
+    
+    Available libraries: pandas as pd, numpy as np, json, re, datetime
+    
+    CRITICAL: Always use print() to output results!
+    """
+    if not isinstance(code, str):
+        return "Error: Invalid input. 'code' must be a string."
+    
+    # Safety checks
+    dangerous_patterns = ['__import__', 'eval(', 'compile(', 'subprocess', 'os.system', 'exec(']
+    code_lower = code.lower()
+    for pattern in dangerous_patterns:
+        if pattern in code_lower:
+            return f"Error: Potentially dangerous operation '{pattern}' is not allowed."
+    
+    if 'open(' in code_lower and any(mode in code for mode in ["'w'", '"w"', "'a'", '"a"', "'wb'", '"wb"']):
+        return "Error: Writing files is not allowed in code_interpreter. Use write_file tool instead."
+    
+    print(f"💻 Executing code...")
+    output_stream = io.StringIO()
+    error_stream = io.StringIO()
+    
+    try:
+        with contextlib.redirect_stdout(output_stream), contextlib.redirect_stderr(error_stream):
+            safe_globals = {
+                "pd": pd,
+                "np": np,
+                "json": json,
+                "re": re,
+                "__builtins__": __builtins__
+            }
+            exec(code, safe_globals, {})
+            
+        stdout = output_stream.getvalue()
+        stderr = error_stream.getvalue()
+        
+        if stderr:
+            return f"Error in execution:\n{stderr}\n\nStdout (if any):\n{stdout}"
+        
+        if stdout:
+            if len(stdout) > MAX_MESSAGE_LENGTH:
+                stdout = stdout[:MAX_MESSAGE_LENGTH] + f"\n...[truncated, {len(stdout)} total chars]"
+            return f"{stdout}"
+        
+        return "Code executed but produced no output. Remember to use print() to display results!"
+        
+    except Exception as e:
+        tb_str = traceback.format_exc()
+        return f"Execution failed:\n{tb_str}"
+
+
+class ReadFileInput(BaseModel):
+    path: str = Field(description="Path to the file to read")
+
+@tool(args_schema=ReadFileInput)
+def read_file(path: str) -> str:
+    """Reads a file from the filesystem."""
+    if not isinstance(path, str) or not path.strip():
+        return "Error: Invalid input. 'path' must be a non-empty string."
+    
+    print(f"📄 Reading file: {path}")
+    
+    file_path = find_file(path)
+    if not file_path:
+        cwd_files = os.listdir(".")
+        return (f"Error: File not found: '{path}'\n"
+                f"Files in current directory: {cwd_files}")
+    
+    try:
+        content = file_path.read_text(encoding='utf-8')
+        return truncate_if_needed(content)
+    except UnicodeDecodeError:
+        size = file_path.stat().st_size
+        ext = file_path.suffix
+        return (f"File appears to be binary ({size} bytes). Cannot display as text.\n"
+                f"File type: {ext}\n"
+                f"Consider using audio_transcription_tool for audio files.")
+    except Exception as e:
+        return f"Error reading file: {str(e)}"
+
+
+class WriteFileInput(BaseModel):
+    path: str = Field(description="Path where file should be written")
+    content: str = Field(description="Content to write to the file")
+
+@tool(args_schema=WriteFileInput)
+def write_file(path: str, content: str) -> str:
+    """Writes content to a file."""
+    if not isinstance(path, str) or not path.strip():
+        return "Error: Invalid input. 'path' must be a non-empty string."
+    if not isinstance(content, str):
+        return "Error: Invalid input. 'content' must be a string."
+    
+    print(f"✍️ Writing file: {path}")
+    
+    try:
+        file_path = Path.cwd() / path
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        file_path.write_text(content, encoding='utf-8')
+        return f"Successfully wrote {len(content)} characters to '{path}'."
+    except Exception as e:
+        return f"Error writing file '{path}': {str(e)}"
+
+
+class ListDirInput(BaseModel):
+    path: str = Field(description="Directory path to list", default=".")
+
+@tool(args_schema=ListDirInput)
+def list_directory(path: str = ".") -> str:
+    """Lists files and directories in a path."""
+    print(f"📁 Listing directory: {path}")
+    
+    try:
+        dir_path = Path.cwd() / path if path != "." else Path.cwd()
+        
+        if not dir_path.is_dir():
+            return f"Error: '{path}' is not a valid directory."
+        
+        items = sorted(dir_path.iterdir())
+        
+        if not items:
+            return f"Directory '{path}' is empty."
+        
+        files, directories = [], []
+        
+        for item in items:
+            if item.is_dir():
+                directories.append(f"📁 {item.name}/")
+            else:
+                size = item.stat().st_size
+                files.append(f"📄 {item.name} ({size} bytes)")
+        
+        result = f"Contents of '{path}':\n\n"
+        if directories:
+            result += "Directories:\n" + "\n".join(directories) + "\n\n"
+        if files:
+            result += "Files:\n" + "\n".join(files)
+        
+        return result
+    except Exception as e:
+        return f"Error listing directory '{path}': {str(e)}"
+
+
+class AudioInput(BaseModel):
+    file_path: str = Field(description="Path to audio file to transcribe")
+
+@tool(args_schema=AudioInput)
+def audio_transcription_tool(file_path: str) -> str:
+    """Transcribes audio files to text using Whisper."""
+    if not isinstance(file_path, str) or not file_path.strip():
+        return "Error: Invalid input. 'file_path' must be a non-empty string."
+    
+    print(f"🎤 Transcribing audio: {file_path}")
+    
+    if asr_pipeline is None:
+        return "Error: ASR pipeline is not available."
+    
+    audio_path = find_file(file_path)
+    if not audio_path:
+        return f"Error: Audio file not found: '{file_path}'"
+    
+    try:
+        transcription = asr_pipeline(str(audio_path))
+        result_text = transcription.get("text", "")
+        
+        if not result_text:
+            return "Error: Transcription produced no text."
+        
+        return f"Tra
 # =============================================================================
 # CONFIGURATION
 # =============================================================================
@@ -1179,72 +1699,6 @@ Turn 7: final_answer_tool("1.796 trillion")
         print("✅ Planning & Reflection Agent graph compiled successfully.")
 
 
-    def __call__(self, question: str) -> str:
-        print(f"\n--- Starting Agent Run for Question ---")
-        print(f"Agent received question (first 100 chars): {question[:100]}...")
-        
-        graph_input = {
-            "messages": [
-                SystemMessage(content=self.system_prompt),
-                HumanMessage(content=question)
-            ],
-            "turn": 0
-        }
-        
-        final_answer = "AGENT FAILED TO PRODUCE ANSWER"
-        try:
-            config = {"recursion_limit": MAX_TURNS + 5}
-            for event in self.graph.stream(graph_input, stream_mode="values", config=config):
-
-                if event.get('messages'): # Ensure messages exist
-                    last_message = event["messages"][-1]
-                else:
-                    continue # Skip if no messages yet
-                
-                # Check for final answer extraction
-                if isinstance(last_message, AIMessage) and last_message.tool_calls:
-                    if last_message.tool_calls[0].get("name") == "final_answer_tool":
-                        final_answer_args = last_message.tool_calls[0].get('args', {})
-                        if 'answer' in final_answer_args:
-                             final_answer = final_answer_args['answer']
-                             print(f"--- Final Answer Captured from tool call: '{final_answer}' ---")
-                             break 
-                        else:
-                             print(f"⚠️ Final Answer tool called without 'answer' argument: {final_answer_args}")
-                             final_answer = "ERROR: FINAL_ANSWER_TOOL CALLED WITHOUT ANSWER"
-                             break 
-
-                elif isinstance(last_message, ToolMessage):
-                    print(f"Tool Result ({last_message.tool_call_id}): {last_message.content[:500]}...")
-                elif isinstance(last_message, AIMessage) and not last_message.tool_calls:
-                    print(f"AI Message (Reasoning): {last_message.content[:500]}...")
-                elif isinstance(last_message, SystemMessage):
-                     print(f"System Message: {last_message.content[:500]}...")
-
-
-            # --- Final Answer Cleaning ---
-            cleaned_answer = str(final_answer).strip()
-            prefixes_to_remove = ["The answer is:", "Here is the answer:", "Based on the information:", "Final Answer:", "Answer:"]
-            original_cleaned = cleaned_answer
-            for prefix in prefixes_to_remove:
-                if cleaned_answer.lower().startswith(prefix.lower()):
-                    potential_answer = cleaned_answer[len(prefix):].strip()
-                    if potential_answer: 
-                        cleaned_answer = potential_answer
-                        break 
-            
-            cleaned_answer = remove_fences_simple(cleaned_answer)
-            if cleaned_answer.startswith("`") and cleaned_answer.endswith("`"):
-                    cleaned_answer = cleaned_answer[1:-1].strip()
-                    
-            print(f"Agent returning final answer (cleaned): '{cleaned_answer}'")
-            return cleaned_answer
-            
-        except Exception as e:
-            print(f"Error running agent graph: {e}")
-            tb_str = traceback.format_exc()
-            print(tb_str)
-            return f"AGENT GRAPH ERROR: {e}"
 
 
 # =============================================================================