Saltar al contenido
> 💻 🧠 Código 1001 > LLM: artículos, materiales prácticos y recursos > Guía completa: any-agent + Tavily para crear agentes de búsqueda inteligentes

Guía completa: any-agent + Tavily para crear agentes de búsqueda inteligentes

  • por

En el mundo moderno de los agentes de IA, uno de los principales problemas es el acceso a información actualizada. La mayoría de los modelos de lenguaje están entrenados con datos hasta una fecha de corte específica y no pueden proporcionar información sobre los últimos acontecimientos, precios actuales o noticias.

En este artículo, exploraremos cómo resolver este problema utilizando la combinación de any-agent (una biblioteca universal para trabajar con agentes de Mozilla.ai) y Tavily (un motor de búsqueda especializado para IA).

In Questo Articolo
  1. Parte 1: Introducción a any-agent
  2. Parte 2: Introducción a Tavily
  3. Parte 3: Integración de any-agent con Tavily
  4. Parte 4: Ejemplo completo
  5. Parte 5: Funcionalidades avanzadas
  6. Parte 6: Mejores prácticas y recomendaciones
  7. Parte 7: Casos de uso prácticos
  8. Conclusión
  9. Apéndice A: Código completo del agente de producción
  10. Apéndice B: Archivos de configuración
  11. Apéndice B: Pruebas
  12. Recomendaciones finales
  13. Observaciones finales

Parte 1: Introducción a any-agent

¿Qué es any-agent?

any-agent es una biblioteca de Python de Mozilla.ai que proporciona una interfaz unificada para trabajar con varios frameworks de agentes. La idea principal es permitir a los desarrolladores cambiar fácilmente entre diferentes frameworks de agentes (AutoGen, CrewAI, LangGraph, etc.) sin reescribir código.

Ventajas clave:

  • API unificada: una interfaz para todos los frameworks
  • Registro estandarizado: trazas idénticas independientemente del framework
  • Comparación sencilla: posibilidad de probar diferentes configuraciones de agentes
  • Extensibilidad: soporte para añadir nuevos frameworks

Requisitos del sistema e instalación

<em># Requisitos del sistema</em>
<em># Python 3.11+</em>

<em># Instalación básica (solo TinyAgent)</em>
pip install any-agent

<em># Instalación con soporte para todos los frameworks</em>
pip install any-agent[all]

<em># O con un framework específico</em>
pip install any-agent[langchain]

Primer ejemplo de agente

from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent

<em># Creación de un agente simple</em>
agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[],
    system_prompt="Eres un asistente útil"
)

<em># Ejecución del agente</em>
result = agent.run("¡Hola! Cuéntame sobre ti.")
print(result)

Importante para Jupyter Notebook:

import nest_asyncio
nest_asyncio.apply()
<em># Ahora se puede usar any-agent</em>

Parte 2: Introducción a Tavily

¿Qué es Tavily?

Tavily es un motor de búsqueda creado específicamente para agentes de IA y sistemas RAG. A diferencia de los motores de búsqueda tradicionales, Tavily está optimizado para proporcionar resultados estructurados y fácticos que son fácilmente procesables por los modelos de lenguaje.

Características clave de Tavily:

  • Optimización para IA: resultados adaptados para LLM
  • Enfoque en hechos: énfasis en la precisión y verificabilidad
  • Alta velocidad: respuestas rápidas en tiempo real
  • Salida estructurada: formato JSON con metadatos

Registro y obtención de la clave API

  1. Vaya a https://app.tavily.com/
  2. Cree una cuenta
  3. Obtenga su clave API en su área personal

Uso básico de Tavily

<em># Instalación del cliente</em>
pip install tavily-python

from tavily import TavilyClient

<em># Inicialización del cliente</em>
client = TavilyClient(api_key="tvly-YOUR_API_KEY")

<em># Búsqueda simple</em>
response = client.search("Últimas noticias sobre Python 3.12")

print(f"Consulta: {response['query']}")
print(f"Respuesta: {response['answer']}")
print(f"Fuentes: {len(response['results'])} resultados")

Estructura de la respuesta de Tavily

{
  "query": "Su consulta de búsqueda",
  "answer": "Respuesta breve basada en la información encontrada",
  "images": [],
  "results": [
    {
      "title": "Título de la fuente",
      "url": "https://example.com",
      "content": "Contenido...",
      "score": 0.81025416,
      "favicon": "https://example.com/favicon.png"
    }
  ],
  "auto_parameters": {
    "topic": "general",
    "search_depth": "basic"
  },
  "response_time": "1.67"
}

Parte 3: Integración de any-agent con Tavily

Enfoque 1: Creación de una herramienta personalizada

La forma más flexible es crear su propia herramienta de búsqueda:

import os
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
from tavily import TavilyClient
from typing import Dict, Optional

class TavilySearchTool:
    """Herramienta personalizada para buscar a través de Tavily"""
    
    def __init__(self, api_key: str):
        self.client = TavilyClient(api_key=api_key)
        self.name = "tavily_search"
        self.description = "Búsqueda de información actualizada en internet"
    
    def __call__(self, query: str, search_depth: str = "basic") -> Dict:
        """Realizar búsqueda"""
        try:
            response = self.client.search(
                query=query,
                search_depth=search_depth
            )
            return {
                "success": True,
                "data": response
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

<em># Creación de un agente con Tavily</em>
tavily_tool = TavilySearchTool(api_key="tvly-YOUR_API_KEY")

agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[tavily_tool],
    system_prompt="""
    Eres un asistente de IA con acceso a información actualizada de internet.
    Utiliza la herramienta tavily_search para obtener datos recientes.
    Siempre indica las fuentes de información en tus respuestas.
    """
)

<em># Uso</em>
result = agent.run("Encuentra las últimas noticias sobre el desarrollo de IA de OpenAI")
print(result)

Enfoque 2: Integración a través de LangChain

Si utilizas un framework compatible con LangChain:

from langchain_tavily import TavilySearchAPIWrapper
from any_agent import AnyAgent

<em># Configuración de la búsqueda a través de LangChain</em>
search_tool = TavilySearchAPIWrapper(
    tavily_api_key="tvly-YOUR_API_KEY"
)

<em># Creación del agente</em>
agent = AnyAgent(
    framework=YourLangChainFramework,
    model="gpt-4",
    tools=[search_tool],
    system_prompt="Asistente con acceso a búsqueda en tiempo real"
)

Parte 4: Ejemplo completo

Crearemos una aplicación completa para trabajar con agentes de búsqueda:

