feat: Implemented Registry Pattern

This commit is contained in:
2026-02-15 15:58:21 -05:00
parent 548d1eed2b
commit e28c42ddd8
6 changed files with 1276 additions and 4 deletions

View File

@@ -1,15 +1,16 @@
from langgraph.graph import StateGraph, START, END from langgraph.graph import StateGraph, START, END
from langchain_core.messages import SystemMessage, ToolMessage, BaseMessage from langchain_core.messages import SystemMessage, ToolMessage, BaseMessage
from typing import Literal, Callable, Any from typing import Literal, Callable, Any, Union
from .schemas import MessagesState from .schemas import MessagesState
from dataclasses import dataclass from dataclasses import dataclass
from ..tools.tool_registry import ToolRegistry, BaseTool
@dataclass @dataclass
class AgentConfig: class AgentConfig:
"""Configuración inmutable del agente para fácil testing.""" """Configuración inmutable del agente para fácil testing."""
system_prompt: str = ( system_prompt: str = (
"Eres Naliia un agente de servicio al cliente que permite a los usuarios hacer gestion de su agenda." "helpful assistant that can call tools when needed. Always respond with a message. "
) )
max_iterations: int = 10 max_iterations: int = 10
timeout_seconds: float = 30.0 timeout_seconds: float = 30.0
@@ -18,17 +19,37 @@ class AgentConfig:
class Agent: class Agent:
""" """
Agente conversacional basado en LangGraph. Agente conversacional basado en LangGraph.
Soporta tools usando el patrón Registry para una gestión centralizada
y extensible de herramientas.
""" """
def __init__( def __init__(
self, self,
model: Any, model: Any,
config: AgentConfig | None = None, config: AgentConfig | None = None,
tools: list | None = None tools: Union[list, ToolRegistry, None] = None
): ):
"""
Inicializa el agente.
Args:
model: Modelo LLM a usar
config: Configuración del agente (AgentConfig)
tools: Lista de tools, ToolRegistry, o None
- Si es None: sin herramientas
- Si es list: lista de herramientas
- Si es ToolRegistry: se extrae la lista de herramientas
"""
self._model = model self._model = model
self._config = config or AgentConfig() self._config = config or AgentConfig()
# Normalizar tools: convertir ToolRegistry a lista si es necesario
if isinstance(tools, ToolRegistry):
self._tools = tools.get_all_tools()
else:
self._tools = tools or [] self._tools = tools or []
self._tools_by_name: dict[str, Any] = { self._tools_by_name: dict[str, Any] = {
tool.name: tool for tool in self._tools tool.name: tool for tool in self._tools
} }

View File

@@ -0,0 +1,20 @@
"""
Módulo de herramientas para el agente.
Implementa el patrón Registry para registrar y gestionar herramientas
de forma centralizada y extensible.
Uso:
from src.naliiabot.bot.tools import ToolRegistry, BaseTool
registry = ToolRegistry()
registry.register(MiHerramienta())
tools = registry.get_all_tools_as_langchain()
"""
from .tool_registry import ToolRegistry, BaseTool
__all__ = [
"ToolRegistry",
"BaseTool",
]

View File

