src/workflow/gaia_workflow.py

#!/usr/bin/env python3
"""
GAIA Agent LangGraph Workflow
Main orchestration workflow for the GAIA benchmark agent system
"""

import logging
from typing import Dict, Any, List, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

from agents.state import GAIAAgentState, AgentRole, QuestionType
from agents.router import RouterAgent
from agents.web_researcher import WebResearchAgent
from agents.file_processor_agent import FileProcessorAgent
from agents.reasoning_agent import ReasoningAgent
from agents.synthesizer import SynthesizerAgent
from models.qwen_client import QwenClient

logger = logging.getLogger(__name__)

class GAIAWorkflow:
    """
    Main GAIA agent workflow using LangGraph
    Orchestrates router → specialized agents → synthesizer pipeline
    """
    
    def __init__(self, llm_client: QwenClient):
        self.llm_client = llm_client
        
        # Initialize all agents
        self.router = RouterAgent(llm_client)
        self.web_researcher = WebResearchAgent(llm_client)
        self.file_processor = FileProcessorAgent(llm_client)
        self.reasoning_agent = ReasoningAgent(llm_client)
        self.synthesizer = SynthesizerAgent(llm_client)
        
        # Create workflow graph
        self.workflow = self._create_workflow()
        
        # Compile workflow with memory
        self.app = self.workflow.compile(checkpointer=MemorySaver())
        
    def _create_workflow(self) -> StateGraph:
        """Create the LangGraph workflow"""
        
        # Define the workflow graph
        workflow = StateGraph(GAIAAgentState)
        
        # Add nodes
        workflow.add_node("router", self._router_node)
        workflow.add_node("web_researcher", self._web_researcher_node)
        workflow.add_node("file_processor", self._file_processor_node)
        workflow.add_node("reasoning_agent", self._reasoning_agent_node)
        workflow.add_node("synthesizer", self._synthesizer_node)
        
        # Define entry point
        workflow.set_entry_point("router")
        
        # Add conditional edges from router to agents
        workflow.add_conditional_edges(
            "router",
            self._route_to_agents,
            {
                "web_researcher": "web_researcher",
                "file_processor": "file_processor", 
                "reasoning_agent": "reasoning_agent",
                "multi_agent": "web_researcher",  # Start with web researcher for multi-agent
                "synthesizer": "synthesizer"  # Direct to synthesizer if no agents needed
            }
        )
        
        # Add edges from agents to synthesizer
        workflow.add_edge("web_researcher", "synthesizer")
        workflow.add_edge("file_processor", "synthesizer") 
        workflow.add_edge("reasoning_agent", "synthesizer")
        
        # Add conditional edges for multi-agent scenarios
        workflow.add_conditional_edges(
            "synthesizer",
            self._check_if_complete,
            {
                "complete": END,
                "need_more_agents": "file_processor"  # Route to next agent if needed
            }
        )
        
        return workflow
    
    def _router_node(self, state: GAIAAgentState) -> GAIAAgentState:
        """Router node - classifies question and selects agents"""
        logger.info("🧭 Executing router node")
        return self.router.route_question(state)
    
    def _web_researcher_node(self, state: GAIAAgentState) -> GAIAAgentState:
        """Web researcher node"""
        logger.info("🌐 Executing web researcher node")
        return self.web_researcher.process(state)
    
    def _file_processor_node(self, state: GAIAAgentState) -> GAIAAgentState:
        """File processor node"""
        logger.info("📁 Executing file processor node")
        return self.file_processor.process(state)
    
    def _reasoning_agent_node(self, state: GAIAAgentState) -> GAIAAgentState:
        """Reasoning agent node"""
        logger.info("🧠 Executing reasoning agent node")
        return self.reasoning_agent.process(state)
    
    def _synthesizer_node(self, state: GAIAAgentState) -> GAIAAgentState:
        """Synthesizer node - combines agent results"""
        logger.info("🔗 Executing synthesizer node")
        return self.synthesizer.process(state)
    
    def _route_to_agents(self, state: GAIAAgentState) -> str:
        """Determine which agent(s) to route to based on router decision"""
        
        selected_agents = state.selected_agents
        
        # Remove synthesizer from routing decision (it's always last)
        agent_roles = [agent for agent in selected_agents if agent != AgentRole.SYNTHESIZER]
        
        if not agent_roles:
            # No specific agents selected, go directly to synthesizer
            return "synthesizer"
        elif len(agent_roles) == 1:
            # Single agent selected
            agent = agent_roles[0]
            if agent == AgentRole.WEB_RESEARCHER:
                return "web_researcher"
            elif agent == AgentRole.FILE_PROCESSOR:
                return "file_processor"
            elif agent == AgentRole.REASONING_AGENT:
                return "reasoning_agent"
            else:
                return "synthesizer"
        else:
            # Multiple agents - start with web researcher
            # The workflow will handle additional agents in subsequent steps
            return "multi_agent"
    
    def _check_if_complete(self, state: GAIAAgentState) -> str:
        """Check if processing is complete or if more agents are needed"""
        
        # If synthesis is complete, we're done
        if state.is_complete:
            return "complete"
        
        # Check if we need to run additional agents
        selected_agents = state.selected_agents
        executed_agents = set(state.agent_results.keys())
        
        # Find agents that haven't been executed yet
        remaining_agents = [
            agent for agent in selected_agents 
            if agent not in executed_agents and agent != AgentRole.SYNTHESIZER
        ]
        
        if remaining_agents:
            # Route to next agent
            next_agent = remaining_agents[0]
            if next_agent == AgentRole.FILE_PROCESSOR:
                return "need_more_agents"  # This will route to file_processor
            elif next_agent == AgentRole.REASONING_AGENT:
                return "need_more_agents"  # Would need additional routing logic
            else:
                return "complete"
        else:
            return "complete"
    
    def process_question(self, question: str, file_path: str = None, file_name: str = None, 
                        task_id: str = None, difficulty_level: int = 1) -> GAIAAgentState:
        """
        Process a GAIA question through the complete workflow
        
        Args:
            question: The question to process
            file_path: Optional path to associated file
            file_name: Optional name of associated file
            task_id: Optional task identifier
            difficulty_level: Question difficulty (1-3)
            
        Returns:
            GAIAAgentState with final results
        """
        
        logger.info(f"🚀 Processing question: {question[:100]}...")
        
        # Initialize state
        initial_state = GAIAAgentState(
            question=question,
            question_id=task_id or f"workflow_{hash(question) % 10000}",
            file_name=file_name,
            file_content=None
        )
        initial_state.file_path = file_path
        initial_state.difficulty_level = difficulty_level
        
        try:
            # Execute workflow
            final_state = self.app.invoke(
                initial_state,
                config={"configurable": {"thread_id": initial_state.task_id}}
            )
            
            logger.info(f"✅ Workflow complete: {final_state.final_answer[:100]}...")
            return final_state
            
        except Exception as e:
            error_msg = f"Workflow execution failed: {str(e)}"
            logger.error(error_msg)
            
            # Create error state
            initial_state.add_error(error_msg)
            initial_state.final_answer = "Workflow execution failed"
            initial_state.final_confidence = 0.0
            initial_state.final_reasoning = error_msg
            initial_state.is_complete = True
            initial_state.requires_human_review = True
            
            return initial_state
    
    def get_workflow_visualization(self) -> str:
        """Get a text representation of the workflow"""
        return """
GAIA Agent Workflow:

┌─────────────┐
│   Router    │ ← Entry Point
└──────┬──────┘
       │
       ├─ Web Researcher ──┐
       ├─ File Processor ──┤
       ├─ Reasoning Agent ─┤
       │                   │
       ▼                   ▼
┌─────────────┐    ┌──────────────┐
│ Synthesizer │ ←──┤ Agent Results │
└──────┬──────┘    └──────────────┘
       │
       ▼
┌─────────────┐
│   END       │
└─────────────┘

Flow:
1. Router classifies question and selects appropriate agent(s)
2. Selected agents process question in parallel/sequence
3. Synthesizer combines results into final answer
4. Workflow completes with final state
"""

