azure redis rag langgraph document-intelligence embeddings

Семантические вопросы и ответы в формате PDF с использованием технологий Document Intelligence, эмбеддингов и Redis

LangGraph извлекает PDF-файлы с расписанием тарифов, разбивает их на части и встраивает в Azure OpenAI, а также отвечает на вопросы о тарифах на естественном языке с использованием локального векторного индекса Redis.

Издатель: REACTAPP.TOPДобавлено на сайт 26 сент. 2026 г.11 мин чтения2476 слов

Источники: Building a PDF Q&A Pipeline with Azure AI and Redis Vector Search

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

Система организована как граф состояний LangGraph и включает следующие шаги:

Извлечение текста с помощью Azure AI Document Intelligence (обычный текст и таблицы).
Разбиение на перекрывающиеся фрагменты размером, соответствующим возможностям модели.
Генерация векторов с использованием Azure OpenAI text-embedding-3-small.
Хранение и поиск векторов в локальном Redis (или удаленном Redis) с помощью RediSearch.

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

Для тестирования с конца до начала использовалась реальная таблица комиссий за использование кредитной карты U.S. Bank в одной странице. Вопрос: «Какова комиссия за снятие денег из банкомата вне сети?» — комиссия в размере 2,50 доллара показала наивысшую степень схожести.

Почему Redis в качестве хранилища векторов

Redis известен как хранилище типа KV в оперативной памяти; RediSearch (в составе redis-stack) добавляет возможность индексации векторов. Причины его подходящести:

Скорость — низкая задержка при работе в оперативной памяти для интерактивного поиска
Одна система, две функции — многие приложения уже используют Redis для кэширования и хранения сессий
Гибкая индексация — алгоритмы FLAT или HNSW
Гибридный поиск — векторы в сочетании с тегами, числовыми и полнотекстовыми фильтрами

Работает где угодно — контейнер для ноутбука, работающий в управляемых облаках с одинаковой структурой запросов

Альтернативы поиску в Redis

Управляемые векторные БД и Azure AI Search могут служить отличным обновлением позже; Redis сохраняет процесс обучения локально и быстро.

Предварительные требования

Docker; Azure CLI (az login); доступ к Azure OpenAI; Python 3.12 (версия 3.14 нарушила эту стек зависимостей во время тестирования); uv для управления средой.

Подготовка ресурсов в Azure

Группа ресурсов rg-pdf-rag (например, eastus)
Document Intelligence docintel-pdf-rag (FormRecognizer/S0)
Azure OpenAI openai-pdf-rag с развертыванием эмбеддингов

Запуск Redis локально

docker compose up -d

mkdir chat-file-search-inmemory && cd chat-file-search-inmemory
uv init .
touch azure_resource.sh inspect_redis.py persist_vector_graph.py search.py
code .

sed -i 's/requires-python = ">=3.14"/requires-python = ">=3.12"/' pyproject.toml
uv python pin 3.12
uv python install 3.12
rm -rf .venv uv.lock
uv sync
uv run python --version

uv add langgraph langchain langchain-openai langchain-text-splitters \
       azure-ai-documentintelligence redis redisvl numpy python-dotenv

AZURE_DOCINTEL_ENDPOINT=https://docintel-pdf-rag-78ee2.cognitiveservices.azure.com/
AZURE_DOCINTEL_KEY=<rotated>
AZURE_OPENAI_ENDPOINT=https://openai-pdf-rag-302a6.openai.azure.com/
AZURE_OPENAI_API_KEY=<rotated>
AZURE_OPENAI_API_VERSION=2023-05-15
AZURE_OPENAI_EMBEDDING_DEPLOYMENT=text-embedding-3-small
REDIS_URL=redis://localhost:6379
REDIS_INDEX_NAME=pdf_docs

import os
import sys
from typing import TypedDict, List, Optional

import numpy as np
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import AnalyzeDocumentRequest
from azure.core.credentials import AzureKeyCredential

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import AzureOpenAIEmbeddings

from redisvl.index import SearchIndex
from redisvl.schema import IndexSchema

load_dotenv()

REDIS_URL = os.environ["REDIS_URL"]
REDIS_INDEX_NAME = os.environ.get("REDIS_INDEX_NAME", "pdf_docs")

EMBEDDING_DIMS = 1536