<em>## \file src/agents/smart_search_agent.py</em>
<em># -*- coding: utf-8 -*-</em>
<em>#! .pyenv/bin/python3</em>

"""
Módulo de agente de búsqueda inteligente
======================================

Integración de any-agent con Tavily para crear un agente
con acceso a información actualizada de internet.
"""

from typing import Dict, List, Optional
from dataclasses import dataclass
import json
from datetime import datetime

from tavily import TavilyClient
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent


@dataclass
class SearchResult:
    """Estructura del resultado de búsqueda"""
    query: str
    answer: str
    sources: List[Dict]
    timestamp: str
    response_time: float


class SmartSearchAgent:
    """
    Agente de búsqueda inteligente con integración de Tavily
    
    El agente es capaz de realizar búsquedas de información actualizada,
    analizar los resultados y proporcionar respuestas estructuradas.
    """
    
    def __init__(
        self,
        tavily_api_key: str, 
        model: str = "gpt-4",
        search_depth: str = "basic"
    ):
        """
        Inicialización del agente
        
        Args:
            tavily_api_key: Clave API para Tavily
            model: Modelo para el agente
            search_depth: Profundidad de búsqueda ('basic' o 'advanced')
        """
        self.tavily_client = TavilyClient(api_key=tavily_api_key)
        self.search_depth = search_depth
        self.agent = self._create_agent(model)
        self.search_history: List[SearchResult] = []
    
    def _create_agent(self, model: str) -> AnyAgent:
        """Creación de un agente con herramientas configuradas"""
        return AnyAgent(
            framework=TinyAgent,
            model=model,
            tools=[self._search_tool],
            system_prompt=self._get_system_prompt()
        )
    
    def _get_system_prompt(self) -> str:
        """Prompt del sistema para el agente"""
        return """
        Eres un asistente de IA experto con acceso a información actualizada de internet.
        
        Tus capacidades:
        - Búsqueda de información actualizada a través de Tavily
        - Análisis y síntesis de datos de múltiples fuentes
        - Provisión de respuestas fácticas con indicación de fuentes
        
        Principios de funcionamiento:
        1. Siempre utiliza la búsqueda para obtener información actualizada
        2. Indica las fuentes para todas las afirmaciones fácticas
        3. Si la información es contradictoria, señálalo
        4. Sé preciso y objetivo
        5. Estructura las respuestas para una mejor comprensión
        """
    
    def _search_tool(self, query: str) -> Dict:
        """Herramienta de búsqueda para el agente
        
        Args:
            query: Consulta de búsqueda
            
        Returns:
            Resultados de búsqueda en formato estructurado
        """
        try:
            start_time = datetime.now()
            
            response = self.tavily_client.search(
                query=query,
                search_depth=self.search_depth
            )
            
            end_time = datetime.now()
            response_time = (end_time - start_time).total_seconds()
            
            <em># Guardar el resultado en el historial</em>
            search_result = SearchResult(
                query=query,
                answer=response.get('answer', ''),
                sources=response.get('results', []),
                timestamp=datetime.now().isoformat(),
                response_time=response_time
            )
            self.search_history.append(search_result)
            
            return {
                "success": True,
                "query": query,
                "answer": response.get('answer', ''),
                "sources": response.get('results', []),
                "source_count": len(response.get('results', [])),
                "response_time": response.get('response_time', response_time)
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": f"Error de búsqueda: {str(e)}",
                "query": query
            }
    
    def search(self, query: str) -> str:
        """
        Ejecución de una consulta de búsqueda a través del agente
        
        Args:
            query: Consulta del usuario
            
        Returns:
            Respuesta del agente con información actualizada
        """
        try:
            result = self.agent.run(query)
            return result
        except Exception as e:
            return f"Ocurrió un error al procesar la solicitud: {str(e)}"
    
    def get_search_history(self) -> List[SearchResult]:
        """Obtener el historial de consultas de búsqueda"""
        return self.search_history.copy()
    
    def clear_history(self):
        """Borrar el historial de búsqueda"""
        self.search_history.clear()
    
    def export_history(self, filename: str):
        """
        Exportar el historial de búsqueda a un archivo JSON
        
        Args:
            filename: Nombre del archivo para exportar
        """
        try:
            history_data = [
                {
                    "query": result.query,
                    "answer": result.answer,
                    "sources": result.sources,
                    "timestamp": result.timestamp,
                    "response_time": result.response_time
                }
                for result in self.search_history
            ]
            
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(history_data, f, ensure_ascii=False, indent=2)
            
            print(f"Historial exportado a {filename}")
            
        except Exception as e:
            print(f"Error de exportación: {str(e)}")


<em># Ejemplo de uso</em>
def main():
    """Demostración del funcionamiento del agente de búsqueda inteligente"""
    
    <em># Inicialización del agente</em>
    agent = SmartSearchAgent(
        tavily_api_key="tvly-YOUR_API_KEY",
        model="gpt-4",
        search_depth="basic"
    )
    
    <em># Ejemplos de consultas</em>
    queries = [
        "Últimas noticias sobre el desarrollo de GPT-5",
        "Precios actuales de Bitcoin y Ethereum",
        "¿Qué hay de nuevo en Python 3.12?",
        "Últimos avances en computadoras cuánticas"
    ]
    
    print("=== Demostración del agente de búsqueda inteligente ===\n")
    
    for i, query in enumerate(queries, 1):
        print(f"Consulta {i}: {query}")
        print("-" * 50)
        
        result = agent.search(query)
        print(result)
        print("\n" + "="*80 + "\n")
    
    <em># Exportar historial</em>
    agent.export_history("search_history.json")
    
    <em># Estadísticas</em>
    history = agent.get_search_history()
    print(f"Total de consultas realizadas: {len(history)}")
    
    if history:
        avg_response_time = sum(r.response_time for r in history) / len(history)
        print(f"Tiempo promedio de respuesta: {avg_response_time:.2f} seg")


if __name__ == "__main__":
    main()

Parte 5: Funcionalidades avanzadas

Sistemas Multi-Agente

any-agent soporta la creación de sistemas Multi-Agente, donde los agentes pueden interactuar entre sí:

