RAG-системы: как создать корпоративную базу знаний на основе AI

Проблема, которая стоит бизнесу миллионы

По данным McKinsey, сотрудники тратят до 20% рабочего времени на поиск информации внутри компании. Для организации из 100 человек это эквивалентно зарплате 20 штатных сотрудников в год — и эти деньги буквально уходят на бесконечное листание папок, переписку «а где найти...» и ожидание ответа от коллег.

Знакомая картина?

• Новый менеджер неделю ищет актуальный регламент работы с клиентами

• Техподдержка каждый день отвечает на одни и те же вопросы

• Важные решения откладываются, потому что «нужно уточнить в документации»

• Знания уходят вместе с увольняющимися сотрудниками

Традиционный поиск по документам здесь не спасает. Он находит файлы по ключевым словам, но не понимает смысл вопроса. Спросите «как оформить возврат товара» — и получите сотню документов, где встречаются эти слова. А нужен был один конкретный ответ.

Именно эту проблему решают RAG-системы — технология, которая объединила возможности поиска и генеративного AI.

Что такое RAG: объясняем без технического жаргона

RAG (Retrieval-Augmented Generation) — это подход, при котором AI-модель сначала находит релевантную информацию в ваших документах, а затем формулирует ответ на основе найденного.

Представьте умного ассистента, который:

1. Прочитал и запомнил всю документацию компании

2. Понимает вопросы на естественном языке

3. Находит нужную информацию за секунды

4. Отвечает своими словами, ссылаясь на источники

В отличие от обычного ChatGPT, RAG-система работает только с вашими данными. Она не выдумывает факты и не галлюцинирует — каждый ответ подкреплён конкретными документами из вашей базы знаний.

Простая аналогия

Представьте библиотекаря, который:

• Обычный поиск — показывает все книги, где встречается слово «договор»

• RAG-система — понимает, что вам нужен шаблон договора аренды для юрлица, находит именно его и сразу показывает нужный раздел

Архитектура RAG: от данных до ответа

Прежде чем погружаться в детали, посмотрим на общую архитектуру системы:

Рассмотрим каждый этап подробнее на примере нашего реального проекта.

┌─────────────────────────────────────────────────────────────────────────┐
│                         ИСТОЧНИКИ ДАННЫХ                                │
├─────────────────────────────────────────────────────────────────────────┤
│  База данных  │  Документы (PDF, Word)  │  API внешних систем           │
└───────────┬───────────────┬─────────────────────────┬───────────────────┘
            │               │                         │
            ▼               ▼                         ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      ЭТАП 1: ВЕРИФИКАЦИЯ И ОБОГАЩЕНИЕ                   │
├─────────────────────────────────────────────────────────────────────────┤
│  • Проверка качества данных                                             │
│  • Определение результата (успех/неудача)                               │
│  • Добавление контекста и метаданных                                    │
└───────────────────────────────────┬─────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      ЭТАП 2: ГЕНЕРАЦИЯ ДОКУМЕНТОВ                       │
├─────────────────────────────────────────────────────────────────────────┤
│  Преобразование структурированных данных в текст, понятный AI           │
└───────────────────────────────────┬─────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      ЭТАП 3: ВЕКТОРНОЕ ХРАНИЛИЩЕ                        │
├─────────────────────────────────────────────────────────────────────────┤
│  ChromaDB / Pinecone / Qdrant — индексация и быстрый поиск              │
└───────────────────────────────────┬─────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                      ЭТАП 4: RAG API SERVICE                            │
├─────────────────────────────────────────────────────────────────────────┤
│  Поиск похожих документов → Формирование контекста → LLM → Ответ        │
└─────────────────────────────────────────────────────────────────────────┘

Этап 1: Верификация данных — фундамент качества

RAG работает по принципу «мусор на входе — мусор на выходе». Если ваши документы содержат устаревшую или неточную информацию, AI будет давать неточные ответы.

В нашем проекте мы работали с базой из 2.3 миллиона записей. Первый шаг — верификация: проверка каждой записи на соответствие реальным результатам.

Структура данных для верификации

# Таблица для хранения результатов верификации
CREATE TABLE IF NOT EXISTS document_outcomes (
    id BIGSERIAL PRIMARY KEY,
    document_id BIGINT NOT NULL,
    outcome VARCHAR(20) NOT NULL,        -- 'success' или 'fail'
    accuracy_percent DECIMAL(10,4),       -- Точность предсказания
    time_to_result_minutes INT,           -- Время до результата
    metadata JSONB,                        -- Дополнительные данные
    verified_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(document_id)
);

