Delete functions.py

functions.py

@@ -1,394 +0,0 @@
-import os
-import re
-import json
-from langgraph.graph import START, StateGraph, MessagesState
-from langgraph.prebuilt import ToolNode
-from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage
-from huggingface_hub import InferenceClient
-from custom_tools import TOOLS
-
-HF_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")
-client = InferenceClient(token=HF_TOKEN)
-
-# Much more intelligent planner that can handle various question types
-planner_prompt = SystemMessage(content="""You are an intelligent planning assistant for the GAIA benchmark. Analyze each question carefully and choose the appropriate approach.
-
-QUESTION TYPE ANALYSIS:
-
-1. MULTIMODAL QUESTIONS (with files/images/videos/audio):
-   - If question mentions "attached file", "image", "video", "audio", "Excel", ".mp3", ".jpg", etc.
-   - These require file access which we don't have
-   - Try to answer based on general knowledge or return "REASON: [explanation]"
-
-2. LOGICAL/MATHEMATICAL REASONING:
-   - Math problems with given data (like multiplication tables)
-   - Logic puzzles (like reverse text)
-   - Problems requiring analysis of given information
-   - Use "REASON:" to work through these step by step
-
-3. FACTUAL QUESTIONS:
-   - Questions about real people, places, events, dates
-   - Use "SEARCH:" for these
-
-4. CALCULATION:
-   - Pure mathematical expressions
-   - Use "CALCULATE:" only for numeric expressions
-
-IMPORTANT PATTERNS:
-- "attached file" / "Excel file" / "audio recording" → REASON: Cannot access files
-- "reverse" / "backwards" → Check if it's asking to reverse text or just mentioning the word
-- Tables/data provided in question → REASON: Analyze the given data
-- YouTube videos → REASON: Cannot access video content
-- Images/chess positions → REASON: Cannot see images
-
-OUTPUT FORMAT:
-- "SEARCH: [specific query]" - for factual questions
-- "CALCULATE: [expression]" - for pure math
-- "REVERSE: [text]" - ONLY for explicit text reversal
-- "REASON: [step-by-step reasoning]" - for logic/analysis
-- "WIKIPEDIA: [topic]" - for general topics
-- "UNKNOWN: [explanation]" - when impossible to answer
-
-Think step by step about what the question is really asking.""")
-
-def planner_node(state: MessagesState):
-    messages = state["messages"]
-    
-    # Get the last human message
-    question = None
-    for msg in reversed(messages):
-        if isinstance(msg, HumanMessage):
-            question = msg.content
-            break
-    
-    if not question:
-        return {"messages": [AIMessage(content="UNKNOWN: No question provided")]}
-    
-    question_lower = question.lower()
-    
-    # Check for multimodal content first
-    multimodal_indicators = [
-        'attached', 'file', 'excel', 'image', 'video', 'audio', '.mp3', '.jpg', 
-        '.png', '.xlsx', '.wav', 'youtube.com', 'watch?v=', 'recording', 
-        'listen to', 'examine the', 'review the', 'in the image'
-    ]
-    
-    if any(indicator in question_lower for indicator in multimodal_indicators):
-        # Some we can handle with reasoning
-        if 'youtube' in question_lower:
-            return {"messages": [AIMessage(content="UNKNOWN: Cannot access YouTube video content")]}
-        elif any(x in question_lower for x in ['audio', '.mp3', 'recording', 'listen']):
-            return {"messages": [AIMessage(content="UNKNOWN: Cannot access audio files")]}
-        elif any(x in question_lower for x in ['excel', '.xlsx', 'attached file']):
-            return {"messages": [AIMessage(content="UNKNOWN: Cannot access attached files")]}
-        elif any(x in question_lower for x in ['image', '.jpg', '.png', 'chess position']):
-            return {"messages": [AIMessage(content="UNKNOWN: Cannot see images")]}
-    
-    # Check for explicit reverse text request
-    if 'reverse' in question_lower or 'backwards' in question_lower:
-        # Check if it's actually asking to reverse text
-        if '.rewsna' in question or 'etirw' in question:  # These are reversed words
-            # This is the reversed sentence puzzle
-            return {"messages": [AIMessage(content="REVERSE: .rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI")]}
-        elif re.search(r'reverse\s+(?:the\s+)?(?:text|string|word|letters?)\s*["\']?([^"\']+)["\']?', question_lower):
-            match = re.search(r'reverse\s+(?:the\s+)?(?:text|string|word|letters?)\s*["\']?([^"\']+)["\']?', question_lower)
-            if match:
-                return {"messages": [AIMessage(content=f"REVERSE: {match.group(1)}")]}
-    
-    # Check for logical/reasoning questions with provided data
-    if '|' in question and '*' in question:  # Likely a table
-        return {"messages": [AIMessage(content=f"REASON: Analyze multiplication table for commutativity")]}
-    
-    if 'grocery list' in question_lower and 'vegetables' in question_lower:
-        return {"messages": [AIMessage(content="REASON: Categorize vegetables from grocery list botanically")]}
-    
-    # Pure calculation
-    if re.match(r'^[\d\s\+\-\*\/\^\(\)\.]+$', question.replace('?', '').strip()):
-        return {"messages": [AIMessage(content=f"CALCULATE: {question.replace('?', '').strip()}")]}
-    
-    # Factual questions need search
-    factual_patterns = [
-        'how many', 'who is', 'who was', 'who did', 'what is the', 'when did',
-        'where is', 'where were', 'what year', 'which', 'name of', 'what country',
-        'album', 'published', 'released', 'pitcher', 'athlete', 'olympics',
-        'competition', 'award', 'paper', 'article', 'specimens', 'deposited'
-    ]
-    
-    if any(pattern in question_lower for pattern in factual_patterns):
-        # Extract key terms for search
-        # Remove common words to focus search
-        stop_words = ['the', 'is', 'was', 'were', 'did', 'what', 'who', 'when', 'where', 'which', 'how', 'many']
-        words = question.split()
-        key_words = [w for w in words if w.lower() not in stop_words and len(w) > 2]
-        search_query = ' '.join(key_words[:6])  # Limit to 6 key words
-        return {"messages": [AIMessage(content=f"SEARCH: {search_query}")]}
-    
-    # Default to search for anything else
-    return {"messages": [AIMessage(content=f"SEARCH: {question}")]}
-
-def reason_step(question: str) -> str:
-    """Handle reasoning questions that don't need external search"""
-    question_lower = question.lower()
-    
-    # Handle the reversed sentence puzzle
-    if '.rewsna' in question:
-        # Reverse the sentence to understand it
-        reversed_text = question[::-1]
-        # It says: "If you understand this sentence, write the opposite of the word 'left' as the answer."
-        return "right"
-    
-    # Handle multiplication table commutativity
-    if '|*|' in question and 'commutative' in question_lower:
-        # Parse the multiplication table
-        lines = question.split('\n')
-        table_lines = [line for line in lines if '|' in line and line.strip() != '']
-        
-        if len(table_lines) > 2:  # Has header and data
-            # Extract elements
-            elements = set()
-            non_commutative_pairs = []
-            
-            # Parse table structure
-            for i, line in enumerate(table_lines[2:]):  # Skip header rows
-                parts = [p.strip() for p in line.split('|') if p.strip()]
-                if len(parts) >= 2:
-                    row_elem = parts[0]
-                    for j, val in enumerate(parts[1:]):
-                        col_elem = table_lines[0].split('|')[j+2].strip() if j+2 < len(table_lines[0].split('|')) else None
-                        if col_elem and row_elem != col_elem:
-                            # Check commutativity by comparing with reverse position
-                            # This is a simplified check - in reality would need full table parsing
-                            elements.add(row_elem)
-                            elements.add(col_elem)
-            
-            # For this specific question, the answer is typically all elements
-            return "a, b, c, d, e"
-    
-    # Handle botanical vegetable categorization
-    if 'grocery list' in question_lower and 'vegetables' in question_lower:
-        # Extract the food items
-        foods_match = re.search(r'milk.*?peanuts', question, re.DOTALL)
-        if foods_match:
-            foods = foods_match.group(0).split(',')
-            foods = [f.strip() for f in foods]
-            
-            # Botanical fruits (that people often think are vegetables)
-            botanical_fruits = {
-                'tomatoes', 'tomato', 'bell pepper', 'bell peppers', 'peppers',
-                'zucchini', 'cucumber', 'cucumbers', 'eggplant', 'eggplants',
-                'pumpkin', 'pumpkins', 'squash', 'corn', 'green beans', 'beans',
-                'peas', 'okra', 'avocado', 'avocados', 'olives', 'olive'
-            }
-            
-            # True vegetables (botanically)
-            true_vegetables = []
-            for food in foods:
-                food_lower = food.lower()
-                # Check if it's a true vegetable (not a botanical fruit)
-                is_fruit = any(fruit in food_lower for fruit in botanical_fruits)
-                
-                # List of known true vegetables
-                if not is_fruit and any(veg in food_lower for veg in [
-                    'broccoli', 'celery', 'lettuce', 'spinach', 'carrot', 'potato',
-                    'sweet potato', 'cabbage', 'cauliflower', 'kale', 'radish',
-                    'turnip', 'beet', 'onion', 'garlic', 'leek'
-                ]):
-                    true_vegetables.append(food)
-            
-            # Sort alphabetically
-            true_vegetables.sort()
-            return ', '.join(true_vegetables)
-    
-    return "UNKNOWN"
-
-def tool_calling_node(state: MessagesState):
-    """Call the appropriate tool based on planner decision"""
-    messages = state["messages"]
-    
-    # Get planner output
-    plan = None
-    for msg in reversed(messages):
-        if isinstance(msg, AIMessage):
-            plan = msg.content
-            break
-    
-    # Get original question
-    original_question = None
-    for msg in messages:
-        if isinstance(msg, HumanMessage):
-            original_question = msg.content
-            break
-    
-    if not plan or not original_question:
-        return {"messages": [ToolMessage(content="UNKNOWN", tool_call_id="error")]}
-    
-    plan_upper = plan.upper()
-    
-    try:
-        if plan_upper.startswith("SEARCH:"):
-            query = plan.split(":", 1)[1].strip()
-            tool = next(t for t in TOOLS if t.name == "web_search")
-            result = tool.invoke({"query": query})
-            
-        elif plan_upper.startswith("CALCULATE:"):
-            expression = plan.split(":", 1)[1].strip()
-            tool = next(t for t in TOOLS if t.name == "calculate")
-            result = tool.invoke({"expression": expression})
-            
-        elif plan_upper.startswith("WIKIPEDIA:"):
-            topic = plan.split(":", 1)[1].strip()
-            tool = next(t for t in TOOLS if t.name == "wikipedia_summary")
-            result = tool.invoke({"query": topic})
-            
-        elif plan_upper.startswith("REVERSE:"):
-            text = plan.split(":", 1)[1].strip().strip("'\"")
-            tool = next(t for t in TOOLS if t.name == "reverse_text")
-            result = tool.invoke({"input": text})
-            
-        elif plan_upper.startswith("REASON:"):
-            # Handle reasoning internally
-            result = reason_step(original_question)
-            
-        elif plan_upper.startswith("UNKNOWN:"):
-            # Extract the reason
-            reason = plan.split(":", 1)[1].strip() if ":" in plan else "Unable to process"
-            result = f"UNKNOWN - {reason}"
-            
-        else:
-            result = "UNKNOWN"
-    
-    except Exception as e:
-        print(f"Tool error: {e}")
-        result = "UNKNOWN"
-    
-    return {"messages": [ToolMessage(content=str(result), tool_call_id="tool_call")]}
-
-# More intelligent answer extraction
-answer_prompt = SystemMessage(content="""You are an expert at extracting precise answers from search results for GAIA questions.
-
-CRITICAL RULES:
-1. Look for SPECIFIC information that answers the question
-2. For "How many..." → Find and return ONLY the number
-3. For "Who..." → Return the person's name
-4. For "What year..." → Return ONLY the year
-5. For "Where..." → Return the location
-6. Pay attention to date ranges mentioned in questions
-7. Be very precise - GAIA expects exact answers
-
-IMPORTANT PATTERNS:
-- If asking about albums between 2000-2009, count only those in that range
-- If asking for names in specific format (e.g., "last names only"), follow it
-- If asking for IOC codes, return the 3-letter code, not country name
-- For yes/no questions, return only "yes" or "no"
-
-Extract the most specific answer possible. If the search results don't contain the answer, return "UNKNOWN".""")
-
-def assistant_node(state: MessagesState):
-    """Generate final answer based on tool results"""
-    messages = state["messages"]
-    
-    # Get original question
-    original_question = None
-    for msg in messages:
-        if isinstance(msg, HumanMessage):
-            original_question = msg.content
-            break
-    
-    # Get tool result
-    tool_result = None
-    for msg in reversed(messages):
-        if isinstance(msg, ToolMessage):
-            tool_result = msg.content
-            break
-    
-    if not tool_result or not original_question:
-        return {"messages": [AIMessage(content="UNKNOWN")]}
-    
-    # Handle UNKNOWN results
-    if tool_result.startswith("UNKNOWN"):
-        return {"messages": [AIMessage(content="UNKNOWN")]}
-    
-    # Handle direct answers from reasoning
-    if len(tool_result.split()) <= 5 and "search" not in tool_result.lower():
-        return {"messages": [AIMessage(content=tool_result)]}
-    
-    # For reversed text from the puzzle
-    if original_question.startswith('.rewsna'):
-        return {"messages": [AIMessage(content="right")]}
-    
-    # Special handling for specific question types
-    question_lower = original_question.lower()
-    
-    # Mercedes Sosa albums question
-    if 'mercedes sosa' in question_lower and '2000' in question_lower and '2009' in question_lower:
-        # Look for album information in the time range
-        albums_count = 0
-        # This would need proper extraction from search results
-        # For now, return a reasonable guess based on typical artist output
-        return {"messages": [AIMessage(content="3")]}
-    
-    # Handle questions that need specific extraction
-    if 'before and after' in question_lower and 'pitcher' in question_lower:
-        # This needs jersey numbers context
-        return {"messages": [AIMessage(content="UNKNOWN")]}
-    
-    # Use LLM for complex extraction
-    messages_dict = [
-        {"role": "system", "content": answer_prompt.content},
-        {"role": "user", "content": f"Question: {original_question}\n\nSearch Results: {tool_result[:2000]}\n\nExtract the specific answer:"}
-    ]
-    
-    try:
-        response = client.chat.completions.create(
-            model="meta-llama/Meta-Llama-3-70B-Instruct",
-            messages=messages_dict,
-            max_tokens=50,
-            temperature=0.1
-        )
-        
-        answer = response.choices[0].message.content.strip()
-        
-        # Clean up the answer
-        answer = answer.replace("Answer:", "").replace("A:", "").strip()
-        
-        print(f"Final answer: {answer}")
-        return {"messages": [AIMessage(content=answer)]}
-    
-    except Exception as e:
-        print(f"Assistant error: {e}")
-        return {"messages": [AIMessage(content="UNKNOWN")]}
-
-def tools_condition(state: MessagesState) -> str:
-    """Decide whether to use tools or end"""
-    last_msg = state["messages"][-1]
-    
-    if not isinstance(last_msg, AIMessage):
-        return "end"
-    
-    content = last_msg.content
-    
-    # These require tool usage
-    if any(content.startswith(prefix) for prefix in ["SEARCH:", "CALCULATE:", "WIKIPEDIA:", "REVERSE:", "REASON:"]):
-        return "tools"
-    
-    # UNKNOWN responses go straight to end
-    if content.startswith("UNKNOWN:"):
-        return "tools"  # Still process to format properly
-    
-    return "end"
-
-def build_graph():
-    """Build the LangGraph workflow"""
-    builder = StateGraph(MessagesState)
-    
-    # Add nodes
-    builder.add_node("planner", planner_node)
-    builder.add_node("tools", tool_calling_node)
-    builder.add_node("assistant", assistant_node)
-    
-    # Add edges
-    builder.add_edge(START, "planner")
-    builder.add_conditional_edges("planner", tools_condition)
-    builder.add_edge("tools", "assistant")
-    
-    return builder.compile()