RAG для закупок: Qdrant и LlamaIndex в локальном контуре

Меня зовут Владимир, и недавно я решил изучить новую (для себя) технологию – LlamaIndex. А тут и задачка подвернулась – надоело копаться в Положении о закупках, поэтому понадобился RAG для ответов по ФЗ-44, ФЗ-223, ну и локальному положению.

В этой статье разберу, как создать простенький RAG, не выходящий из локального контура, на базе LlamaIndex + Qdrant, напишем к нему API и UI на Gradio. Поехали.

Конфигурация проекта

Для сервиса понадобятся LLM, векторная БД Qdrant, эмбеддер и реранкер результатов. Сразу код конфигурации:

from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class ServiceConfig(BaseModel):
    name: str = 'ProcurementRAG'
    host: str = '0.0.0.0'
    port: Annotated[int, Field(ge=1, le=65535)] = 8000
    loglevel: str = 'INFO'
    log_to_file: bool = False
    trace_debug: bool = False
    enable_gui: bool = False

class LLMConfig(BaseModel):
    host: str = 'localhost'
    port: Annotated[int | None, Field(ge=1, le=65535)] = 11434
    model_name: str = 'qwen3:14b'
    api_key: SecretStr = SecretStr('EMPTY')
    temperature: Annotated[float, Field(ge=0.0, le=1.1)] = 0.1
    @property
    def url(self) -> str:
        if self.port is not None:
            return f'http://{self.host}:{self.port}/v1'
        return f'https://{self.host}/v1'

class QdrantConfig(BaseModel):
    host: str = 'localhost'
    port: Annotated[int, Field(ge=1, le=65535)] = 6333
    collection: str = 'procurement_docs'
    @property
    def url(self) -> str:
        return f'http://{self.host}:{self.port}'

class EmbeddingConfig(BaseModel):
    model_name: str = 'intfloat/multilingual-e5-base'
    dimension: Annotated[int, Field(ge=1)] = 768

class RerankConfig(BaseModel):
    enabled: bool = False
    model_name: str = 'cross-encoder/ms-marco-MiniLM-L-6-v2'
    top_n: Annotated[int, Field(ge=1, le=50)] = 5

class Settings(BaseSettings):
    service: ServiceConfig = Field(default_factory=ServiceConfig, validation_alias='RAG')
    llm: LLMConfig = Field(default_factory=LLMConfig)
    qdrant: QdrantConfig = Field(default_factory=QdrantConfig)
    embedding: EmbeddingConfig = Field(default_factory=EmbeddingConfig)
    rerank: RerankConfig = Field(default_factory=RerankConfig)

    model_config = SettingsConfigDict(
        env_file=_ENV_FILE,
        env_file_encoding='utf-8',
        extra='ignore',
        case_sensitive=False,
        env_nested_delimiter='__',
    )

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

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

Индексация документов

Перед тем, как добавлять документы в индекс Qdrant, их необходимо преобразовать в текст и нарезать на чанки. Предполагается, что документы в сервис будут попадать в формате PDF по сети, поэтому встроенные в LlamaIndex средства преобразования PDF, такие как llama_index.readers.file.PDFReader и llama_index.readers.file.PyMuPDFReader не подходят. Пишем свой:

class PdfReader:
    def __init__(self, min_text_chars_per_page: int = 50) -> None:
        self._min_text_chars_per_page = min_text_chars_per_page

В конструкторе класса задаём порог, относительно которого будем оценивать наличие текста на странице. Дальше метод извлечения данных.

from dataclasses import dataclass

@dataclass
class PdfDocument:
    file_name: str
    text: str
    used_ocr: bool

class PdfReader:
    # Предыдущий код
    def read_bytes(self, content: bytes, file_name: str) -> PdfDocument:
        doc = pymupdf.open(stream=content, filetype='pdf')
        return self._extract(doc, file_name=file_name, file_path=file_name)

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