class ResearchTeam:
    """Equipo de agentes de investigación"""
    
    def __init__(self, tavily_api_key: str):
        <em># Agente investigador</em>
        self.researcher = SmartSearchAgent(
            tavily_api_key=tavily_api_key,
            model="gpt-4"
        )
        
        <em># Agente analista</em>
        self.analyst = AnyAgent(
            framework=TinyAgent,
            model="gpt-4",
            system_prompt="""
            Eres un experto analista. Tu tarea es analizar 
            la información obtenida y sacar conclusiones.
            """
        )
        
        <em># Agente escritor</em>
        self.writer = AnyAgent(
            framework=TinyAgent,
            model="gpt-4",
            system_prompt="""
            Eres un escritor profesional. Crea 
            informes estructurados basados en el análisis.
            """
        )
    
    def research_topic(self, topic: str) -> str:
        """Investigación compleja de un tema"""
        
        <em># Etapa 1: Búsqueda de información</em>
        research_data = self.researcher.search(f"Información detallada sobre {topic}")
        
        <em># Etapa 2: Análisis de datos</em>
        analysis = self.analyst.run(f"Analiza esta información: {research_data}")
        
        <em># Etapa 3: Redacción del informe</em>
        report = self.writer.run(f"Crea un informe estructurado basado en el análisis: {analysis}")
        
        return report

<em># Uso</em>
team = ResearchTeam("tvly-YOUR_API_KEY")
report = team.research_topic("Impacto de la IA en el mercado laboral en 2024")
print(report)

Configuración de la profundidad de búsqueda

Tavily ofrece diferentes niveles de profundidad de búsqueda:

<em># Búsqueda rápida (basic)</em>
basic_agent = SmartSearchAgent(
    tavily_api_key="your-key",
    search_depth="basic"  <em># Más rápido, menos fuentes</em>
)

<em># Búsqueda profunda (advanced)</em>
advanced_agent = SmartSearchAgent(
    tavily_api_key="your-key", 
    search_depth="advanced"  <em># Más lento, más fuentes</em>
)

Filtrado de resultados

def create_specialized_search_tool(domains: List[str] = None):
    """Creación de una herramienta de búsqueda especializada"""
    
    def search_tool(query: str) -> Dict:
        client = TavilyClient(api_key="your-key")
        
        search_params = {
            "query": query,
            "search_depth": "advanced"
        }
        
        <em># Adición de filtros por dominios</em>
        if domains:
            search_params["include_domains"] = domains
        
        response = client.search(**search_params)
        return response
    
    return search_tool

<em># Creación de un agente para buscar solo en fuentes científicas</em>
science_agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[create_specialized_search_tool([
        "arxiv.org", 
        "nature.com", 
        "science.org"
    ])],
    system_prompt="Experto en información científica"
)

Parte 6: Mejores prácticas y recomendaciones

Seguridad y limitaciones

⚠️ Advertencias importantes:

  1. Recursos computacionales: Los sistemas de agentes requieren significativamente más recursos que las simples llamadas a LLM
  2. Seguridad: Los agentes pueden realizar acciones impredecibles
  3. Costo: Cada consulta de búsqueda y llamada a LLM se factura

Optimización del rendimiento

class OptimizedSearchAgent:
    """Agente de búsqueda optimizado"""
    
    def __init__(self, tavily_api_key: str):
        self.client = TavilyClient(api_key=tavily_api_key)
        self.cache = {}  <em># Caché para solicitudes repetidas</em>
        self.rate_limit_delay = 1  <em># Retraso entre solicitudes</em>
    
    def search_with_cache(self, query: str) -> Dict:
        """Búsqueda con caché de resultados"""
        
        <em># Verificación de caché</em>
        if query in self.cache:
            print(f"Resultado obtenido de la caché para: {query}")
            return self.cache[query]
        
        <em># Realizar búsqueda</em>
        time.sleep(self.rate_limit_delay)  <em># Cumplimiento de límites de tasa</em>
        result = self.client.search(query)
        
        <em># Guardar en caché</em>
        self.cache[query] = result
        
        return result

Manejo de errores

class RobustSearchAgent:
    """Agente de búsqueda resistente a errores"""
    
    def __init__(self, tavily_api_key: str, max_retries: int = 3):
        self.client = TavilyClient(api_key=tavily_api_key)
        self.max_retries = max_retries
    
    def search_with_retry(self, query: str) -> Dict:
        """Búsqueda con reintentos"""
        
        for attempt in range(self.max_retries):
            try:
                return self.client.search(query)
            
            except Exception as e:
                print(f"Intento {attempt + 1} fallido: {str(e)}")
                
                if attempt == self.max_retries - 1:
                    return {
                        "error": f"La búsqueda falló después de {self.max_retries} intentos",
                        "query": query
                    }
                
                time.sleep(2 ** attempt)  <em># Retraso exponencial</em>

Monitoreo y registro

import logging
from datetime import datetime

