Update app.py

app.py

@@ -37,526 +37,6 @@ 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
 # =============================================================================
@@ -658,7 +138,6 @@ def find_file(path: str) -> Optional[Path]:
     
     return None
 
-
 # =============================================================================
 # PLANNING & REFLECTION TOOLS
 # =============================================================================
@@ -822,7 +301,6 @@ def validate_answer(proposed_answer: str, original_question: str) -> str:
     return "✅ VALIDATION PASSED: Answer looks good! Proceed with final_answer_tool now."
 
 
-# =============================================================================
 # =============================================================================
 # CORE TOOLS
 # =============================================================================
@@ -1016,6 +494,7 @@ def write_file(path: str, content: str) -> str:
 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."""
@@ -1229,7 +708,8 @@ def final_answer_tool(answer: str) -> str:
 # DEFINED TOOLS LIST
 # =============================================================================
 defined_tools = [
-    # Planning & Reflection (use these first!)
+    # Planning & Reflection (use these strategically!)
+    think_through_logic,  # NEW: For logic puzzles
     create_plan,
     reflect_on_progress,
     validate_answer,
@@ -1254,6 +734,7 @@ defined_tools = [
 ]
 
 
+
 # =============================================================================
 # AGENT STATE
 # =============================================================================
@@ -1372,6 +853,41 @@ def parse_tool_call_from_string(content: str, tools: List) -> List[ToolCall]:
     return []
 
 
+# =============================================================================
+# CONDITIONAL EDGE FUNCTION
+# =============================================================================
+def should_continue(state: AgentState):
+    """Decide whether to continue, call tools, or end."""
+    last_message = state['messages'][-1]
+    current_turn = state.get('turn', 0)
+
+    # Check for final_answer_tool
+    if isinstance(last_message, AIMessage) and last_message.tool_calls:
+        for tool_call in last_message.tool_calls:
+            if tool_call.get("name") == "final_answer_tool":
+                print("--- Condition: final_answer_tool called, ending. ---")
+                return END
+
+    # Check turn limit
+    if current_turn >= MAX_TURNS:
+        print(f"--- Condition: Max turns ({MAX_TURNS}) reached. Ending. ---")
+        return END
+
+    # Route to tools if tool calls exist
+    if isinstance(last_message, AIMessage) and last_message.tool_calls:
+        print("--- Condition: Tools called, routing to tools node. ---")
+        return "tools"
+    
+    # Loop prevention
+    if len(state['messages']) > 2 and isinstance(last_message, AIMessage) and isinstance(state['messages'][-2], AIMessage):
+        print(f"--- Condition: Detected 2+ consecutive AI messages (Turn {current_turn}). Ending to prevent loop. ---")
+        return END
+
+    # Loop back to agent
+    print(f"--- Condition: No tool call (Turn {current_turn}). Continuing to agent. ---")
+    return "agent"
+
+
 # =============================================================================
 # ENHANCED AGENT CLASS WITH PLANNING & REFLECTION
 # =============================================================================
@@ -1674,7 +1190,7 @@ Turn 7: final_answer_tool("1.796 trillion")
             
             return result
         
-        # Build Graph
+# Build Graph
         print("Building Planning & Reflection Agent graph...")
         graph_builder = StateGraph(AgentState)
         
@@ -1697,9 +1213,7 @@ Turn 7: final_answer_tool("1.796 trillion")
         
         self.graph = graph_builder.compile()
         print("✅ Planning & Reflection Agent graph compiled successfully.")
-
-
-
+        
 
 # =============================================================================
 # GLOBAL AGENT INSTANTIATION