File size: 5,485 Bytes
4f36157 8d859d2 4f36157 512cd61 4f36157 9223edf 4f36157 4f25d46 4f36157 c08e15d 4f36157 0c1043b 9223edf 4f36157 54adbbb 9223edf 4f36157 9223edf 4f36157 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
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]}...")
# Extract metadata from kwargs to provide context to the LLM
context = self._build_context_from_kwargs(kwargs)
print(context)
# Use LLM to decide which agent should handle this question
agent_type = self._llm_decide_agent(question, context)
# Delegate to the selected specialized agent
if agent_type in self.agents:
print(f"Agent Router picked: {agent_type}.")
return self.agents[agent_type](question, file_name, **kwargs)
# Fallback to research agent if available
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)
# Ultimate fallback
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
"""
# Build descriptions of available agents to help the LLM decide
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)
# Clean up the response to extract just the agent type
response = response.strip().lower()
# Extract just the agent type from potential verbose responses
for agent_type in self.agents.keys():
if agent_type in response:
return agent_type
# Default to research question if no clear match
return "research_question" |