-- Индексы для быстрого поиска
CREATE INDEX idx_outcomes_outcome ON document_outcomes(outcome);
CREATE INDEX idx_outcomes_accuracy ON document_outcomes(accuracy_percent);

Процесс верификации

def verify_document(doc_id, actual_result):
    """
    Верифицирует документ по реальному результату.
    Возвращает outcome: 'success' или 'fail'
    """
    # Получаем предсказание из документа
    prediction = get_document_prediction(doc_id)

    # Сравниваем с реальным результатом
    if prediction_matches_result(prediction, actual_result):
        outcome = 'success'
        accuracy = calculate_accuracy(prediction, actual_result)
    else:
        outcome = 'fail'
        accuracy = 0

    # Сохраняем результат верификации
    save_outcome(doc_id, outcome, accuracy)

    return outcome

Результаты верификации нашего проекта

После обработки 2.2 миллиона записей мы получили статистику:

Эти данные критически важны: теперь RAG-система знает не только «что рекомендовать», но и «насколько надёжна эта рекомендация» на основе исторических данных.

Этап 2: Генерация документов для RAG

Структурированные данные из базы нужно преобразовать в текст, который AI сможет понять и использовать для ответов.

Формат документа

def generate_document_text(record, metadata):
    """
    Преобразует запись из БД в текст для RAG.
    Текст должен быть информативным и содержать все важные детали.
    """

    # Базовая информация
    text_parts = [
        f"{record['category']} в контексте {record['context']}, "
        f"{format_datetime(record['timestamp'])}."
    ]

    # Параметры и индикаторы
    indicators = []

    if 'confidence_score' in metadata:
        level = classify_confidence(metadata['confidence_score'])
        indicators.append(f"Уровень уверенности: {metadata['confidence_score']}% ({level})")

    if 'related_factors' in metadata:
        indicators.append(f"Связанные факторы: {metadata['related_factors']}")

    if indicators:
        text_parts.append("Параметры: " + ", ".join(indicators) + ".")

    # Результат (для верифицированных данных)
    outcome_text = "УСПЕШНО" if record['outcome'] == 'success' else "НЕУСПЕШНО"
    accuracy = record.get('accuracy_percent', 0)

    text_parts.append(
        f"РЕЗУЛЬТАТ: {outcome_text}, точность {accuracy}%."
    )

    return " ".join(text_parts)

Пример сгенерированного документа

{
  "id": "doc_847291",
  "category": "technical_support",
  "context": "server_configuration",
  "text": "Техническая поддержка в контексте настройки сервера, 2025-01-15 14:30. Параметры: Уровень уверенности: 87% (высокий), Связанные факторы: memory_optimization, cache_settings. Факторы: превышение лимита памяти; некорректные настройки кэша. РЕЗУЛЬТАТ: УСПЕШНО, точность 94%.",
  "outcome": "success",
  "confidence": 87,
  "accuracy_percent": 94
}

Пакетная обработка больших объёмов

При работе с миллионами записей важна оптимизация:

def process_documents_batch(batch_size=10000):
    """
    Обрабатывает документы пакетами для экономии памяти.
    """
    offset = 0
    total_processed = 0

    while True:
        # Получаем пакет записей
        records = fetch_records_batch(
            limit=batch_size,
            offset=offset
        )

        if not records:
            break

        # Генерируем документы
        documents = []
        for record in records:
            doc = generate_document_text(record, record['metadata'])
            documents.append({
                'id': f"doc_{record['id']}",
                'text': doc,
                'metadata': extract_metadata(record)
            })

        # Сохраняем пакет
        save_documents_jsonl(documents, 'rag_documents.jsonl')

        total_processed += len(records)
        offset += batch_size

        logger.info(f"Обработано: {total_processed}")

    return total_processed

Этап 3: Построение векторного индекса

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

Как работает векторизация

Каждый текст преобразуется в вектор — массив из сотен чисел. Похожие по смыслу тексты получают похожие векторы:

"Как настроить сервер?"     → [0.12, -0.34, 0.56, ..., 0.78]
"Конфигурация сервера"      → [0.11, -0.32, 0.55, ..., 0.79]  ← похожи!
"Заказ пиццы"               → [-0.45, 0.67, -0.12, ..., 0.23] ← далеко

Создание индекса в ChromaDB