@@ -0,0 +1,268 @@
"""
Patrón Registry para herramientas del agente.
Este módulo proporciona una forma centralizada y extensible de registrar
y gestionar herramientas usando el patrón Registry.
"""
from abc import ABC, abstractmethod
from typing import Any, Callable, Optional, Dict, List
from langchain_core.tools import Tool
import json
class BaseTool(ABC):
"""
Clase base para todas las herramientas del agente.
Define la interfaz que deben implementar todas las tools.
"""
@property
@abstractmethod
def name(self) -> str:
"""Nombre único de la herramienta."""
pass
@property
@abstractmethod
def description(self) -> str:
"""Descripción de qué hace la herramienta."""
pass
@property
@abstractmethod
def args_schema(self) -> Dict[str, Any]:
"""
Schema de argumentos esperados (JSON Schema).
Returns:
Dict con format JSON Schema que describe los argumentos.
"""
pass
@abstractmethod
def invoke(self, **kwargs) -> Any:
"""
Ejecuta la herramienta.
Args:
**kwargs: Argumentos específicos de la herramienta
Returns:
Resultado de la ejecución
"""
pass
def to_langchain_tool(self) -> Tool:
"""
Convierte la herramienta a formato LangChain Tool.
Returns:
Tool de LangChain lista para ser usada con el modelo
"""
return Tool(
name=self.name,
description=self.description,
func=self.invoke,
args_schema=self.args_schema
)
class ToolRegistry:
"""
Registro centralizado de herramientas.
Implementa el patrón Registry para:
- Registrar nuevas herramientas
- Recuperar herramientas por nombre
- Listar todas las herramientas
- Validar herramientas antes de registrar
Ejemplo de uso:
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(GreeterTool())
tools = registry.get_all_tools()
calc_tool = registry.get_tool("calculator")
"""
def __init__(self):
"""Inicializa el registro vacío."""
self._tools: Dict[str, BaseTool] = {}
def register(self, tool: BaseTool) -> None:
"""
Registra una nueva herramienta.
Args:
tool: Instancia de BaseTool a registrar
Raises:
ValueError: Si la herramienta ya está registrada
TypeError: Si no es una instancia de BaseTool
"""
if not isinstance(tool, BaseTool):
raise TypeError(
f"Tool must be instance of BaseTool, got {type(tool).__name__}"
)
if tool.name in self._tools:
raise ValueError(
f"Tool with name '{tool.name}' is already registered"
)
self._tools[tool.name] = tool
def register_multiple(self, tools: List[BaseTool]) -> None:
"""
Registra múltiples herramientas de una vez.
Args:
tools: Lista de BaseTool a registrar
Raises:
ValueError: Si alguna herramienta ya está registrada
TypeError: Si algún elemento no es BaseTool
"""
for tool in tools:
self.register(tool)
def unregister(self, tool_name: str) -> None:
"""
Desregistra una herramienta.
Args:
tool_name: Nombre de la herramienta a remover
Raises:
KeyError: Si la herramienta no existe
"""
if tool_name not in self._tools:
raise KeyError(f"Tool '{tool_name}' not found in registry")
del self._tools[tool_name]
def get_tool(self, tool_name: str) -> Optional[BaseTool]:
"""
Obtiene una herramienta por nombre.
Args:
tool_name: Nombre de la herramienta
Returns:
La herramienta si existe, None en caso contrario
"""
return self._tools.get(tool_name)
def get_all_tools(self) -> List[BaseTool]:
"""
Obtiene todas las herramientas registradas.
Returns:
Lista de todas las herramientas
"""
return list(self._tools.values())
def get_langchain_tools(self) -> List[Tool]:
"""
Obtiene todas las herramientas en formato LangChain.
Returns:
Lista de herramientas convertidas a LangChain Tool
"""
return [tool.to_langchain_tool() for tool in self._tools.values()]
def has_tool(self, tool_name: str) -> bool:
"""
Verifica si una herramienta está registrada.
Args:
tool_name: Nombre de la herramienta
Returns:
True si la herramienta existe, False en caso contrario
"""
return tool_name in self._tools
def list_tools(self) -> Dict[str, str]:
"""
Lista todas las herramientas con sus descripciones.
Returns:
Dict con nombre de herramienta como key y descripción como value
"""
return {
tool.name: tool.description
for tool in self._tools.values()
}
def clear(self) -> None:
"""Limpia el registro, removiendo todas las herramientas."""
self._tools.clear()
def __len__(self) -> int:
"""Retorna la cantidad de herramientas registradas."""
return len(self._tools)
def __contains__(self, tool_name: str) -> bool:
"""Permite usar 'in' para verificar si una herramienta existe."""
return tool_name in self._tools
def __repr__(self) -> str:
"""Representación en string del registro."""
tool_names = ", ".join(self._tools.keys())
return f"ToolRegistry(tools=[{tool_names}])"
class ToolRegistryBuilder:
"""
Constructor fluido para ToolRegistry.
Permite una forma conveniente de crear y configurar un registro.
Ejemplo:
registry = (ToolRegistryBuilder()
.add(CalculatorTool())
.add(GreeterTool())
.build())
"""
def __init__(self):
"""Inicializa el constructor."""
self._registry = ToolRegistry()
def add(self, tool: BaseTool) -> "ToolRegistryBuilder":
"""
Añade una herramienta al registro.
Args:
tool: Herramienta a añadir
Returns:
Self para permitir encadenamiento
"""
self._registry.register(tool)
return self
def add_multiple(self, tools: List[BaseTool]) -> "ToolRegistryBuilder":
"""
Añade múltiples herramientas.
Args:
tools: Lista de herramientas
Returns:
Self para permitir encadenamiento
"""
self._registry.register_multiple(tools)
return self
def build(self) -> ToolRegistry:
"""
Construye y retorna el ToolRegistry.
Returns:
ToolRegistry configurado
"""
return self._registry

