src/gaia/agent/graph.py

"""
LangGraph workflow definition for the GAIA agent.

This module defines the workflow graph for the GAIA agent using LangGraph.
It implements nodes for question analysis, planning, tool selection,
tool execution, and answer formulation, as well as the edges between them.

The graph is designed to be integrated with the GAIAAgent class and can be
extended with additional nodes and edges as needed.
"""

import logging
import json
import re
import traceback
import openai
from typing import Dict, Any, List, Tuple, Optional, Annotated, TypedDict, Literal, Union, cast

from langchain.prompts import PromptTemplate
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.output_parsers import StrOutputParser

from langgraph.graph import StateGraph, END

from src.gaia.agent.config import get_model_config, get_agent_config, VERBOSE
from src.gaia.agent.tool_registry import (
    ToolRegistry,
    create_default_registry,
    search,
    analyze_query,
    resolve_question_type
)
from src.gaia.memory import WorkingMemory, ResultCache

logger = logging.getLogger("gaia_agent.graph")

class QuestionAnalysis(TypedDict):
    """Type for question analysis results."""
    question_type: str
    required_tools: List[str]
    information_sources: List[str]
    complexity: str

class PlanStep(TypedDict):
    """Type for a single step in the plan."""
    step_number: int
    description: str
    tool: Optional[str]
    tool_input: Optional[Dict[str, Any]]

class ToolResult(TypedDict):
    """Type for tool execution results."""
    tool_name: str
    success: bool
    result: Any
    error: Optional[str]

class AgentState(TypedDict):
    """Type for agent state."""
    question: str
    analysis: Optional[QuestionAnalysis]
    plan: Optional[List[PlanStep]]
    current_step: Optional[int]
    tool_results: Optional[List[ToolResult]]
    reasoning: Optional[str]
    answer: Optional[str]

def analyze_question(state: AgentState) -> AgentState:
    """
    Analyze the question to determine its type and required tools.
    
    Args:
        state: Current agent state
        
    Returns:
        Updated agent state with analysis
    """
    logger.info("Analyzing question")
    
    question = state["question"]
    
    # Detect question type using the tool_registry analyze function
    question_analysis = analyze_query(question)
    question_type = resolve_question_type(question)
    
    # Determine required tools based on question analysis
    required_tools = []
    
    # Web search is needed for most questions
    required_tools.append("web_search")
    
    # Add specific tools based on question type
    if "video" in question.lower() or "youtube" in question.lower():
        required_tools.append("youtube_video")
    
    if "article" in question.lower() or "paper" in question.lower() or "research" in question.lower():
        required_tools.append("arxiv_search")
    
    # For complex questions, add perplexity
    if question_analysis["depth_needed"] == "high":
        required_tools.append("perplexity_search")
    
    # Check for reversed text
    if question_type == "reversed_text":
        required_tools.append("text_manipulation")
    
    # Create analysis object
    analysis = {
        "question_type": question_type,
        "required_tools": required_tools,
        "information_sources": question_analysis["preferred_sources"],
        "complexity": question_analysis["depth_needed"]
    }
    
    logger.info(f"Question analysis: {analysis}")
    
    # Update state
    return {
        **state,
        "analysis": analysis
    }