class MonitoredSearchAgent:
    """Agente con monitoreo y registro"""
    
    def __init__(self, tavily_api_key: str):
        self.client = TavilyClient(api_key=tavily_api_key)
        self.setup_logging()
        self.stats = {
            "total_searches": 0,
            "successful_searches": 0,
            "failed_searches": 0,
            "total_response_time": 0
        }
    
    def setup_logging(self):
        """Configuración del sistema de registro"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('search_agent.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('SearchAgent')
    
    def search_with_monitoring(self, query: str) -> Dict:
        """Búsqueda con monitoreo completo"""
        start_time = datetime.now()
        self.stats["total_searches"] += 1
        
        try:
            self.logger.info(f"Iniciando búsqueda: {query}")
            
            result = self.client.search(query)
            
            end_time = datetime.now()
            response_time = (end_time - start_time).total_seconds()
            
            self.stats["successful_searches"] += 1
            self.stats["total_response_time"] += response_time
            
            self.logger.info(f"Búsqueda exitosa en {response_time:.2f}s: {query}")
            
            return result
            
        except Exception as e:
            self.stats["failed_searches"] += 1
            self.logger.error(f"Error de búsqueda {query}: {str(e)}")
            
            return {"error": str(e), "query": query}
    
    def get_statistics(self) -> Dict:
        """Obtener estadísticas de operación"""
        success_rate = (
            self.stats["successful_searches"] / self.stats["total_searches"] * 100
            if self.stats["total_searches"] > 0 else 0
        )
        
        avg_response_time = (
            self.stats["total_response_time"] / self.stats["successful_searches"]
            if self.stats["successful_searches"] > 0 else 0
        )
        
        return {
            **self.stats,
            "success_rate": f"{success_rate:.1f}%",
            "average_response_time": f"{avg_response_time:.2f}s"
        }

Parte 7: Casos de uso prácticos

Caso 1: Agregador de noticias

class NewsAggregatorAgent:
    """Agente para la agregación de noticias por temas"""
    
    def __init__(self, tavily_api_key: str):
        self.search_agent = SmartSearchAgent(tavily_api_key)
    
    def get_daily_digest(self, topics: List[str]) -> str:
        """Creación de un resumen diario de noticias"""
        
        digest_parts = []
        
        for topic in topics:
            query = f"Últimas noticias de hoy: {topic}"
            news = self.search_agent.search(query)
            
            digest_parts.append(f"## {topic}\n{news}\n")
        
        return "\n".join(digest_parts)

<em># Uso</em>
news_agent = NewsAggregatorAgent("your-api-key")
digest = news_agent.get_daily_digest([
    "Tecnologías de IA",
    "Criptomonedas", 
    "Investigación espacial"
])

Caso 2: Asistente de investigación

class ResearchAssistant:
    """Asistente para investigación científica"""
    
    def __init__(self, tavily_api_key: str):
        self.agent = SmartSearchAgent(tavily_api_key)
    
    def literature_review(self, topic: str, num_sources: int = 10) -> Dict:
        """Revisión de literatura sobre un tema"""
        
        queries = [
            f"{topic} recent research 2024",
            f"{topic} scientific papers",
            f"{topic} review articles",
            f"{topic} latest findings"
        ]
        
        all_sources = []
        
        for query in queries:
            result = self.agent.search(query)
            <em># Extracción de fuentes del resultado</em>
            <em># ... lógica de procesamiento</em>
        
        return {
            "topic": topic,
            "sources_found": len(all_sources),
            "summary": "...",
            "key_findings": "...",
            "sources": all_sources
        }

Caso 3: Analista de negocios

class BusinessAnalyst:
    """Agente para análisis de negocios"""
    
    def __init__(self, tavily_api_key: str):
        self.agent = SmartSearchAgent(tavily_api_key)
    
    def market_analysis(self, company: str, competitors: List[str]) -> str:
        """Análisis de mercado y competidores"""
        
        <em># Análisis de la empresa principal</em>
        company_analysis = self.agent.search(
            f"Resultados financieros y noticias {company} 2024"
        )
        
        <em># Análisis de competidores</em>
        competitor_analyses = []
        for competitor in competitors:
            analysis = self.agent.search(
                f"Resultados financieros {competitor} comparación con {company}"
            )
            competitor_analyses.append(analysis)
        
        <em># Síntesis del informe</em>
        report_query = f"""
        Crea un informe de negocios basado en los datos:
        Empresa: {company_analysis}
        Competidores: {competitor_analyses}
        """
        
        return self.agent.search(report_query)

Conclusión

La integración de any-agent con Tavily abre amplias posibilidades para crear agentes inteligentes con acceso a información actualizada. Esta combinación es especialmente útil para:

Ventajas clave:

  • Actualidad: acceso a información fresca en tiempo real
  • Flexibilidad: una interfaz unificada para diferentes frameworks de agentes
  • Fiabilidad: búsqueda especializada, optimizada para IA
  • Escalabilidad: posibilidad de crear sistemas multi-agente complejos

Recomendaciones de uso:

  1. Empiece con lo simple: domine la integración básica antes de pasar a sistemas complejos
  2. Monitoree los costos: supervise el uso de la API y los recursos computacionales
  3. Pruebe a fondo: los sistemas de agentes pueden comportarse de forma impredecible
  4. Utilice el almacenamiento en caché: evite solicitudes repetidas de la misma información
  5. Planifique el manejo de errores: siempre prevea escenarios de respaldo

Próximos pasos:

  • Explore frameworks adicionales compatibles con any-agent
  • Experimente con diferentes configuraciones de profundidad de búsqueda de Tavily
  • Considere la integración con otras herramientas (bases de datos, API de servicios externos)
  • Explore las posibilidades de crear agentes especializados para su dominio

Esta combinación tecnológica representa una herramienta poderosa para el desarrollo de aplicaciones de IA modernas, capaces de trabajar con datos actualizados y proporcionar a los usuarios información relevante.

Apéndice A: Código completo del agente de producción

<em>## \file src/agents/production_search_agent.py</em>
<em># -*- coding: utf-8 -*-</em>
<em>#! .pyenv/bin/python3</em>
"""
Agente de búsqueda de producción
===============================
Agente con todas las funciones para uso en producción con un conjunto completo
de capacidades: monitoreo, almacenamiento en caché, manejo de errores.
"""
import asyncio
import json
import time
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass, asdict
from enum import Enum
import logging
from tavily import TavilyClient
from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent
class SearchDepth(Enum):
    """Profundidad de búsqueda"""
    BASIC = "basic"
    ADVANCED = "advanced"
class SearchStatus(Enum):
    """Estado de la consulta de búsqueda"""
    SUCCESS = "success"
    ERROR = "error"
    CACHED = "cached"
    RATE_LIMITED = "rate_limited"
@dataclass
class SearchRequest:
    """Estructura de la consulta de búsqueda"""
    query: str
    depth: SearchDepth = SearchDepth.BASIC
    timestamp: str = None
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    
    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.now().isoformat()
    
    def get_cache_key(self) -> str:
        """Generación de clave para el almacenamiento en caché"""
        content = f"{self.query}:{self.depth.value}"
        return hashlib.md5(content.encode()).hexdigest()
@dataclass
class SearchResponse:
    """Estructura de la respuesta de búsqueda"""
    request: SearchRequest
    status: SearchStatus
    answer: str = ""
    sources: List[Dict] = None
    error_message: str = ""
    response_time: float = 0.0
    from_cache: bool = False
    
    def __post_init__(self):
        if self.sources is None:
            self.sources = []
class CacheManager:
    """Gestor de caché para consultas de búsqueda"""
    
    def __init__(self, cache_dir: str = "./cache", ttl_hours: int = 24):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)
    
    def _get_cache_file(self, cache_key: str) -> Path:
        """Obtener la ruta al archivo de caché"""
        return self.cache_dir / f"{cache_key}.json"
    
    def get(self, cache_key: str) -> Optional[Dict]:
        """Obtener datos de la caché"""
        cache_file = self._get_cache_file(cache_key)
        
        if not cache_file.exists():
            return None
        
        try:
            with open(cache_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
            
            <em># Check TTL</em>
            cached_time = datetime.fromisoformat(data['cached_at'])
            if datetime.now() - cached_time > self.ttl:
                cache_file.unlink()  <em># Delete expired cache</em>
                return None
            
            return data['content']
            
        except Exception:
            return None
    
    def set(self, cache_key: str, content: Dict):
        """Guardar datos en la caché"""
        cache_file = self._get_cache_file(cache_key)
        
        try:
            cache_data = {
                'cached_at': datetime.now().isoformat(),
                'content': content
            }
            
            with open(cache_file, 'w', encoding='utf-8') as f:
                json.dump(cache_data, f, ensure_ascii=False, indent=2)
                
        except Exception as e:
            logging.warning(f"No se pudo guardar en la caché: {e}")
    
    def clear_expired(self):
        """Limpiar caché caducada"""
        for cache_file in self.cache_dir.glob("*.json"):
            try:
                with open(cache_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                
                cached_time = datetime.fromisoformat(data['cached_at'])
                if datetime.now() - cached_time > self.ttl:
                    cache_file.unlink()
                    
            except Exception:
                continue
class RateLimiter:
    """Limitador de frecuencia de solicitudes"""
    
    def __init__(self, max_requests: int = 60, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = []
    
    def can_make_request(self) -> bool:
        """Verificar la posibilidad de realizar una solicitud"""
        now = time.time()
        
        <em># Clear expired requests</em>
        self.requests = [req_time for req_time in self.requests 
                        if now - req_time < self.time_window]
        
        return len(self.requests) < self.max_requests
    
    def register_request(self):
        """Registrar solicitud realizada"""
        self.requests.append(time.time())
    
    def get_wait_time(self) -> float:
        """Obtener tiempo de espera hasta la próxima solicitud"""
        if not self.requests:
            return 0.0
        
        oldest_request = min(self.requests)
        wait_time = self.time_window - (time.time() - oldest_request)
        
        return max(0.0, wait_time)
class ProductionSearchAgent:
    """
    Agente de búsqueda de producción
    
    Un agente con todas las funciones para uso en producción con:
    - Almacenamiento en caché de resultados
    - Limitación de frecuencia
    - Monitoreo y registro
    - Manejo de errores
    - Estadísticas de uso
    """
    
    def __init__(
        self,
        tavily_api_key: str,
        model: str = "gpt-4",
        cache_ttl_hours: int = 24,
        rate_limit_per_minute: int = 60,
        max_retries: int = 3,
        log_level: str = "INFO"
    ):
        <em># Inicialización de componentes</em>
        self.tavily_client = TavilyClient(api_key=tavily_api_key)
        self.cache_manager = CacheManager(ttl_hours=cache_ttl_hours)
        self.rate_limiter = RateLimiter(max_requests=rate_limit_per_minute)
        self.max_retries = max_retries
        
        <em># Configuración de registro</em>
        self.setup_logging(log_level)
        
        <em># Creación del agente</em>
        self.agent = self._create_agent(model)
        
        <em># Estadísticas</em>
        self.stats = {
            "total_requests": 0,
            "successful_requests": 0,
            "cached_requests": 0,
            "failed_requests": 0,
            "rate_limited_requests": 0,
            "total_response_time": 0.0,
            "start_time": datetime.now().isoformat()
        }
    
    def setup_logging(self, log_level: str):
        """Configurar el sistema de registro"""
        logging.basicConfig(
            level=getattr(logging, log_level.upper()),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('production_search_agent.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('ProductionSearchAgent')
    
    def _create_agent(self, model: str) -> AnyAgent:
        """Crear un agente con configuración de producción"""
        return AnyAgent(
            framework=TinyAgent,
            model=model,
            tools=[self._search_tool],
            system_prompt=self._get_production_system_prompt()
        )
    
    def _get_production_system_prompt(self) -> str:
        """Prompt del sistema para producción"""
        return """
        Eres un asistente de IA profesional de nivel empresarial con acceso 
        a información actualizada de internet.
        Tus principios operativos:
        1. PRECISIÓN: Siempre verifica los hechos a través de la búsqueda
        2. FUENTES: Cita las fuentes para todas las afirmaciones
        3. OBJETIVIDAD: Presenta diferentes puntos de vista
        4. ESTRUCTURA: Organiza la información de forma lógica
        5. PUNTUALIDAD: Utiliza solo datos recientes
        Formato de respuesta:
        - Resumen breve (2-3 oraciones)
        - Información clave con fuentes
        - Fuentes adicionales para estudio posterior
        Si la información es contradictoria o poco fiable, asegúrate de señalarlo.
        """
    
    def _search_tool(self, query: str, depth: str = "basic") -> Dict:
        """Herramienta de búsqueda con manejo completo de errores"""
        search_request = SearchRequest(
            query=query,
            depth=SearchDepth(depth)
        )
        
        response = self._execute_search(search_request)
        
        if response.status == SearchStatus.SUCCESS:
            return {
                "success": True,
                "query": query,
                "answer": response.answer,
                "sources": response.sources,
                "from_cache": response.from_cache,
                "response_time": response.response_time
            }
        else:
            return {
                "success": False,
                "query": query,
                "error": response.error_message,
                "status": response.status.value
            }
    
    def _execute_search(self, request: SearchRequest) -> SearchResponse:
        """Realizar búsqueda con manejo completo"""
        start_time = time.time()
        self.stats["total_requests"] += 1
        
        try:
            <em># Verificar caché</em>
            cache_key = request.get_cache_key()
            cached_result = self.cache_manager.get(cache_key)
            
            if cached_result:
                self.stats["cached_requests"] += 1
                self.logger.info(f"Resultado obtenido de la caché: {request.query}")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.CACHED,
                    answer=cached_result.get('answer', ''),
                    sources=cached_result.get('sources', []),
                    response_time=time.time() - start_time,
                    from_cache=True
                )
            
            <em># Verificar límite de frecuencia</em>
            if not self.rate_limiter.can_make_request():
                wait_time = self.rate_limiter.get_wait_time()
                self.stats["rate_limited_requests"] += 1
                self.logger.warning(f"Límite de frecuencia excedido. Tiempo de espera: {wait_time:.1f}s")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.RATE_LIMITED,
                    error_message=f"Límite de frecuencia excedido. Inténtelo de nuevo en {wait_time:.1f} segundos",
                    response_time=time.time() - start_time
                )
            
            <em># Realizar búsqueda con reintentos</em>
            result = None
            last_error = None
            
            for attempt in range(self.max_retries):
                try:
                    self.rate_limiter.register_request()
                    
                    result = self.tavily_client.search(
                        query=request.query,
                        search_depth=request.depth.value
                    )
                    break
                    
                except Exception as e:
                    last_error = e
                    if attempt < self.max_retries - 1:
                        wait_time = (2 ** attempt) * 1  <em># Retroceso exponencial</em>
                        self.logger.warning(f"Intento {attempt + 1} fallido: {e}. Reintentando en {wait_time}s...")
                        time.sleep(wait_time)
            
            if result is None:
                self.stats["failed_requests"] += 1
                self.logger.error(f"Todos los intentos de búsqueda fallaron para: {request.query}")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.ERROR,
                    error_message=f"La búsqueda falló después de {self.max_retries} intentos: {last_error}",
                    response_time=time.time() - start_time
                )
            
            <em># Manejar resultado exitoso</em>
            response_time = time.time() - start_time
            self.stats["successful_requests"] += 1
            self.stats["total_response_time"] += response_time
            
            <em># Guardar en caché</em>
            cache_data = {
                'answer': result.get('answer', ''),
                'sources': result.get('results', [])
            }
            self.cache_manager.set(cache_key, cache_data)
            
            self.logger.info(f"Búsqueda exitosa en {response_time:.2f}s: {request.query}")
            
            return SearchResponse(
                request=request,
                status=SearchStatus.SUCCESS,
                answer=result.get('answer', ''),
                sources=result.get('results', []),
                response_time=response_time,
                from_cache=False
            )
            
        except Exception as e:
            self.stats["failed_requests"] += 1
            self.logger.error(f"Error inesperado durante la búsqueda: {e}", exc_info=True)
            
            return SearchResponse(
                request=request,
                status=SearchStatus.ERROR,
                error_message=f"Error inesperado: {str(e)}",
                response_time=time.time() - start_time
            )
    
    def search(self, query: str, depth: SearchDepth = SearchDepth.BASIC) -> str:
        """
        Método de búsqueda principal para usuarios
        
        Args:
            query: Consulta de búsqueda
            depth: Profundidad de búsqueda (básica/avanzada)
            
        Returns:
            Respuesta del agente con la información encontrada
        """
        try:
            self.logger.info(f"Procesando consulta de usuario: {query}")
            
            <em># Crear contexto para el agente</em>
            search_context = f"Realiza una búsqueda y proporciona una respuesta detallada a la consulta: {query}"
            if depth == SearchDepth.ADVANCED:
                search_context += " (utiliza búsqueda avanzada para un análisis más profundo)"
            
            result = self.agent.run(search_context)
            return result
            
        except Exception as e:
            self.logger.error(f"Error al procesar la consulta: {e}", exc_info=True)
            return f"Lo siento, ocurrió un error al procesar la solicitud: {str(e)}"
    
    def get_statistics(self) -> Dict[str, Any]:
        """Obtener estadísticas operativas detalladas"""
        uptime = datetime.now() - datetime.fromisoformat(self.stats["start_time"])
        
        success_rate = (
            self.stats["successful_requests"] / self.stats["total_requests"] * 100
            if self.stats["total_requests"] > 0 else 0
        )
        
        cache_hit_rate = (
            self.stats["cached_requests"] / self.stats["total_requests"] * 100
            if self.stats["total_requests"] > 0 else 0
        )
        
        avg_response_time = (
            self.stats["total_response_time"] / self.stats["successful_requests"]
            if self.stats["successful_requests"] > 0 else 0
        )
        
        return {
            **self.stats,
            "uptime": str(uptime),
            "success_rate": f"{success_rate:.1f}%",
            "cache_hit_rate": f"{cache_hit_rate:.1f}%",
            "average_response_time": f"{avg_response_time:.3f}s",
            "requests_per_hour": self.stats["total_requests"] / (uptime.total_seconds() / 3600) if uptime.total_seconds() > 0 else 0
        }
    
    def health_check(self) -> Dict[str, Any]:
        """Verificación de estado del sistema"""
        try:
            <em># Verificar funcionalidad básica</em>
            test_response = self._execute_search(SearchRequest("test query"))
            
            return {
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "tavily_api": "conectado",
                "cache": "operacional",
                "rate_limiter": "operacional",
                "last_test_status": test_response.status.value
            }
            
        except Exception as e:
            return {
                "status": "unhealthy",
                "timestamp": datetime.now().isoformat(),
                "error": str(e)
            }
    
    def cleanup(self):
        """Limpiar recursos al apagar"""
        self.logger.info("Iniciando proceso de limpieza...")
        
        <em># Limpiar caché caducada</em>
        self.cache_manager.clear_expired()
        
        <em># Guardar estadísticas</em>
        stats_file = Path("agent_statistics.json")
        try:
            with open(stats_file, 'w', encoding='utf-8') as f:
                json.dump(self.get_statistics(), f, ensure_ascii=False, indent=2)
            self.logger.info(f"Estadísticas guardadas en {stats_file}")
        except Exception as e:
            self.logger.error(f"Error al guardar estadísticas: {e}")
        
        self.logger.info("Limpieza completada")
<em># Ejemplo de uso en producción</em>
async def main():
    """Demostración del agente de producción"""
    
    <em># Inicialización del agente</em>
    agent = ProductionSearchAgent(
        tavily_api_key="tvly-YOUR_API_KEY",
        model="mistral/mistral-small-latest",
        cache_ttl_hours=12,
        rate_limit_per_minute=30,
        log_level="INFO"
    )
    
    try:
        <em># Verificación de estado</em>
        health = agent.health_check()
        print(f"Estado del agente: {health['status']}")
        
        <em># Consultas de prueba</em>
        test_queries = [
            "Latest achievements in quantum computing",
            "Current trends in AI development 2024",
            "Impact of climate change on the economy",
        ]
        
        print("\n=== Demostración del agente de producción ===\n")
        
        for i, query in enumerate(test_queries, 1):
            print(f"Consulta {i}: {query}")
            print("-" * 60)
            
            result = agent.search(query, SearchDepth.BASIC)
            print(result[:500] + "..." if len(result) > 500 else result)
            
            print(f"\nEstadísticas:")
            stats = agent.get_statistics()
            print(f"- Solicitudes totales: {stats['total_requests']}")
            print(f"- Exitosas: {stats['success_rate']}")
            print(f"- Desde caché: {stats['cache_hit_rate']}")
            print(f"- Tiempo promedio de respuesta: {stats['average_response_time']}")
            print("\n" + "="*80 + "\n")
    
    <em># Estadísticas finales</em>
    print("Estadísticas finales:")
    final_stats = agent.get_statistics()
    for key, value in final_stats.items():
        print(f"{key}: {value}")
    
finally:
    <em># Limpieza de recursos</em>
    agent.cleanup()
if __name__ == "__main__":
    asyncio.run(main())

Apéndice B: Archivos de configuración

config/agent_config.json

{
  "tavily": {
    "api_key": "tvly-YOUR_API_KEY",
    "default_search_depth": "basic",
    "timeout_seconds": 30
  },
  "agent": {
    "model": "gpt-4",
    "framework": "TinyAgent",
    "max_tokens": 4000,
    "temperature": 0.1
  },
  "cache": {
    "enabled": true,
    "ttl_hours": 24,
    "max_size_mb": 100,
    "directory": "./cache"
  },
  "rate_limiting": {
    "enabled": true,
    "requests_per_minute": 60,
    "burst_allowance": 10
  },
  "logging": {
    "level": "INFO",
    "file": "agent.log",
    "max_file_size_mb": 10,
    "backup_count": 5
  },
  "monitoring": {
    "enabled": true,
    "metrics_interval_seconds": 300,
    "health_check_interval_seconds": 60
  }
}

docker/Dockerfile

FROM python:3.11-slim

WORKDIR /app

<em># Install system dependencies</em>
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

<em># Copy dependency files</em>
COPY requirements.txt .

<em># Install Python dependencies</em>
RUN pip install --no-cache-dir -r requirements.txt

<em># Copy application code</em>
COPY . .

<em># Create necessary directories</em>
RUN mkdir -p /app/cache /app/logs

<em># Configure non-root user</em>
RUN useradd -m -u 1000 agent && \
    chown -R agent:agent /app
USER agent

<em># Expose ports</em>
EXPOSE 8000

<em># Startup command</em>
CMD ["python", "-m", "src.agents.production_search_agent"]

docker-compose.yml

version: '3.8'

services:
  search-agent:
    build: .
    container_name: production-search-agent
    environment:
      - TAVILY_API_KEY=${TAVILY_API_KEY}
      - LOG_LEVEL=INFO
    volumes:
      - ./cache:/app/cache
      - ./logs:/app/logs
      - ./config:/app/config
    ports:
      - "8000:8000"
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python", "-c", "import requests; requests.get('http://localhost:8000/health')"]
      interval: 30s
      timeout: 10s
      retries: 3

  redis:
    image: redis:7-alpine
    container_name: agent-redis
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    container_name: agent-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped

volumes:
  redis_data:

Apéndice B: Pruebas

tests/test_production_agent.py

import pytest
import asyncio
from unittest.mock import Mock, patch
from src.agents.production_search_agent import (
    ProductionSearchAgent,
    SearchRequest,
    SearchDepth,
    SearchStatus
)


@pytest.fixture
def mock_agent():
    """Fixture para crear un agente simulado"""
    with patch('src.agents.production_search_agent.TavilyClient'):
        agent = ProductionSearchAgent(
            tavily_api_key="test-key",
            model="gpt-4"
        )
        return agent


class TestProductionSearchAgent:
    
    def test_agent_initialization(self, mock_agent):
        """Probar la inicialización del agente"""
        assert mock_agent is not None
        assert mock_agent.stats["total_requests"] == 0
        assert mock_agent.max_retries == 3
    
    def test_search_request_creation(self):
        """Probar la creación de la solicitud de búsqueda"""
        request = SearchRequest(
            query="test query",
            depth=SearchDepth.BASIC
        )
        
        assert request.query == "test query"
        assert request.depth == SearchDepth.BASIC
        assert request.timestamp is not None
        
        cache_key = request.get_cache_key()
        assert len(cache_key) == 32  <em># Hash MD5</em>
    
    @patch('src.agents.production_search_agent.TavilyClient')
    def test_successful_search(self, mock_tavily, mock_agent):
        """Probar búsqueda exitosa"""
        <em># Configuración del mock</em>
        mock_tavily_instance = Mock()
        mock_tavily.return_value = mock_tavily_instance
        mock_tavily_instance.search.return_value = {
            'answer': 'Test answer',
            'results': [{'title': 'Test', 'url': 'http://test.com'}]
        }
        
        <em># Realizar búsqueda</em>
        request = SearchRequest("test query")
        response = mock_agent._execute_search(request)
        
        <em># Afirmaciones</em>
        assert response.status == SearchStatus.SUCCESS
        assert response.answer == 'Test answer'
        assert len(response.sources) == 1
        assert not response.from_cache
    
    def test_cache_functionality(self, mock_agent):
        """Probar la funcionalidad de la caché"""
        cache_key = "test_key"
        test_data = {"answer": "cached answer", "sources": []}
        
        <em># Guardar en caché</em>
        mock_agent.cache_manager.set(cache_key, test_data)
        
        <em># Obtener de la caché</em>
        cached_data = mock_agent.cache_manager.get(cache_key)
        
        assert cached_data is not None
        assert cached_data["answer"] == "cached answer"
    
    def test_rate_limiting(self, mock_agent):
        """Probar la limitación de frecuencia de solicitudes"""
        rate_limiter = mock_agent.rate_limiter
        
        <em># Llenar límite</em>
        for _ in range(rate_limiter.max_requests):
            rate_limiter.register_request()
        
        <em># Verificar bloqueo</em>
        assert not rate_limiter.can_make_request()
        
        <em># Verificar tiempo de espera</em>
        wait_time = rate_limiter.get_wait_time()
        assert wait_time > 0
    
    def test_statistics_tracking(self, mock_agent):
        """Probar el seguimiento de estadísticas"""
        initial_stats = mock_agent.get_statistics()
        assert initial_stats["total_requests"] == 0
        assert initial_stats["success_rate"] == "0.0%"
        
        <em># Simular solicitudes</em>
        mock_agent.stats["total_requests"] = 10
        mock_agent.stats["successful_requests"] = 8
        mock_agent.stats["cached_requests"] = 2
        
        updated_stats = mock_agent.get_statistics()
        assert updated_stats["total_requests"] == 10
        assert updated_stats["success_rate"] == "80.0%"
        assert updated_stats["cache_hit_rate"] == "20.0%"
    
    def test_health_check(self, mock_agent):
        """Probar la verificación de estado del sistema"""
        health = mock_agent.health_check()
        
        assert "status" in health
        assert "timestamp" in health
        assert health["status"] in ["healthy", "unhealthy"]


@pytest.mark.asyncio
async def test_async_functionality():
    """Probar la funcionalidad asíncrona"""
    <em># Las operaciones asíncronas se pueden probar aquí</em>
    <em># si se añaden en el futuro</em>
    pass


if __name__ == "__main__":
    pytest.main([__file__, "-v"])

Recomendaciones finales

1. Despliegue en producción

Contenerización: Utilice Docker para el aislamiento de dependencias y un despliegue simplificado.

Monitoreo: Configure un sistema de monitoreo (Prometheus + Grafana) para rastrear el rendimiento.

Registro: Utilice un registro centralizado (pila ELK o similar).

2. Seguridad

<em># Almacenamiento seguro de claves API</em>
import os
from cryptography.fernet import Fernet

class SecureConfig:
    """Gestión segura de la configuración"""
    
    def __init__(self, key_file: str = ".encryption_key"):
        self.key_file = key_file
        self.key = self._load_or_create_key()
        self.cipher = Fernet(self.key)
    
    def _load_or_create_key(self) -> bytes:
        """Cargar o crear clave de cifrado"""
        if os.path.exists(self.key_file):
            with open(self.key_file, 'rb') as f:
                return f.read()
        else:
            key = Fernet.generate_key()
            with open(self.key_file, 'wb') as f:
                f.write(key)
            return key
    
    def encrypt_api_key(self, api_key: str) -> str:
        """Cifrar clave API"""
        return self.cipher.encrypt(api_key.encode()).decode()
    
    def decrypt_api_key(self, encrypted_key: str) -> str:
        """Descifrar clave API"""
        return self.cipher.decrypt(encrypted_key.encode()).decode()

<em># Uso</em>
config = SecureConfig()
encrypted_key = config.encrypt_api_key("tvly-YOUR_API_KEY")
<em># Guardar encrypted_key en la configuración</em>
<em># Al usar:</em>
api_key = config.decrypt_api_key(encrypted_key)

3. Escalado

<em># Ejemplo de arquitectura de microservicios</em>
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="API de Agente de Búsqueda", version="1.0.0")

class SearchQuery(BaseModel):
    query: str
    depth: str = "basic"
    user_id: str = None

class SearchResponse(BaseModel):
    query: str
    answer: str
    sources: list
    response_time: float
    from_cache: bool

<em># Instancia global del agente</em>
agent = ProductionSearchAgent(tavily_api_key=os.getenv("TAVILY_API_KEY"))

@app.post("/search", response_model=SearchResponse)
async def search_endpoint(query: SearchQuery, background_tasks: BackgroundTasks):
    """Endpoint para la consulta de búsqueda"""
    try:
        result = agent.search(query.query, SearchDepth(query.depth))
        
        <em># Registro asíncrono</em>
        background_tasks.add_task(log_search_request, query, result)
        
        return SearchResponse(
            query=query.query,
            answer=result,
            sources=[],  <em># Rellenar con fuentes reales</em>
            response_time=0.0,  <em># Rellenar con tiempo real</em>
            from_cache=False  <em># Rellenar con estado real</em>
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Verificación de estado del servicio"""
    return agent.health_check()

