Логика и части кода приводятся из функция № 598 в составе платформы ProTalk
Представьте, что вам нужно найти ответ на конкретный вопрос — не в интернете, а внутри закрытой корпоративной базы: сотни PDF-файлов, таблиц Excel, презентаций и внутренних сайтов. Обычный поиск по ключевым словам даст кучу мусора или не найдёт ничего, если вы написали вопрос «не теми словами».
Именно для этого была создана технология RAG — Retrieval-Augmented Generation, «поиск с последующей генерацией». Это не просто поиск и не просто ИИ — это их комбинация. Система сначала находит нужные фрагменты текста, а потом объясняет их на человеческом языке.
«Представьте опытного референта, который перед ответом на ваш вопрос сначала прочитал всю нужную документацию — и теперь говорит своими словами, но строго по источникам.»
1. Подготовка: документы превращаются в числа
Прежде чем отвечать на вопросы, система должна «прочитать» все документы и запомнить их особым образом. Это происходит один раз при первой загрузке файлов.
Шаг 1. Извлечение текста
Программа умеет читать практически любой формат: PDF, Word, Excel, PowerPoint, CSV-таблицы, HTML-страницы, даже исходный код. У каждого формата своя «читалка» — но на выходе всегда получается обычный текст.
# Словарь «читалок» для разных форматов файлов SUPPORTED = { '.pdf': lambda f: ''.join(p.extract_text() or '' for p in PyPDF2.PdfReader(f).pages), '.docx': lambda f: '\n'.join(p.text for p in Document(f).paragraphs), '.pptx': lambda f: '\n'.join( shape.text for slide in Presentation(f).slides for shape in slide.shapes if hasattr(shape, "text") and shape.text.strip() ), '.xlsx': extract_xlsx, '.csv': extract_csv, # и ещё десяток форматов — .txt, .json, .py, .js и т.д. }
Аналогия. Это как если бы вы получили стопку документов разного вида — газеты, рукописи, распечатки — и переписали всё в единый список. Формат не важен, важно содержание.
Шаг 2. Нарезка на фрагменты
Целый документ нельзя отдать ИИ целиком — это слишком много. Поэтому текст нарезается на чанки — небольшие смысловые кусочки по несколько абзацев. Но нарезка делается умно, не механически.
Сначала система определяет тип контента:
def detect_content_type(text): """Определяет тип контента: table, code или prose.""" # Это HTML-таблица или TSV-данные (много табуляций)? if re.search(r'<table|</td>|<tr>', text, re.IGNORECASE): return 'table' tab_lines = sum(1 for line in text.splitlines() if '\t' in line) if tab_lines > max(3, len(text.splitlines()) * 0.4): return 'table' # Это программный код? if re.search(r'function\s+\w+\s*\(|\bdef\s+\w+\s*\(|\bclass\s+\w+\b', text): return 'code' # Иначе — обычный текст (prose) return 'prose'
В зависимости от типа применяется разная стратегия нарезки:
Таблица → режем по строкам, но копируем заголовок в каждый кусок (чтобы данные не потеряли контекст)
Код → режем по пустым строкам между функциями/классами, не разрывая логику
Обычный текст → ищем смысловые границы между темами (подробнее ниже)
# Для таблиц: сохраняем заголовок в каждом чанке def split_table_by_rows(text, chunk_size): lines = [l for l in text.splitlines() if l.strip()] header = lines[0] # первая строка — заголовок chunks = [] cur_lines = [header] for line in lines[1:]: if sum(len(l) for l in cur_lines) + len(line) >= chunk_size: chunks.append('\n'.join(cur_lines)) cur_lines = [header, line] # ← заголовок всегда остаётся else: cur_lines.append(line) if cur_lines: chunks.append('\n'.join(cur_lines)) return chunks
Аналогия. Представьте, что вы готовите картотеку. Каждая карточка — один чанк. Вы не хотите, чтобы начало одной темы попало на карточку с концом другой. Поэтому вы аккуратно разделяете текст там, где меняется смысл.
Шаг 3. Семантическая нарезка: ищем смысловые границы
Для обычного текста нарезка делается не по символам, а по смыслу. Алгоритм разбивает текст на предложения, получает эмбеддинги каждого (подробнее о них — в следующем шаге) и ищет места, где смысл резко меняется.
# Упрощённая схема семантической нарезки sims = [] for i in range(len(sentences) - 1): # Косинусное сходство между соседними предложениями sims.append(float(np.dot(sent_embs[i], sent_embs[i + 1]))) # Адаптивный порог: нижний квартиль сходств adaptive_threshold = float(np.percentile(sims, 25)) # Там, где сходство падает ниже порога — граница между темами boundaries = {i + 1 for i, sim in enumerate(sims) if sim < adaptive_threshold}
Если сходство между двумя соседними предложениями резко падает — значит, тема сменилась. Здесь и проходит граграница чанка.
Шаг 4. Превращение текста в числа (эмбеддинги)
Каждый фрагмент отправляется в специальную нейросеть, которая возвращает вектор — список из 1536 чисел. Этот вектор — математическое «описание смысла» фрагмента.
def get_embeddings(texts, batch_size=100): embeddings = [] for i in range(0, len(texts), batch_size): resp = requests.post( "https://openrouter.ai/api/v1/embeddings", headers=HEADERS, json={ "model": "openai/text-embedding-3-small", "input": texts[i:i + batch_size], "encoding_format": "float" }, ).json() embeddings.extend(d["embedding"] for d in resp.get("data", [])) return np.array(embeddings)
Аналогия. Представьте систему координат, где каждая тема — это направление в пространстве. «Бухгалтерия» — одно направление, «технологии» — другое. Текст про налоги окажется близко к «бухгалтерии». Два похожих по смыслу текста — близко друг к другу в этом пространстве, даже если написаны разными словами. Это и есть суть смыслового поиска.
Все векторы сохраняются в базу данных вместе с исходным текстом чанков. Подготовка завершена.
2. Поиск: как система находит нужное
Когда пользователь задаёт вопрос, начинается многоэтапный поиск. Просто сравнить слова — недостаточно. Система использует сразу несколько методов.
Шаг 1. Вопрос тоже превращается в вектор
Вопрос пользователя проходит через ту же нейросеть. Теперь у нас есть «математический портрет» вопроса — и мы можем искать похожие портреты среди чанков.
Шаг 2. Переформулировки вопроса
ИИ генерирует 3 разных версии вашего вопроса — с другими словами, но тем же смыслом. Это помогает найти фрагменты, которые используют иную терминологию.
# Запрос к ИИ: придумай 3 переформулировки exp_resp, _ = call_ai_with_retry( "Сгенерируй 3 разных переформулировки следующего вопроса для поиска в базе знаний. " "Пиши только переформулировки, каждую с новой строки, без нумерации.\nВопрос: " + query, rerank_model, ) # Затем все варианты тоже превращаются в векторы и ищутся параллельно expansion_variants = [v.strip() for v in exp_resp["choices"][0]["message"]["content"].split('\n')]
Шаг 3. Двойной поиск — смысловой и по ключевым словам
Система ищет одновременно двумя способами:
Смысловой поиск (векторный): находит фрагменты, близкие по смыслу, даже с другими словами.
Поиск по ключевым словам (BM25): классический алгоритм, похожий на поиск Google. Хорошо находит точные совпадения, числа, имена.
# BM25: поиск по ключевым словам # k1 и b — стандартные параметры алгоритма k1, b = 1.5, 0.75 avgdl = sum(len(c['content'].split()) for c in chunks_data) / N for chunk in chunks_data: text_words = re.findall(r'\w+', chunk['content'].lower()) tf = defaultdict(int) for w in text_words: tf[w] += 1 doc_len = len(text_words) # Итоговый BM25-score для чанка score = sum( idf.get(w, 0) * (tf[w] * (k1 + 1)) / (tf[w] + k1 * (1 - b + b * doc_len / avgdl)) for w in query_words )
Шаг 4. Объединение результатов через RRF
Результаты обоих поисков объединяются по формуле RRF (Reciprocal Rank Fusion) — она «голосует» между методами. Фрагменты, которые хорошо себя показали сразу в нескольких поисках, поднимаются наверх.
# RRF: чем выше ранг в каждом поиске — тем больше очков # k_rrf — параметр сглаживания (60 по умолчанию) for rank, idx in enumerate(sorted_by_vector): rrf_scores[idx] += 1.0 / (k_rrf + rank) for rank, idx in enumerate(sorted_by_bm25): rrf_scores[idx] += bm25_normalized[idx] * (1.0 / (k_rrf + rank))
Аналогия. Представьте, что у вас два эксперта и каждый составил свой рейтинг нужных страниц. RRF — это как взять оба рейтинга и выдать победителя по сумме мест: кто стабильно высоко у обоих, тот и первый.
Шаг 5. Финальная фильтрация через ИИ (ре-ранкинг)
Топ-50 кандидатов отправляются ИИ с вопросом: «Какие из этих фрагментов реально полезны для ответа?» ИИ выбирает только самые релевантные — обычно 5–10 штук.
# Отправляем кандидатов ИИ для финального отбора candidate_texts = [f"[{i+1}]: {chunk['content'][:3000]}" for i, chunk in enumerate(top_candidates)] prompt = ( "Ты эксперт. Выбери строго минимальное количество лучших фрагментов, " "достаточных для ответа на вопрос.\n" f"Вопрос: {query}\n\n" f"Фрагменты:\n{chr(10).join(candidate_texts)}\n\n" f"Верни номера (через запятую) ТОЛЬКО самых релевантных фрагментов (максимум {ask_count}). " "Если релевантных нет — верни 'NONE'.\nОтвет:" )
Особый случай: цитаты и точные числа
Если в вопросе есть число или фраза в кавычках — система принудительно включает все фрагменты с точным совпадением, независимо от рейтинга.
# Если в вопросе есть кавычки — ищем точное совпадение в чанках quote_matches = re.findall(r'«([^»]+)»|"([^"]+)"', query) for match in quote_matches: quote = (match[0] or match[1]).strip() if len(quote) > 5: for i, chunk in enumerate(chunks_data): if quote in chunk['content']: forced_indices.append(i) # принудительно включаем этот чанк break # Аналогично для чисел из вопроса query_numbers = re.findall(r'\b\d+\b', query) for num in query_numbers: pattern = re.compile(rf'\b(№|N|номер)\s*{num}\b', re.IGNORECASE) for i, chunk in enumerate(chunks_data): if pattern.search(chunk['content']): forced_indices.append(i)
«Система знает, когда вы ищете конкретный факт — и ведёт себя иначе, чем когда вы просите объяснить принцип.»
3. Умные надстройки
Помимо базового поиска, система включает ряд продвинутых механизмов.
Иерархический поиск: сначала документ, потом фрагмент
Если документов очень много, полный перебор всех чанков был бы медленным. Система сначала находит нужные документы, а потом ищет фрагменты только внутри них.
# Этап 1: находим топ-5 наиболее релевантных документов # (берём первый чанк каждого как "аннотацию") doc_summaries = [chunks_data_by_doc[doc_id][0]['content'][:500] for doc_id in all_doc_ids] doc_embs = get_embeddings(doc_summaries) sims = np.dot(doc_embs_normalized, query_embedding_normalized) top_doc_ids = [all_doc_ids[i] for i in np.argsort(sims)[::-1][:5]] # Этап 2: ищем чанки только среди выбранных документов chunks_data = [ch for ch in chunks_data if ch['doc_id'] in top_doc_ids]
Аналогия. Как искать в нужном ящике картотеки, а не перебирать все ящики подряд.
Итеративный поиск: несколько раундов
Если вопрос сложный («сравни А и Б» или «почему...»), одного поиска может не хватить. Система делает несколько раундов.
# После первого поиска спрашиваем ИИ: чего ещё не хватает? prompt = ( "Если информации достаточно — верни 'NONE'. " "Если не хватает, сформулируй уточняющий под-запрос.\n" f"Исходный вопрос: {base_query}\n\nКонтекст:\n{combined_context}\n\nПод-запрос или NONE:" ) resp, _ = call_ai_with_retry(prompt, ai_model) follow_up = resp["choices"][0]["message"]["content"].strip() if "NONE" not in follow_up.upper(): # Делаем ещё один поиск по уточнённому запросу follow_context, follow_used, _ = get_top_context_advanced( chunks_data, follow_up, get_embeddings([follow_up])[0], ... ) combined_context += "\n\n" + follow_context
Расширение контекста: берём соседей
Когда нашёлся нужный фрагмент, система автоматически прихватывает соседние — если они по смыслу близки. Это помогает не потерять важный контекст вокруг ответа.
def get_adaptive_neighbors(idx, sim_threshold=0.7, max_window=3): """Расширяем контекст в обе стороны, пока соседние чанки остаются достаточно похожи по смыслу.""" chunk = chunks_data[idx] doc_id = chunk['doc_id'] cur_idx = chunk['chunk_index'] neighbors = [] for step in range(1, max_window + 1): next_idx = cur_idx + step # Считаем косинусное сходство с предыдущим чанком sim = float(np.dot(chunk_embs_norm[cur_idx + step - 1], chunk_embs_norm[cur_idx + step])) if sim < sim_threshold: break # тема сменилась — дальше не идём neighbors.append(next_idx) return neighbors
Разнообразие результатов (MMR)
Алгоритм MMR (Maximal Marginal Relevance) следит, чтобы выбранные фрагменты не повторяли друг друга. Если два куска говорят об одном — берётся только один.
mmr_selected = [] for idx in sorted_by_rrf_score: if len(mmr_selected) >= 50: break if not mmr_selected: mmr_selected.append(idx) continue # Проверяем: насколько этот чанк похож на уже выбранные? current_emb = chunk_embeddings_norm[idx] is_diverse = all( np.dot(current_emb, chunk_embeddings_norm[sel]) < 0.85 for sel in mmr_selected ) if is_diverse: mmr_selected.append(idx) # берём только если достаточно уникален
HyDE: поиск через гипотетический ответ
ИИ сначала фантазирует: «Как мог бы выглядеть идеальный ответ?» — и ищет реальные фрагменты, похожие на эту фантазию. Помогает при нестандартных или абстрактных вопросах.
# Генерируем гипотетический ответ hyde_resp, _ = call_ai_with_retry( "Напиши подробный гипотетический ответ на следующий вопрос, " "как если бы ты нашёл его в базе знаний. " f"Вопрос: {query}", rerank_model, ) hyde_text = hyde_resp["choices"][0]["message"]["content"].strip() # Превращаем его в вектор и добавляем к поиску hyde_emb = get_embeddings([hyde_text])[0] expanded_embeddings.append(hyde_emb)
Кэширование повторных запросов
Если тот же вопрос уже задавали — система мгновенно отдаёт сохранённый результат без повторного поиска.
import hashlib def make_cache_key(query_text, vs_id): # Уникальный ключ = хэш от вопроса + ID базы знаний return hashlib.sha256(f"{query_text}|{vs_id}".encode("utf-8")).hexdigest() # Проверяем кэш перед поиском _cache_key = make_cache_key(query, vector_store_id) _cached = cache_get(_cache_key) # смотрим в базе данных if _cached: context, used_chunks = _cached # мгновенный ответ! else: context, used_chunks = iterative_retrieve(...) # полный поиск cache_set(_cache_key, query, vector_store_id, context, used_chunks)
4. Финальный ответ
После того как лучшие фрагменты найдены, они передаются языковой модели вместе с исходным вопросом. Модель получает чёткую инструкцию: работать строго по источникам.
final_prompt = ( "Ты — эксперт по базам знаний. Используй ТОЛЬКО информацию из контекста ниже.\n" "Дай развернутый ответ на вопрос. Если нужно сравнить что-то — " "опиши различия, преимущества и т.д.\n" "Не добавляй ничего от себя.\n\n" f"Контекст:\n{context}\n\n" f"Вопрос: {query}\n" f"Ответ:" ) resp, model_used = call_ai_with_retry(final_prompt, ai_model)
Это принципиально важно: ИИ здесь работает не как «всезнайка», а как переводчик — он берёт найденную информацию и излагает её понятным языком. Это защищает от «галлюцинаций» — когда ИИ выдумывает несуществующие факты.
Если ни одна модель не ответила (сбой, лимиты), система возвращает найденный контекст как есть — без потери данных:
if resp is None: return { "answer": None, "fallback": True, "fallback_reason": "all_ai_requests_failed", "context": context, # контекст всё равно возвращается "chunks": used_chunks, }
Итоговая аналогия. Представьте двух специалистов. Первый — опытный библиотекарь, который мгновенно находит нужные страницы в тысячах книг. Второй — блестящий редактор, который читает эти страницы и объясняет суть простым языком. RAG — это их совместная работа.
Главное за 30 секунд
Документы загружаются один раз. Система читает файлы, нарезает их на фрагменты и превращает каждый в математический «отпечаток смысла».
Вопрос тоже получает свой «отпечаток». Система ищет фрагменты, чей смысл математически близок к вопросу — даже если слова другие.
Несколько методов поиска работают вместе. Смысловой поиск, ключевые слова, ИИ-фильтрация — результаты объединяются для максимальной точности.
ИИ строго по источникам составляет ответ. Никаких догадок — только то, что нашлось в документах, пересказанное понятным языком.
В результате пользователь получает точный ответ со ссылкой на источник — за секунды, вместо часов ручного поиска по документации.
Бот ответит, опираясь строго на загруженные материалы — именно так, как описано в этой статье.
