Рекурсивный RAG в LangGraph: Когда остановить цикл, чтобы не сжечь бюджет

Рекурсивный RAG: Когда хорошая идея превращается в бесконечный цикл

Вы построили RAG-систему. Она работает. Находит документы, генерирует ответы. Пока не сталкивается со сложным запросом, требующим глубины. Тогда вы добавляете рекурсию — и система начинает искать всё глубже и глубже. Пока не сжигает весь бюджет на API-вызовы или не зависает в бесконечном цикле. Знакомо?

Рекурсивный RAG в LangGraph — это мощно. Особенно с появлением LangGraph 0.2.8 в 2025 году и его стабильной версии 1.0.0 к февралю 2026. Но без правильных критериев остановки это как дать ребёнку кредитную карту в игрушечном магазине.

Почему ваша первая реализация рекурсивного RAG сломается

Вы читаете документацию LangGraph, видите пример с рекурсивным поиском, копируете код. Запускаете. Работает на простых запросах. Потом приходит запрос: "Какие изменения в ГОСТ Р 12.4.026-2025 влияют на требования к средствам индивидуальной защиты при работе с искусственным интеллектом?".

Система начинает:

1. Искать про ГОСТ Р 12.4.026-2025
2. Находит упоминание про СИЗ
3. Ищет про СИЗ и ИИ
4. Находит документ про ИИ в промышленности
5. Ищет про промышленность и ГОСТы...

Через 15 итераций у вас 45 вызовов LLM, 120 поисков в векторе, счёт на $8.73, а ответа всё нет. Потому что вы забыли главное — рекурсия должна где-то остановиться.

Архитектура, которая не съест ваш бюджет

Перед кодом — понимание. Рекурсивный RAG в LangGraph строится на графах состояний. Каждый узел — шаг. Ребра — переходы. Цикл — это возврат к предыдущему узлу с новыми данными.

Основные компоненты на 22.02.2026:

• Retriever: Qdrant 1.9.x с поддержкой sparse-dense эмбеддингов или Pinecone с их новым гибридным поиском
• LLM: GPT-4.5 Turbo (вышел в январе 2026) или открытые альтернативы типа DeepSeek-R1
• Graph: LangGraph 1.0.0 с улучшенной поддержкой циклов
• Оркестрация: собственный StateGraph с кастомными состояниями

1 Базовый граф без остановки (как НЕ делать)

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
from langchain_openai import ChatOpenAI
from langchain_qdrant import QdrantVectorStore

class GraphState(TypedDict):
    question: str
    context: List[str]
    answer: str
    iteration: int  # Забыли добавить счётчик!

def retrieve(state: GraphState):
    # Поиск документов
    results = vectorstore.similarity_search(state["question"])
    return {"context": [doc.page_content for doc in results]}

def generate(state: GraphState):
    # Генерация ответа
    llm = ChatOpenAI(model="gpt-4.5-turbo")
    prompt = f"Context: {state['context']}\nQuestion: {state['question']}"
    response = llm.invoke(prompt)
    return {"answer": response.content}

def should_continue(state: GraphState):
    # ВСЕГДА возвращает "retrieve" - бесконечный цикл!
    return "retrieve"

# Создаём граф
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.set_entry_point("retrieve")
workflow.add_conditional_edges(
    "generate",
    should_continue,
    {"retrieve": "retrieve", "end": END}
)
workflow.add_edge("retrieve", "generate")

app = workflow.compile()
# ОПАСНО: этот граф будет выполняться вечно

Видите проблему? should_continue всегда возвращает "retrieve". Нет условия выхода. Это как автомобиль без тормозов — рано или поздно врежется в стену.

2 Добавляем критерии остановки: минимальный набор для production

Начнём с трёх обязательных критериев, без которых не стоит выпускать систему даже в staging:

from typing import Literal
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class ProductionGraphState(TypedDict):
    question: str
    context: List[str]
    answer: str
    previous_answer: str  # Для проверки сходимости
    iteration: int
    scores: List[float]  # Оценки релевантности документов
    max_iterations: int = 5  # Жёсткий лимит

def should_continue_production(state: ProductionGraphState) -> Literal["retrieve", "end"]:
    """Умная проверка продолжения с тремя критериями"""
    
    # 1. Жёсткий лимит итераций
    if state["iteration"] >= state["max_iterations"]:
        print(f"Остановка по лимиту итераций: {state['iteration']}")
        return "end"
    
    # 2. Проверка сходимости ответов
    if state["previous_answer"]:
        # Эмбеддинги для сравнения (упрощённо, лучше через ту же модель)
        curr_embed = get_embedding(state["answer"])
        prev_embed = get_embedding(state["previous_answer"])
        similarity = cosine_similarity([curr_embed], [prev_embed])[0][0]
        
        if similarity > 0.85:  # Ответы практически идентичны
            print(f"Остановка по сходимости: similarity={similarity:.3f}")
            return "end"
    
    # 3. Минимальный прирост релевантности
    if len(state["scores"]) >= 2:
        last_score = state["scores"][-1]
        prev_score = state["scores"][-2]
        improvement = (last_score - prev_score) / prev_score
        
        if improvement < 0.05:  # Менее 5% улучшения
            print(f"Остановка по минимальному приросту: {improvement:.1%}")
            return "end"
    
    # Все проверки пройдены - продолжаем
    return "retrieve"

def get_embedding(text: str) -> List[float]:
    """Упрощённая функция для эмбеддингов"""
    # В реальности используйте ту же модель, что и для основного пайплайна
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer('all-MiniLM-L6-v2')
    return model.encode(text).tolist()

Расширенные критерии для сложных систем

Три базовых критерия защитят от катастрофы. Но для enterprise-систем, особенно в регуляторных областях, нужен более тонкий контроль. Вот что мы добавили в систему для эксперта по охране труда после инцидента с бесконечным поиском:

Circuit Breaker Pattern для LLM-вызовов

Вдохновлено статьёй "Stop-First RAG: Как перестать платить за глупости". Если LLM три раза подряд говорит "не могу найти ответ в предоставленных документах", прекращаем цикл.

class CircuitBreakerState(TypedDict):
    question: str
    context: List[str]
    answer: str
    iteration: int
    not_found_count: int  # Счётчик "не найдено"
    max_not_found: int = 3  # Лимит перед разрывом цепи

def generate_with_circuit_breaker(state: CircuitBreakerState):
    llm = ChatOpenAI(model="gpt-4.5-turbo", temperature=0)
    
    prompt = f"""Context: {state['context']}
Question: {state['question']}

Answer based ONLY on context. If answer not in context, say \"I cannot find answer in documents\"."""
    
    response = llm.invoke(prompt)
    answer = response.content
    
    # Проверяем, нашёл ли модель ответ
    if "cannot find" in answer.lower() or "not in context" in answer.lower():
        state["not_found_count"] += 1
    else:
        state["not_found_count"] = 0  # Сбрасываем при успешном ответе
    
    return {"answer": answer, "not_found_count": state["not_found_count"]}

def should_continue_with_breaker(state: CircuitBreakerState):
    # Circuit breaker срабатывает первым
    if state["not_found_count"] >= state["max_not_found"]:
        print(f"Circuit breaker: {state['not_found_count']} consecutive 'not found'")
        return "end"
    
    # Затем обычные проверки
    if state["iteration"] >= 5:
        return "end"
        
    return "retrieve"

Динамическое управление глубиной на основе сложности запроса

Не все запросы одинаковы. "Что такое СИЗ?" требует одной итерации. "Как применять п. 4.5 ГОСТ Р 12.0.230-2024 в свете изменений ТК РФ ст. 214?" требует больше. Определяем сложность через LLM и адаптируем max_iterations.

def analyze_query_complexity(query: str) -> dict:
    """Анализирует запрос и возвращает рекомендованную глубину"""
    complexity_prompt = f"""Analyze this query complexity:
Query: {query}

Return JSON with:
- complexity_score: 1-10
- suggested_iterations: 1-5
- reasoning: brief explanation"""
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # Дешёвая модель для классификации
    response = llm.invoke(complexity_prompt)
    
    try:
        import json
        return json.loads(response.content)
    except:
        # Fallback значения
        return {"complexity_score": 5, "suggested_iterations": 3, "reasoning": "fallback"}

# Использование в инициализации состояния
initial_state = {
    "question": user_query,
    "iteration": 0,
    "max_iterations": analyze_query_complexity(user_query)["suggested_iterations"],
    # ... другие поля
}

Parent-Child Retrieval с памятью о пройденном пути

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

class ParentChildState(TypedDict):
    question: str
    context: List[str]
    answer: str
    iteration: int
    visited_doc_ids: Set[str]  # ID уже посещённых документов
    parent_child_map: Dict[str, List[str]]  # Граф связей документов

def retrieve_with_memory(state: ParentChildState):
    """Поиск с учётом уже посещённых документов"""
    
    # Базовый поиск
    base_results = vectorstore.similarity_search(
        state["question"], 
        k=10,
        filter={"doc_id": {"$nin": list(state["visited_doc_ids"])}}  # Исключаем посещённые
    )
    
    # Если есть родительские документы, ищем их детей
    child_docs = []
    if state["parent_child_map"]:
        recent_parents = list(state["parent_child_map"].keys())[-3:]  # 3 последних родителя
        for parent_id in recent_parents:
            children = state["parent_child_map"].get(parent_id, [])
            child_docs.extend(get_docs_by_ids(children))
    
    # Объединяем и убираем дубликаты
    all_docs = deduplicate_documents(base_results + child_docs)
    
    # Обновляем посещённые документы
    new_visited_ids = {doc.metadata["doc_id"] for doc in all_docs}
    state["visited_doc_ids"].update(new_visited_ids)
    
    # Если новых документов нет - флаг для остановки
    if not all_docs or len(all_docs) == 0:
        state["no_new_docs"] = True
    
    return {
        "context": [doc.page_content for doc in all_docs],
        "visited_doc_ids": state["visited_doc_ids"],
        "no_new_docs": state.get("no_new_docs", False)
    }

def should_continue_with_memory(state: ParentChildState):
    """Останавливаемся, если нет новых документов"""
    if state.get("no_new_docs", False):
        print("Остановка: нет новых документов для поиска")
        return "end"
    
    if state["iteration"] >= 5:
        return "end"
        
    return "retrieve"

Идемпотентность: почему ваши чанки дублируются

Самая коварная проблема рекурсивного RAG — идемпотентность индексирования. Если на каждом шаге вы добавляете в контекст результаты поиска и снова их индексируете, через 3-4 итерации у вас будет 80% дубликатов.

Решение — content-based хеширование и дедупликация перед индексированием:

import hashlib
from typing import List, Dict

def get_content_hash(content: str) -> str:
    """Хеш содержимого для дедупликации"""
    return hashlib.md5(content.encode()).hexdigest()

def deduplicate_before_indexing(documents: List[Dict]) -> List[Dict]:
    """Удаляет дубликаты перед отправкой в векторную БД"""
    seen_hashes = set()
    unique_docs = []
    
    for doc in documents:
        content_hash = get_content_hash(doc["page_content"])
        
        if content_hash not in seen_hashes:
            seen_hashes.add(content_hash)
            # Добавляем хеш в метаданные для будущих проверок
            doc["metadata"]["content_hash"] = content_hash
            unique_docs.append(doc)
        else:
            print(f"Пропущен дубликат: {content_hash}")
    
    return unique_docs

