Перейти к содержимому
davidka.net > 💻 🧠 Код 1001 > LLM: статьи, практические материалы > any-agent + Tavily для создания умных поисковых агентов

any-agent + Tavily для создания умных поисковых агентов

  • автор:

В современном мире ИИ-агентов одной из главных проблем является доступ к актуальной информации. Большинство языковых моделей обучены на данных с определенной датой среза и не могут предоставить информацию о последних событиях, текущих ценах или новостях.

В этой статье мы рассмотрим, как решить эту проблему с помощью связки any-agent (универсальная библиотека для работы с агентами от Mozilla.ai) и Tavily (специализированная поисковая система для ИИ).

In Questo Articolo
  1. Часть 1: Знакомство с any-agent
  2. Часть 2: Знакомство с Tavily
  3. Часть 3: Интеграция any-agent с Tavily
  4. Часть 4: Полнофункциональный пример
  5. Часть 5: Продвинутые возможности
  6. Часть 6: Лучшие практики и рекомендации
  7. Часть 7: Практические кейсы использования
  8. Заключение
  9. Приложение А: Полный код производственного агента
  10. Приложение Б: Конфигурационные файлы
  11. Приложение В: Тестирование
  12. Заключительные рекомендации
  13. Итоговые выводы

Часть 1: Знакомство с any-agent

Что такое any-agent?

any-agent — это Python-библиотека от Mozilla.ai, которая предоставляет единый интерфейс для работы с различными агентными фреймворками. Главная идея — позволить разработчикам легко переключаться между разными фреймворками агентов (AutoGen, CrewAI, LangGraph и др.) без переписывания кода.

Ключевые преимущества:

  • Унифицированный API: один интерфейс для всех фреймворков
  • Стандартизированное логирование: одинаковые трейсы независимо от фреймворка
  • Легкое сравнение: возможность тестировать разные настройки агентов
  • Расширяемость: поддержка добавления новых фреймворков

Системные требования и установка

<em># Системные требования</em>
<em># Python 3.11+</em>

<em># Базовая установка (только TinyAgent)</em>
pip install any-agent

<em># Установка с поддержкой всех фреймворков</em>
pip install any-agent[all]

<em># Или с конкретным фреймворком</em>
pip install any-agent[langchain]

Первый пример агента

from any_agent import AnyAgent
from any_agent.frameworks import TinyAgent

<em># Создание простого агента</em>
agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[],
    system_prompt="Вы полезный ассистент"
)

<em># Запуск агента</em>
result = agent.run("Привет! Расскажи о себе.")
print(result)

Важно для Jupyter Notebook:

import nest_asyncio
nest_asyncio.apply()
<em># Теперь можно использовать any-agent</em>

Часть 2: Знакомство с Tavily

Что такое Tavily?

Tavily — это поисковая система, созданная специально для ИИ-агентов и RAG-систем. В отличие от обычных поисковых систем, Tavily оптимизирована для предоставления структурированных, фактических результатов, которые легко обрабатываются языковыми моделями.

Ключевые особенности Tavily:

  • Оптимизация для ИИ: результаты адаптированы для LLM
  • Фокус на фактах: акцент на точность и проверяемость
  • Высокая скорость: быстрые ответы в реальном времени
  • Структурированный вывод: JSON-формат с метаданными

Регистрация и получение API-ключа

  1. Перейдите на https://app.tavily.com/
  2. Создайте аккаунт
  3. Получите API-ключ в личном кабинете

Базовое использование Tavily

<em># Установка клиента</em>
pip install tavily-python

from tavily import TavilyClient

<em># Инициализация клиента</em>
client = TavilyClient(api_key="tvly-YOUR_API_KEY")

<em># Простой поиск</em>
response = client.search("Последние новости о Python 3.12")

print(f"Запрос: {response['query']}")
print(f"Ответ: {response['answer']}")
print(f"Источники: {len(response['results'])} результатов")

Структура ответа Tavily

{
  "query": "Ваш поисковый запрос",
  "answer": "Краткий ответ на основе найденной информации",
  "images": [],
  "results": [
    {
      "title": "Заголовок источника",
      "url": "https://example.com",
      "content": "Содержимое...",
      "score": 0.81025416,
      "favicon": "https://example.com/favicon.png"
    }
  ],
  "auto_parameters": {
    "topic": "general",
    "search_depth": "basic"
  },
  "response_time": "1.67"
}

Часть 3: Интеграция any-agent с Tavily

Подход 1: Создание пользовательского инструмента

Самый гибкий способ — создать собственный инструмент поиска:

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

class TavilySearchTool:
    """Пользовательский инструмент для поиска через Tavily"""
    
    def __init__(self, api_key: str):
        self.client = TavilyClient(api_key=api_key)
        self.name = "tavily_search"
        self.description = "Поиск актуальной информации в интернете"
    
    def __call__(self, query: str, search_depth: str = "basic") -> Dict:
        """Выполнение поиска"""
        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># Создание агента с Tavily</em>
tavily_tool = TavilySearchTool(api_key="tvly-YOUR_API_KEY")

agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[tavily_tool],
    system_prompt="""
    Вы ИИ-ассистент с доступом к актуальной информации из интернета.
    Используйте инструмент tavily_search для получения свежих данных.
    Всегда указывайте источники информации в своих ответах.
    """
)

<em># Использование</em>
result = agent.run("Найди последние новости о разработке ИИ от OpenAI")
print(result)

Подход 2: Интеграция через LangChain

Если вы используете фреймворк, совместимый с LangChain:

from langchain_tavily import TavilySearchAPIWrapper
from any_agent import AnyAgent

<em># Настройка поиска через LangChain</em>
search_tool = TavilySearchAPIWrapper(
    tavily_api_key="tvly-YOUR_API_KEY"
)

<em># Создание агента</em>
agent = AnyAgent(
    framework=YourLangChainFramework,
    model="gpt-4",
    tools=[search_tool],
    system_prompt="Ассистент с доступом к поиску в реальном времени"
)

Часть 4: Полнофункциональный пример

Создадим полноценное приложение для работы с поисковыми агентами:

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

"""
Модуль умного поискового агента
==============================

Интеграция any-agent с Tavily для создания агента
с доступом к актуальной информации из интернета.
"""

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:
    """Структура результата поиска"""
    query: str
    answer: str
    sources: List[Dict]
    timestamp: str
    response_time: float


class SmartSearchAgent:
    """
    Умный поисковый агент с интеграцией Tavily
    
    Агент способен выполнять поиск актуальной информации,
    анализировать результаты и предоставлять структурированные ответы.
    """
    
    def __init__(
        self, 
        tavily_api_key: str, 
        model: str = "gpt-4",
        search_depth: str = "basic"
    ):
        """
        Инициализация агента
        
        Args:
            tavily_api_key: API-ключ для Tavily
            model: Модель для агента
            search_depth: Глубина поиска ('basic' или '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:
        """Создание агента с настроенными инструментами"""
        return AnyAgent(
            framework=TinyAgent,
            model=model,
            tools=[self._search_tool],
            system_prompt=self._get_system_prompt()
        )
    
    def _get_system_prompt(self) -> str:
        """Системный промпт для агента"""
        return """
        Вы — экспертный ИИ-ассистент с доступом к актуальной информации из интернета.
        
        Ваши возможности:
        - Поиск актуальной информации через Tavily
        - Анализ и синтез данных из множественных источников
        - Предоставление фактических ответов с указанием источников
        
        Принципы работы:
        1. Всегда используйте поиск для получения актуальной информации
        2. Указывайте источники для всех фактических утверждений
        3. Если информация противоречива, укажите на это
        4. Будьте точными и объективными
        5. Структурируйте ответы для лучшего восприятия
        """
    
    def _search_tool(self, query: str) -> Dict:
        """
        Инструмент поиска для агента
        
        Args:
            query: Поисковый запрос
            
        Returns:
            Результаты поиска в структурированном формате
        """
        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># Сохранение результата в историю</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"Ошибка поиска: {str(e)}",
                "query": query
            }
    
    def search(self, query: str) -> str:
        """
        Выполнение поискового запроса через агента
        
        Args:
            query: Запрос пользователя
            
        Returns:
            Ответ агента с актуальной информацией
        """
        try:
            result = self.agent.run(query)
            return result
        except Exception as e:
            return f"Произошла ошибка при обработке запроса: {str(e)}"
    
    def get_search_history(self) -> List[SearchResult]:
        """Получение истории поисковых запросов"""
        return self.search_history.copy()
    
    def clear_history(self):
        """Очистка истории поиска"""
        self.search_history.clear()
    
    def export_history(self, filename: str):
        """
        Экспорт истории поиска в JSON файл
        
        Args:
            filename: Имя файла для экспорта
        """
        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"История экспортирована в {filename}")
            
        except Exception as e:
            print(f"Ошибка экспорта: {str(e)}")


<em># Пример использования</em>
def main():
    """Демонстрация работы умного поискового агента"""
    
    <em># Инициализация агента</em>
    agent = SmartSearchAgent(
        tavily_api_key="tvly-YOUR_API_KEY",
        model="gpt-4",
        search_depth="basic"
    )
    
    <em># Примеры запросов</em>
    queries = [
        "Последние новости о разработке GPT-5",
        "Текущие цены на Bitcoin и Ethereum",
        "Что нового в Python 3.12?",
        "Последние достижения в области квантовых компьютеров"
    ]
    
    print("=== Демонстрация умного поискового агента ===\n")
    
    for i, query in enumerate(queries, 1):
        print(f"Запрос {i}: {query}")
        print("-" * 50)
        
        result = agent.search(query)
        print(result)
        print("\n" + "="*80 + "\n")
    
    <em># Экспорт истории</em>
    agent.export_history("search_history.json")
    
    <em># Статистика</em>
    history = agent.get_search_history()
    print(f"Всего выполнено запросов: {len(history)}")
    
    if history:
        avg_response_time = sum(r.response_time for r in history) / len(history)
        print(f"Среднее время ответа: {avg_response_time:.2f} сек")


if __name__ == "__main__":
    main()

Часть 5: Продвинутые возможности

Multi-Agent системы

any-agent поддерживает создание Multi-Agent систем, где агенты могут взаимодействовать друг с другом:

class ResearchTeam:
    """Команда исследовательских агентов"""
    
    def __init__(self, tavily_api_key: str):
        <em># Агент-исследователь</em>
        self.researcher = SmartSearchAgent(
            tavily_api_key=tavily_api_key,
            model="gpt-4"
        )
        
        <em># Агент-аналитик</em>
        self.analyst = AnyAgent(
            framework=TinyAgent,
            model="gpt-4",
            system_prompt="""
            Вы эксперт-аналитик. Ваша задача — анализировать 
            полученную информацию и делать выводы.
            """
        )
        
        <em># Агент-писатель</em>
        self.writer = AnyAgent(
            framework=TinyAgent,
            model="gpt-4",
            system_prompt="""
            Вы профессиональный писатель. Создавайте 
            структурированные отчеты на основе анализа.
            """
        )
    
    def research_topic(self, topic: str) -> str:
        """Комплексное исследование темы"""
        
        <em># Этап 1: Поиск информации</em>
        research_data = self.researcher.search(f"Подробная информация о {topic}")
        
        <em># Этап 2: Анализ данных</em>
        analysis = self.analyst.run(f"Проанализируй эту информацию: {research_data}")
        
        <em># Этап 3: Написание отчета</em>
        report = self.writer.run(f"Создай структурированный отчет на основе анализа: {analysis}")
        
        return report

<em># Использование</em>
team = ResearchTeam("tvly-YOUR_API_KEY")
report = team.research_topic("Влияние ИИ на рынок труда в 2024 году")
print(report)

Настройка глубины поиска

Tavily предлагает разные уровни глубины поиска:

<em># Быстрый поиск (basic)</em>
basic_agent = SmartSearchAgent(
    tavily_api_key="your-key",
    search_depth="basic"  <em># Быстрее, меньше источников</em>
)

<em># Глубокий поиск (advanced)</em>
advanced_agent = SmartSearchAgent(
    tavily_api_key="your-key", 
    search_depth="advanced"  <em># Медленнее, больше источников</em>
)

Фильтрация результатов

def create_specialized_search_tool(domains: List[str] = None):
    """Создание специализированного поискового инструмента"""
    
    def search_tool(query: str) -> Dict:
        client = TavilyClient(api_key="your-key")
        
        search_params = {
            "query": query,
            "search_depth": "advanced"
        }
        
        <em># Добавление фильтров по доменам</em>
        if domains:
            search_params["include_domains"] = domains
        
        response = client.search(**search_params)
        return response
    
    return search_tool

<em># Создание агента для поиска только по научным источникам</em>
science_agent = AnyAgent(
    framework=TinyAgent,
    model="gpt-4",
    tools=[create_specialized_search_tool([
        "arxiv.org", 
        "nature.com", 
        "science.org"
    ])],
    system_prompt="Эксперт по научной информации"
)

Часть 6: Лучшие практики и рекомендации

Безопасность и ограничения

⚠️ Важные предупреждения:

  1. Вычислительные ресурсы: Агентные системы требуют значительно больше ресурсов, чем простые вызовы LLM
  2. Безопасность: Агенты могут выполнять непредсказуемые действия
  3. Стоимость: Каждый поисковый запрос и вызов LLM тарифицируется

Оптимизация производительности

class OptimizedSearchAgent:
    """Оптимизированный поисковый агент"""
    
    def __init__(self, tavily_api_key: str):
        self.client = TavilyClient(api_key=tavily_api_key)
        self.cache = {}  <em># Кеш для повторяющихся запросов</em>
        self.rate_limit_delay = 1  <em># Задержка между запросами</em>
    
    def search_with_cache(self, query: str) -> Dict:
        """Поиск с кешированием результатов"""
        
        <em># Проверка кеша</em>
        if query in self.cache:
            print(f"Результат взят из кеша для: {query}")
            return self.cache[query]
        
        <em># Выполнение поиска</em>
        time.sleep(self.rate_limit_delay)  <em># Соблюдение rate limits</em>
        result = self.client.search(query)
        
        <em># Сохранение в кеш</em>
        self.cache[query] = result
        
        return result

Обработка ошибок

class RobustSearchAgent:
    """Устойчивый к ошибкам поисковый агент"""
    
    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:
        """Поиск с повторными попытками"""
        
        for attempt in range(self.max_retries):
            try:
                return self.client.search(query)
            
            except Exception as e:
                print(f"Попытка {attempt + 1} неудачна: {str(e)}")
                
                if attempt == self.max_retries - 1:
                    return {
                        "error": f"Поиск не удался после {self.max_retries} попыток",
                        "query": query
                    }
                
                time.sleep(2 ** attempt)  <em># Экспоненциальная задержка</em>

Мониторинг и логирование

import logging
from datetime import datetime

class MonitoredSearchAgent:
    """Агент с мониторингом и логированием"""
    
    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):
        """Настройка системы логирования"""
        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:
        """Поиск с полным мониторингом"""
        start_time = datetime.now()
        self.stats["total_searches"] += 1
        
        try:
            self.logger.info(f"Начинаем поиск: {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"Поиск успешен за {response_time:.2f}c: {query}")
            
            return result
            
        except Exception as e:
            self.stats["failed_searches"] += 1
            self.logger.error(f"Ошибка поиска {query}: {str(e)}")
            
            return {"error": str(e), "query": query}
    
    def get_statistics(self) -> Dict:
        """Получение статистики работы"""
        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"
        }

Часть 7: Практические кейсы использования

Кейс 1: Новостной агрегатор

class NewsAggregatorAgent:
    """Агент для агрегации новостей по темам"""
    
    def __init__(self, tavily_api_key: str):
        self.search_agent = SmartSearchAgent(tavily_api_key)
    
    def get_daily_digest(self, topics: List[str]) -> str:
        """Создание ежедневной сводки новостей"""
        
        digest_parts = []
        
        for topic in topics:
            query = f"Последние новости за сегодня: {topic}"
            news = self.search_agent.search(query)
            
            digest_parts.append(f"## {topic}\n{news}\n")
        
        return "\n".join(digest_parts)

<em># Использование</em>
news_agent = NewsAggregatorAgent("your-api-key")
digest = news_agent.get_daily_digest([
    "Технологии ИИ",
    "Криптовалюты", 
    "Космические исследования"
])

Кейс 2: Исследовательский помощник

class ResearchAssistant:
    """Помощник для научных исследований"""
    
    def __init__(self, tavily_api_key: str):
        self.agent = SmartSearchAgent(tavily_api_key)
    
    def literature_review(self, topic: str, num_sources: int = 10) -> Dict:
        """Обзор литературы по теме"""
        
        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># Извлечение источников из результата</em>
            <em># ... логика обработки</em>
        
        return {
            "topic": topic,
            "sources_found": len(all_sources),
            "summary": "...",
            "key_findings": "...",
            "sources": all_sources
        }

Кейс 3: Бизнес-аналитик

class BusinessAnalyst:
    """Агент для бизнес-анализа"""
    
    def __init__(self, tavily_api_key: str):
        self.agent = SmartSearchAgent(tavily_api_key)
    
    def market_analysis(self, company: str, competitors: List[str]) -> str:
        """Анализ рынка и конкурентов"""
        
        <em># Анализ основной компании</em>
        company_analysis = self.agent.search(
            f"Финансовые результаты и новости {company} 2024"
        )
        
        <em># Анализ конкурентов</em>
        competitor_analyses = []
        for competitor in competitors:
            analysis = self.agent.search(
                f"Финансовые результаты {competitor} сравнение с {company}"
            )
            competitor_analyses.append(analysis)
        
        <em># Синтез отчета</em>
        report_query = f"""
        Создай бизнес-отчет на основе данных:
        Компания: {company_analysis}
        Конкуренты: {competitor_analyses}
        """
        
        return self.agent.search(report_query)

Заключение

Интеграция any-agent с Tavily открывает широкие возможности для создания умных агентов с доступом к актуальной информации. Эта связка особенно полезна для:

Ключевые преимущества:

  • Актуальность: доступ к свежей информации в реальном времени
  • Гибкость: единый интерфейс для разных агентных фреймворков
  • Надежность: специализированный поиск, оптимизированный для ИИ
  • Масштабируемость: возможность создания сложных мульти-агентных систем

Рекомендации по использованию:

  1. Начните с простого: освойте базовую интеграцию перед переходом к сложным системам
  2. Мониторьте затраты: следите за использованием API и вычислительных ресурсов
  3. Тестируйте тщательно: агентные системы могут вести себя непредсказуемо
  4. Используйте кеширование: избегайте повторных запросов одинаковой информации
  5. Планируйте обработку ошибок: всегда предусматривайте fallback-сценарии

Следующие шаги:

  • Изучите дополнительные фреймворки, поддерживаемые any-agent
  • Экспериментируйте с различными настройками глубины поиска Tavily
  • Рассмотрите интеграцию с другими инструментами (базы данных, API внешних сервисов)
  • Изучите возможности создания специализированных агентов для вашей предметной области

Эта технологическая связка представляет собой мощный инструмент для разработки современных ИИ-приложений, способных работать с актуальными данными и предоставлять пользователям релевантную информацию.

Приложение А: Полный код производственного агента

<em>## \file src/agents/production_search_agent.py</em>
<em># -*- coding: utf-8 -*-</em>
<em>#! .pyenv/bin/python3</em>
"""
Производственный поисковый агент
===============================
Полнофункциональный агент для работы в продакшене с полным
набором возможностей: мониторинг, кеширование, обработка ошибок.
"""
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):
    """Глубина поиска"""
    BASIC = "basic"
    ADVANCED = "advanced"
class SearchStatus(Enum):
    """Статус поискового запроса"""
    SUCCESS = "success"
    ERROR = "error"
    CACHED = "cached"
    RATE_LIMITED = "rate_limited"
@dataclass
class SearchRequest:
    """Структура поискового запроса"""
    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:
        """Генерация ключа для кеширования"""
        content = f"{self.query}:{self.depth.value}"
        return hashlib.md5(content.encode()).hexdigest()
@dataclass
class SearchResponse:
    """Структура ответа поиска"""
    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:
    """Менеджер кеша для поисковых запросов"""
    
    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:
        """Получение пути к файлу кеша"""
        return self.cache_dir / f"{cache_key}.json"
    
    def get(self, cache_key: str) -> Optional[Dict]:
        """Получение данных из кеша"""
        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># Проверка TTL</em>
            cached_time = datetime.fromisoformat(data['cached_at'])
            if datetime.now() - cached_time > self.ttl:
                cache_file.unlink()  <em># Удаляем устаревший кеш</em>
                return None
            
            return data['content']
            
        except Exception:
            return None
    
    def set(self, cache_key: str, content: Dict):
        """Сохранение данных в кеш"""
        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"Не удалось сохранить в кеш: {e}")
    
    def clear_expired(self):
        """Очистка устаревшего кеша"""
        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:
    """Ограничитель частоты запросов"""
    
    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:
        """Проверка возможности выполнения запроса"""
        now = time.time()
        
        <em># Очистка устаревших запросов</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):
        """Регистрация выполненного запроса"""
        self.requests.append(time.time())
    
    def get_wait_time(self) -> float:
        """Получение времени ожидания до следующего запроса"""
        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:
    """
    Производственный поисковый агент
    
    Полнофункциональный агент для работы в продакшене с:
    - Кешированием результатов
    - Rate limiting
    - Мониторингом и логированием
    - Обработкой ошибок
    - Статистикой использования
    """
    
    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># Инициализация компонентов</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># Настройка логирования</em>
        self.setup_logging(log_level)
        
        <em># Создание агента</em>
        self.agent = self._create_agent(model)
        
        <em># Статистика</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):
        """Настройка системы логирования"""
        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:
        """Создание агента с производственными настройками"""
        return AnyAgent(
            framework=TinyAgent,
            model=model,
            tools=[self._search_tool],
            system_prompt=self._get_production_system_prompt()
        )
    
    def _get_production_system_prompt(self) -> str:
        """Системный промпт для продакшена"""
        return """
        Вы — профессиональный ИИ-ассистент корпоративного уровня с доступом 
        к актуальной информации из интернета.
        Ваши принципы работы:
        1. ТОЧНОСТЬ: Всегда проверяйте факты через поиск
        2. ИСТОЧНИКИ: Указывайте источники для всех утверждений
        3. ОБЪЕКТИВНОСТЬ: Представляйте разные точки зрения
        4. СТРУКТУРА: Организуйте информацию логично
        5. АКТУАЛЬНОСТЬ: Используйте только свежие данные
        Формат ответа:
        - Краткое резюме (2-3 предложения)
        - Основная информация с источниками
        - Ключевые выводы
        - Дополнительные источники для изучения
        Если информация противоречива или недостоверна - обязательно укажите на это.
        """
    
    def _search_tool(self, query: str, depth: str = "basic") -> Dict:
        """Инструмент поиска с полной обработкой ошибок"""
        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:
        """Выполнение поиска с полной обработкой"""
        start_time = time.time()
        self.stats["total_requests"] += 1
        
        try:
            <em># Проверка кеша</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"Результат получен из кеша: {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># Проверка rate limit</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"Rate limit exceeded. Wait time: {wait_time:.1f}s")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.RATE_LIMITED,
                    error_message=f"Rate limit exceeded. Try again in {wait_time:.1f} seconds",
                    response_time=time.time() - start_time
                )
            
            <em># Выполнение поиска с повторными попытками</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># Экспоненциальная задержка</em>
                        self.logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
                        time.sleep(wait_time)
            
            if result is None:
                self.stats["failed_requests"] += 1
                self.logger.error(f"All search attempts failed for: {request.query}")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.ERROR,
                    error_message=f"Search failed after {self.max_retries} attempts: {last_error}",
                    response_time=time.time() - start_time
                )
            
            <em># Обработка успешного результата</em>
            response_time = time.time() - start_time
            self.stats["successful_requests"] += 1
            self.stats["total_response_time"] += response_time
            
            <em># Сохранение в кеш</em>
            cache_data = {
                'answer': result.get('answer', ''),
                'sources': result.get('results', [])
            }
            self.cache_manager.set(cache_key, cache_data)
            
            self.logger.info(f"Search successful in {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"Unexpected error during search: {e}", exc_info=True)
            
            return SearchResponse(
                request=request,
                status=SearchStatus.ERROR,
                error_message=f"Unexpected error: {str(e)}",
                response_time=time.time() - start_time
            )
    
    def search(self, query: str, depth: SearchDepth = SearchDepth.BASIC) -> str:
        """
        Основной метод поиска для пользователей
        
        Args:
            query: Поисковый запрос
            depth: Глубина поиска (basic/advanced)
            
        Returns:
            Ответ агента с найденной информацией
        """
        try:
            self.logger.info(f"Processing user query: {query}")
            
            <em># Создание контекста для агента</em>
            search_context = f"Выполни поиск и предоставь подробный ответ на запрос: {query}"
            if depth == SearchDepth.ADVANCED:
                search_context += " (используй расширенный поиск для более глубокого анализа)"
            
            result = self.agent.run(search_context)
            return result
            
        except Exception as e:
            self.logger.error(f"Error processing query: {e}", exc_info=True)
            return f"Извините, произошла ошибка при обработке запроса: {str(e)}"
    
    def get_statistics(self) -> Dict[str, Any]:
        """Получение подробной статистики работы"""
        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]:
        """Проверка состояния системы"""
        try:
            <em># Проверка базовой функциональности</em>
            test_response = self._execute_search(SearchRequest("test query"))
            
            return {
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "tavily_api": "connected",
                "cache": "operational",
                "rate_limiter": "operational",
                "last_test_status": test_response.status.value
            }
            
        except Exception as e:
            return {
                "status": "unhealthy",
                "timestamp": datetime.now().isoformat(),
                "error": str(e)
            }
    
    def cleanup(self):
        """Очистка ресурсов при завершении работы"""
        self.logger.info("Starting cleanup process...")
        
        <em># Очистка устаревшего кеша</em>
        self.cache_manager.clear_expired()
        
        <em># Сохранение статистики</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"Statistics saved to {stats_file}")
        except Exception as e:
            self.logger.error(f"Failed to save statistics: {e}")
        
        self.logger.info("Cleanup completed")
<em># Пример использования в продакшене</em>
async def main():
    """Демонстрация работы производственного агента"""
    
    <em># Инициализация агента</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># Проверка состояния</em>
        health = agent.health_check()
        print(f"Agent health: {health['status']}")
        
        <em># Тестовые запросы</em>
        test_queries = [
            "Последние достижения в области квантовых вычислений",
            "Текущие тренды в разработке ИИ 2024",
            "Влияние изменения климата на экономику",
        ]
        
        print("\n=== Демонстрация работы производственного агента ===\n")
        
        for i, query in enumerate(test_queries, 1):
            print(f"Запрос {i}: {query}")
            print("-" * 60)
            
            result = agent.search(query, SearchDepth.BASIC)
            print(result[:500] + "..." if len(result) > 500 else result)
            
            print(f"\nСтатистика:")
            stats = agent.get_statistics()
            print(f"- Всего запросов: {stats['total_requests']}")
            print(f"- Успешных: {stats['success_rate']}")
            print(f"- Из кеша: {stats['cache_hit_rate']}")
            print(f"- Среднее время ответа: {stats['average_response_time']}")
            print("\n" + "="*80 + "\n")
        
        <em># Финальная статистика</em>
        print("Итоговая статистика:")
        final_stats = agent.get_statistics()
        for key, value in final_stats.items():
            print(f"{key}: {value}")
    
    finally:
        <em># Очистка ресурсов</em>
        agent.cleanup()
if __name__ == "__main__":
    asyncio.run(main())

Приложение Б: Конфигурационные файлы

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># Установка системных зависимостей</em>
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

<em># Копирование файлов зависимостей</em>
COPY requirements.txt .

<em># Установка Python зависимостей</em>
RUN pip install --no-cache-dir -r requirements.txt

<em># Копирование кода приложения</em>
COPY . .

<em># Создание необходимых директорий</em>
RUN mkdir -p /app/cache /app/logs

<em># Настройка пользователя без root</em>
RUN useradd -m -u 1000 agent && \
    chown -R agent:agent /app
USER agent

<em># Экспорт портов</em>
EXPOSE 8000

<em># Команда запуска</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:

Приложение В: Тестирование

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():
    """Фикстура для создания мокового агента"""
    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):
        """Тест инициализации агента"""
        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):
        """Тест создания поискового запроса"""
        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># MD5 hash</em>
    
    @patch('src.agents.production_search_agent.TavilyClient')
    def test_successful_search(self, mock_tavily, mock_agent):
        """Тест успешного поиска"""
        <em># Настройка мока</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># Выполнение поиска</em>
        request = SearchRequest("test query")
        response = mock_agent._execute_search(request)
        
        <em># Проверки</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):
        """Тест работы кеша"""
        cache_key = "test_key"
        test_data = {"answer": "cached answer", "sources": []}
        
        <em># Сохранение в кеш</em>
        mock_agent.cache_manager.set(cache_key, test_data)
        
        <em># Получение из кеша</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):
        """Тест ограничения частоты запросов"""
        rate_limiter = mock_agent.rate_limiter
        
        <em># Заполнение лимита</em>
        for _ in range(rate_limiter.max_requests):
            rate_limiter.register_request()
        
        <em># Проверка блокировки</em>
        assert not rate_limiter.can_make_request()
        
        <em># Проверка времени ожидания</em>
        wait_time = rate_limiter.get_wait_time()
        assert wait_time > 0
    
    def test_statistics_tracking(self, mock_agent):
        """Тест отслеживания статистики"""
        initial_stats = mock_agent.get_statistics()
        assert initial_stats["total_requests"] == 0
        assert initial_stats["success_rate"] == "0.0%"
        
        <em># Симуляция запросов</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):
        """Тест проверки состояния системы"""
        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():
    """Тест асинхронной функциональности"""
    <em># Здесь можно тестировать асинхронные операции</em>
    <em># если они будут добавлены в будущем</em>
    pass


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

Заключительные рекомендации

1. Развертывание в продакшене

Контейнеризация: Используйте Docker для изоляции зависимостей и упрощения развертывания.

Мониторинг: Настройте систему мониторинга (Prometheus + Grafana) для отслеживания производительности.

Логирование: Используйте централизованное логирование (ELK stack или аналоги).

2. Безопасность

<em># Безопасное хранение API ключей</em>
import os
from cryptography.fernet import Fernet

class SecureConfig:
    """Безопасное управление конфигурацией"""
    
    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:
        """Загрузка или создание ключа шифрования"""
        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:
        """Шифрование API ключа"""
        return self.cipher.encrypt(api_key.encode()).decode()
    
    def decrypt_api_key(self, encrypted_key: str) -> str:
        """Расшифровка API ключа"""
        return self.cipher.decrypt(encrypted_key.encode()).decode()

<em># Использование</em>
config = SecureConfig()
encrypted_key = config.encrypt_api_key("tvly-YOUR_API_KEY")
<em># Сохранить encrypted_key в конфиге</em>
<em># При использовании:</em>
api_key = config.decrypt_api_key(encrypted_key)

3. Масштабирование

<em># Пример микросервисной архитектуры</em>
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="Search Agent API", 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># Глобальный экземпляр агента</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):
    """Эндпоинт для поискового запроса"""
    try:
        result = agent.search(query.query, SearchDepth(query.depth))
        
        <em># Асинхронное логирование</em>
        background_tasks.add_task(log_search_request, query, result)
        
        return SearchResponse(
            query=query.query,
            answer=result,
            sources=[],  <em># Заполнить реальными источниками</em>
            response_time=0.0,  <em># Заполнить реальным временем</em>
            from_cache=False  <em># Заполнить реальным статусом</em>
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Проверка состояния сервиса"""
    return agent.health_check()

@app.get("/stats")
async def get_statistics():
    """Получение статистики"""
    return agent.get_statistics()

def log_search_request(query: SearchQuery, result: str):
    """Асинхронное логирование запроса"""
    <em># Логирование в базу данных или внешнюю систему</em>
    pass

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

4. Мониторинг и алертинг

<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: "High error rate detected"
          description: "Error rate is {{ $value | humanizePercentage }}"

      - alert: SlowResponseTime
        expr: search_agent_avg_response_time_seconds > 5
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Slow response time"
          description: "Average response time is {{ $value }}s"

5. Непрерывная интеграция

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

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

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt
      
      - name: Run tests
        run: |
          pytest tests/ --cov=src --cov-report=xml
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Run security scan
        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: Build Docker image
        run: |
          docker build -t search-agent:latest .
      
      - name: Push to registry
        run: |
          echo ${{ secrets.DOCKER_PASSWORD }} | docker login -u ${{ secrets.DOCKER_USERNAME }} --password-stdin
          docker push search-agent:latest

6. Документация API

<em># Расширенная документация 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="Smart Search Agent API",
        version="2.0.0",
        description="""
        ## Умный поисковый агент с интеграцией Tavily

        Этот API предоставляет доступ к умному поисковому агенту,
        который использует актуальную информацию из интернета.

        ### Возможности:
        - Поиск актуальной информации в реальном времени
        - Кеширование результатов для повышения производительности
        - Ограничение частоты запросов
        - Подробная статистика использования
        - Мониторинг состояния системы

        ### Аутентификация:
        Используйте API ключ в заголовке `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

Итоговые выводы

Интеграция any-agent с Tavily создает мощную платформу для разработки современных ИИ-приложений. Эта технологическая связка решает критическую проблему доступа к актуальной информации и открывает новые возможности для создания интеллектуальных систем.

Ключевые достижения:

🎯 Универсальность: Единый интерфейс для работы с различными агентными фреймворками

⚡ Актуальность: Доступ к свежей информации в реальном времени через специализированную поисковую систему

🔧 Гибкость: Возможность создания как простых агентов, так и сложных мульти-агентных систем

📊 Надежность: Полный набор инструментов для продакшена: мониторинг, кеширование, обработка ошибок

🚀 Масштабируемость: Готовность к развертыванию в корпоративной среде

Применение в реальных проектах:

  • Корпоративные ассистенты с доступом к актуальной информации
  • Системы анализа рынка для финансовых организаций
  • Исследовательские платформы для научных учреждений
  • Новостные агрегаторы и аналитические системы
  • Образовательные помощники с актуальными знаниями

Будущее развитие:

Технологии ИИ-агентов стремительно развиваются. Комбинация any-agent + Tavily создает прочную основу для адаптации к новым фреймворкам и возможностям, обеспечивая долгосрочную ценность инвестиций в разработку.

Данное руководство предоставляет полный цикл разработки — от базовых примеров до производственных решений. Используйте его как справочник для создания собственных умных агентов с доступом к актуальной информации.

Удачи в разработке! 🤖✨

Метки:

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *