azure redis rag langgraph document-intelligence embeddings

Semantische PDF-Fragen- und -Antwortfunktion mit Dokumentintelligenz, Embeddings und Redis

LangGraph extrahiert PDFs mit Gebührenplänen, teilt sie in Blöcke auf und integriert sie mit Azure OpenAI, wobei es gleichzeitig Fragen zu Gebühren in natürlicher Sprache aus einem lokalen Redis-Vektorindex beantwortet.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 26. Sept. 202611 Min. Lesezeit2476 Wörter

Quellen: Building a PDF Q&A Pipeline with Azure AI and Redis Vector Search

Organisationen sind überflutet mit unstrukturierten PDFs – Gebührenplänen, Offenlegungen, Verträgen, Richtlinien – bei denen Suchfunktionen nach Schlüsselwörtern nur unzureichend funktionieren. Dieses Projekt entwickelt einen kleinen Pipeline, der es ermöglicht, PDFs semantisch zu durchsuchen: Eine Frage in natürlicher Sprache liefert die relevantesten Abschnitte, selbst wenn die Formulierung unterschiedlich ist.

Als LangGraph-Zustandsgraph organisiert, umfasst der Pipeline folgende Schritte:

Auswählen des Textes mithilfe von Azure AI Document Intelligence (einfacher Text und Tabellen).
Teilen in überschneidende Abschnitte der Größe des Modells.
Einfügen in Vektoren mit Azure OpenAI text-embedding-3-small.
Speichern und suchen der Vektoren in lokalen Redis (oder entfernten Redis) über RediSearch.

Halten Sie es absichtlich klein: ein paar Python-Dateien sowie lokale Infrastruktur auf dem Laptop – abgesehen von den beiden Aufrufen bei Azure AI.

Der End-to-End-Test nutzte eine echte, einseitige Gebührenübersicht der U.S. Bank Expense Card. Frage: „Wie hoch sind die Gebühren für Geldabhebungen an nicht vernetzten Geldautomaten?“ – die Gebühr von 2,50 Dollar erhielt aufgrund der Ähnlichkeit den ersten Platz.

Warum Redis als Vektorlager

Redis ist bekannt als In-Memory-KV-Lager; RediSearch (in redis-stack) fügt Vektorindexierung hinzu. Gründe, warum es geeignet ist:

Geschwindigkeit – In-Memory-Latenz für interaktive Abfragen
Eines System, zwei Funktionen – Viele Anwendungen nutzen bereits Redis für Caching/Sessions
Flexible Indexierung – FLAT oder HNSW
Hybride Suche – Vektoren in Kombination mit Tags, numerischen und Volltextfiltern

Funktioniert überall – Laptop-Container bis zum verwalteten Cloud-Dienst mit derselben Abfragesyntax

Alternativen zu Redis für Suchfunktionen

Verwaltete Vektor-Datenbanken sowie Azure AI Search eignen sich gut als spätere Erweiterungen; Redis hält den Lernprozess lokal und schnell ablaufen.

Voraussetzungen

Docker; Azure CLI (az login); Zugang zu Azure OpenAI; Python 3.12 (3.14 brach in Tests diese Abhängigkeitskette); uv zur Verwaltung der Umgebung.

Azure-Ressourcen bereitstellen

Ressourcengruppe rg-pdf-rag (zum Beispiel eastus)
Document Intelligence docintel-pdf-rag (FormRecognizer/S0)
Azure OpenAI openai-pdf-rag mit einer Embeddings-Implementierung

Redis lokal starten

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-Umgebung mit `uv`

Pinnen Sie die Versionen sorgfältig. Die veralteten LangChain-Redis-Integrationen stellten ein echtes Problem dar – wählen Sie lieber die im begleitenden Repository dokumentierten, gepflegten Paketsets und überprüfen Sie die Importe vor dem Erstellen von Graphenodes.

Umgebungsvariablen

Weisen Sie auf den Document Intelligence-Endpunkt/Schlüssel, den Azure OpenAI-Endpunkt/Schlüssel, den Namen der Embedding-Deployment (nicht den Namen des Basismodells) sowie die Redis-URL hin.

Vorlage-Dokument und Pipeline-Ausführung

