#!/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() initial_state.task_id = task_id or f"workflow_{hash(question) % 10000}" initial_state.question = question initial_state.file_path = file_path initial_state.file_name = file_name 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() state.task_id = task_id or f"simple_{hash(question) % 10000}" state.question = question state.file_path = file_path state.file_name = file_name 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