def create_vector_index(documents, collection_name="knowledge_base"):
    """
    Создаёт векторный индекс из документов.
    """
    import chromadb
    from chromadb.config import Settings

    # Инициализация клиента с persistent storage
    client = chromadb.PersistentClient(
        path="./chroma_db",
        settings=Settings(anonymized_telemetry=False)
    )

    # Создаём коллекцию
    collection = client.get_or_create_collection(
        name=collection_name,
        metadata={"description": "Corporate knowledge base"}
    )

    # Индексируем пакетами (для больших объёмов)
    batch_size = 1000

    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]

        collection.add(
            ids=[doc['id'] for doc in batch],
            documents=[doc['text'] for doc in batch],
            metadatas=[{
                'category': doc.get('category', ''),
                'outcome': doc.get('outcome', ''),
                'confidence': doc.get('confidence', 0)
            } for doc in batch]
        )

    logger.info(f"Индекс создан: {collection.count()} документов")
    return collection

Статистика нашего индекса

Этап 4: RAG API Service — мозг системы

Теперь собираем всё вместе в единый сервис, который принимает вопросы и возвращает ответы.

Архитектура сервиса

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List, Dict, Any

app = FastAPI(title="RAG Knowledge Base API")

class QueryRequest(BaseModel):
    """Запрос к RAG системе"""
    query: str                           # Вопрос пользователя
    category: Optional[str] = None       # Фильтр по категории
    context: Optional[str] = None        # Дополнительный контекст
    top_k: int = 50                       # Количество похожих документов

class RAGResponse(BaseModel):
    """Ответ RAG системы"""
    success: bool
    query: str
    similar_found: int
    success_rate: float                  # Процент успешных среди похожих
    recommendation: str                  # PROCEED / SKIP / REVIEW
    analysis: str                        # Развёрнутый анализ от LLM
    sources: List[Dict]                  # Ссылки на источники

Поиск похожих документов

Ключевая особенность нашего подхода — поиск без предварительной фильтрации по результату. Это даёт реальную статистику:

def search_similar(self, query: str, category: Optional[str] = None,
                   top_k: int = 50) -> List[Dict]:
    """
    Ищет похожие документы в векторной базе.
    Возвращает реальное соотношение успешных/неуспешных.
    """

    # Формируем фильтры
    where_filter = None
    if category:
        where_filter = {"category": category}

    # Поиск в ChromaDB
    results = self.collection.query(
        query_texts=[query],
        n_results=top_k,
        where=where_filter
    )

    # Форматируем результаты
    similar_docs = []
    for doc_id, text, metadata in zip(
        results['ids'][0],
        results['documents'][0],
        results['metadatas'][0]
    ):
        similar_docs.append({
            'id': doc_id,
            'text': text,
            'category': metadata.get('category', ''),
            'outcome': metadata.get('outcome', ''),
            'confidence': metadata.get('confidence', 0)
        })

    return similar_docs

Анализ и генерация ответа

def analyze_query(self, request: QueryRequest) -> RAGResponse:
    """
    Полный цикл анализа: поиск → статистика → LLM → ответ
    """

    # 1. Ищем похожие документы
    similar_docs = self.search_similar(
        query=request.query,
        category=request.category,
        top_k=request.top_k
    )

    # 2. Считаем статистику
    success_docs = [d for d in similar_docs if d['outcome'] == 'success']
    fail_docs = [d for d in similar_docs if d['outcome'] == 'fail']

    total = len(similar_docs)
    success_rate = (len(success_docs) / total * 100) if total > 0 else 50

    # 3. Определяем рекомендацию
    if success_rate >= 70:
        recommendation = "PROCEED"
    elif success_rate <= 40:
        recommendation = "SKIP"
    else:
        recommendation = "REVIEW"

    # 4. Формируем контекст для LLM
    context = self._build_context(success_docs, fail_docs, request.query)

    # 5. Генерируем анализ через LLM
    analysis = self._generate_analysis(context, success_rate, recommendation)

    return RAGResponse(
        success=True,
        query=request.query,
        similar_found=total,
        success_rate=round(success_rate, 1),
        recommendation=recommendation,
        analysis=analysis,
        sources=[{'id': d['id'], 'outcome': d['outcome']} for d in similar_docs[:10]]
    )

Интеграция с LLM (GigaChat)

def _generate_analysis(self, context: str, success_rate: float,
                       recommendation: str) -> str:
    """
    Генерирует развёрнутый анализ с помощью LLM.
    """
    from langchain_gigachat import GigaChat

    llm = GigaChat(
        credentials=os.getenv('GIGACHAT_API_KEY'),
        model="GigaChat-Max",  # Используем максимальную модель
        verify_ssl_certs=False
    )

    prompt = f"""Ты — эксперт-аналитик. Проанализируй запрос на основе исторических данных.

ЗАПРОС ПОЛЬЗОВАТЕЛЯ:
{context['query']}

ИСТОРИЧЕСКИЕ ДАННЫЕ:
{context['similar_cases']}

СТАТИСТИКА:
- Найдено похожих случаев: {context['total']}
- Успешных: {context['success_count']} ({success_rate:.1f}%)
- Неуспешных: {context['fail_count']}

Дай краткий анализ (3-4 абзаца):
1. Оцени вероятность успеха на основе исторических данных
2. На какие факторы стоит обратить внимание
3. Возможные риски
4. Итоговая рекомендация: {recommendation}
"""

    response = llm.invoke(prompt)
    return response.content