@app.get("/stats")
async def get_statistics():
    """Obtener estadísticas"""
    return agent.get_statistics()

def log_search_request(query: SearchQuery, result: str):
    """Registro asíncrono de la solicitud"""
    <em># Registrar en la base de datos o sistema externo</em>
    pass

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

4. Monitoreo y Alertas

<em># monitoring/prometheus.yml</em>
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'search-agent'
    static_configs:
      - targets: ['search-agent:8000']
    metrics_path: '/metrics'
    scrape_interval: 30s

rule_files:
  - "alert_rules.yml"

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - alertmanager:9093

<em># monitoring/alert_rules.yml</em>
groups:
  - name: search_agent_alerts
    rules:
      - alert: HighErrorRate
        expr: (rate(search_agent_failed_requests_total[5m]) / rate(search_agent_total_requests_total[5m])) > 0.1
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Tasa de error alta detectada"
          description: "La tasa de error es {{ $value | humanizePercentage }}"

      - alert: SlowResponseTime
        expr: search_agent_avg_response_time_seconds > 5
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Tiempo de respuesta lento"
          description: "El tiempo promedio de respuesta es {{ $value }}s"

5. Integración Continua

<em># .github/workflows/ci.yml</em>
name: Pipeline CI/CD

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Configurar Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Instalar dependencias
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt
      
      - name: Ejecutar pruebas
        run: |
          pytest tests/ --cov=src --cov-report=xml
      
      - name: Subir cobertura
        uses: codecov/codecov-action@v3

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Ejecutar escaneo de seguridad
        run: |
          pip install bandit safety
          bandit -r src/
          safety check

  build:
    needs: [test, security]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v3
      
      - name: Construir imagen Docker
        run: |
          docker build -t search-agent:latest .
      
      - name: Enviar a registro
        run: |
          echo ${{ secrets.DOCKER_PASSWORD }} | docker login -u ${{ secrets.DOCKER_USERNAME }} --password-stdin
          docker push search-agent:latest