def get_index_schema() -> IndexSchema:
    return IndexSchema.from_dict({
        "index": {"name": REDIS_INDEX_NAME, "prefix": REDIS_INDEX_NAME},
        "fields": [
            {"name": "content", "type": "text"},
            {"name": "source", "type": "tag"},
            {"name": "chunk_index", "type": "numeric"},
            {
                "name": "embedding",
                "type": "vector",
                "attrs": {
                    "dims": EMBEDDING_DIMS,
                    "distance_metric": "cosine",
                    "algorithm": "flat",
                    "datatype": "float32",
                },
            },
        ],
    })


def vector_to_bytes(vector: List[float]) -> bytes:
    return np.array(vector, dtype=np.float32).tobytes()


class PipelineState(TypedDict):
    file_path: str
    text: Optional[str]
    chunks: Optional[List[str]]
    num_stored: Optional[int]


def extract_text_node(state: PipelineState) -> PipelineState:
    client = DocumentIntelligenceClient(
        endpoint=os.environ["AZURE_DOCINTEL_ENDPOINT"],
        credential=AzureKeyCredential(os.environ["AZURE_DOCINTEL_KEY"]),
    )

    with open(state["file_path"], "rb") as f:
        poller = client.begin_analyze_document(
            "prebuilt-layout",
            AnalyzeDocumentRequest(bytes_source=f.read()),
        )
    result = poller.result()

    state["text"] = result.content
    return state


def chunk_text_node(state: PipelineState) -> PipelineState:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150,
    )
    state["chunks"] = splitter.split_text(state["text"])
    return state


def embed_and_store_node(state: PipelineState) -> PipelineState:
    embeddings = AzureOpenAIEmbeddings(
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_key=os.environ["AZURE_OPENAI_API_KEY"],
        api_version=os.environ["AZURE_OPENAI_API_VERSION"],
        azure_deployment=os.environ["AZURE_OPENAI_EMBEDDING_DEPLOYMENT"],
    )

    chunks = state["chunks"]
    vectors = embeddings.embed_documents(chunks)

    index = SearchIndex(get_index_schema(), redis_url=REDIS_URL)
    index.create(overwrite=False)

    records = [
        {
            "content": chunk,
            "source": state["file_path"],
            "chunk_index": i,
            "embedding": vector_to_bytes(vector),
        }
        for i, (chunk, vector) in enumerate(zip(chunks, vectors))
    ]

    index.load(records)

    state["num_stored"] = len(records)
    return state


def build_graph():
    graph = StateGraph(PipelineState)

    graph.add_node("extract_text", extract_text_node)
    graph.add_node("chunk_text", chunk_text_node)
    graph.add_node("embed_and_store", embed_and_store_node)

    graph.set_entry_point("extract_text")
    graph.add_edge("extract_text", "chunk_text")
    graph.add_edge("chunk_text", "embed_and_store")
    graph.add_edge("embed_and_store", END)

    return graph.compile()


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python persist_vector_graph.py path/to/file.pdf")
        sys.exit(1)

    pdf_path = sys.argv[1]
    app = build_graph()
    result = app.invoke({"file_path": pdf_path})

    print(f"Extracted {len(result['text'])} characters")
    print(f"Split into {len(result['chunks'])} chunks")
    print(f"Stored {result['num_stored']} vectors in Redis index '{REDIS_INDEX_NAME}'")

curl -L -o usbank_expense_card_fee_schedule.pdf \
  "https://www.usbankexpensecard.com/documents/792269/792586/Fee+Schedule_Cash_197196658_7.23.19.pdf/8388d9e3-8225-8017-152a-c9cb601d626f"

uv run python persist_vector_graph.py ./usbank_expense_card_fee_schedule.pdf
# Execution results
Extracted 5466 characters
Split into 7 chunks
Stored 7 vectors in Redis index 'pdf_docs'

uv run inspect_redis.py

# Execution results

Index: pdf_docs
  Documents in index: 7
  Fields: ['content', 'source', 'chunk_index', 'embedding']