# Применение в пайплайне индексирования
def recursive_indexing_pipeline(initial_docs, max_depth=3):
    """Индексирование с рекурсивным расширением, но без дубликатов"""
    all_docs = []
    current_docs = initial_docs
    
    for depth in range(max_depth):
        # Дедупликация перед индексированием
        unique_docs = deduplicate_before_indexing(current_docs)
        
        # Индексируем только уникальные
        vectorstore.add_documents(unique_docs)
        all_docs.extend(unique_docs)
        
        # Находим связанные документы (например, по ссылкам)
        current_docs = find_related_documents(unique_docs)
        
        if not current_docs:
            break
    
    return all_docs

Production-рекомендации: что работает в 2026

Собрали рекомендации из десятков production-систем, которые пережили 2024-2025:

1. Мониторинг — прежде всего
   Каждая итерация должна логироваться с метриками:
   • Количество извлечённых документов
   • Средняя релевантность (score)
   • Время выполнения
   • Токены использованные
   • Причина остановки (если остановилась)
2. Динамические лимиты на основе бюджета
   Если у вас $10 на запрос и каждая итерация стоит $0.50, максимальная глубина = $10 / $0.50 = 20 итераций. Но лучше установить лимит 10 и оставить запас для сложных случаев.
3. A/B тестирование критериев
   Запустите параллельно две версии: с простым лимитом итераций и с полным набором критериев. Сравните:

   Метрика	Простой лимит	Умные критерии
   Средняя глубина	5.0 (всегда максимум)	2.8 (адаптивно)
   Точность ответов	78%	84%
   Стоимость/запрос	$2.45	$1.72

4. Circuit Breaker как must-have
   После инцидента в январе 2026, где система сделала 147 итераций по одному запросу (стоимость $412), circuit breaker стал обязательным для всех наших проектов.
5. Визуализация графа поиска
   Для дебагга и понимания, что происходит:

   def visualize_search_path(state_history):
       """Визуализирует путь поиска системы"""
       import networkx as nx
       import matplotlib.pyplot as plt
       
       G = nx.DiGraph()
       
       for i, state in enumerate(state_history):
           G.add_node(i, 
                      question=state["question"][:50],
                      docs=len(state["context"]),
                      iteration=state["iteration"])
           
           if i > 0:
               G.add_edge(i-1, i)
       
       plt.figure(figsize=(12, 8))
       pos = nx.spring_layout(G)
       nx.draw(G, pos, with_labels=True, node_color='lightblue')
       plt.show()

Чеклист для запуска в production

• ✅ Лимит итераций: Установлен и проверен на максимально сложных запросах
• ✅ Circuit breaker: Реагирует на повторяющиеся "не найдено"
• ✅ Мониторинг затрат: Алёрт при превышении $X за запрос
• ✅ Дедупликация: Content-based хеширование работает
• ✅ Логирование пути: Можете восстановить, почему система сделала N итераций
• ✅ Fallback механизм: При остановке по лимиту возвращается лучший найденный ответ
• ✅ Тест на абсурдных запросах: "Найди мне связь между квантовой физикой и трудовым кодексом" не должен вызывать 100 итераций
• ✅ A/B тестирование готово: Можете сравнить с версией без рекурсии

Что дальше? Эволюция рекурсивного RAG

К февралю 2026 рекурсивный RAG перестал быть экспериментальной технологией. Это стандарт для сложных поисковых систем. Но развитие продолжается:

1. Прогнозирование оптимальной глубины: Модели, которые предсказывают нужное количество итераций до начала поиска
2. Мультимодальная рекурсия: Поиск не только в текстах, но в таблицах, изображениях, схемах
3. Межъязыковой поиск: Запрос на русском → поиск в английских документах → ответ на русском с учётом найденного
4. Коллаборативная фильтрация: Учёт похожих запросов других пользователей для определения глубины поиска

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

Начните с простого лимита итераций. Добавьте circuit breaker. Затем — проверку сходимости. Через месяц у вас будет production-система, а не экспериментальный прототип, который сжигает бюджет.

А если хотите глубже погрузиться в архитектурные решения для сложных RAG-систем, посмотрите "RAG 2026: От гибридного поиска до production" — там разбираем полный roadmap от MVP до масштабируемой системы.