Создаем ИИ-агента на стероидах: RAG + SQL + Поиск в Интернет на smolagents

Помните тот момент, когда понимаешь, что твой ИИ-агент — это просто чат-бот с прикрученным поиском? Он отвечает на простые вопросы, но стоит попросить его проанализировать данные из базы и сверить с актуальной информацией из сети — и он теряется. Знакомое чувство? Я через это проходил десятки раз.

Сегодня соберем агента, который не просто отвечает. Он думает, планирует, запрашивает данные из SQL-базы, ищет актуальную информацию в интернете и синтезирует ответ на основе всего этого. Не очередной RAG с векторным поиском, а полноценный исследовательский инструмент.

Почему smolagents, а не LangChain или LlamaIndex?

Открою секрет: я перепробовал все популярные фреймворки. LangChain — монстр с кучей абстракций, где простой агент требует 500 строк кода. LlamaIndex — специалист по RAG, но с агентами у него не очень. Smolagents — минималистичный, но мощный. Он не пытается быть всем для всех, а делает одну вещь хорошо: создает агентов, которые умеют работать с инструментами.

Архитектура нашего монстра

Представьте агента как продвинутого аналитика. У него есть три источника информации:

• SQL-база с внутренними данными (продажи, пользователи, логи)
• Векторная база с документацией и историческими данными (RAG)
• Поиск в интернете для актуальной информации

Агент сам решает, к какому источнику обратиться, как комбинировать данные и когда остановиться. Это не последовательная цепочка вызовов, а интеллектуальный планировщик.

Подготовка: ставим все необходимое

1 Устанавливаем smolagents и зависимости

pip install smolagents==0.4.2
pip install chromadb==0.5.0  # для векторной базы
pip install duckdb==1.0.0    # легковесная SQL база
pip install beautifulsoup4==4.13.0  # для парсинга веб-страниц
pip install google-search-results==2.4.2  # для поиска в интернете

2 Настраиваем модель LLM

Smolagents поддерживает все популярные провайдеры. Я буду использовать OpenAI, потому что у них стабильный API и хорошая поддержка function calling. Но можете взять Anthropic Claude 3.7 или Google Gemini 2.5 — smolagents работает со всеми.

import os
from smolagents import OpenAIServerModel

# Для локальных моделей используйте LiteLLM или vLLM
# from smolagents import LiteLLMModel

os.environ["OPENAI_API_KEY"] = "ваш_ключ"

# GPT-4o-mini быстрее и дешевле для агентных задач
model = OpenAIServerModel(
    model="gpt-4o-mini",
    api_base="https://api.openai.com/v1",
    temperature=0.1  # Низкая температура для детерминированных ответов
)

Создаем инструменты: SQL, RAG и поиск

Инструменты в smolagents — это функции Python с декоратором @tool. Агент видит их описание и решает, когда использовать.

3 SQL-инструмент для работы с базами данных

import duckdb
from smolagents import tool

