azure redis rag langgraph document-intelligence embeddings

Pytania i odpowiedzi semantyczne w formacie PDF z wykorzystaniem Document Intelligence, embeddingów oraz Redis

LangGraph wydobywa pliki PDF z harmonogramem opłat, dzieli je na fragmenty i integruje z Azure OpenAI, a także odpowiada na pytania dotyczące opłat w języku naturalnym za pomocą lokalnego indeksu wektorowego Redis.

Wydawca: REACTAPP.TOPDodane na tej stronie 26 wrz 202611 min czytania2476 słów

Źródła: Building a PDF Q&A Pipeline with Azure AI and Redis Vector Search

Organizacje toną w nieuporządkowanych plikach PDF – tabelach opłat, informacjach ujawnieniowych, umowach, politykach – których wyszukiwanie za pomocą słów kluczowych sprawdza się słabo. Ten projekt tworzy niewielki pipeline, który umożliwia semantyczne wyszukiwanie w plikach PDF: pytanie w języku naturalnym zwraca najbardziej istotne fragmenty, nawet gdy sformułowanie się różni.

Pipeline ten jest zorganizowany jako graf stanów LangGraph i składa się z następujących kroków:

Wydobywanie tekstu za pomocą Azure AI Document Intelligence (tekst prosty i tabele).
Dzielenie tekstu na nakładające się fragmenty odpowiadające rozmiarowi modelu.
Wkładanie danych do formatu wektorowego za pomocą Azure OpenAI text-embedding-3-small.
Zachowywanie i wyszukiwanie wektorów w lokalnym Redis (lub zdalnym Redis) za pośrednictwem RediSearch.

Zachowaj strukturę celowo prostą: kilka plików w Pythonie, lokalna infrastruktura na laptopie oprócz dwóch wywołań do Azure AI.

Test od początku do końca wykorzystywał rzeczywisty harmonogram opłat kart kredytowych banku U.S. Bank składający się z jednej strony. Pytanie brzmiało: „jaka jest opłata za wypłatę z bankomatu poza siecią?” — opłata 2,50 dolarów uzyskała najwyższą ocenę pod względem podobieństwa.

Dlaczego Redis jako magazyn wektorów

Redis jest znany jako magazyn typu KV pamięci operacyjnej; RediSearch (w pakiecie redis-stack) dodaje funkcję indeksowania wektorów. Powody, dla których jest odpowiedni:

Szybkość — niska latencja pamięci operacyjnej umożliwiająca szybkie wyszukiwanie
Jeden system, dwa zastosowania — wiele aplikacji już używa Redis do przechowywania cache’ów i sesji
Elastyczne indeksowanie — opcje FLAT lub HNSW
Hybrydowe wyszukiwanie — wektory w połączeniu z tagami, filtrami numerycznymi i filtrom tekstu pełnego

Działa wszędzie — kontener laptopa do zarządzanego chmury przy użyciu tej samej struktury zapytań

Aльтernatywy do wyszukiwania w Redis

Zarządzane bazy danych wektorowych oraz Azure AI Search to dobre opcje na późniejsze ulepszenia; Redis utrzymuje proces nauki w lokalnym środowisku i sprawia, że jest szybki.

Wymagania wstępne

Docker; Azure CLI (az login); dostęp do Azure OpenAI; Python 3.12 (wersja 3.14 spowodowała problemy z tą ścieżką zależności podczas testów); uv do zarządzania środowiskiem.

Załóż zasoby w Azure

Grupa zasobów rg-pdf-rag (na przykład eastus)
Document Intelligence docintel-pdf-rag (FormRecognizer/S0)
Azure OpenAI openai-pdf-rag z implementacją embeddings

Rozpocznij działanie Redis lokalnie

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.
...

Środowisko Python z `uv`

Należy starannie ustalać wersje bibliotek. Zastarzałe integracje LangChain z Redis stanowiły poważny problem – lepiej używać utrzymywanego zestawu pakietów opisanego w towarzyszącym repozytorium oraz sprawdzać importy przed tworzeniem węzłów grafu.

Zmienne środowiskowe

Wskazuj na endpoint/klucz Document Intelligence, endpoint/klucz Azure OpenAI, nazwę implementacji embeddingów (a nie nazwę modelu bazowego) oraz adres URL Redis.

Przykładowy dokument i uruchomienie pipeline’a