def create_plan(state: AgentState) -> AgentState:
    """
    Create a plan for answering the question.
    
    Args:
        state: Current agent state
        
    Returns:
        Updated agent state with plan
    """
    logger.info("Creating plan")
    
    question = state["question"]
    analysis = state["analysis"]
    
    if not analysis:
        logger.warning("No question analysis available, creating default plan")
        analysis = {
            "question_type": "factual",
            "required_tools": ["web_search"],
            "information_sources": ["web"],
            "complexity": "medium"
        }
    
    # Create plan based on question type and required tools
    plan = []
    step_num = 1
    
    # Handle reversed text questions
    if analysis["question_type"] == "reversed_text":
        # If text is reversed, handle it first
        plan.append({
            "step_number": step_num,
            "description": "Reverse the text in the question to understand it",
            "tool": "text_manipulation",
            "tool_input": {"text": question, "operation": "reverse"}
        })
        step_num += 1
    
    # Web search is the primary tool for most questions
    if "web_search" in analysis["required_tools"]:
        plan.append({
            "step_number": step_num,
            "description": "Search for information using web search",
            "tool": "web_search",
            "tool_input": {"query": question}
        })
        step_num += 1
    
    # Add specialized tools as needed
    if "youtube_video" in analysis["required_tools"]:
        plan.append({
            "step_number": step_num,
            "description": "Analyze video content using YouTube tool",
            "tool": "youtube_video",
            "tool_input": {"query": question}
        })
        step_num += 1
    
    if "arxiv_search" in analysis["required_tools"]:
        plan.append({
            "step_number": step_num,
            "description": "Search for academic papers on arXiv",
            "tool": "arxiv_search",
            "tool_input": {"query": question}
        })
        step_num += 1
    
    if "perplexity_search" in analysis["required_tools"]:
        plan.append({
            "step_number": step_num,
            "description": "Use Perplexity for in-depth information",
            "tool": "perplexity_search",
            "tool_input": {"query": question}
        })
        step_num += 1
    
    # Always add analysis and answer formulation step
    plan.append({
        "step_number": step_num,
        "description": "Analyze all retrieved information and formulate answer",
        "tool": None,
        "tool_input": None
    })
    
    logger.info(f"Created plan with {len(plan)} steps")
    
    # Update state
    return {
        **state,
        "plan": plan,
        "current_step": 0,
        "tool_results": []  # Initialize tool_results as an empty list
    }

def execute_tool(state: AgentState) -> AgentState:
    """
    Execute the current tool in the plan.
    
    Args:
        state: Current agent state
        
    Returns:
        Updated agent state with tool results
    """
    current_step = state["current_step"]
    plan = state["plan"]
    
    if current_step is None or plan is None or current_step >= len(plan):
        return state
    
    step = plan[current_step]
    tool_name = step.get("tool")
    
    logger.info(f"Executing step {current_step}: {step}")
    
    if not tool_name:
        return {
            **state,
            "current_step": current_step + 1
        }
    
    tool_input = step.get("tool_input", {})
    
    # Get existing tool results or initialize as empty list if None
    tool_results = state.get("tool_results", []) or []
    
    try:
        logger.info(f"Executing tool: {tool_name}")
        
        # Create tool registry
        registry = create_default_registry()
        
        # Execute the appropriate tool
        result = None
        if tool_name == "web_search":
            query = tool_input.get("query", state["question"])
            search_results = search(registry, query, format_type="unified")
            result = search_results
        
        elif tool_name == "youtube_video":
            query = tool_input.get("query", state["question"])
            # Extract YouTube URL
            youtube_urls = re.findall(r'https?://(?:www\.)?youtube\.com/watch\?v=[\w-]+', query)
            if not youtube_urls:
                youtube_urls = re.findall(r'https?://(?:www\.)?youtu\.be/[\w-]+', query)
            
            if youtube_urls:
                video_url = youtube_urls[0]
                if registry.get_tool("youtube_video"):
                    result = registry.execute_tool("youtube_video", url=video_url)
                else:
                    raise Exception("YouTube video tool not available")
            else:
                raise Exception("No YouTube URL found in query")
        
        elif tool_name == "arxiv_search":
            query = tool_input.get("query", state["question"])
            if registry.get_tool("arxiv_search"):
                result = registry.execute_tool("arxiv_search", query=query)
            else:
                raise Exception("ArXiv search tool not available")
        
        elif tool_name == "perplexity_search":
            query = tool_input.get("query", state["question"])
            if registry.get_tool("perplexity_search"):
                result = registry.execute_tool("perplexity_search", query=query)
            else:
                raise Exception("Perplexity search tool not available")
                
        elif tool_name == "text_manipulation":
            text = tool_input.get("text", state["question"])
            operation = tool_input.get("operation", "reverse")
            
            if operation == "reverse":
                result = {'reversed_text': text[::-1]}
            else:
                result = {'original_text': text}
        
        # Create tool result object
        tool_result = {
            "tool_name": tool_name,
            "success": True,
            "result": result,
            "error": None
        }
        
        logger.info(f"Tool {tool_name} executed successfully")
        
    except Exception as e:
        logger.error(f"Error executing tool {tool_name}: {str(e)}")
        logger.debug(traceback.format_exc())
        
        # Record the error
        tool_result = {
            "tool_name": tool_name,
            "success": False,
            "result": None,
            "error": str(e)
        }
    
    # Update state
    return {
        **state,
        "tool_results": tool_results + [tool_result],
        "current_step": current_step + 1
    }

