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, комісії за ATM“, а не на випадкові уривки тексту. Ці метадані також дозволяють використовувати гібридні фільтри, коли кілька PDF-файлів мають спільний індекс.

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

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

← Усі статті Надіслати бриф