class DatabaseAgent:
    def __init__(self, db_path="company_data.db"):
        self.conn = duckdb.connect(db_path)
        # Создаем тестовые таблицы если их нет
        self._setup_database()
    
    def _setup_database(self):
        self.conn.execute("""
        CREATE TABLE IF NOT EXISTS sales (
            id INTEGER PRIMARY KEY,
            product_name VARCHAR,
            amount DECIMAL(10,2),
            sale_date DATE,
            region VARCHAR
        )""")
        
        self.conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            user_id INTEGER PRIMARY KEY,
            email VARCHAR,
            signup_date DATE,
            plan VARCHAR
        )""")
    
    @tool
    def query_database(self, sql_query: str) -> str:
        """
        Выполняет SQL запрос к базе данных компании.
        Используй для получения данных о продажах, пользователях, метриках.
        
        Args:
            sql_query: SQL запрос для выполнения
        """
        try:
            result = self.conn.execute(sql_query).fetchall()
            return str(result)
        except Exception as e:
            return f"Ошибка выполнения запроса: {str(e)}"
    
    @tool
    def get_table_schema(self, table_name: str = None) -> str:
        """
        Возвращает схему таблиц в базе данных.
        Полезно перед написанием SQL запросов.
        """
        if table_name:
            query = f"DESCRIBE {table_name}"
        else:
            query = "SHOW TABLES"
        
        result = self.conn.execute(query).fetchall()
        return str(result)

4 RAG-инструмент с векторным поиском

Здесь многие совершают ошибку — создают один огромный RAG на все документы. Лучше разделить по темам. Вот минимальный работающий вариант:

import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
import hashlib

class RAGAgent:
    def __init__(self, collection_name="company_docs"):
        self.client = chromadb.Client(Settings(
            chroma_db_impl="duckdb+parquet",
            persist_directory="./chroma_db"
        ))
        
        # Используем современную модель для эмбеддингов
        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
        
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}
        )
    
    @tool
    def search_documents(self, query: str, n_results: int = 5) -> str:
        """
        Ищет релевантные документы в базе знаний компании.
        Используй для поиска документации, инструкций, исторических данных.
        
        Args:
            query: Поисковый запрос
            n_results: Количество результатов (по умолчанию 5)
        """
        query_embedding = self.embedder.encode(query).tolist()
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results
        )
        
        if not results["documents"]:
            return "Документы не найдены"
        
        formatted_results = []
        for i, (doc, metadata) in enumerate(zip(results["documents"][0], 
                                               results["metadatas"][0])):
            formatted_results.append(
                f"Документ {i+1}:\n"
                f"Содержание: {doc[:500]}...\n"
                f"Источник: {metadata.get('source', 'Неизвестно')}\n"
                f"---"
            )
        
        return "\n\n".join(formatted_results)
    
    def add_document(self, text: str, metadata: dict = None):
        """Вспомогательный метод для добавления документов"""
        doc_id = hashlib.md5(text.encode()).hexdigest()
        embedding = self.embedder.encode(text).tolist()
        
        self.collection.add(
            documents=[text],
            embeddings=[embedding],
            metadatas=[metadata or {}],
            ids=[doc_id]
        )

5 Инструмент поиска в интернете

from serpapi import GoogleSearch
import requests
from bs4 import BeautifulSoup

class WebSearchAgent:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.getenv("SERPAPI_KEY")
    
    @tool
    def search_web(self, query: str, num_results: int = 3) -> str:
        """
        Ищет актуальную информацию в интернете.
        Используй для получения свежих данных, новостей, проверки фактов.
        
        Args:
            query: Поисковый запрос
            num_results: Количество результатов (1-10)
        """
        if not self.api_key:
            return "API ключ для поиска не настроен"
        
        search = GoogleSearch({
            "q": query,
            "api_key": self.api_key,
            "num": num_results
        })
        
        try:
            results = search.get_dict()
            organic_results = results.get("organic_results", [])
            
            if not organic_results:
                return "Результаты поиска не найдены"
            
            formatted = []
            for i, result in enumerate(organic_results[:num_results]):
                title = result.get("title", "Без названия")
                snippet = result.get("snippet", "Без описания")
                link = result.get("link", "#")
                
                # Пытаемся получить больше контента
                try:
                    page_content = self._fetch_page_content(link)
                    content_preview = page_content[:500] + "..." if len(page_content) > 500 else page_content
                except:
                    content_preview = "Не удалось загрузить содержимое"
                
                formatted.append(
                    f"Результат {i+1}:\n"
                    f"Заголовок: {title}\n"
                    f"Сниппет: {snippet}\n"
                    f"Контент: {content_preview}\n"
                    f"Ссылка: {link}\n"
                    f"---"
                )
            
            return "\n\n".join(formatted)
            
        except Exception as e:
            return f"Ошибка поиска: {str(e)}"
    
    def _fetch_page_content(self, url: str) -> str:
        """Загружает и парсит содержимое страницы"""
        headers = {
            "User-Agent": "Mozilla/5.0 (compatible; ResearchBot/1.0)"
        }
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.content, "html.parser")
        
        # Удаляем скрипты и стили
        for script in soup(["script", "style"]):
            script.decompose()
        
        text = soup.get_text()
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
        text = " ".join(chunk for chunk in chunks if chunk)
        
        return text

Собираем агента воедино

Теперь самый интересный момент — создаем агента, который будет использовать все три инструмента. Smolagents делает это элегантно:

from smolagents import Agent, Tool

# Создаем экземпляры наших агентов
db_agent = DatabaseAgent()
rag_agent = RAGAgent()
web_agent = WebSearchAgent()

# Собираем все инструменты в один список
tools = [
    Tool.from_function(db_agent.query_database),
    Tool.from_function(db_agent.get_table_schema),
    Tool.from_function(rag_agent.search_documents),
    Tool.from_function(web_agent.search_web)
]

# Создаем системный промпт, который объясняет агенту его возможности
system_prompt = """Ты — аналитический ассистент компании. У тебя есть доступ к:
1. SQL базе данных с продажами и пользователями
2. Векторной базе знаний с документацией
3. Поиску в интернете для актуальной информации