def formulate_answer(state: AgentState) -> AgentState:
    """
    Formulate an answer based on tool results and reasoning.
    
    Args:
        state: Current agent state
        
    Returns:
        Updated agent state with answer
    """
    logger.info("Formulating answer")
    logger.info(f"State: {state}")
    
    question = state["question"]
    tool_results = state["tool_results"] or []
    
    # Extract relevant information from tool results
    web_search_data = []
    youtube_data = None
    arxiv_data = []
    perplexity_data = None
    reversed_text = None
    
    for tool_result in tool_results:
        if not tool_result["success"]:
            continue
        
        if tool_result["tool_name"] == "web_search":
            result = tool_result["result"]
            if result and "results" in result:
                web_search_data = result["results"]
        
        elif tool_result["tool_name"] == "youtube_video":
            youtube_data = tool_result["result"]
        
        elif tool_result["tool_name"] == "arxiv_search":
            arxiv_data = tool_result["result"]
        
        elif tool_result["tool_name"] == "perplexity_search":
            result = tool_result["result"]
            if result and "content" in result:
                perplexity_data = result["content"]
        
        elif tool_result["tool_name"] == "text_manipulation":
            result = tool_result["result"]
            if result and "reversed_text" in result:
                reversed_text = result["reversed_text"]
    
    # First check if this is a reversed text question
    if reversed_text:
        logger.info("Processing reversed text question")
        # The reversed text is the original question in the correct order
        # Parse it to extract what's being asked
        original_question = reversed_text
        
        # Example pattern: "If you understand this sentence, write the opposite of the word "left" as the answer."
        # Look for common patterns in reversed text questions
        if "opposite" in original_question and "word" in original_question:
            # Extract the word to find the opposite of
            match = re.search(r'opposite of the word (?:"|")?(\w+)(?:"|")?', original_question)
            if match:
                word = match.group(1).lower()
                
                # Define common opposites
                opposites = {
                    "left": "right",
                    "right": "left",
                    "up": "down",
                    "down": "up",
                    "black": "white",
                    "white": "black",
                    "yes": "no",
                    "no": "yes",
                    "hot": "cold",
                    "cold": "hot",
                    "big": "small",
                    "small": "big",
                    "tall": "short",
                    "short": "tall"
                }
                
                if word in opposites:
                    answer = opposites[word]
                    return {
                        **state,
                        "reasoning": f"This is a reversed text question asking for the opposite of '{word}'.",
                        "answer": answer
                    }
    
    # Now process normally based on collected data
    reasoning = "Based on the information I've gathered:\n\n"
    
    # Process web search data
    if web_search_data:
        reasoning += "From web search results:\n"
        for i, result in enumerate(web_search_data[:3]):
            reasoning += f"- {result.get('title', 'Untitled')}: {result.get('snippet', 'No snippet')}...\n"
    
    # Process YouTube data
    if youtube_data:
        reasoning += "\nFrom the YouTube video:\n"
        if isinstance(youtube_data, dict):
            reasoning += f"- Title: {youtube_data.get('title', 'Unknown')}\n"
            reasoning += f"- Content: {youtube_data.get('content', 'No content extracted')}...\n"
        elif isinstance(youtube_data, str):
            reasoning += f"- Content: {youtube_data}...\n"
    
    # Process arXiv data
    if arxiv_data:
        reasoning += "\nFrom academic papers:\n"
        for i, paper in enumerate(arxiv_data[:2]):
            if isinstance(paper, dict):
                reasoning += f"- {paper.get('title', 'Untitled')}: {paper.get('summary', 'No summary')[:100]}...\n"
    
    # Process Perplexity data
    if perplexity_data:
        reasoning += "\nFrom Perplexity AI:\n"
        reasoning += f"{perplexity_data[:500]}...\n"
    
    # Formulate answer based on collected data
    answer = ""
    
    # If we have Perplexity data, it's often comprehensive enough
    if perplexity_data:
        # Extract the most relevant parts of the perplexity answer
        answer = perplexity_data.split('\n')[0]  # First paragraph
        
        # For specific question types, extract the specific answer
        if "how many" in question.lower():
            # Look for numbers in the perplexity data
            numbers = re.findall(r'\d+', perplexity_data)
            if numbers:
                # Typically the first number is the answer
                answer = numbers[0]
        
        elif "who" in question.lower():
            # Look for names (capitalized words)
            names = re.findall(r'[A-Z][a-z]+ [A-Z][a-z]+', perplexity_data)
            if names:
                answer = names[0]
    
    # If no answer yet, use web search data
    if not answer and web_search_data:
        # Extract the most relevant information from the top result
        if web_search_data:
            top_result = web_search_data[0]
            answer = top_result.get('snippet', '')
    
    # If we have YouTube data and the question is about a video
    if "video" in question.lower() and youtube_data:
        if isinstance(youtube_data, dict) and "content" in youtube_data:
            answer = youtube_data["content"]
    
    # If still no answer, provide a generic response
    if not answer:
        answer = "I couldn't find enough information to answer your question accurately."
    
    # Ensure we have a proper answer
    answer = answer.strip()
    if not answer.endswith((".", "!", "?")):
        answer += "."
    
    logger.info("Answer formulation complete")
    
    # Update state
    return {
        **state,
        "reasoning": reasoning,
        "answer": answer
    }

