azure redis rag langgraph document-intelligence embeddings

PDF semantic Q&A with Document Intelligence, embeddings, and Redis

LangGraph extracts fee-schedule PDFs, chunks and embeds with Azure OpenAI, and answers natural-language fee questions from a local Redis vector index.

Publisher: REACTAPP.TOPAdded to this site Sep 26, 202611 min read2476 words

This article is published in English.

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

Organizations drown in unstructured PDFs—fee schedules, disclosures, contracts, policies—that keyword search handles poorly. This project builds a small pipeline that makes a PDF semantically searchable: a natural-language question returns the most relevant passages even when wording differs.

Orchestrated as a LangGraph state graph, the pipeline:

Extract text with Azure AI Document Intelligence (plain text and tables).
Chunk into overlapping, model-sized passages.
Embed with Azure OpenAI text-embedding-3-small.
Store and search vectors in local Redis (or remote Redis) via RediSearch.

Keep it deliberately small: a handful of Python files, laptop-local infra aside from the two Azure AI calls.

End-to-end test used a real one-page U.S. Bank Expense Card fee schedule. Question: “what is the ATM withdrawal fee out of network?” — the $2.50 passage ranked first by similarity.

Why Redis as the vector store

Redis is famous as an in-memory KV store; RediSearch (in redis-stack) adds vector indexing. Reasons it fits:

Speed — in-memory latency for interactive retrieval
One system, two jobs — many apps already run Redis for cache/sessions
Flexible indexing — FLAT or HNSW
Hybrid search — vectors plus tags/numeric/full-text filters
Runs anywhere — laptop container to managed cloud with the same query shape

Redis search alternatives

Managed vector DBs and Azure AI Search are fine upgrades later; Redis keeps the teaching loop local and fast.

Prerequisites

Docker; Azure CLI (az login); Azure OpenAI access; Python 3.12 (3.14 broke this dependency stack in testing); uv for env management.

Provision Azure resources

Resource group rg-pdf-rag (for example eastus)
Document Intelligence docintel-pdf-rag (FormRecognizer/S0)
Azure OpenAI openai-pdf-rag with an embeddings deployment

Start Redis locally

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 environment with `uv`

Pin versions carefully. Deprecated LangChain Redis integrations were a real footgun—prefer the maintained package set documented in the companion repo and verify imports before writing graph nodes.

Environment variables

Point at Document Intelligence endpoint/key, Azure OpenAI endpoint/key, embedding deployment name (not base model name), and Redis URL.

Sample document and pipeline run

Feed the fee-schedule PDF through extract → chunk → embed → upsert. Confirm Redis key counts match chunk counts before searching.

Run a semantic search

Embed the question with the same deployment, run KNN against the index, print top passages with scores. Validate that the ATM fee answer is rank-1 on the golden question.

Lessons learned

Python version skew breaks native wheels quietly—pin 3.12 for this stack.
Deployment name ≠ model name in Azure OpenAI clients.
Overlap in chunking matters for table-like fee rows that span chunk boundaries.
Local Redis is perfect for demos; production still needs persistence, auth, and backups.
LangGraph earns its keep when you want inspectable node state between extract and upsert.

What’s next?

Swap Redis for Azure AI Search, add PII guardrails on query/context/answer, and promote the graph to a hosted API. Keep the golden fee-schedule question in CI so migrations cannot silently break retrieval.

Chunking tip for fee tables: prefer slightly larger overlaps when Document Intelligence returns cell-oriented text, and store page numbers in Redis metadata so answers can cite “page 1, ATM fees” instead of floating snippets. That metadata also enables hybrid filters when multiple PDFs share one index.

Operational checklist before calling the pipeline “done”: Redis module loaded (FT._LIST works), embedding dimension matches the index field, Document Intelligence returns non-empty content for the sample PDF, and the golden ATM-fee query stays rank-1 after a cold restart of the container. Automate those four checks so dependency upgrades cannot silently break the demo.

← All articles Send the brief