| """ |
| LLM Orchestrator Agent that uses pure LLM reasoning to delegate questions to specialized agents. |
| """ |
| from typing import Dict, Any, List, Optional, Tuple |
| import re |
| import json |
| import os |
| from base_agent import BaseAgent |
|
|
| class RouterAgent(BaseAgent): |
| """ |
| Pure LLM-based orchestrator that analyzes questions and delegates to specialized agents. |
| """ |
| def __init__(self): |
| """ |
| Initialize the LLM orchestrator agent with an advanced model for better reasoning. |
| """ |
| DEFAULT_MODEL = os.getenv("DEFAULT_LLM_MODEL") |
| print("Initializing Router Agent with " + DEFAULT_MODEL) |
| super().__init__(model=DEFAULT_MODEL) |
| self.agents = {} |
| self.agent_descriptions = {} |
| |
| def register_agent(self, agent_type: str, agent: BaseAgent, description: str) -> None: |
| """ |
| Register an agent for a specific question type along with a description of its capabilities. |
| |
| Args: |
| agent_type: The type of questions this agent handles |
| agent: The agent instance |
| description: Description of what this agent does and when to use it |
| """ |
| self.agents[agent_type] = agent |
| self.agent_descriptions[agent_type] = description |
| |
| def __call__(self, question: str, file_name: str = None, **kwargs) -> str: |
| """ |
| Process the question using pure LLM reasoning to select the appropriate agent. |
| |
| Args: |
| question: The question to process |
| |
| Returns: |
| The answer from the appropriate specialized agent |
| """ |
| print(f"LLM Orchestrator analyzing question: {question[:100]}...") |
| |
| |
| context = self._build_context_from_kwargs(kwargs) |
|
|
| print(context) |
| |
| |
| agent_type = self._llm_decide_agent(question, context) |
| |
| |
| if agent_type in self.agents: |
| print(f"Agent Router picked: {agent_type}.") |
| return self.agents[agent_type](question, file_name, **kwargs) |
| |
| |
| if "research_question" in self.agents: |
| print(f"Agent '{agent_type}' not found. Falling back to research_question agent.") |
| return self.agents["research_question"](question, **kwargs) |
|
|
| |
| |
| return f"No appropriate agent found for question type: {agent_type}" |
| |
| def _build_context_from_kwargs(self, kwargs: Dict[str, Any]) -> str: |
| """ |
| Build context information from kwargs to help the LLM make a better decision. |
| |
| Args: |
| kwargs: Additional keyword arguments |
| |
| Returns: |
| Context information as a string |
| """ |
| context_parts = [] |
| |
| if kwargs.get('has_image', False): |
| context_parts.append("This question involves an image file.") |
| |
| if kwargs.get('has_audio', False): |
| context_parts.append("This question involves an audio file.") |
| |
| if kwargs.get('has_excel', False): |
| context_parts.append("This question involves an Excel or CSV file.") |
| |
| if kwargs.get('has_python', False): |
| context_parts.append("This question involves Python code.") |
| |
| return " ".join(context_parts) |
| |
| def _llm_decide_agent(self, question: str, context: str = "") -> str: |
| """ |
| Use the LLM to decide which agent should handle the question. |
| |
| Args: |
| question: The question to analyze |
| context: Additional context about the question |
| |
| Returns: |
| The selected agent type |
| """ |
| |
| agent_descriptions_text = "\n".join([ |
| f"- {agent_type}: {description}" |
| for agent_type, description in self.agent_descriptions.items() |
| ]) |
| |
| system_prompt = f""" |
| You are an expert question analyzer and orchestrator for a multi-agent system. |
| |
| The following specialized agents are available: |
| {agent_descriptions_text} |
| |
| Your task is to analyze the given question and determine which specialized agent would be best suited to handle it. |
| Consider the question's content, type, and any contextual clues provided. |
| |
| Return ONLY the agent type identifier as a single word, with no additional explanation or text. |
| For example: "research_question" or "audio_transcript" |
| """ |
| |
| user_prompt = f""" |
| Question: {question} |
| |
| {context if context else ""} |
| |
| Based on this question, which agent should handle it? |
| """ |
| |
| response = self.llm_call(system_prompt, user_prompt) |
| |
| |
| response = response.strip().lower() |
| |
| |
| for agent_type in self.agents.keys(): |
| if agent_type in response: |
| return agent_type |
| |
| |
| return "research_question" |