Перед ответом на вопрос:
1. Подумай, какие данные нужны
2. Проверь, есть ли информация в базе знаний компании
3. Если нужны свежие данные — поищи в интернете
4. Синтезируй ответ на основе всех источников

Будь кратким, но информативным. Если данные противоречат друг другу — укажи на это.
"""

# Создаем агента
agent = Agent(
    model=model,
    tools=tools,
    max_steps=10,  # Максимальное количество шагов
    add_base_tools=True,  # Добавляем базовые инструменты (калькулятор и т.д.)
    system_prompt=system_prompt
)

Тестируем на реальных задачах

Давайте посмотрим, как наш агент справляется с комплексными запросами:

# Пример 1: Анализ продаж с контекстом
query = """
Проанализируй продажи за последний квартал. 
Сравни с аналогичным периодом прошлого года.
Учти текущие рыночные тренды в нашей отрасли.
"""

result = agent.run(query)
print(f"Ответ агента: {result}")
print(f"Использованные инструменты: {agent.traces}")

Что происходит внутри:

1. Агент анализирует запрос и понимает, что нужны данные из SQL
2. Сначала запрашивает схему таблиц (get_table_schema)
3. Формирует SQL запрос для получения продаж за последний квартал
4. Параллельно ищет в RAG исторические отчеты
5. Ищет в интернете текущие рыночные тренды
6. Синтезирует ответ на основе всех источников

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

Проблема 1: Агент слишком часто обращается к поиску

Решение — добавляем квоты и приоритеты:

class SmartWebSearchAgent(WebSearchAgent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.search_count = 0
        self.max_searches_per_session = 3
    
    @tool
    def search_web(self, query: str, num_results: int = 3) -> str:
        """
        Ищет в интернете. Используй экономно — максимум 3 запроса за сессию.
        Перед использованием проверь, есть ли информация в локальных базах.
        """
        if self.search_count >= self.max_searches_per_session:
            return "Лимит поисковых запросов исчерпан. Используй локальные данные."
        
        self.search_count += 1
        return super().search_web(query, num_results)

Проблема 2: SQL инъекции и опасные запросы

class SafeDatabaseAgent(DatabaseAgent):
    SAFE_TABLES = {"sales", "users", "products"}
    
    @tool
    def query_database(self, sql_query: str) -> str:
        """
        Выполняет SQL запрос. Разрешены только SELECT запросы.
        Доступные таблицы: sales, users, products
        """
        # Проверяем, что это SELECT запрос
        if not sql_query.strip().upper().startswith("SELECT"):
            return "Разрешены только SELECT запросы"
        
        # Проверяем, что запрашиваются только разрешенные таблицы
        query_upper = sql_query.upper()
        for table in self.SAFE_TABLES:
            if f"FROM {table.upper()}" in query_upper or f"JOIN {table.upper()}" in query_upper:
                return super().query_database(sql_query)
        
        return f"Доступ запрещен. Разрешены только таблицы: {', '.join(self.SAFE_TABLES)}"

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

Когда агент начинает тормозить (а он начнет), вот что проверять:

Распространенные ошибки (и как их избежать)

Я наступил на эти грабли, чтобы вам не пришлось:

Куда развивать дальше

Собранный агент — хорошее начало, но в production нужно больше. Вот что добавить:

• Кэширование результатов: Не искать одно и то же дважды. Redis или простой dict с TTL
• Логирование и мониторинг: Записывать все шаги агента. Потом анализировать, где он ошибается
• Human-in-the-loop: Для критических операций (перевод денег, изменение данных) запрашивать подтверждение
• Мультиагентная архитектура: Специализированные агенты для разных задач, как в статье про Agent Skills

Самое важное — начать с простого. Не пытайтесь сразу построить DeepResearch от Яндекса. Сначала сделайте агента, который отвечает на один тип вопросов идеально. Потом масштабируйте.

Через месяц использования такого агента вы поймете странную вещь: 80% его работы — это не генерация гениальных инсайтов, а банальное соединение данных из разных источников. Но именно эта рутина занимает у аналитиков 90% времени. И именно здесь ИИ показывает свою настоящую ценность — не как творческий гений, а как неутомимый ассистент, который не устает искать, сопоставлять, проверять.

Теперь у вас есть рабочий агент. Не идеальный, но работающий. Дальше — экспериментируйте, добавляйте инструменты, учите его на своих ошибках. Через пару недель он станет незаменимым помощником. А через месяц вы будете удивляться, как раньше жили без него.