def should_continue(state: AgentState) -> Literal["continue", "complete"]:
    """
    Determine if the agent should continue executing the plan or is done.
    
    Args:
        state: Current agent state
        
    Returns:
        "continue" if there are more steps to execute, "complete" otherwise
    """
    current_step = state.get("current_step")
    plan = state.get("plan", [])
    
    if current_step is None:
        return "continue"
    
    if current_step < len(plan):
        return "continue"
    
    return "complete"

def run_agent_graph(initial_state: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Run the agent graph with the initial state.
    
    Args:
        initial_state: Initial state dictionary, must include "question"
        config: Agent configuration dictionary
        
    Returns:
        Final state with answer and reasoning
    """
    logger.info("Running agent graph")
    logger.info(f"Initial state: {initial_state}")
    logger.info(f"Config: {config}")
    
    try:
        # Create initial state
        state: AgentState = {
            "question": initial_state.get("question", ""),
            "analysis": None,
            "plan": None,
            "current_step": None,
            "tool_results": [],  # Initialize as empty list rather than None
            "reasoning": None,
            "answer": None
        }
        
        # Define the graph
        state = analyze_question(state)
        state = create_plan(state)
        
        # Execute tools until done
        while should_continue(state) == "continue":
            state = execute_tool(state)
        
        state = formulate_answer(state)
        
        logger.info("Agent graph completed successfully")
        
        # Return the final state
        final_result = {
            "answer": state.get("answer", "I couldn't find an answer to your question."),
            "reasoning": state.get("reasoning", ""),
            "plan": state.get("plan", []),
            "tool_results": state.get("tool_results", [])
        }
        
        logger.info(f"Final result: {final_result['answer'][:100]}...")
        
        return final_result
        
    except Exception as e:
        logger.error(f"Error running agent graph: {str(e)}")
        logger.debug(traceback.format_exc())
        
        return {
            "answer": f"I encountered an error while processing your question: {str(e)}",
            "reasoning": f"An error occurred: {str(e)}",
            "plan": [],
            "tool_results": []
        }