Skip to main content

Agent Tools

Tool definitions and registry system for RAG agents, providing retrieval, web search, escalation, and custom tool capabilities.

Core Classes​

RetrievalTool​

Description: Tool for retrieving documents from vector stores using hybrid retrieval

Parameters:

  • retriever (HybridRetriever): Hybrid retriever instance
  • name (str): Tool name (default: "retrieval")
  • description (str): Tool description

Returns: LangChain BaseTool instance

Example:

from recoagent.agents.tools import RetrievalTool
from recoagent.rag.retrievers import HybridRetriever

# Create retriever
retriever = HybridRetriever(
    vector_retriever=vector_retriever,
    bm25_retriever=bm25_retriever
)

# Create tool
retrieval_tool = RetrievalTool(retriever=retriever)

# Use in agent
result = retrieval_tool.run("machine learning algorithms")

WebSearchTool​

Description: Tool for performing web searches using external APIs

Parameters:

  • api_key (str): API key for search service
  • search_engine (str): Search engine to use
  • name (str): Tool name (default: "web_search")
  • description (str): Tool description

Returns: LangChain BaseTool instance

Example:

from recoagent.agents.tools import WebSearchTool

# Create web search tool
web_tool = WebSearchTool(
    api_key="your_api_key",
    search_engine="google"
)

# Use in agent
result = web_tool.run("latest AI research papers")

EscalateTool​

Description: Tool for escalating queries to human agents

Parameters:

  • escalation_endpoint (str): Endpoint for escalation
  • name (str): Tool name (default: "escalate")
  • description (str): Tool description

Returns: LangChain BaseTool instance

Example:

from recoagent.agents.tools import EscalateTool

# Create escalation tool
escalate_tool = EscalateTool(
    escalation_endpoint="https://api.company.com/escalate"
)

# Use in agent
result = escalate_tool.run("complex technical question")

ToolRegistry​

Description: Registry for managing and organizing agent tools

Parameters:

  • tools (List[BaseTool]): List of tools to register
  • categories (Dict[str, List[str]]): Tool categories

Returns: ToolRegistry instance

Example:

from recoagent.agents.tools import ToolRegistry, RetrievalTool, WebSearchTool

# Create tools
retrieval_tool = RetrievalTool(retriever=retriever)
web_tool = WebSearchTool(api_key="key")

# Create registry
registry = ToolRegistry(
    tools=[retrieval_tool, web_tool],
    categories={
        "retrieval": ["retrieval"],
        "search": ["web_search"],
        "escalation": ["escalate"]
    }
)

# Get tools by category
retrieval_tools = registry.get_tools_by_category("retrieval")

Usage Examples​

Basic Tool Setup​

from recoagent.agents.tools import RetrievalTool, WebSearchTool, ToolRegistry
from recoagent.rag.retrievers import HybridRetriever

# Create retrievers
vector_retriever = VectorRetriever(vector_store)
bm25_retriever = BM25Retriever(index)
hybrid_retriever = HybridRetriever(
    vector_retriever=vector_retriever,
    bm25_retriever=bm25_retriever
)

# Create tools
retrieval_tool = RetrievalTool(retriever=hybrid_retriever)
web_tool = WebSearchTool(api_key="your_key")

# Register tools
registry = ToolRegistry(tools=[retrieval_tool, web_tool])

Custom Tool Creation​

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from recoagent.agents.tools import ToolRegistry

class CustomInput(BaseModel):
    """Input for custom tool."""
    text: str = Field(description="Text to process")
    operation: str = Field(description="Operation to perform")

class CustomTool(BaseTool):
    """Custom tool for text processing."""
    
    name = "custom_processor"
    description = "Process text with custom operations"
    args_schema = CustomInput
    
    def _run(self, text: str, operation: str) -> str:
        """Execute custom processing."""
        if operation == "uppercase":
            return text.upper()
        elif operation == "lowercase":
            return text.lower()
        else:
            return text
    
    async def _arun(self, text: str, operation: str) -> str:
        """Async execution."""
        return self._run(text, operation)

# Create and register custom tool
custom_tool = CustomTool()
registry = ToolRegistry(tools=[custom_tool])

Tool with Reranking​

from recoagent.agents.tools import RetrievalTool
from recoagent.rag.rerankers import CrossEncoderReranker

# Create reranker
reranker = CrossEncoderReranker(
    model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"
)

# Create retrieval tool with reranking
retrieval_tool = RetrievalTool(
    retriever=hybrid_retriever,
    reranker=reranker
)

# Tool will automatically rerank results
result = retrieval_tool.run("query", k=10)  # Returns top 5 after reranking

Advanced Tool Registry​

from recoagent.agents.tools import ToolRegistry

# Create comprehensive registry
registry = ToolRegistry(
    tools=[retrieval_tool, web_tool, escalate_tool, custom_tool],
    categories={
        "retrieval": ["retrieval"],
        "search": ["web_search"],
        "escalation": ["escalate"],
        "custom": ["custom_processor"]
    }
)

# Get tools by category
retrieval_tools = registry.get_tools_by_category("retrieval")

# Get tool by name
specific_tool = registry.get_tool("retrieval")

# List all tools
all_tools = registry.list_tools()

# Add new tool
registry.add_tool(new_tool, category="custom")

API Reference​

RetrievalTool Methods​

run(query: str, k: int = 5) -> Dict​

Execute retrieval with given query

Parameters:

  • query (str): Search query
  • k (int): Number of documents to retrieve

Returns: Dictionary with documents and metadata

arun(query: str, k: int = 5) -> Dict​

Async execution of retrieval

Parameters:

  • query (str): Search query
  • k (int): Number of documents to retrieve

Returns: Dictionary with documents and metadata

WebSearchTool Methods​

run(query: str, num_results: int = 5) -> Dict​

Execute web search

Parameters:

  • query (str): Search query
  • num_results (int): Number of results

Returns: Dictionary with search results and metadata

ToolRegistry Methods​

get_tools_by_category(category: str) -> List[BaseTool]​

Get tools by category

Parameters:

  • category (str): Tool category

Returns: List of tools in category

get_tool(name: str) -> BaseTool​

Get specific tool by name

Parameters:

  • name (str): Tool name

Returns: Tool instance

add_tool(tool: BaseTool, category: str) -> None​

Add new tool to registry

Parameters:

  • tool (BaseTool): Tool to add
  • category (str): Tool category

list_tools() -> List[BaseTool]​

List all registered tools

Returns: List of all tools

See Also​