Финальный проект: Архитектура масштабируемого RAG-сервиса

Введение: От прототипа к продакшену

Добро пожаловать на финальный урок курса! Мы прошли долгий путь от первых запросов к API до создания автономных агентов. Сегодня мы займемся самым важным этапом для любого инженера: архитектурой масштабируемого решения.

Многие RAG-системы (Retrieval-Augmented Generation) умирают на этапе PoC (Proof of Concept). Почему? Потому что написать скрипт на 50 строк, который ищет по трем PDF-файлам в локальной папке — это одно. А создать систему, которая обрабатывает терабайты корпоративной документации, отвечает за секунды сотням пользователей одновременно и не разоряет компанию на токенах Gemini — это совершенно другая задача.

В этом уроке мы спроектируем архитектуру RAG-сервиса уровня Enterprise, используя возможности Gemini 3. Мы сосредоточимся не просто на коде, а на LLMOps практиках: наблюдаемости, оценке качества и оптимизации затрат.

Архитектурный паттерн: Разделение Ingestion и Serving

Первое правило масштабируемого RAG — никогда не смешивайте конвейер загрузки данных (Ingestion) и конвейер ответов пользователю (Serving) в одном процессе. Это частая ошибка новичков.

1. Ingestion Pipeline (ETL): Это асинхронный процесс. Документы (PDF, HTML, TXT) поступают в систему, очищаются, разбиваются на чанки (chunking) и векторизуются. Этот процесс ресурсоемкий, но не требует мгновенной реакции.

2. Serving API: Это синхронный процесс. Пользователь задает вопрос, система ищет контекст и генерирует ответ с помощью Gemini. Здесь критична каждая миллисекунда.

Давайте рассмотрим, как правильно организовать разбиение текста на фрагменты (чанкинг), так как это фундамент качественного поиска.

import asyncio
from typing import List
import google.generativeai as genai
from dataclasses import dataclass

# Конфигурация Gemini 3
genai.configure(api_key="YOUR_API_KEY")

@dataclass
class Document:
    content: str
    metadata: dict

class SmartChunker:
    """
    Продвинутый чанкер, который учитывает семантические границы.
    Для Gemini 3 с его большим окном контекста мы можем позволить себе
    более крупные чанки, что улучшает связность.
    """
    def __init__(self, chunk_size: int = 1000, overlap: int = 200):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def split_text(self, text: str) -> List[str]:
        # Упрощенная логика: в реальности здесь стоит использовать 
        # рекурсивный сплиттер или сплиттер на основе spaCy/nltk
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            # Пытаемся найти конец предложения, чтобы не разрывать смысл
            if end < len(text):
                next_period = text.find('.', end)
                if next_period != -1 and next_period - end < 100:
                    end = next_period + 1
            
            chunks.append(text[start:end])
            start = end - self.overlap
        return chunks

async def generate_embeddings_batch(chunks: List[str], model: str = "models/text-embedding-004") -> List[List[float]]:
    """
    Асинхронная генерация эмбеддингов с учетом Rate Limits.
    Используем модель text-embedding-004, оптимизированную для RAG.
    """
    # В продакшене здесь должен быть механизм повторных попыток (retry policy)
    result = genai.embed_content(
        model=model,
        content=chunks,
        task_type="retrieval_document"
    )
    return result['embedding']

# Пример использования
text_data = "Длинный текст договора... " * 100
chunker = SmartChunker(chunk_size=2000) # Gemini любит контекст побольше
chunks = chunker.split_text(text_data)
# embeddings = await generate_embeddings_batch(chunks) # Запуск в асинхронном лупе

Дилемма контекста: RAG против Long Context Window

С выходом Gemini 3, обладающей контекстным окном в миллионы токенов, возникает закономерный вопрос: «Зачем нам вообще нужен RAG? Почему просто не 'скормить' всю документацию в промпт?»

Это отличный вопрос для архитектора. Вот критерии выбора:

• Стоимость (Cost): Отправлять 1 миллион токенов в каждом запросе невероятно дорого. Если у вас 1000 пользователей в день, бюджет закончится к обеду. RAG позволяет отправлять только релевантные 5-10 тысяч токенов.
• Задержка (Latency): Обработка огромного контекста занимает время (Time to First Token). Для чат-бота ожидание в 30 секунд недопустимо. RAG обеспечивает быстрый ответ.
• Точность (Needle in a Haystack): Хотя Gemini великолепно справляется с поиском в большом контексте, классический поиск (особенно гибридный) часто дает более предсказуемый результат для специфических фактов.

Гибридная стратегия: Мы используем векторный поиск, чтобы найти 10-20 самых релевантных документов, и подаем их в Gemini. Это «золотая середина».

Слой извлечения (Retrieval Layer): Reranking

Одной из самых частых проблем векторного поиска является потеря нюансов. Косинусное сходство (Cosine Similarity) находит семантически близкие тексты, но не всегда те, которые лучше всего отвечают на конкретный вопрос.

Для решения этой проблемы в продакшене внедряют этап Reranking (Переранжирование).