6. Documentación de la API

<em># Documentación extendida de FastAPI</em>
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title="API de Agente de Búsqueda Inteligente",
        version="2.0.0",
        description="""
        ## Agente de Búsqueda Inteligente con Integración Tavily

        Esta API proporciona acceso a un agente de búsqueda inteligente,
        que utiliza información actualizada de internet.

        ### Capacidades:
        - Búsqueda en tiempo real de información actualizada
        - Almacenamiento en caché de resultados para un rendimiento mejorado
        - Limitación de frecuencia de solicitudes
        - Estadísticas de uso detalladas
        - Monitoreo del estado del sistema

        ### Autenticación:
        Utilice la clave API en el encabezado `X-API-Key`
        """,
        routes=app.routes,
    )
    
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

Observaciones finales

La integración de any-agent con Tavily crea una potente plataforma para el desarrollo de aplicaciones de IA modernas. Esta combinación tecnológica resuelve el problema crítico del acceso a información actualizada y abre nuevas posibilidades para la creación de sistemas inteligentes.

Logros clave:

🎯 Versatilidad: Una interfaz unificada para trabajar con varios frameworks de agentes

⚡ Oportunidad: Acceso a información fresca en tiempo real a través de un motor de búsqueda especializado

🔧 Flexibilidad: Capacidad para crear tanto agentes simples como sistemas multi-agente complejos

📊 Fiabilidad: Un conjunto completo de herramientas para producción: monitoreo, almacenamiento en caché, manejo de errores

🚀 Escalabilidad: Listo para su despliegue en un entorno empresarial

Aplicación en proyectos reales:

  • Asistentes corporativos con acceso a información actualizada
  • Sistemas de análisis de mercado para organizaciones financieras
  • Plataformas de investigación para instituciones científicas
  • Agregadores de noticias y sistemas analíticos
  • Asistentes educativos con conocimientos actuales

Desarrollo futuro:

Las tecnologías de agentes de IA están evolucionando rápidamente. La combinación de any-agent + Tavily crea una base sólida para adaptarse a nuevos frameworks y capacidades, asegurando un valor a largo plazo para las inversiones en desarrollo.

Esta guía proporciona un ciclo de desarrollo completo, desde ejemplos básicos hasta soluciones de producción. Úsela como referencia para crear sus propios agentes inteligentes con acceso a información actualizada.

¡Mucha suerte con su desarrollo! 🤖✨

Etiquetas:

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *