Перейти до вмісту
> 💻 🧠 Код 1001 > > Повний посібник: 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. Додаток A: Повний код виробничого агента
  10. Додаток B: Файли конфігурації
  11. Додаток B: Тести
  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: Розширені можливості

Мультиагентні системи

any-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># Дотримання лімітів частоти</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}с: {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} останні дослідження 2024",
            f"{topic} наукові статті",
            f"{topic} оглядові статті",
            f"{topic} останні відкриття"
        ]
        
        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. Плануйте обробку помилок: завжди передбачайте резервні сценарії

Наступні кроки:

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

Ця технологічна комбінація є потужним інструментом для розробки сучасних ІІ-додатків, здатних працювати з актуальними даними та надавати користувачам релевантну інформацію.

Додаток A: Повний код виробничого агента

<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:
    """Виробничий пошуковий агент
    
    Повний агент для використання у виробництві з:
    - Кешуванням результатів
    - Обмеженням частоти
    - Моніторингом та логуванням
    - Обробкою помилок
    - Статистикою використання
    """
    
    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># Перевірити ліміт частоти</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"Перевищено ліміт частоти. Час очікування: {wait_time:.1f}с")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.RATE_LIMITED,
                    error_message=f"Перевищено ліміт частоти. Повторіть спробу через {wait_time:.1f} секунд",
                    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 + 1} невдала: {e}. Повторна спроба через {wait_time}с...")
                        time.sleep(wait_time)
            
            if result is None:
                self.stats["failed_requests"] += 1
                self.logger.error(f"Усі спроби пошуку не вдалися для: {request.query}")
                
                return SearchResponse(
                    request=request,
                    status=SearchStatus.ERROR,
                    error_message=f"Пошук не вдався після {self.max_retries} спроб: {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"Пошук успішно завершено за {response_time:.2f}с: {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"Неочікувана помилка під час пошуку: {e}", exc_info=True)
            
            return SearchResponse(
                request=request,
                status=SearchStatus.ERROR,
                error_message=f"Неочікувана помилка: {str(e)}",
                response_time=time.time() - start_time
            )
    
    def search(self, query: str, depth: SearchDepth = SearchDepth.BASIC) -> str:
        """
        Основний метод пошуку для користувачів
        
        Args:
            query: Пошуковий запит
            depth: Глибина пошуку (базова/розширена)
            
        Returns:
            Відповідь агента зі знайденою інформацією
        """
        try:
            self.logger.info(f"Обробка запиту користувача: {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"Помилка під час обробки запиту: {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}с",
            "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("тестовий запит"))
            
            return {
                "status": "здоровий",
                "timestamp": datetime.now().isoformat(),
                "tavily_api": "підключено",
                "cache": "працює",
                "rate_limiter": "працює",
                "last_test_status": test_response.status.value
            }
            
        except Exception as e:
            return {
                "status": "нездоровий",
                "timestamp": datetime.now().isoformat(),
                "error": str(e)
            }
    
    def cleanup(self):
        """Очистити ресурси при зупинці"""
        self.logger.info("Запуск процесу очищення...")
        
        <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"Статистика збережена в {stats_file}")
        except Exception as e:
            self.logger.error(f"Помилка під час збереження статистики: {e}")
        
        self.logger.info("Очищення завершено")


<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"Стан агента: {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())

Додаток B: Файли конфігурації

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:

Додаток B: Тести

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="тестовий запит",
            depth=SearchDepth.BASIC
        )
        
        assert request.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 хеш</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': 'Тестова відповідь',
            'results': [{'title': 'Тест', 'url': 'http://test.com'}]
        }
        
        <em># Виконати пошук</em>
        request = SearchRequest("тестовий запит")
        response = mock_agent._execute_search(request)
        
        <em># Твердження</em>
        assert response.status == SearchStatus.SUCCESS
        assert response.answer == 'Тестова відповідь'
        assert len(response.sources) == 1
        assert not response.from_cache
    
    def test_cache_functionality(self, mock_agent):
        """Тестування функціональності кешу"""
        cache_key = "тестовий_ключ"
        test_data = {"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"] == "відповідь з кешу"
    
    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 ["здоровий", "нездоровий"]


@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 або подібне).

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="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: "Виявлено високий рівень помилок"
          description: "Рівень помилок становить {{ $value | humanizePercentage }}"

      - alert: SlowResponseTime
        expr: search_agent_avg_response_time_seconds > 5
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Повільний час відповіді"
          description: "Середній час відповіді становить {{ $value }}с"

5. Безперервна інтеграція

<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: Налаштувати Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Встановити залежності
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt
      
      - name: Виконати тести
        run: |
          pytest tests/ --cov=src --cov-report=xml
      
      - name: Завантажити покриття
        uses: codecov/codecov-action@v3

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Виконати аналіз безпеки
        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: Зібрати образ Docker
        run: |
          docker build -t search-agent:latest .
      
      - name: Відправити в реєстр
        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="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 створює міцну основу для адаптації до нових фреймворків та можливостей, забезпечуючи довгострокову цінність для інвестицій у розробку.

Цей посібник надає повний цикл розробки, від базових прикладів до виробничих рішень. Використовуйте його як довідник для створення власних інтелектуальних агентів з доступом до актуальної інформації.

Успіхів у вашій розробці! 🤖✨

Позначки:

Залишити відповідь

Ваша e-mail адреса не оприлюднюватиметься. Обов’язкові поля позначені *