# Simplified workflow for cases where we don't need full LangGraph
class SimpleGAIAWorkflow:
    """
    Simplified workflow that doesn't require LangGraph for basic cases
    Useful for testing and lightweight deployments
    """
    
    def __init__(self, llm_client: QwenClient):
        self.llm_client = llm_client
        self.router = RouterAgent(llm_client)
        self.web_researcher = WebResearchAgent(llm_client)
        self.file_processor = FileProcessorAgent(llm_client)
        self.reasoning_agent = ReasoningAgent(llm_client)
        self.synthesizer = SynthesizerAgent(llm_client)
    
    def process_question(self, question: str, file_path: str = None, file_name: str = None, 
                        task_id: str = None, difficulty_level: int = 1) -> GAIAAgentState:
        """Process question with simplified sequential workflow"""
        
        # Initialize state
        state = GAIAAgentState(
            question=question,
            question_id=task_id or f"simple_{hash(question) % 10000}",
            file_name=file_name,
            file_content=None
        )
        state.file_path = file_path
        state.difficulty_level = difficulty_level
        
        try:
            # Step 1: Route
            state = self.router.route_question(state)
            
            # Step 2: Execute agents
            for agent_role in state.selected_agents:
                if agent_role == AgentRole.WEB_RESEARCHER:
                    state = self.web_researcher.process(state)
                elif agent_role == AgentRole.FILE_PROCESSOR:
                    state = self.file_processor.process(state)
                elif agent_role == AgentRole.REASONING_AGENT:
                    state = self.reasoning_agent.process(state)
                # Skip synthesizer for now
            
            # Step 3: Synthesize
            state = self.synthesizer.process(state)
            
            return state
            
        except Exception as e:
            error_msg = f"Simple workflow failed: {str(e)}"
            state.add_error(error_msg)
            state.final_answer = "Processing failed"
            state.final_confidence = 0.0
            state.final_reasoning = error_msg
            state.is_complete = True
            return state 

def create_gaia_workflow(llm_client, tools_dict):
    """
    Create an enhanced GAIA workflow with multi-phase planning and iterative refinement
    """
    
    # Initialize agents with enhanced capabilities
    router = RouterAgent(llm_client)
    web_researcher = WebResearchAgent(llm_client) 
    file_processor = FileProcessorAgent(llm_client)
    reasoning_agent = ReasoningAgent(llm_client)
    synthesizer = SynthesizerAgent(llm_client)
    
    # Enhanced workflow nodes with multi-step processing
    def router_node(state: GAIAAgentState) -> GAIAAgentState:
        """Enhanced router with multi-phase analysis"""
        logger.info("🧭 Router: Starting multi-phase analysis")
        return router.process(state)
    
    def web_researcher_node(state: GAIAAgentState) -> GAIAAgentState:
        """Web researcher with multi-step planning"""
        logger.info("🌐 Web Researcher: Starting enhanced research")
        return web_researcher.process(state)
    
    def file_processor_node(state: GAIAAgentState) -> GAIAAgentState:
        """File processor with step-by-step analysis"""
        logger.info("📁 File Processor: Starting file analysis")
        return file_processor.process(state)
    
    def reasoning_agent_node(state: GAIAAgentState) -> GAIAAgentState:
        """Reasoning agent with systematic approach"""
        logger.info("🧠 Reasoning Agent: Starting analysis")
        return reasoning_agent.process(state)
    
    def synthesizer_node(state: GAIAAgentState) -> GAIAAgentState:
        """Enhanced synthesizer with verification"""
        logger.info("🎯 Synthesizer: Starting GAIA-compliant synthesis")
        return synthesizer.process(state)
    
    def should_continue_to_next_agent(state: GAIAAgentState) -> str:
        """
        Enhanced routing logic that follows the planned agent sequence
        """
        # Get the planned sequence from router
        agent_sequence = getattr(state, 'agent_sequence', [])
        
        if not agent_sequence:
            logger.warning("No agent sequence found, using fallback routing")
            # Fallback to basic routing
            if not state.agent_results:
                return "web_researcher"
            return "synthesizer"
        
        # Count how many agents have been executed
        executed_count = len(state.agent_results)
        
        # Check if we've executed all planned agents
        if executed_count >= len(agent_sequence):
            return "synthesizer"
        
        # Get next agent in sequence
        next_agent = agent_sequence[executed_count]
        
        # Map string names to node names
        agent_mapping = {
            'web_researcher': 'web_researcher',
            'file_processor': 'file_processor', 
            'reasoning_agent': 'reasoning_agent',
            'synthesizer': 'synthesizer'
        }
        
        return agent_mapping.get(next_agent, 'synthesizer')
    
    def check_quality_and_refinement(state: GAIAAgentState) -> str:
        """
        Check if results need refinement before synthesis
        """
        if not state.agent_results:
            return "synthesizer"
        
        # Check overall quality of results
        avg_confidence = sum(r.confidence for r in state.agent_results) / len(state.agent_results)
        
        # If confidence is very low and we haven't tried refinement yet
        if avg_confidence < 0.3 and not getattr(state, 'refinement_attempted', False):
            logger.info(f"Low confidence ({avg_confidence:.2f}), attempting refinement")
            state.refinement_attempted = True
            return "refine_approach"
        
        return "synthesizer"
    
    def refinement_node(state: GAIAAgentState) -> GAIAAgentState:
        """
        Attempt to refine the approach when initial results are poor
        """
        logger.info("🔄 Attempting result refinement")
        state.add_processing_step("Workflow: Attempting refinement due to low confidence")
        
        # Analyze what went wrong and try a different approach
        router_analysis = getattr(state, 'router_analysis', {})
        
        if router_analysis:
            # Try alternative strategy from router analysis
            strategy = router_analysis.get('strategy', {})
            fallback_strategies = strategy.get('fallback_needed', True)
            
            if fallback_strategies:
                # Try web research if it wasn't the primary approach
                if not any(r.agent_role == AgentRole.WEB_RESEARCHER for r in state.agent_results):
                    return web_researcher.process(state)
                # Try reasoning if web search was done
                elif not any(r.agent_role == AgentRole.REASONING_AGENT for r in state.agent_results):
                    return reasoning_agent.process(state)
        
        # Fallback: try reasoning agent for additional analysis
        return reasoning_agent.process(state)
    
    # Create workflow graph with enhanced routing
    workflow = StateGraph(GAIAAgentState)
    
    # Add nodes
    workflow.add_node("router", router_node)
    workflow.add_node("web_researcher", web_researcher_node)
    workflow.add_node("file_processor", file_processor_node)
    workflow.add_node("reasoning_agent", reasoning_agent_node)
    workflow.add_node("refine_approach", refinement_node)
    workflow.add_node("synthesizer", synthesizer_node)
    
    # Set entry point
    workflow.set_entry_point("router")
    
    # Enhanced routing edges
    workflow.add_conditional_edges(
        "router",
        should_continue_to_next_agent,
        {
            "web_researcher": "web_researcher",
            "file_processor": "file_processor",
            "reasoning_agent": "reasoning_agent", 
            "synthesizer": "synthesizer"
        }
    )
    
    # Progressive routing with quality checks
    workflow.add_conditional_edges(
        "web_researcher",
        should_continue_to_next_agent,
        {
            "file_processor": "file_processor",
            "reasoning_agent": "reasoning_agent",
            "synthesizer": "synthesizer",
            "refine_approach": "refine_approach"
        }
    )
    
    workflow.add_conditional_edges(
        "file_processor", 
        should_continue_to_next_agent,
        {
            "web_researcher": "web_researcher",
            "reasoning_agent": "reasoning_agent",
            "synthesizer": "synthesizer",
            "refine_approach": "refine_approach"
        }
    )
    
    workflow.add_conditional_edges(
        "reasoning_agent",
        should_continue_to_next_agent,
        {
            "web_researcher": "web_researcher",
            "file_processor": "file_processor", 
            "synthesizer": "synthesizer",
            "refine_approach": "refine_approach"
        }
    )
    
    # Quality check before synthesis
    workflow.add_conditional_edges(
        "refine_approach",
        check_quality_and_refinement,
        {
            "synthesizer": "synthesizer",
            "refine_approach": "refine_approach"  # Allow multiple refinement attempts
        }
    )
    
    # Synthesizer is the final step
    workflow.add_edge("synthesizer", END)
    
    return workflow.compile()

