Привет Хабр! Меня зовут Владимир, и недавно я решил изучить новую (для себя) технологию - 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 = '\n\n'.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 пару узлов накидал.
Код проекта тут.