Pipeline обновления данных в реальном времени

Статичная база знаний быстро устаревает. Настоящая ценность RAG — в постоянном обучении на новых данных.

Архитектура Feedback Loop

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Новый запрос  │────▶│   RAG анализ    │───▶│  Рекомендация   │
└─────────────────┘     └─────────────────┘     └────────┬────────┘
                                                         │
                                                         ▼
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│ Обновление RAG  │◀────│   Верификация   │◀───│ Реальный результат
└─────────────────┘     └─────────────────┘     └─────────────────┘
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│              RAG становится умнее с каждым циклом               │
└─────────────────────────────────────────────────────────────────┘

Сервис автоматического обновления

class RAGUpdateService:
    """
    Сервис для автоматического обновления RAG-базы новыми данными.
    """

    def __init__(self, collection, update_interval_hours=1):
        self.collection = collection
        self.update_interval = update_interval_hours * 3600
        self.last_update_id = self._get_last_processed_id()

    async def process_new_results(self):
        """
        Обрабатывает новые результаты и добавляет в RAG.
        Запускается по расписанию или по событию.
        """
        # Получаем новые верифицированные записи
        new_records = await self._fetch_new_verified_records(
            since_id=self.last_update_id
        )

        if not new_records:
            logger.info("Новых записей нет")
            return 0

        # Генерируем документы
        documents = []
        for record in new_records:
            doc_text = generate_document_text(record, record['metadata'])
            documents.append({
                'id': f"doc_{record['id']}",
                'text': doc_text,
                'metadata': {
                    'category': record['category'],
                    'outcome': record['outcome'],
                    'confidence': record['confidence'],
                    'added_at': datetime.now().isoformat()
                }
            })

        # Добавляем в индекс
        self._add_to_index(documents)

        # Обновляем указатель
        self.last_update_id = new_records[-1]['id']
        self._save_last_processed_id(self.last_update_id)

        logger.info(f"Добавлено {len(documents)} новых документов в RAG")
        return len(documents)

    def _add_to_index(self, documents):
        """Добавляет документы в существующий индекс."""
        self.collection.add(
            ids=[doc['id'] for doc in documents],
            documents=[doc['text'] for doc in documents],
            metadatas=[doc['metadata'] for doc in documents]
        )

Инкрементальное обновление vs полная переиндексация

# Пример расписания обновлений
schedule = {
    'incremental': '0 * * * *',      # Каждый час
    'full_reindex': '0 3 * * 0',     # Воскресенье в 3:00
    'cleanup_old': '0 4 1 * *'       # 1-е число месяца в 4:00
}

Мониторинг качества RAG

class RAGQualityMonitor:
    """
    Отслеживает качество рекомендаций RAG-системы.
    """

    def track_recommendation(self, query_id: str, recommendation: str,
                            predicted_success_rate: float):
        """Сохраняет рекомендацию для последующей проверки."""
        self.pending_verifications[query_id] = {
            'recommendation': recommendation,
            'predicted_rate': predicted_success_rate,
            'timestamp': datetime.now()
        }

    def verify_recommendation(self, query_id: str, actual_outcome: str):
        """Проверяет, насколько точной была рекомендация."""
        if query_id not in self.pending_verifications:
            return

        pred = self.pending_verifications[query_id]

        # Рекомендация была верной?
        was_correct = (
            (pred['recommendation'] == 'PROCEED' and actual_outcome == 'success') or
            (pred['recommendation'] == 'SKIP' and actual_outcome == 'fail')
        )

        # Сохраняем метрику
        self.metrics.record({
            'query_id': query_id,
            'predicted': pred['recommendation'],
            'actual': actual_outcome,
            'correct': was_correct,
            'predicted_rate': pred['predicted_rate']
        })

        del self.pending_verifications[query_id]

    def get_accuracy_report(self, period_days=7):
        """Генерирует отчёт о точности за период."""
        recent = self.metrics.get_since(days=period_days)

        total = len(recent)
        correct = sum(1 for m in recent if m['correct'])

        return {
            'period_days': period_days,
            'total_recommendations': total,
            'correct_recommendations': correct,
            'accuracy_percent': (correct / total * 100) if total > 0 else 0,
            'by_recommendation': self._group_by_recommendation(recent)
        }