1. Retrieval: Быстро достаем 50 кандидатов из векторной базы данных (Qdrant/Weaviate/Pinecone).
2. Reranking: Используем Cross-Encoder модель (или специализированное API), которая медленнее, но гораздо точнее оценивает релевантность каждой пары «Вопрос — Документ».
3. Top-K: Оставляем топ-5 документов после переранжирования для отправки в Gemini.

class RAGService:
    def __init__(self, vector_db_client, gemini_model):
        self.vector_db = vector_db_client
        self.model = gemini_model
        
    async def retrieve_and_rerank(self, query: str) -> List[str]:
        # 1. Быстрый поиск (Retrieval)
        # Получаем больше кандидатов, чем нужно (например, 20)
        initial_results = await self.vector_db.search(query, limit=20)
        
        # 2. Переранжирование (Reranking)
        # Здесь может использоваться сторонний Cross-Encoder или специфический промпт к Gemini
        # Для примера используем Gemini как Reranker (дорогой, но мощный способ)
        rerank_prompt = f"""
        Оцени релевантность следующих отрывков для вопроса: '{query}'.
        Верни JSON список индексов наиболее полезных отрывков (максимум 5), 
        отсортированных по убыванию полезности.
        
        Отрывки:
        {self._format_candidates(initial_results)}
        """
        
        response = self.model.generate_content(rerank_prompt, generation_config={"response_mime_type": "application/json"})
        top_indices = self._parse_json(response.text)
        
        return [initial_results[i] for i in top_indices]

    async def generate_answer(self, query: str, context: List[str]):
        system_instruction = """
        Ты - экспертный аналитик. Используй ТОЛЬКО предоставленный контекст для ответа.
        Если информации недостаточно, так и скажи. Ссылайся на источники.
        """
        
        full_prompt = f"Контекст:\n{''.join(context)}\n\nВопрос: {query}"
        
        # Используем стриминг для лучшего UX
        response = self.model.generate_content_stream(full_prompt)
        return response

LLMOps: Оценка качества (Evaluation)

Как понять, что ваше изменение в промпте или размере чанка улучшило систему, а не сломало её? На глаз? В продакшене это недопустимо.

Вам нужен автоматизированный пайплайн оценки. Стандарт индустрии — фреймворк RAGAS (Retrieval Augmented Generation Assessment) или его аналоги. Мы оцениваем три метрики:

• Faithfulness (Достоверность): Не выдумал ли Gemini что-то, чего нет в контексте (борьба с галлюцинациями).
• Answer Relevance (Релевантность ответа): Ответил ли бот на поставленный вопрос.
• Context Precision (Точность контекста): Насколько полезным был найденный в базе мусор или золото.

В проекте мы будем использовать подход «LLM-as-a-Judge»: мы просим более мощную модель (например, Gemini 1.5 Pro) оценить ответы более быстрой модели (Gemini 1.5 Flash), которую используем в рантайме.

python
import google.generativeai as genai
from google.api_core.exceptions import ResourceExhausted

def robust_generate(query, context, model_primary, enable_web_search=True):
    try:
        # Попытка 1: Строгий RAG
        prompt = f"Контекст: {context}\nВопрос: {query}"
        response = model_primary.generate_content(prompt)
        
        # Проверка на отказ отвечать из-за отсутствия данных
        if "информации недостаточно" in response.text.lower():
            raise ValueError("Not enough context")
            
        return response.text

    except ValueError:
        # Попытка 2: Grounding (Веб-поиск)
        if enable_web_search:
            print("Переключение на веб-поиск...")
            # Настройка инструментов для поиска
            tools_config = {'tools': [{'google_search_retrieval': {}}]}
            fallback_model = genai.GenerativeModel('gemini-1.5-pro', tools=tools_config['tools'])
            
            fallback_response = fallback_model.generate_content(query)
            return f"На основе веб-поиска: {fallback_response.text}"
        else:
            return "Извините, во внутренних документах нет информации, а доступ в интернет отключен."

    except ResourceExhausted:
        return "Система перегружена, попробуйте позже."
            
    except Exception as e:
        # Логирование реальной ошибки для разработчиков
        print(f"CRITICAL ERROR: {e}")
        return "Произошла непредвиденная ошибка."

Финальные штрихи: Кэширование и Трейсинг

Чтобы снизить затраты и ускорить ответы на частые вопросы (например, «Как оформить отпуск?»), необходимо внедрить Семантическое Кэширование (Semantic Caching). В отличие от обычного кэша (по точному совпадению ключа), семантический кэш понимает, что «Как взять отгул» и «Как оформить отпуск» — это одно и то же, и возвращает сохраненный ответ без обращения к LLM.

И наконец, для продакшена обязателен Трейсинг (Tracing). Инструменты вроде LangSmith, Arize Phoenix или Weights & Biases позволяют видеть цепочку вызовов: Вопрос -> Эмбеддинг -> DB Search -> Rerank -> LLM -> Ответ. Это единственный способ отлаживать сложные RAG-системы.