View File

@@ -0,0 +1,54 @@
"""
Implementaciones concretas de herramientas del agente.
Este módulo contiene las herramientas concretas que extienden BaseTool
y pueden ser registradas en el ToolRegistry.
Ejemplo de cómo crear una herramienta personalizada:
from typing import Any, Dict
from .tool_registry import BaseTool
class MiHerramienta(BaseTool):
@property
def name(self) -> str:
return "mi_herramienta"
@property
def description(self) -> str:
return "Descripción de qué hace mi herramienta"
@property
def args_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"param1": {
"type": "string",
"description": "Descripción del parámetro"
}
},
"required": ["param1"]
}
def invoke(self, **kwargs) -> str:
param1 = kwargs.get("param1")
# Implementa la lógica de tu herramienta
return f"Resultado: {param1}"
Luego registra tu herramienta:
from .tool_registry import ToolRegistry
from .tools import MiHerramienta
registry = ToolRegistry()
registry.register(MiHerramienta())
tools = registry.get_all_tools_as_langchain()
"""
from typing import Any, Dict
from .tool_registry import BaseTool
# Define aquí tus herramientas personalizadas
# Ejemplo: class MiHerramienta(BaseTool): ...

View File

@@ -0,0 +1,475 @@
import pytest
from unittest.mock import Mock, MagicMock
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from src.naliiabot.bot.agent.agent import Agent, AgentConfig
"""
Tests robustos para las tools/herramientas de la clase Agent.
Estrategia:
- Tests de ejecución de herramientas individuales
- Tests de manejo de errores
- Tests del nodo de tools
- Tests de integración con el flujo del agente
"""
# ============================================================================
# TESTS DE EJECUCION DE HERRAMIENTAS INDIVIDUALES
# ============================================================================
class TestExecuteTool:
"""Pruebas para la ejecución de herramientas individuales."""
def test_execute_tool_success(self, agent_with_tool):
"""
DADO: Una herramienta registrada en el agente
CUANDO: Se ejecuta exitosamente
ENTONCES: Retorna ToolMessage con status 'success'
"""
tool_call = {
"name": "calculator",
"id": "tool_123",
"args": {"x": 5, "y": 7}
}
result = agent_with_tool._execute_tool(tool_call)
assert isinstance(result, ToolMessage)
assert result.status == "success"
assert result.tool_call_id == "tool_123"
assert result.content == "42"
def test_execute_tool_not_found(self, agent_with_tool):
"""
DADO: Una herramienta que no existe
CUANDO: Se intenta ejecutar
ENTONCES: Retorna ToolMessage con error
"""
tool_call = {
"name": "nonexistent_tool",
"id": "tool_456",
"args": {}
}
result = agent_with_tool._execute_tool(tool_call)
assert isinstance(result, ToolMessage)
assert result.status == "error"
assert "not found" in result.content.lower()
def test_execute_tool_with_exception(self, mock_model, default_config):
"""
DADO: Una herramienta que lanza excepción
CUANDO: Se ejecuta
ENTONCES: Captura el error y retorna ToolMessage con estado error
"""
failing_tool = Mock()
failing_tool.name = "failing_tool"
failing_tool.invoke = Mock(side_effect=ValueError("Tool error"))
agent = Agent(
model=mock_model,
config=default_config,
tools=[failing_tool]
)
tool_call = {
"name": "failing_tool",
"id": "tool_789",
"args": {"param": "value"}
}
result = agent._execute_tool(tool_call)
assert isinstance(result, ToolMessage)
assert result.status == "error"
assert "Tool error" in result.content
def test_execute_tool_with_no_args(self, agent_with_tool):
"""
DADO: Una herramienta sin argumentos en tool_call
CUANDO: Se ejecuta
ENTONCES: Usa dict vacío como argumentos
"""
tool_call = {
"name": "calculator",
"id": "tool_abc"
# Sin 'args'
}
result = agent_with_tool._execute_tool(tool_call)
assert isinstance(result, ToolMessage)
assert result.tool_call_id == "tool_abc"
# ============================================================================
# TESTS DEL NODO DE HERRAMIENTAS (TOOL NODE)
# ============================================================================
class TestToolNode:
"""Pruebas para el nodo que ejecuta herramientas."""
def test_tool_node_with_tool_calls(self, agent_with_tool):
"""
DADO: Un estado con mensaje AIMessage que contiene tool_calls
CUANDO: Se ejecuta el tool_node
ENTONCES: Retorna estado actualizado con ToolMessages
"""
# Mock del modelo para generar tool_call
ai_message = AIMessage(
content="Using calculator",
tool_calls=[
{
"name": "calculator",
"id": "call_1",
"args": {"x": 10}
}
]
)
state = {
"messages": [
HumanMessage(content="Calculate"),
ai_message
]
}
result = agent_with_tool._tool_node(state)
assert "messages" in result
assert len(result["messages"]) == 3 # original 2 + 1 ToolMessage
assert isinstance(result["messages"][-1], ToolMessage)
def test_tool_node_without_tool_calls(self, agent):
"""
DADO: Un estado con mensaje sin tool_calls
CUANDO: Se ejecuta el tool_node
ENTONCES: Retorna estado sin cambios
"""
state = {
"messages": [
HumanMessage(content="Hello"),
AIMessage(content="Hi there!")
]
}
result = agent._tool_node(state)
assert result["messages"] == state["messages"]
def test_tool_node_multiple_tool_calls(self, mock_model, default_config):
"""
DADO: Un AIMessage con múltiples tool_calls
CUANDO: Se ejecuta el tool_node
ENTONCES: Ejecuta todas las herramientas y retorna múltiples ToolMessages
"""
tool1 = Mock()
tool1.name = "tool_a"
tool1.invoke = Mock(return_value="Result A")
tool2 = Mock()
tool2.name = "tool_b"
tool2.invoke = Mock(return_value="Result B")
agent = Agent(
model=mock_model,
config=default_config,
tools=[tool1, tool2]
)
ai_message = AIMessage(
content="Using tools",
tool_calls=[
{"name": "tool_a", "id": "call_1", "args": {}},
{"name": "tool_b", "id": "call_2", "args": {}}
]
)
state = {
"messages": [ai_message]
}
result = agent._tool_node(state)
# 1 AIMessage + 2 ToolMessages
assert len(result["messages"]) == 3
assert all(isinstance(msg, (AIMessage, ToolMessage)) for msg in result["messages"])
def test_tool_node_post_hook_called(self, agent_with_tool):
"""
DADO: Un agente con post_tool_hook registrado
CUANDO: Se ejecuta una herramienta
ENTONCES: Se invoca el hook después de cada ejecución
"""
hook = Mock()
agent_with_tool.set_post_tool_hook(hook)
ai_message = AIMessage(
content="Test",
tool_calls=[
{
"name": "calculator",
"id": "call_1",
"args": {}
}
]
)
state = {
"messages": [ai_message]
}
agent_with_tool._tool_node(state)
# Hook debe ser llamado una vez por herramienta
assert hook.call_count == 1
# ============================================================================
# TESTS DE INTEGRACIÓN CON FLUJO COMPLETO
# ============================================================================
class TestToolsIntegration:
"""Pruebas de integración de tools en el flujo completo del agente."""
def test_agent_invokes_tool_in_flow(self, agent_with_tool):
"""
DADO: Un agente con herramientas configurado
CUANDO: Se invocan herramientas en la conversación
ENTONCES: El estado se actualiza con los resultados
"""
# Mock del modelo para retornar un AIMessage con tool_calls
ai_response = AIMessage(
content="I'll use the calculator",
tool_calls=[
{
"name": "calculator",
"id": "call_x",
"args": {"result": "42"}
}
]
)
agent_with_tool._model.bind_tools().invoke.return_value = ai_response
state = {
"messages": [HumanMessage(content="Calculate 2+2")]
}
result = agent_with_tool.invoke(state)
assert "messages" in result
# Debe contener original + AIMessage + ToolMessage
assert len(result["messages"]) >= 2
def test_tools_are_bound_to_model(self, agent_with_tool):
"""
DADO: Un agente con herramientas
CUANDO: Se invoca el LLM
ENTONCES: Las herramientas están vinculadas al modelo
"""
initial_state = {
"messages": [HumanMessage(content="Hi")]
}
try:
agent_with_tool.invoke(initial_state)
except:
# Es ok si falla por mocks incompletos
pass
# Verificar que bind_tools fue llamado
agent_with_tool._model.bind_tools.assert_called()
def test_tools_property_returns_copy(self, agent_with_tool):
"""
DADO: Un agente con herramientas
CUANDO: Se accede a la propiedad tools
ENTONCES: Retorna una copia de la lista de herramientas
"""
tools_copy = agent_with_tool.tools
assert len(tools_copy) == 1
assert tools_copy[0].name == "calculator"
# Modificar la copia no afecta al agente
tools_copy.clear()
assert len(agent_with_tool.tools) == 1
def test_multiple_tools_independently(self, mock_model, default_config):
"""
DADO: Un agente con múltiples herramientas diferentes
CUANDO: Se ejecutan herramientas
ENTONCES: Cada una se ejecuta de forma independiente
"""
weather_tool = Mock()
weather_tool.name = "get_weather"
weather_tool.invoke = Mock(return_value="Sunny 25°C")
time_tool = Mock()
time_tool.name = "get_time"
time_tool.invoke = Mock(return_value="14:30")
agent = Agent(
model=mock_model,
config=default_config,
tools=[weather_tool, time_tool]
)
# Ejecutar weather_tool
result1 = agent._execute_tool({
"name": "get_weather",
"id": "w1",
"args": {}
})
assert "Sunny" in result1.content
# Ejecutar time_tool
result2 = agent._execute_tool({
"name": "get_time",
"id": "t1",
"args": {}
})
assert "14:30" in result2.content
# ============================================================================
# TESTS DE EDGE CASES
# ============================================================================
class TestToolsEdgeCases:
"""Pruebas de casos extremos y edge cases."""
def test_tool_returns_none(self, mock_model, default_config):
"""
DADO: Una herramienta que retorna None
CUANDO: Se ejecuta
ENTONCES: Se convierte a string adecuadamente
"""
tool = Mock()
tool.name = "none_tool"
tool.invoke = Mock(return_value=None)
agent = Agent(
model=mock_model,
config=default_config,
tools=[tool]
)
result = agent._execute_tool({
"name": "none_tool",
"id": "n1",
"args": {}
})
assert result.content == "None"
assert result.status == "success"
def test_tool_returns_complex_object(self, mock_model, default_config):
"""
DADO: Una herramienta que retorna objeto complejo
CUANDO: Se ejecuta
ENTONCES: Se serializa a string correctamente
"""
tool = Mock()
tool.name = "complex_tool"
tool.invoke = Mock(return_value={"key": "value", "nested": {"data": 123}})
agent = Agent(
model=mock_model,
config=default_config,
tools=[tool]
)
result = agent._execute_tool({
"name": "complex_tool",
"id": "c1",
"args": {}
})
assert isinstance(result.content, str)
assert "key" in result.content
assert "value" in result.content
def test_empty_tools_list(self, mock_model, default_config):
"""
DADO: Un agente sin herramientas
CUANDO: Se intenta ejecutar una herramienta
ENTONCES: Retorna error apropiado
"""
agent = Agent(
model=mock_model,
config=default_config,
tools=[]
)
result = agent._execute_tool({
"name": "any_tool",
"id": "e1",
"args": {}
})
assert result.status == "error"
assert "not found" in result.content.lower()
def test_tool_with_special_characters_in_response(self, mock_model, default_config):
"""
DADO: Una herramienta que retorna strings especiales
CUANDO: Se ejecuta
ENTONCES: Se preservan los caracteres especiales
"""
tool = Mock()
tool.name = "special_tool"
tool.invoke = Mock(return_value="Response with special chars: ñ, é, 中文, 🚀")
agent = Agent(
model=mock_model,
config=default_config,
tools=[tool]
)
result = agent._execute_tool({
"name": "special_tool",
"id": "s1",
"args": {}
})
assert "ñ" in result.content
assert "中文" in result.content
def test_tool_call_missing_name(self, agent):
"""
DADO: Un tool_call sin campo 'name'
CUANDO: Se ejecuta
ENTONCES: Maneja el error gracefully
"""
result = agent._execute_tool({
"id": "m1",
"args": {}
# Sin 'name'
})
assert result.status == "error"
def test_tool_call_empty_args_dict(self, agent_with_tool):
"""
DADO: Un tool_call con args vacío
CUANDO: Se ejecuta
ENTONCES: Invoca la herramienta con dict vacío
"""
agent_with_tool._tools[0].invoke = Mock(return_value="Empty args result")
result = agent_with_tool._execute_tool({
"name": "calculator",
"id": "ea1",
"args": {}
})
agent_with_tool._tools[0].invoke.assert_called_with({})
assert result.status == "success"