Führen Sie das PDF mit der Gebührensatzliste durch die Schritte extract → chunk → embed → upsert. Überprüfen Sie vor der Suche, ob die Anzahl der Redis-Schlüssel mit der Anzahl der Blöcke übereinstimmt.

Durchführen einer semantischen Suche

Embedden Sie die Frage mit derselben Bereitstellung, führen Sie KNN gegen den Index durch und geben Sie die besten Passagen mit ihren Scores aus. Überprüfen Sie, ob die Antwort zur ATM-Gebühr bei der „goldenen“ Frage auf Rang 1 liegt.

Gelernte Lektionen

Eine abweichende Python-Version kann native Pakete heimlich beeinträchtigen – verwenden Sie für diesen Stack fest 3.12.
Der Name der Bereitstellung unterscheidet sich vom Modellnamen in den Azure OpenAI-Clientn.
Die Überschneidung bei der Aufteilung in Blöcke ist wichtig für gebührenbezogene Zeilen, die Grenzen zwischen den Blöcken überschreiten.
Ein lokaler Redis eignet sich perfekt für Demonstrationen; in der Produktion sind jedoch Persistenz, Authentifizierung und Backups weiterhin notwendig.
LangGraph ist nützlich, wenn Sie einen überprüfbaren Knotenzustand zwischen Extraktion und Aktualisierung benötigen.

Was kommt als Nächstes?

Ersetzen Sie Redis durch Azure AI Search, fügen Sie Schutzmechanismen für PII zu Abfragen, Kontexten und Antworten hinzu und heben Sie das Graphen in eine gehostete API auf. Behalten Sie die „goldene“ Frage zur Gebührenstruktur in der CI-Phase bei, damit Migrationen nicht heimlich die Abfragefunktion stören können.

Tipp zum Aufteilen von Gebührentabellen: Wählen Sie bei Texten, die von Document Intelligence in zellorientierter Form zurückgegeben werden, leicht größere Überschneidungen, und speichern Sie die Seitenzahlen in den Redis-Metadaten, damit Antworten auf „Seite 1, ATM-Gebühren“ verweisen können anstelle von losen Auszügen. Diese Metadaten ermöglichen außerdem hybride Filter, wenn mehrere PDFs denselben Index teilen.

Tipp zur Aufteilung in Blöcke für Gebührentabellen: Wählen Sie bei Texten, die von Document Intelligence in zellorientierter Form zurückgegeben werden, leicht größere Überschneidungen, und speichern Sie die Seitenzahlen in den Redis-Metadaten, damit Antworten beispielsweise „Seite 1, ATM-Gebühren“ angeben können anstelle von losen Auszügen. Diese Metadaten ermöglichen außerdem hybride Filter, wenn mehrere PDFs denselben Index teilen.

Tipp zur Aufteilung in Blöcke für Gebührentabellen: Wählen Sie bei Texten, die von Document Intelligence zellbasiert zurückgegeben werden, leicht größere Überschneidungen, und speichern Sie die Seitenzahlen in den Redis-Metadaten, damit Antworten beispielsweise „Seite 1, ATM-Gebühren“ angeben können anstelle von losen Auszügen. Diese Metadaten ermöglichen außerdem hybride Filter, wenn mehrere PDFs denselben Index teilen.

Operative Überprülliste, bevor der Pipeline-Status als „abgeschlossen“ markiert wird: Das Redis-Modul ist geladen (FT._LIST funktioniert), die Eingabedimension entspricht dem Indexfeld, Document Intelligence liefert für das Beispiel-PDF inhaltliche Daten, und die Schlüsselabfrage zu den ATM-Gebühren bleibt nach einem Neustart des Containers weiterhin auf Rang 1. Automatisieren Sie diese vier Überprüfungen, damit Upgrade-Aktualisierungen die Demo nicht heimlich stören können.

Die Betriebskontrollliste vor dem Markieren der Pipeline als „abgeschlossen“ sieht wie folgt aus: Das Redis-Modul ist geladen (FT._LIST funktioniert), die Eingabedimension entspricht dem Indexfeld, Document Intelligence liefert für das Beispiel-PDF nicht leeres Inhalt, und die Referenzabfrage zu den ATM-Gebühren bleibt nach einem kaltstart des Containers auf Rang 1. Automatisieren Sie diese vier Überprüfungen, damit Upgrade-Aktualisierungen die Demo nicht heimlich stören.

← Alle Artikel Brief senden