class PdfReader:
    # Предыдущий код
    def _extract(self, doc: pymupdf.Document, file_name: str, file_path: str) -> PdfDocument:
        pages_text: list[str] = []
        used_ocr = False

        for page in doc:
            page_text = page.get_text('text').strip()
            if len(page_text) < self._min_text_chars_per_page:
                logger.info(f'Страница {page.number + 1} требует OCR: {file_name}')
                ocr_text = self._ocr_page(page)
                if ocr_text is None:
                    logger.warning(f'OCR не выполнен для страницы {page.number + 1} ({file_name})')
                else:
                    used_ocr = True
                    page_text = ocr_text.strip() or page_text
            pages_text.append(page_text)

        doc.close()
        full_text = 'nn'.join(pages_text)
        return PdfDocument(file_name=file_path, text=full_text, used_ocr=used_ocr)    

Проходим циклом по всем полученным страницам документа. Если получили со страницы меньше self._min_text_chars_per_page символов, то вызываем OCR (через Tesseract):

class PdfReader:
    # Предыдущий код
    def _ocr_page(self, page: pymupdf.Page) -> str | None:
        try:
            pix = page.get_pixmap(dpi=300)
            image = Image.frombytes('RGB', (pix.width, pix.height), pix.samples)
            return pytesseract.image_to_string(image, lang='rus+eng')
        except pytesseract.TesseractNotFoundError:
            logger.error('Tesseract не установлен или не найден в PATH')
            return None
        except pytesseract.TesseractError as e:
            logger.warning(f'Ошибка Tesseract на странице {page.number + 1}: {e}')
            return None

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

ARTICLE_RE = re.compile(r'Статьяs+(d+(?:.d+)?)', re.IGNORECASE)
POINT_RE = re.compile(r'^(d+).s', re.MULTILINE)

Первое правило ищет слово “Статья” и выделяет её номер, а второе – число с точкой, которая обычно обозначает номер пункта НПА. Теперь создадим два класса для чанков – один с метаданными, другой – класс-чанк:

@dataclass
class LegalChunkMetadata:
    source: str
    doc_type: str
    article: str | None
    point: str | None
    citation: str
    chunk_index: int

@dataclass
class TextChunk:
    text: str
    metadata: LegalChunkMetadata

Начнём чанкирование текста:

class LegalStructuralChunker:
    def chunk(self, text: str, file_name: str) -> list[TextChunk]:
        source, doc_type = (
            ('ФЗ-44', 'federal_law') if '44' in file_name 
            else ('ФЗ-223', 'federal_law') if '223' in file_name 
            else ('Положение', 'internal_regulation'))
        
        chunks: list[TextChunk] = []
        chunk_index = 0

        article_splits = list(ARTICLE_RE.finditer(text))
        if not article_splits:
            return [
                TextChunk(
                    text=text,
                    metadata=LegalChunkMetadata(
                        source=source, doc_type=doc_type, article=None, point=None, citation='', chunk_index=chunk_index,
                ))]

Чанкирование начнем с определения источника информации (для фильтрации) и типа документа – просто ищем в имени файла номер закона. Если не нашли – значит это локальное Положение. Дальше ищем все вхождения слова Статья в тексте. Если не нашли – возвращаем весь текст целиком в виде одного чанка. Если Статья нашлась, то продолжаем обработку:

class LegalStructuralChunker:
    def chunk(self, text: str, file_name: str) -> list[TextChunk]:
        # Предыдущий код
        for i, article_match in enumerate(article_splits):
            article_num = article_match.group(1)
            article_label = f'Статья {article_num}'
            start = article_match.start()
            end = article_splits[i + 1].start() if i + 1 < len(article_splits) else len(text)
            article_text = text[start:end]

            point_chunks = self._split_by_points(
                article_text=article_text, source=source, doc_type=doc_type,
                article=article_label, start_index=chunk_index,)
            if point_chunks:
                chunks.extend(point_chunks)
                chunk_index = point_chunks[-1].metadata.chunk_index + 1
                continue

            chunks.append(
                TextChunk(
                    text=article_text.strip(),
                    metadata=LegalChunkMetadata(
                        source=source, doc_type=doc_type, article=article_label, point=None,
                        citation=build_citation(article_label, None), chunk_index=chunk_index,
            )))
            chunk_index += 1

        return chunks     

Для каждого найденного вхождения мы извлекаем кусок текста статьи из исходного текста (с помощью метода re.Match.start()) и отправляем его на дальнейший чанкинг по пунктам:

class LegalStructuralChunker:
    # Предыдущий код
    def _split_by_points(self, article_text: str, source: str, doc_type: str, article: str, start_index: int) -> list[TextChunk]:
        chunks: list[TextChunk] = []
        point_matches = list(POINT_RE.finditer(article_text))

        if not point_matches:
            return []

        chunk_index = start_index
        for i, point_match in enumerate(point_matches):
            point_num = point_match.group(1)
            point_label = f'Пункт {point_num}'
            start = point_match.start()
            end = point_matches[i + 1].start() if i + 1 < len(point_matches) else len(article_text)
            point_text = article_text[start:end].strip()

            chunks.append(
                TextChunk(
                    text=point_text,
                    metadata=LegalChunkMetadata(
                        source=source, doc_type=doc_type, article=article, point=point_label,
                        citation=build_citation(article, point_label), chunk_index=chunk_index,
            )))
            chunk_index += 1

        return chunks

Фактически метод повторяет код из основного чанкера, но с другим паттерном. Теперь соберём наши методы в пайплайн.

Опишем сначала вход-выход пайплайна:

@dataclass
class UploadedFile:
    filename: str
    content: bytes

@dataclass
class IndexingResult:
    documents_processed: int
    chunks_indexed: int
    filenames: list[str]

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

class IndexingPipeline:
    def __init__(self, qdrant_client: QdrantClient) -> None:
        self._qdrant_client = qdrant_client
        self._pdf_reader = PdfReader()
        self._chunker = LegalStructuralChunker()

    def index_files(self, files: list[UploadedFile], recreate_collection: bool = False) -> IndexingResult:
        if not files:
            return IndexingResult(documents_processed=0, chunks_indexed=0, filenames=[])

        documents = [self._pdf_reader.read_bytes(file.content, file.filename) for file in files]
        return self._index_documents(documents, recreate_collection)

В метод индексации вместе с файлами передаём флаг пересоздания коллекции. В теле проверяем наличие файлов, получаем их текст и запускаем процесс индексации:

class IndexingPipeline:
    # Предыдущий код
    def _index_documents(self, documents: list[PdfDocument], recreate_collection: bool) -> IndexingResult:
        self._initialize_collection(recreate_collection)
        nodes = self._convert_to_nodes(documents)
        filenames = [doc.file_name for doc in documents]

        if not nodes:
            return IndexingResult(documents_processed=len(documents), chunks_indexed=0, filenames=filenames)

        vector_store = QdrantVectorStore(client=self._qdrant_client, collection_name=settings.qdrant.collection)
        index = VectorStoreIndex.from_vector_store(vector_store=vector_store)
        
        index.insert_nodes(nodes)
        logger.info(f'Проиндексировано {len(nodes)} чанков из {len(documents)} документов')
        return IndexingResult(documents_processed=len(documents), chunks_indexed=len(nodes), filenames=filenames)

Метод вызывает два других метода – для создания коллекции и получения TextNode LlamaIndex (единицы контента в LlamaIndex). Далее интегрируем Qdrant в наш LlamaIndex – создаём обёртку над БД (llama_index.vector_stores.qdrant.QdrantVectorStore) и создаём индекс LlamaIndex поверх обёртки. Полученный нами объект llama_index.core.VectorStoreIndex – это ключевой объект LlamaIndex в нашем пайплайне. У него есть ссылка на наше хранилище, он знает, как считать эмбеддинги (благодаря LlamaSettings.embed_model, который будем биндить в main.py), умеет добавлять ноды в базу (а при поиске — и искать по ним). Короче, мини конвейер по работе с БД, который берёт на себя все промежуточные действия.

Метод IndexingPipeline._initialize_collection() проверяет, существует ли заданная коллекция в базе данных и, при необходимости, удаляет её (по флагу recreate) и создаёт (если коллекция отсутствует).

class IndexingPipeline:
    # Предыдущий код
    def _initialize_collection(self, recreate: bool = False) -> None:
        collection_name = settings.qdrant.collection
        exists = any(c.name == collection_name for c in self._qdrant_client.get_collections().collections)

        if exists and recreate:
            self._qdrant_client.delete_collection(collection_name)
            logger.info(f'Коллекция "{collection_name}" удалена')
            exists = False

        if not exists:
            self._qdrant_client.create_collection(
                collection_name=collection_name,
                vectors_config=VectorParams(
                    size=settings.embedding.dimension,
                    distance=Distance.COSINE),
            )
            logger.info(f'Коллекция "{collection_name}" создана')

При флаге recreate=False метод позволяет убедиться, что коллекция существует. Метод IndexingPipeline._convert_to_nodes() преобразует чанки текста с метаданными в TextNode:

class IndexingPipeline:
    # Предыдущий код
    def _convert_to_nodes(self, documents: list[PdfDocument]) -> list[TextNode]:
        nodes: list[TextNode] = []
        for document in documents:
            logger.info(f'Обработка документа: {document.file_name}')
            chunks = self._chunker.chunk(document.text, document.file_name)
            for chunk in chunks:
                nodes.append(
                    TextNode(
                        id_=str(uuid.uuid5(uuid.NAMESPACE_URL, f'{chunk.metadata.source}:{chunk.metadata.chunk_index}')),
                        text=self._chunk_text_with_citation(chunk),
                        metadata=asdict(chunk.metadata)))
        return nodes

В методе проходим по всем документам, разбиваем их на чанки и формируем из каждого чанка TextNode. При формировании TextNode:

• поле id_ инициализируется не случайным uuid4, а детерминированным uuid5. В текущей реализации идентификатор строится от source и номера чанка, поэтому повторная индексация одного и того же источника заменяет прежние чанки, и рассчитана, что одновременно в базе данных будет только одна версия документа.

• в поле metadata помещаем метаданные, сформированные на этапе чанкирования

• в поле text помещаем текст с текстовой ссылкой на пункт НПА:

class IndexingPipeline:
    # Предыдущий код
    @staticmethod
    def _chunk_text_with_citation(chunk: TextChunk) -> str:
        meta = chunk.metadata
        if meta.citation:
            header = f'[{meta.source}, {meta.citation}]'
        elif meta.source:
            header = f'[{meta.source}]'
        else:
            return chunk.text
        return f'{header}n{chunk.text}'

Это простейший способ указать источник данных в ответе RAG-системы.

Пайплайн индексации готов. Теперь реализуем поисковый движок.

Поисковый движок

Начнём с конструктора:

class QueryEngineFactory:
    def __init__(self) -> None:
        self._vector_store = QdrantVectorStore(
            client=get_qdrant_client(), aclient=get_async_qdrant_client(), collection_name=settings.qdrant.collection)
        self._index = VectorStoreIndex.from_vector_store(vector_store=self._vector_store)
        self._postprocessors = [
            SentenceTransformerRerank(model=settings.rerank.model_name, top_n=settings.rerank.top_n)
        ] if settings.rerank.enabled else None

В конструкторе сначала создаём индекс LlamaIndex, аналогично IndexingPipeline, затем инициализируем модель для реранкинга, при необходимости. Модель для вычисления эмбеддингов и LLM передаются в LlamaIndex следующим образом:

# Кусочек кода из main.py
from llama_index.core import Settings as LlamaSettings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.openai_like import OpenAILike

LlamaSettings.embed_model = HuggingFaceEmbedding(model_name=settings.embedding.model_name)
LlamaSettings.llm = OpenAILike(
    model=settings.llm.model_name, api_base=settings.llm.url,
    api_key=settings.llm.api_key.get_secret_value(),
    temperature=settings.llm.temperature, is_chat_model=True,
)

llama_index.core.Settings – это глобальный синглтон конфигурации библиотеки LlamaIndex, своеобразный контейнер дефолтных зависимостей, которые LlamaIndex подставляет, если их явно не передать в конкретный вызов.

Вернёмся к движку и реализуем метод поиска:

class QueryEngineFactory:
    # Предыдущий код
    async def aquery(self, query: str, source_filter: str | None = None) -> str:
        filters = MetadataFilters(
            filters=[MetadataFilter(key='source', value=source_filter)]) if source_filter else None
        query_engine = self._index.as_query_engine(
            similarity_top_k=10, filters=filters,
            node_postprocessors=self._postprocessors, text_qa_template=get_prompt_template())
        response = await query_engine.aquery(query)

        sources = [f"{node.node.get_content()[:150]}..." for node in getattr(response, 'source_nodes', []) or []]
        logger.info(f'Запрос обработан, найдено {len(sources)} источников')
        sources_text = 'n'.join(sources)
        logger.info(f'Источники:n{sources_text}')

        return str(response)

В данном методе фильтрация реализована только по полю “Источник”, чтобы поиск был в конкретном ФЗ (или Положении). Методом as_query_engine преобразуем наш индекс в поисковый движок (поэтому класс имеет гордое “Фабрика” в названии), передав в него параметры фильтрации, промпт и метод постобработки (в данном случае реранкер). LlamaIndex, кроме ответа, возвращает ещё и найденные в БД чанки документов. Но в данном коде я их использую просто как отладочную информацию, просто выводя в лог.

Чтобы не поднимать контейнер с Langfuse, промпт для модели просто храню в .yaml файле:

from pathlib import Path
import yaml
from llama_index.core.prompts import PromptTemplate

def get_prompt_template() -> PromptTemplate:
    with Path(__file__).with_name('prompts.yaml').open(encoding='utf-8') as f:
        data = yaml.safe_load(f)
    try:
        return PromptTemplate(data['prompt'].strip())
    except Exception as e:
        logger.error(f'Ошибка при создании PromptTemplate: {e}')
        raise RuntimeError(f'Ошибка при создании PromptTemplate')

Дополнительно реализуем три кэшируемые функции – для синхронного и асинхронного клиента Qdrant и для QueryEngineFactory:

@lru_cache
def get_qdrant_client() -> QdrantClient:
    return QdrantClient(url=settings.qdrant.url)

@lru_cache
def get_async_qdrant_client() -> AsyncQdrantClient:
    return AsyncQdrantClient(url=settings.qdrant.url)

@lru_cache
def get_query_engine_factory() -> QueryEngineFactory:
    return QueryEngineFactory()

Поиск готов. Осталось завернуть его в API.

API

Для работы минимально необходимо два эндпоинта – один для индексации, другой для поиска. Начнём с индексации:

from typing import Annotated
from fastapi import APIRouter, File, Form, UploadFile
from indexing.pipeline import IndexingPipeline, UploadedFile
from query.engine import get_qdrant_client
from schemas.index import IndexResponse

router = APIRouter(tags=['index'])

@router.post('/', summary='Индексация загруженных PDF-документов')
async def index_documents(
    files: Annotated[list[UploadFile], File(description='PDF-файлы для индексации')],
    recreate_collection: Annotated[
        bool, Form(description='Пересоздать коллекцию Qdrant перед индексацией')] = False,
) -> IndexResponse:
    uploaded: list[UploadedFile] = []
    for file in files:
        content = await file.read()
        uploaded.append(UploadedFile(filename=file.filename, content=content))

    pipeline = IndexingPipeline(qdrant_client=get_qdrant_client())

    result = pipeline.index_files(uploaded, recreate_collection=recreate_collection)
    return IndexResponse(
        documents_processed=result.documents_processed,
        chunks_indexed=result.chunks_indexed,
        filenames=result.filenames)

Полученные файлы мы по очереди скачиваем и приводим ко входному формату пайплайна индексации. Далее создаём экземпляр пайплайна и вызываем метод индексирования файлов. Возвращает эндпоинт количество проиндексированных документов и количество полученных чанков. Теперь API для поисковых запросов:

from fastapi import APIRouter
from query.engine import get_query_engine_factory
from schemas.search import SearchRequest, SearchResponse

router = APIRouter(tags=['search'])