def create_simple_workflow(llm_client, tools_dict):
    """
    Enhanced simple workflow with better planning and execution
    """
    # Use same agents as complex workflow for consistency
    router = RouterAgent(llm_client)
    web_researcher = WebResearchAgent(llm_client)
    reasoning_agent = ReasoningAgent(llm_client)
    synthesizer = SynthesizerAgent(llm_client)
    
    def process_with_planning(state: GAIAAgentState) -> GAIAAgentState:
        """Simple but systematic processing with planning"""
        
        logger.info("🚀 Starting simple workflow with enhanced planning")
        
        # Step 1: Analyze and plan
        state = router.process(state)
        
        # Step 2: Execute primary research/reasoning
        agent_sequence = getattr(state, 'agent_sequence', ['web_researcher', 'reasoning_agent'])
        
        for agent_name in agent_sequence:
            if agent_name == 'web_researcher':
                state = web_researcher.process(state)
            elif agent_name == 'reasoning_agent':
                state = reasoning_agent.process(state)
            elif agent_name == 'synthesizer':
                break  # Synthesizer is handled separately
            
            # Early exit if we have high confidence result
            if state.agent_results and state.agent_results[-1].confidence > 0.8:
                logger.info("High confidence result achieved, proceeding to synthesis")
                break
        
        # Step 3: Synthesize results
        state = synthesizer.process(state)
        
        return state
    
    # Create simple workflow graph
    workflow = StateGraph(GAIAAgentState)
    workflow.add_node("process", process_with_planning)
    workflow.set_entry_point("process")
    workflow.add_edge("process", END)
    
    return workflow.compile()