Przesuń plik PDF z harmonogramem opłat przez procesy extract → chunk → embed → upsert. Sprawdź, czy liczba kluczy w Redis odpowiada liczbie fragmentów danych przed przeprowadzaniem wyszukiwania.

Przeprowadzenie wyszukiwania semantycznego

Wbuduj pytanie przy użyciu tej samej konfiguracji, uruchom KNN na indeksie i wydrukuj najlepsze fragmenty wraz z ich ocenami. Sprawdź, czy odpowiedź dotycząca opłaty w bankomacie zajmuje pierwsze miejsce w pytaniu referencyjnym.

Lekcje wyciągnięte

Niezgodność wersji Pythona potajemnie psuje dostępne pakiety — ustal wersję 3.12 dla tej konfiguracji.
Nazwa konfiguracji ≠ nazwa modelu w klientach Azure OpenAI.
Przekrycie przy dzieleniu na fragmenty ma znaczenie dla wierszy z opłatami przecinających granice fragmentów.
Lokalny Redis jest doskonały do demonstracji; w środowisku produkcyjnym nadal potrzebna jest trwałość danych, autoryzacja i kopie zapasowe.
LangGraph okazuje się przydatny, gdy chcesz mieć możliwość inspekcji stanu węzłów pomiędzy procesem wyodrębniania a aktualizacją danych.

Co dalej?

Zastąp Redis usługą Azure AI Search, dodaj zasady ochrony danych osobowych w zapytaniach, kontekście i odpowiedziach, a także przenieś strukturę grafu do hostowanej API. Zachowaj pytanie referencyjne dotyczące harmonogramu opłat w procesie CI, aby migracje nie potajemnie nie zakłócały funkcjonowania systemu wyszukiwania.

Wskazówka dotycząca dzielenia na fragmenty w tabelach opłat: wybieraj nieco większe nakładanie się fragmentów, gdy Document Intelligence zwraca tekst skierowany na komórki, oraz przechowuj numery stron w metadanych Redis, aby odpowiedzi mogły odwoływać się do „strony 1, opłaty za bankomaty” zamiast do losowych fragmentów tekstu. Te metadane umożliwiają również użycie filtrów hybrydowych, gdy kilka plików PDF dzieli się jednym indeksem.

Wskazówka dotycząca dzielenia na fragmenty w tabelach opłat: wybieraj nieco większe nakładanie się fragmentów, gdy Document Intelligence zwraca tekst skierowany na komórki, oraz przechowuj numery stron w metadanych Redis, aby odpowiedzi mogły odwoływać się do „strona 1, opłaty za bankomaty” zamiast do losowych fragmentów tekstu. Te metadane umożliwiają również użycie filtrów hybrydowych, gdy kilka plików PDF dzieli się jednym indeksem.

Wskazówka dotycząca dzielenia na fragmenty w tabelach opłat: wybieraj nieco większe nakładanie się fragmentów, gdy Document Intelligence zwraca tekst skierowany na komórki, a numery stron przechowuj w metadanych Redis, aby odpowiedzi mogły odwoływać się do „strony 1, opłaty za ATM” zamiast do losowych fragmentów tekstu. Te metadane umożliwiają również użycie filtrów hybrydowych, gdy kilka plików PDF dzieli ten sam indeks.

Listwa kontrolna przed uznaniem procesu za zakończony: moduł Redis został załadowany (FT._LIST działa), wymiary embedowania odpowiadają polu indeksu, Document Intelligence zwraca niepusty treść dla przykładowego pliku PDF, a kluczowa zapytanie dotyczące opłat za ATM pozostaje na pierwszym miejscu po ponownym uruchomieniu kontenera. Zautomatyzuj te cztery sprawdzenia, aby aktualizacje zależności nie mogły potajemnie uszkodzić demonstracji.

Listwa kontrolna przed uruchomieniem pipeline jest uznawana za „gotową”, gdy moduł Redis został załadowany (FT._LIST działa), wymiary embeddingów odpowiadają polu indeksu, Document Intelligence zwraca niepuste treści dla przykładowego pliku PDF, a zapytanie dotyczące opłat w bankomatach pozostaje na pierwszym miejscu po zimnym restartie kontenera. Zautomatyzuj te cztery sprawdzenia, aby aktualizacje zależności nie mogły potajemnie uszkodzić demonstracji.

← Wszystkie artykuły Wyślij brief