434
tests/test_tool_registry.py Normal file
View File

@@ -0,0 +1,434 @@
"""
Tests para el patrón Registry de herramientas.
Prueba la funcionalidad del ToolRegistry y la integración con el Agent.
"""
import pytest
from src.naliiabot.bot.tools.tool_registry import (
ToolRegistry,
ToolRegistryBuilder,
BaseTool
)
from src.naliiabot.bot.tools.tools import (
CalculatorTool,
GreeterTool,
WeatherTool,
TimeTool
)
from src.naliiabot.bot.agent.agent import Agent, AgentConfig
# ============================================================================
# TESTS DEL BASETOOL
# ============================================================================
class TestBaseTool:
"""Pruebas para la clase base BaseTool."""
def test_calculator_tool_properties(self):
"""Verificar que CalculatorTool implementa correctamente BaseTool."""
tool = CalculatorTool()
assert tool.name == "calculator"
assert "mathematical" in tool.description.lower()
assert "properties" in tool.args_schema
assert "required" in tool.args_schema
def test_calculator_add(self):
"""Prueba la operación de suma."""
tool = CalculatorTool()
result = tool.invoke(a=5, b=3, operation="add")
assert "8" in result
assert "add" in result
def test_calculator_subtract(self):
"""Prueba la operación de resta."""
tool = CalculatorTool()
result = tool.invoke(a=10, b=4, operation="subtract")
assert "6" in result
def test_calculator_multiply(self):
"""Prueba la operación de multiplicación."""
tool = CalculatorTool()
result = tool.invoke(a=6, b=7, operation="multiply")
assert "42" in result
def test_calculator_divide(self):
"""Prueba la operación de división."""
tool = CalculatorTool()
result = tool.invoke(a=20, b=4, operation="divide")
assert "5" in result
def test_calculator_divide_by_zero(self):
"""Prueba el manejo de división por cero."""
tool = CalculatorTool()
result = tool.invoke(a=10, b=0, operation="divide")
assert "Error" in result or "zero" in result.lower()
def test_greeter_tool_formal(self):
"""Prueba el saludo formal."""
tool = GreeterTool()
result = tool.invoke(name="Alice", tone="formal")
assert "Alice" in result
assert "pleasure" in result.lower()
def test_greeter_tool_casual(self):
"""Prueba el saludo casual."""
tool = GreeterTool()
result = tool.invoke(name="Bob", tone="casual")
assert "Bob" in result
assert "up" in result.lower()
def test_greeter_tool_friendly(self):
"""Prueba el saludo amigable."""
tool = GreeterTool()
result = tool.invoke(name="Charlie", tone="friendly")
assert "Charlie" in result
def test_weather_tool(self):
"""Prueba la herramienta de clima."""
tool = WeatherTool()
result = tool.invoke(location="Madrid", units="celsius")
assert "Madrid" in result
assert "Weather" in result or "temperature" in result.lower()
def test_time_tool_24h_format(self):
"""Prueba la herramienta de hora en formato 24h."""
tool = TimeTool()
result = tool.invoke(format="24h", timezone="UTC")
assert "UTC" in result
assert ":" in result # Contiene al menos un separador de hora
def test_time_tool_12h_format(self):
"""Prueba la herramienta de hora en formato 12h."""
tool = TimeTool()
result = tool.invoke(format="12h", timezone="EST")
assert "EST" in result
assert ("AM" in result or "PM" in result)
# ============================================================================
# TESTS DEL TOOL REGISTRY
# ============================================================================
class TestToolRegistry:
"""Pruebas para el ToolRegistry."""
def test_registry_empty_on_init(self):
"""Verificar que el registro inicia vacío."""
registry = ToolRegistry()
assert len(registry) == 0
assert registry.list_tools() == {}
def test_register_single_tool(self):
"""Registrar una sola herramienta."""
registry = ToolRegistry()
tool = CalculatorTool()
registry.register(tool)
assert len(registry) == 1
assert registry.has_tool("calculator")
assert registry.get_tool("calculator") == tool
def test_register_multiple_tools(self):
"""Registrar múltiples herramientas."""
registry = ToolRegistry()
tools = [CalculatorTool(), GreeterTool(), WeatherTool()]
registry.register_multiple(tools)
assert len(registry) == 3
assert registry.has_tool("calculator")
assert registry.has_tool("greeter")
assert registry.has_tool("get_weather")
def test_register_duplicate_tool_raises_error(self):
"""Intentar registrar una herramienta duplicada debe fallar."""
registry = ToolRegistry()
registry.register(CalculatorTool())
with pytest.raises(ValueError, match="already registered"):
registry.register(CalculatorTool())
def test_register_non_basetool_raises_error(self):
"""Registrar algo que no es BaseTool debe fallar."""
registry = ToolRegistry()
with pytest.raises(TypeError, match="must be instance of BaseTool"):
registry.register("not a tool")
def test_get_tool_existing(self):
"""Obtener una herramienta existente."""
registry = ToolRegistry()
tool = CalculatorTool()
registry.register(tool)
retrieved = registry.get_tool("calculator")
assert retrieved == tool
def test_get_tool_non_existing(self):
"""Obtener una herramienta que no existe retorna None."""
registry = ToolRegistry()
result = registry.get_tool("nonexistent")
assert result is None
def test_get_all_tools(self):
"""Obtener todas las herramientas."""
registry = ToolRegistry()
tools = [CalculatorTool(), GreeterTool()]
registry.register_multiple(tools)
all_tools = registry.get_all_tools()
assert len(all_tools) == 2
assert all(isinstance(t, BaseTool) for t in all_tools)
def test_unregister_tool(self):
"""Desregistrar una herramienta."""
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.unregister("calculator")
assert len(registry) == 0
assert not registry.has_tool("calculator")
def test_unregister_non_existing_raises_error(self):
"""Desregistrar una herramienta que no existe debe fallar."""
registry = ToolRegistry()
with pytest.raises(KeyError, match="not found"):
registry.unregister("nonexistent")
def test_list_tools(self):
"""Listar todas las herramientas con descripciones."""
registry = ToolRegistry()
registry.register(CalculatorTool())
registry.register(GreeterTool())
tools_list = registry.list_tools()
assert isinstance(tools_list, dict)
assert "calculator" in tools_list
assert "greeter" in tools_list
assert len(tools_list) == 2
def test_clear_registry(self):
"""Limpiar el registro."""
registry = ToolRegistry()
registry.register_multiple([
CalculatorTool(),
GreeterTool(),
WeatherTool()
])
registry.clear()
assert len(registry) == 0
def test_contains_operator(self):
"""Usar el operador 'in' para verificar existencia."""
registry = ToolRegistry()
registry.register(CalculatorTool())
assert "calculator" in registry
assert "greeter" not in registry
def test_get_langchain_tools(self):
"""Obtener herramientas en formato LangChain."""
registry = ToolRegistry()
registry.register_multiple([
CalculatorTool(),
GreeterTool()
])
langchain_tools = registry.get_langchain_tools()
assert len(langchain_tools) == 2
assert all(hasattr(t, 'name') for t in langchain_tools)
assert all(hasattr(t, 'description') for t in langchain_tools)
def test_registry_repr(self):
"""Prueba la representación en string del registro."""
registry = ToolRegistry()
registry.register(CalculatorTool())
repr_str = repr(registry)
assert "ToolRegistry" in repr_str
assert "calculator" in repr_str
# ============================================================================
# TESTS DEL TOOL REGISTRY BUILDER
# ============================================================================
class TestToolRegistryBuilder:
"""Pruebas para el ToolRegistryBuilder."""
def test_builder_fluent_interface(self):
"""Verificar la interfaz fluida del builder."""
registry = (ToolRegistryBuilder()
.add(CalculatorTool())
.add(GreeterTool())
.build())
assert isinstance(registry, ToolRegistry)
assert len(registry) == 2
def test_builder_add_multiple(self):
"""Añadir múltiples herramientas con builder."""
registry = (ToolRegistryBuilder()
.add_multiple([
CalculatorTool(),
GreeterTool(),
WeatherTool()
])
.build())
assert len(registry) == 3
def test_builder_mixed_operations(self):
"""Mezclar operaciones add y add_multiple."""
registry = (ToolRegistryBuilder()
.add(CalculatorTool())
.add_multiple([GreeterTool(), WeatherTool()])
.add(TimeTool())
.build())
assert len(registry) == 4
assert registry.has_tool("calculator")
assert registry.has_tool("greeter")
assert registry.has_tool("get_weather")
assert registry.has_tool("get_time")
# ============================================================================
# TESTS DE INTEGRACIÓN CON AGENT
# ============================================================================
class TestAgentWithRegistry:
"""Pruebas de integración del Agent con ToolRegistry."""
def test_agent_accepts_tool_registry(self, mock_model):
"""El agente debe aceptar un ToolRegistry."""
registry = ToolRegistry()
registry.register_multiple([
CalculatorTool(),
GreeterTool()
])
agent = Agent(model=mock_model, tools=registry)
assert len(agent.tools) == 2
def test_agent_registry_list_conversion(self, mock_model):
"""El agente debe convertir ToolRegistry a lista internamente."""
registry = ToolRegistry()
registry.register(CalculatorTool())
agent = Agent(model=mock_model, tools=registry)
# Internamente debe tener la lista de tools
assert len(agent._tools) == 1
def test_agent_with_builder_registry(self, mock_model):
"""Usar Agent con ToolRegistry creado con Builder."""
registry = (ToolRegistryBuilder()
.add(CalculatorTool())
.add(WeatherTool())
.build())
agent = Agent(model=mock_model, tools=registry)
assert len(agent.tools) == 2
def test_agent_still_accepts_list_of_tools(self, mock_model):
"""El agente debe seguir aceptando listas de tools."""
tools = [CalculatorTool(), GreeterTool()]
agent = Agent(model=mock_model, tools=tools)
assert len(agent.tools) == 2
def test_agent_none_tools_default(self, mock_model):
"""El agente debe funcionar sin tools."""
agent = Agent(model=mock_model, tools=None)
assert len(agent.tools) == 0
# ============================================================================
# TESTS DE CASOS DE USO PRÁCTICOS
# ============================================================================
class TestPracticalUseCases:
"""Casos de uso prácticos del ToolRegistry."""
def test_build_customer_service_registry(self):
"""Construir un registro para un agente de servicio al cliente."""
registry = (ToolRegistryBuilder()
.add(GreeterTool())
.add(WeatherTool())
.add(TimeTool())
.build())
assert registry.has_tool("greeter")
assert len(registry) == 3
def test_dynamic_tool_addition(self):
"""Añadir herramientas dinámicamente."""
registry = ToolRegistry()
# Fase 1: registrar herramientas básicas
registry.register(CalculatorTool())
assert len(registry) == 1
# Fase 2: registrar herramientas adicionales
registry.register(GreeterTool())
assert len(registry) == 2
# Fase 3: remover si es necesario
registry.unregister("calculator")
assert len(registry) == 1
def test_tool_registry_documentation(self):
"""Generar documentación de herramientas disponibles."""
registry = (ToolRegistryBuilder()
.add_multiple([
CalculatorTool(),
GreeterTool(),
WeatherTool(),
TimeTool()
])
.build())
tools_doc = registry.list_tools()
assert len(tools_doc) == 4
for tool_name, description in tools_doc.items():
assert isinstance(tool_name, str)
assert isinstance(description, str)
assert len(description) > 0