@router.post('/', summary='Семантический поиск с генерацией ответа')
async def search(body: SearchRequest) -> SearchResponse:
    result = await get_query_engine_factory().aquery(query=body.query, source_filter=body.source)
    return SearchResponse(answer=result)

Кроме доступа через API (для возможных интеграций) решил сделать простенький GUI на Gradio.

GUI на Gradio

Интерфейс будет простым: gr.ChatInterface, но со списком для фильтрации по НПА:

import gradio as gr

from core import SOURCE_VALUES
from query.engine import get_query_engine_factory

class GradioApp:
    def __init__(self):
        self.demo = None

    def build_interface(self):
        with gr.Blocks(title='ProcurementRAG') as self.demo:
            gr.Markdown('# ProcurementRAG')
            gr.Markdown('Семантический поиск по нормативным документам закупок')
            source_dropdown = gr.Dropdown(choices=['Все', *SOURCE_VALUES], value='Все', label='Источник документа')
            chatbot = gr.Chatbot(height=500)
            gr.ChatInterface(fn=self._respond, chatbot=chatbot, additional_inputs=[source_dropdown])

    async def _respond(self, message: str, _: list, source: str) -> str:
        result = await get_query_engine_factory().aquery(
            query=message,
            source_filter=None if source == 'Все' else source)
        return result

def create_gradio_app() -> GradioApp:
    app = GradioApp()
    app.build_interface()
    return app.demo

gr.Dropdown как раз создаёт элемент-выпадающий список, в который мы передаём список необходимых значений. Чат-функция аналогична API для поиска – в поисковый движок передаём запрос из окна чата и параметр фильтрации (если не выбрано “Все”).

Метод create_gradio_app создаёт экземпляр приложения.

Осталось собрать всё это в main.py. Сначала lifespan с зависимостями

from fastapi import FastAPI
from llama_index.core import Settings as LlamaSettings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.openai_like import OpenAILike
from loguru import logger
from core import settings
from query.engine import get_query_engine_factory

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    logger.info(f'Запуск сервиса')
    LlamaSettings.embed_model = HuggingFaceEmbedding(model_name=settings.embedding.model_name)
    LlamaSettings.llm = OpenAILike(
        model=settings.llm.model_name,
        api_base=settings.llm.url,
        api_key=settings.llm.api_key.get_secret_value(),
        temperature=settings.llm.temperature,
        is_chat_model=True,
    )
    _ = get_query_engine_factory()

    yield

    logger.info(f'Остановка сервиса')

Как показывал ранее, прописываем глобальные настройки LlamaIndex, затем вызываем get_query_engine_factory – чтобы модель реранкера скачалась (если реранкинг включен в настройках). Осталось упаковать всё это в FastAPI-приложение.

import gradio as gr
from fastapi import FastAPI
from api import api_router
from core import settings
from gui.gradio_app import create_gradio_app

def create_app() -> FastAPI:
    application = FastAPI(
        title=settings.service.name,
        description='RAG-сервис для нормативных документов в сфере закупок',
        lifespan=lifespan,
    )
    application.include_router(api_router, prefix='/api')
    if settings.service.enable_gui:
        gr.mount_gradio_app(application, create_gradio_app(), path='/')
    return application

app = create_app()

Подключаем к нашему приложению API-интерфейс, при наличии флага enable_gui монтируем в корневой эндпоинт наш Gradio GUI. Всё, можно пользоваться.

Подведение итогов

Вот и получился у нас MVP RAG-а по закупкам. Пока что это реально MVP – простые regex-правила для чанкинга НПА, никакого покрытия подпунктов/частей, отсутствие нормальной стратегии удаления старых чанков.

За кадром я оставил первые попытки сборки пайплайнов со встроенными инструментами LlamaIndex, настройку трассировки LlamaIndex, т.к. пришлось немного поотлаживаться и развёртывание инфраструктуры в докере.

В принципе, проект рабочий, даже пару раз помог оперативно найти ответы. Но основная цель была именно в знакомстве и оценке использования LlamaIndex. И что-то как-то мне не зашло. Может, я просто не так пайплайны собирал, но имея кастомный ридер и чанкер, быстрее бы на LangGraph пару узлов накидал.

Код проекта тут.