Found 7 stored chunks. Showing up to 10:

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASG ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 5
content     : $3.00
This is our fee charged each month after you have not completed a transaction using your card for 180 consecutive days.
Transaction Limits
For security reasons, there are limitations on the number and amount of transactions that you may perform with your U.S. Bank card. Daily limits are based ...

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASC ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 1
content     : . Memorize the Personal Identification Number (PIN) that you will establish . Sign your name in ink on the back of the card
Card Usage Tips: Gas Stations: When purchasing gasoline at a gas station using the pay-at-the-pump option, a maximum hold of $75.00 will be placed on your account to initiate y...

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASD ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 2
content     : Hotels: When making travel reservations with a hotel or similar merchant, ask for the amount of the authorization they will send to your account. These merchants may send an initial authorization amount equal to your entire stay or rental period, plus taxes and incidentals, even though your actual p...

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASE ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 3
content     : ATM Withdrawal (out-of-network)
$2.50
This is our fee per withdrawal. "Out-of-network" refers to all the ATMs outside of the U.S. Bank or MoneyPass ATM networks. You may also be charged a fee by the ATM operator even if you do not complete a transaction.
Teller Cash Withdrawal
$5.00
This is our fee ...

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASH ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 6
content     : 20 transactions and $5,050 per day
Teller withdrawals (at Visa member banks) (Financial Institutions may have lower limits)
3 transactions and $5,050 per day
Maximum daily credits
10 transactions and $10,000 per day
Returns and Refunds
May not exceed 4 transactions per day
USBEXCV9
The U.S. Bank Exp...
the
--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASB ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 0
content     : Welcome! To your new U.S. Bank Expense Card - Cash Access
Your card can be used anywhere Visa debit cards are accepted.
Card Checklist :unselected: Activate your card □ □ :unselected: Set up your online account □ :unselected: Sign up for text1 or email alerts
See the enclosed Usage Guide for more de...

--- pdf_docs:01KZ28WSFJZFGWTPADP0MWZASF ---
source      : ./usbank_expense_card_fee_schedule.pdf
chunk_index : 4
content     : Using your card outside the U.S.
International Transaction
3%
This is our fee which applies when you use your card for purchases at foreign merchants and for cash withdrawals from foreign ATMs and is a percentage of the transaction dollar amount, after any currency conversion. Some transactions, eve...

import os
import sys

import numpy as np
from dotenv import load_dotenv
from langchain_openai import AzureOpenAIEmbeddings

from redisvl.index import SearchIndex
from redisvl.query import VectorQuery

from persist_vector_graph import get_index_schema, vector_to_bytes

load_dotenv()

REDIS_URL = os.environ["REDIS_URL"]
REDIS_INDEX_NAME = os.environ.get("REDIS_INDEX_NAME", "pdf_docs")


def get_embeddings() -> AzureOpenAIEmbeddings:
    return AzureOpenAIEmbeddings(
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_key=os.environ["AZURE_OPENAI_API_KEY"],
        api_version=os.environ["AZURE_OPENAI_API_VERSION"],
        azure_deployment=os.environ["AZURE_OPENAI_EMBEDDING_DEPLOYMENT"],
    )


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print('Usage: python search.py "your question here"')
        sys.exit(1)

    query_text = " ".join(sys.argv[1:])

    embeddings = get_embeddings()
    query_vector = embeddings.embed_query(query_text)

    index = SearchIndex(get_index_schema(), redis_url=REDIS_URL)

    vq = VectorQuery(
        vector=vector_to_bytes(query_vector),
        vector_field_name="embedding",
        return_fields=["content", "source", "chunk_index"],
        num_results=4,
    )

    results = index.query(vq)

    for i, r in enumerate(results, start=1):
        print(f"\n--- Result {i} (score={r.get('vector_distance')}) ---")
        print(f"Source: {r.get('source')} | chunk {r.get('chunk_index')}")
        print(r.get("content", "")[:500])

uv run python search.py "what is the ATM withdrawal fee out of network?"Output:

--- Result 1 (score=0.301177144051) ---
Source: ./usbank_expense_card_fee_schedule.pdf | chunk 3
ATM Withdrawal (out-of-network)
$2.50
This is our fee per withdrawal. "Out-of-network" refers to all the ATMs
outside of the U.S. Bank or MoneyPass ATM networks. You may also be charged
a fee by the ATM operator even if you do not complete a transaction.
Teller Cash Withdrawal
$5.00
...
--- Result 2 (score=0.400232374668) ---
Source: ./usbank_expense_card_fee_schedule.pdf | chunk 4
Using your card outside the U.S.
International Transaction
3%
...
--- Result 3 (score=0.4464854002) ---
Source: ./usbank_expense_card_fee_schedule.pdf | chunk 2
Hotels: When making travel reservations with a hotel or similar merchant...
--- Result 4 (score=0.517238736153) ---
Source: ./usbank_expense_card_fee_schedule.pdf | chunk 5
$3.00
This is our fee charged each month after you have not completed a
transaction using your card for 180 consecutive days.
...

Среда Python с `uv`

Аккуратно задавайте версии компонентов. Устаревшие интеграции LangChain с Redis создавали серьезные проблемы — предпочтите поддерживаемый набор пакетов, описанный в сопутствующем репозитории, и проверяйте импорты перед созданием узлов графа.

Переменные среды

Укажите адрес конечной точки/ключа Document Intelligence, адрес/ключ Azure OpenAI, название реализации эмбеддингов (а не название базовой модели) и URL Redis.

Пример документа и запуск пайплайна

Подавайте PDF с расписанием тарифов через этапы extract → chunk → embed → upsert. Перед поиском убедитесь, что количество ключей в Redis совпадает с количеством чанков.

Запуск семантического поиска

Встроить вопрос с тем же способом развертывания, применить алгоритм KNN к индексу и вывести лучшие результаты с оценками. Проверить, что ответ о комиссии банкомата занимает первое место по рейтингу для «золотого» вопроса.

Уроки, извлеченные

Различия в версиях Python тихо нарушают работу стандартных библиотек — для этой сборки необходимо фиксировать версию 3.12.
В клиентах Azure OpenAI имя развертывания не совпадает с именем модели.
Перекрытие при разбиении на чанки имеет значение для строк с данными о комиссиях, которые простираются через границы чанков.
Локальный Redis идеален для демонстраций; в производстве всё равно требуются механизмы сохранения данных, аутентификация и резервное копирование.
LangGraph оправдывает своё использование, когда необходимо контролировать состояние узлов между этапами извлечения данных и их обновления.

Что дальше?

Заменить Redis на Azure AI Search, добавить механизмы защиты от персональных данных в запросах, контексте и ответах, а также превратить граф в API с хостингом. Сохранить «золотой» вопрос о графике комиссий в процессах CI, чтобы миграции не приводили к незаметному нарушению функций поиска.

Совет по разбиению на части для таблиц с тарифами: при возврате текста, ориентированного на ячейки, средством Document Intelligence следует использовать более крупные перекрытия, а номера страниц хранить в метаданных Redis, чтобы ответы могли указывать «страница 1, тарифы на банкоматы», а не просто отдельные фрагменты текста. Эти метаданные также позволяют применять гибридные фильтры, когда несколько PDF-файлов используют один индекс.

Совет по разбиению на части для таблиц с комиссиями: при возврате текста, ориентированного на ячейки, с помощью Document Intelligence следует использовать более крупные перекрытия, а номера страниц хранить в метаданных Redis, чтобы ответы могли указывать «страница 1, комиссии за операции АТМ», а не просто отдельные фрагменты текста. Эти метаданные также позволяют использовать гибридные фильтры, когда несколько PDF-файлов используют один и тот же индекс.

Проверочный список перед тем, как считать работу конвейера завершенной: модуль Redis загружен (FT._LIST работает), размер встраиваемых данных соответствует полю индекса, Document Intelligence возвращает непустой контент для примерного PDF-файла, и запрос о комиссиях за операции АТМ сохраняет первое место в рейтинге после холодного перезапуска контейнера. Автоматизируйте эти четыре проверки, чтобы обновления зависимостей не могли незаметно нарушить работу демо-версии.

Чек-лист операционной подготовки перед тем, как считать работу потока «завершённой»: загружен модуль Redis (FT._LIST работает), размер встраиваемых данных совпадает с полем индекса, сервис Document Intelligence возвращает непустой контент для примерного PDF, а запрос о стоимости операций в банкомате сохраняет первое место в рейтинге после холодного перезапуска контейнера. Автоматизируйте эти четыре проверки, чтобы обновления зависимостей не могли незаметно нарушить работу демо-версии.

← Все статьи Прислать бриф