|
|
""" |
|
|
Tools module for GAIA implementation. |
|
|
|
|
|
Contains implementations of various tools for web browsing, reasoning, and multimodal processing. |
|
|
Includes a tool registry for dynamic tool selection based on question requirements. |
|
|
""" |
|
|
|
|
|
import logging |
|
|
from typing import Dict, Any, List, Optional, Union, Callable, Type, Set |
|
|
|
|
|
from src.gaia.tools.web_tools import ( |
|
|
DuckDuckGoSearchTool, |
|
|
SerperSearchTool, |
|
|
WebContentExtractor, |
|
|
WebNavigator, |
|
|
create_duckduckgo_search, |
|
|
create_serper_search, |
|
|
create_web_content_extractor, |
|
|
create_web_navigator |
|
|
) |
|
|
|
|
|
from src.gaia.tools.reasoning_tools import ( |
|
|
StepByStepReasoner, |
|
|
MathCalculator, |
|
|
FactVerifier, |
|
|
create_step_by_step_reasoner, |
|
|
create_math_calculator, |
|
|
create_fact_verifier |
|
|
) |
|
|
|
|
|
from src.gaia.tools.perplexity_tool import ( |
|
|
PerplexityTool, |
|
|
create_perplexity_tool |
|
|
) |
|
|
|
|
|
from src.gaia.tools.multimodal_tools import ( |
|
|
ImageAnalyzer, |
|
|
ChartInterpreter, |
|
|
DocumentParser, |
|
|
YouTubeVideoTool, |
|
|
create_image_analyzer, |
|
|
create_chart_interpreter, |
|
|
create_document_parser, |
|
|
create_youtube_video_tool |
|
|
) |
|
|
|
|
|
logger = logging.getLogger("gaia_agent.tools") |
|
|
|
|
|
class ToolMetadata: |
|
|
"""Metadata for a tool in the registry.""" |
|
|
|
|
|
def __init__( |
|
|
self, |
|
|
name: str, |
|
|
description: str, |
|
|
category: str, |
|
|
factory: Callable[[], Any], |
|
|
keywords: List[str], |
|
|
input_schema: Dict[str, Any], |
|
|
output_schema: Dict[str, Any], |
|
|
requires_api_key: bool = False, |
|
|
example_uses: List[str] = None |
|
|
): |
|
|
""" |
|
|
Initialize tool metadata. |
|
|
|
|
|
Args: |
|
|
name: The name of the tool |
|
|
description: A description of what the tool does |
|
|
category: The category of the tool (web, reasoning, multimodal) |
|
|
factory: A factory function that creates an instance of the tool |
|
|
keywords: Keywords for tool selection |
|
|
input_schema: Schema describing the tool's input parameters |
|
|
output_schema: Schema describing the tool's output format |
|
|
requires_api_key: Whether the tool requires an API key |
|
|
example_uses: Example use cases for the tool |
|
|
""" |
|
|
self.name = name |
|
|
self.description = description |
|
|
self.category = category |
|
|
self.factory = factory |
|
|
self.keywords = keywords |
|
|
self.input_schema = input_schema |
|
|
self.output_schema = output_schema |
|
|
self.requires_api_key = requires_api_key |
|
|
self.example_uses = example_uses or [] |
|
|
self._instance = None |
|
|
|
|
|
def get_instance(self) -> Any: |
|
|
""" |
|
|
Get or create an instance of the tool. |
|
|
|
|
|
Returns: |
|
|
An instance of the tool |
|
|
""" |
|
|
if self._instance is None: |
|
|
self._instance = self.factory() |
|
|
return self._instance |
|
|
|
|
|
|
|
|
class ToolRegistry: |
|
|
"""Registry for managing available tools.""" |
|
|
|
|
|
def __init__(self): |
|
|
"""Initialize the tool registry.""" |
|
|
self.tools: Dict[str, ToolMetadata] = {} |
|
|
self.categories: Dict[str, List[str]] = { |
|
|
"web": [], |
|
|
"reasoning": [], |
|
|
"multimodal": [] |
|
|
} |
|
|
|
|
|
def register_tool(self, metadata: ToolMetadata) -> None: |
|
|
""" |
|
|
Register a tool in the registry. |
|
|
|
|
|
Args: |
|
|
metadata: The tool metadata |
|
|
""" |
|
|
self.tools[metadata.name] = metadata |
|
|
|
|
|
if metadata.category in self.categories: |
|
|
self.categories[metadata.category].append(metadata.name) |
|
|
else: |
|
|
self.categories[metadata.category] = [metadata.name] |
|
|
|
|
|
|
|
|
def get_tool(self, name: str) -> Optional[Any]: |
|
|
""" |
|
|
Get a tool instance by name. |
|
|
|
|
|
Args: |
|
|
name: The name of the tool |
|
|
|
|
|
Returns: |
|
|
The tool instance, or None if not found |
|
|
""" |
|
|
if name not in self.tools: |
|
|
logger.warning(f"Tool not found: {name}") |
|
|
return None |
|
|
|
|
|
return self.tools[name].get_instance() |
|
|
|
|
|
def get_tool_metadata(self, name: str) -> Optional[ToolMetadata]: |
|
|
""" |
|
|
Get tool metadata by name. |
|
|
|
|
|
Args: |
|
|
name: The name of the tool |
|
|
|
|
|
Returns: |
|
|
The tool metadata, or None if not found |
|
|
""" |
|
|
return self.tools.get(name) |
|
|
|
|
|
def list_tools(self, category: Optional[str] = None) -> List[str]: |
|
|
""" |
|
|
List available tools, optionally filtered by category. |
|
|
|
|
|
Args: |
|
|
category: Optional category to filter by |
|
|
|
|
|
Returns: |
|
|
List of tool names |
|
|
""" |
|
|
if category: |
|
|
return self.categories.get(category, []) |
|
|
|
|
|
return list(self.tools.keys()) |
|
|
|
|
|
def select_tools(self, requirements: List[str]) -> List[str]: |
|
|
""" |
|
|
Select appropriate tools based on requirements. |
|
|
|
|
|
Args: |
|
|
requirements: List of tool requirements or keywords |
|
|
|
|
|
Returns: |
|
|
List of selected tool names |
|
|
""" |
|
|
selected_tools = set() |
|
|
|
|
|
for req in requirements: |
|
|
req_lower = req.lower() |
|
|
|
|
|
if req_lower in self.tools: |
|
|
selected_tools.add(req_lower) |
|
|
continue |
|
|
|
|
|
for name, metadata in self.tools.items(): |
|
|
if any(keyword.lower() in req_lower for keyword in metadata.keywords): |
|
|
selected_tools.add(name) |
|
|
|
|
|
return list(selected_tools) |
|
|
|
|
|
def get_tools_by_question(self, question: str) -> List[str]: |
|
|
""" |
|
|
Select appropriate tools based on a question. |
|
|
|
|
|
Args: |
|
|
question: The question to analyze |
|
|
|
|
|
Returns: |
|
|
List of selected tool names |
|
|
""" |
|
|
question_lower = question.lower() |
|
|
selected_tools = set() |
|
|
|
|
|
if any(term in question_lower for term in ["search", "find", "look up", "latest", "current", "news"]): |
|
|
selected_tools.add("duckduckgo_search") |
|
|
selected_tools.add("perplexity_search") |
|
|
|
|
|
if any(term in question_lower for term in ["website", "webpage", "url", "link", "extract", "content"]): |
|
|
selected_tools.add("web_content_extractor") |
|
|
|
|
|
if any(term in question_lower for term in ["navigate", "browse", "follow", "links"]): |
|
|
selected_tools.add("web_navigator") |
|
|
|
|
|
if any(term in question_lower for term in ["reason", "explain", "why", "how", "analyze"]): |
|
|
selected_tools.add("step_by_step_reasoner") |
|
|
selected_tools.add("perplexity_search") |
|
|
|
|
|
if any(term in question_lower for term in ["calculate", "compute", "solve", "equation", "math", "formula"]): |
|
|
selected_tools.add("math_calculator") |
|
|
|
|
|
if any(term in question_lower for term in ["verify", "check", "fact", "true", "false", "accurate"]): |
|
|
selected_tools.add("fact_verifier") |
|
|
selected_tools.add("perplexity_search") |
|
|
|
|
|
if any(term in question_lower for term in ["image", "picture", "photo", "describe", "visual", "see"]): |
|
|
selected_tools.add("image_analyzer") |
|
|
|
|
|
if any(term in question_lower for term in ["chart", "graph", "plot", "diagram", "visualization", "trend"]): |
|
|
selected_tools.add("chart_interpreter") |
|
|
|
|
|
if any(term in question_lower for term in ["document", "pdf", "docx", "text", "extract", "parse", "read"]): |
|
|
selected_tools.add("document_parser") |
|
|
|
|
|
if not selected_tools: |
|
|
selected_tools.add("step_by_step_reasoner") |
|
|
selected_tools.add("perplexity_search") |
|
|
|
|
|
return list(selected_tools) |
|
|
|
|
|
|
|
|
registry = ToolRegistry() |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="duckduckgo_search", |
|
|
description="Search the web using DuckDuckGo", |
|
|
category="web", |
|
|
factory=create_duckduckgo_search, |
|
|
keywords=["search", "web", "find", "lookup", "information", "duckduckgo"], |
|
|
input_schema={ |
|
|
"query": { |
|
|
"type": "string", |
|
|
"description": "The search query" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "array", |
|
|
"items": { |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"title": {"type": "string"}, |
|
|
"link": {"type": "string"}, |
|
|
"snippet": {"type": "string"} |
|
|
} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Finding current information about a topic", |
|
|
"Searching for specific facts or data", |
|
|
"Looking up definitions or explanations" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="serper_search", |
|
|
description="Search the web using Serper API", |
|
|
category="web", |
|
|
factory=create_serper_search, |
|
|
keywords=["search", "web", "find", "lookup", "information", "serper", "google"], |
|
|
input_schema={ |
|
|
"query": { |
|
|
"type": "string", |
|
|
"description": "The search query" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "array", |
|
|
"items": { |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"title": {"type": "string"}, |
|
|
"link": {"type": "string"}, |
|
|
"snippet": {"type": "string"} |
|
|
} |
|
|
} |
|
|
}, |
|
|
requires_api_key=True, |
|
|
example_uses=[ |
|
|
"Finding current information about a topic", |
|
|
"Searching for specific facts or data", |
|
|
"Looking up definitions or explanations" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="web_content_extractor", |
|
|
description="Extract content from a web page", |
|
|
category="web", |
|
|
factory=create_web_content_extractor, |
|
|
keywords=["extract", "content", "webpage", "website", "url", "scrape"], |
|
|
input_schema={ |
|
|
"url": { |
|
|
"type": "string", |
|
|
"description": "The URL to extract content from" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"url": {"type": "string"}, |
|
|
"title": {"type": "string"}, |
|
|
"content": {"type": "string"}, |
|
|
"links": {"type": "array"}, |
|
|
"metadata": {"type": "object"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Extracting the main content from an article", |
|
|
"Getting the text content of a webpage", |
|
|
"Analyzing the structure of a webpage" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="web_navigator", |
|
|
description="Navigate and scrape web pages", |
|
|
category="web", |
|
|
factory=create_web_navigator, |
|
|
keywords=["navigate", "browse", "follow", "links", "scrape", "crawl"], |
|
|
input_schema={ |
|
|
"url": { |
|
|
"type": "string", |
|
|
"description": "The URL to navigate to" |
|
|
}, |
|
|
"link_pattern": { |
|
|
"type": "string", |
|
|
"description": "Optional regex pattern to match links", |
|
|
"required": False |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "array", |
|
|
"items": { |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"url": {"type": "string"}, |
|
|
"title": {"type": "string"}, |
|
|
"content": {"type": "string"}, |
|
|
"links": {"type": "array"}, |
|
|
"metadata": {"type": "object"} |
|
|
} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Following links on a webpage", |
|
|
"Crawling a website for specific information", |
|
|
"Navigating through multiple pages" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="step_by_step_reasoner", |
|
|
description="Perform step-by-step reasoning on a question", |
|
|
category="reasoning", |
|
|
factory=create_step_by_step_reasoner, |
|
|
keywords=["reason", "analyze", "step", "logic", "thinking", "solve"], |
|
|
input_schema={ |
|
|
"question": { |
|
|
"type": "string", |
|
|
"description": "The question to reason about" |
|
|
}, |
|
|
"context": { |
|
|
"type": "string", |
|
|
"description": "Optional context information", |
|
|
"required": False |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"steps": { |
|
|
"type": "array", |
|
|
"items": { |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"step_number": {"type": "integer"}, |
|
|
"reasoning": {"type": "string"}, |
|
|
"intermediate_conclusion": {"type": "string"} |
|
|
} |
|
|
} |
|
|
}, |
|
|
"final_conclusion": {"type": "string"}, |
|
|
"confidence": {"type": "number"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Breaking down complex problems", |
|
|
"Analyzing logical arguments", |
|
|
"Solving multi-step reasoning tasks" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="math_calculator", |
|
|
description="Perform mathematical calculations", |
|
|
category="reasoning", |
|
|
factory=create_math_calculator, |
|
|
keywords=["calculate", "compute", "math", "equation", "solve", "formula", "arithmetic"], |
|
|
input_schema={ |
|
|
"expression": { |
|
|
"type": "string", |
|
|
"description": "The mathematical expression to evaluate" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"expression": {"type": "string"}, |
|
|
"result": {"type": "number"}, |
|
|
"steps": {"type": "array"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Evaluating mathematical expressions", |
|
|
"Solving equations", |
|
|
"Performing arithmetic calculations" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="fact_verifier", |
|
|
description="Verify facts and claims", |
|
|
category="reasoning", |
|
|
factory=create_fact_verifier, |
|
|
keywords=["verify", "fact", "check", "claim", "true", "false", "accuracy"], |
|
|
input_schema={ |
|
|
"claim": { |
|
|
"type": "string", |
|
|
"description": "The claim to verify" |
|
|
}, |
|
|
"evidence": { |
|
|
"type": "string", |
|
|
"description": "Optional evidence to use for verification", |
|
|
"required": False |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"is_verifiable": {"type": "boolean"}, |
|
|
"verification_result": {"type": "string"}, |
|
|
"confidence": {"type": "number"}, |
|
|
"reasoning": {"type": "string"}, |
|
|
"missing_information": {"type": "string"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Checking the accuracy of statements", |
|
|
"Verifying claims against evidence", |
|
|
"Determining the reliability of information" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="image_analyzer", |
|
|
description="Analyze and describe images", |
|
|
category="multimodal", |
|
|
factory=create_image_analyzer, |
|
|
keywords=["image", "picture", "photo", "visual", "analyze", "describe"], |
|
|
input_schema={ |
|
|
"image_path": { |
|
|
"type": "string", |
|
|
"description": "Path to the image file" |
|
|
}, |
|
|
"prompt": { |
|
|
"type": "string", |
|
|
"description": "Optional specific prompt for analysis", |
|
|
"required": False |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"description": {"type": "string"}, |
|
|
"subjects": {"type": "array"}, |
|
|
"text_content": {"type": "string"}, |
|
|
"context": {"type": "string"}, |
|
|
"tags": {"type": "array"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Describing the content of images", |
|
|
"Identifying objects in photos", |
|
|
"Extracting text from images", |
|
|
"Analyzing visual content" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="chart_interpreter", |
|
|
description="Interpret charts and graphs", |
|
|
category="multimodal", |
|
|
factory=create_chart_interpreter, |
|
|
keywords=["chart", "graph", "plot", "diagram", "visualization", "interpret"], |
|
|
input_schema={ |
|
|
"chart_path": { |
|
|
"type": "string", |
|
|
"description": "Path to the chart image file" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"chart_type": {"type": "string"}, |
|
|
"title": {"type": "string"}, |
|
|
"axes": {"type": "object"}, |
|
|
"data_points": {"type": "array"}, |
|
|
"trends": {"type": "array"}, |
|
|
"insights": {"type": "string"}, |
|
|
"confidence": {"type": "number"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Interpreting data visualizations", |
|
|
"Extracting insights from charts", |
|
|
"Analyzing trends in graphs", |
|
|
"Understanding statistical plots" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="document_parser", |
|
|
description="Parse and extract information from documents", |
|
|
category="multimodal", |
|
|
factory=create_document_parser, |
|
|
keywords=["document", "pdf", "docx", "text", "extract", "parse"], |
|
|
input_schema={ |
|
|
"document_path": { |
|
|
"type": "string", |
|
|
"description": "Path to the document file" |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"document_path": {"type": "string"}, |
|
|
"file_type": {"type": "string"}, |
|
|
"text_content": {"type": "string"}, |
|
|
"summary": {"type": "string"}, |
|
|
"word_count": {"type": "integer"}, |
|
|
"character_count": {"type": "integer"} |
|
|
} |
|
|
}, |
|
|
example_uses=[ |
|
|
"Extracting text from PDF documents", |
|
|
"Parsing Word documents", |
|
|
"Summarizing document content", |
|
|
"Extracting structured data from documents" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
registry.register_tool( |
|
|
ToolMetadata( |
|
|
name="perplexity_search", |
|
|
description="Search the web and get reasoning using Perplexity API", |
|
|
category="web", |
|
|
factory=create_perplexity_tool, |
|
|
keywords=["search", "web", "find", "lookup", "information", "perplexity", "reasoning", "online"], |
|
|
input_schema={ |
|
|
"query": { |
|
|
"type": "string", |
|
|
"description": "The search query" |
|
|
}, |
|
|
"max_tokens": { |
|
|
"type": "integer", |
|
|
"description": "Maximum number of tokens in the response", |
|
|
"required": False |
|
|
}, |
|
|
"system_prompt": { |
|
|
"type": "string", |
|
|
"description": "Optional system prompt to guide the response", |
|
|
"required": False |
|
|
} |
|
|
}, |
|
|
output_schema={ |
|
|
"type": "object", |
|
|
"properties": { |
|
|
"query": {"type": "string"}, |
|
|
"content": {"type": "string"}, |
|
|
"citations": {"type": "array"}, |
|
|
"usage": {"type": "object"}, |
|
|
"model": {"type": "string"} |
|
|
} |
|
|
}, |
|
|
requires_api_key=True, |
|
|
example_uses=[ |
|
|
"Searching for current information with reasoning", |
|
|
"Getting detailed explanations with citations", |
|
|
"Answering complex questions with web search" |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
def get_tool(name: str) -> Optional[Any]: |
|
|
""" |
|
|
Get a tool instance by name. |
|
|
|
|
|
Args: |
|
|
name: The name of the tool |
|
|
|
|
|
Returns: |
|
|
The tool instance, or None if not found |
|
|
""" |
|
|
return registry.get_tool(name) |
|
|
|
|
|
def list_tools(category: Optional[str] = None) -> List[str]: |
|
|
""" |
|
|
List available tools, optionally filtered by category. |
|
|
|
|
|
Args: |
|
|
category: Optional category to filter by |
|
|
|
|
|
Returns: |
|
|
List of tool names |
|
|
""" |
|
|
return registry.list_tools(category) |
|
|
|
|
|
def select_tools_for_question(question: str) -> List[str]: |
|
|
""" |
|
|
Select appropriate tools based on a question. |
|
|
|
|
|
Args: |
|
|
question: The question to analyze |
|
|
|
|
|
Returns: |
|
|
List of selected tool names |
|
|
""" |
|
|
return registry.get_tools_by_question(question) |
|
|
|
|
|
def get_tool_metadata(name: str) -> Optional[ToolMetadata]: |
|
|
""" |
|
|
Get tool metadata by name. |
|
|
|
|
|
Args: |
|
|
name: The name of the tool |
|
|
|
|
|
Returns: |
|
|
The tool metadata, or None if not found |
|
|
""" |
|
|
return registry.get_tool_metadata(name) |