Результаты внедрения: цифры и факты

До внедрения RAG

• Точность рекомендаций на основе простых правил: 43%

• Время получения релевантной информации: минуты

• Учёт исторического контекста: отсутствует

После внедрения RAG

Пример ответа системы

{
  "success": true,
  "query": "Настройка кэширования для высоконагруженного сервера",
  "similar_found": 47,
  "success_rate": 72.3,
  "recommendation": "PROCEED",
  "analysis": "На основе анализа 47 похожих случаев из истории,
    вероятность успешного решения составляет 72.3%.

    Ключевые факторы успеха:
    - Корректная настройка TTL (в 89% успешных случаев)
    - Мониторинг hit rate (в 76% успешных случаев)

    Риски:
    - Недостаточный объём памяти (причина 34% неудач)

    Рекомендация: ПРИСТУПИТЬ к настройке с учётом указанных факторов.",
  "sources": [
    {"id": "doc_847291", "outcome": "success"},
    {"id": "doc_623847", "outcome": "success"},
    {"id": "doc_991823", "outcome": "fail"}
  ]
}

Примеры применения RAG в бизнесе

1. Внутренний помощник для сотрудников

Задача: Сотрудники постоянно спрашивают HR, бухгалтерию, IT-отдел одни и те же вопросы.

Решение: RAG-бот в корпоративном мессенджере, который мгновенно отвечает на типовые вопросы и показывает, как решались похожие ситуации в прошлом.

Эффект: HR-отдел освобождается от рутины, сотрудники получают ответы 24/7.

2. Умный FAQ на сайте

Задача: Техподдержка перегружена однотипными вопросами, клиенты ждут ответа часами.

Решение: Виджет на сайте, который понимает вопросы клиентов, отвечает на основе базы знаний и показывает процент успешных решений для похожих проблем.

Эффект: 60-70% обращений закрываются автоматически.

3. Анализ рисков на основе истории

Задача: Принятие решений без учёта исторического опыта приводит к повторению ошибок.

Решение: RAG-система, которая для каждого нового случая находит похожие исторические и показывает статистику успехов/неудач.

Эффект: Снижение ошибок на 25-40% за счёт учёта исторического контекста.

Какие документы подходят для RAG

Стоимость и сроки внедрения

Компоненты стоимости

Сроки

ROI

Типичная окупаемость — 6-12 месяцев при внедрении на 50+ сотрудников.

RAG vs обычный поиск

Частый вопрос: «Почему вы используете GigaChat, а не GPT-4 или Claude?»

Ответ кроется в архитектуре RAG. Языковая модель выполняет последние 10% работы — просто переводит уже найденный контекст в человекочитаемый текст.

90% работы — RAG-система (ваша инфраструктура):           

• Верификация данных по реальным результатам              

• Векторизация и индексация документов                     

• Семантический поиск похожих случаев                      

• Подсчёт статистики и формирование рекомендации          

• Подготовка контекста для LLM 

10% работы — LLM:                                          

"Вот 50 похожих случаев, 35 успешных, 15 неудачных.       

Напиши анализ и объясни рекомендацию."

LLM не принимает решение — решение уже принято на основе данных. Модель лишь объясняет его человеческим языком.

Трансграничная передача данных

Для российского бизнеса это критически важный фактор:

Ключевой момент: При использовании RAG вы отправляете в LLM только обезличенный контекст — выжимку из найденных документов. Конфиденциальные данные остаются в вашей инфраструктуре:

• Векторная база — на вашем сервере

• Исходные документы — на вашем сервере

• В LLM уходит только: «Вот статистика по похожим случаям, сформулируй ответ»

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

Заключение

RAG-системы — это не футуристическая технология, а рабочий инструмент, который уже сегодня экономит компаниям миллионы рублей и тысячи человеко-часов.

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

1. Работает с вашими данными — без передачи конфиденциальной информации вовне

2. Учится на истории — каждый новый результат улучшает качество рекомендаций

3. Масштабируется — от тысяч до миллионов документов

4. Интегрируется — с любыми корпоративными системами

Если ваши сотрудники тратят больше 10 минут в день на поиск информации — RAG окупится. Вопрос не «нужно ли внедрять», а «когда начинать».

MediaTen — разрабатываем AI-решения для бизнеса. Поможем внедрить RAG-систему под ваши задачи: от анализа требований до production-ready решения с автоматическим обновлением данных.