azure redis rag langgraph document-intelligence embeddings

Preguntas y respuestas semánticas en PDF con inteligencia de documentos, embeddings y Redis

LangGraph extrae los PDFs con el calendario de tarifas, los divide en fragmentos y los integra con Azure OpenAI, además de responder preguntas sobre tarifas en lenguaje natural a partir de un índice vectorial local de Redis.

Editor: REACTAPP.TOPAñadido a este sitio 26 sept 202611 min de lectura2476 palabras

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

Las organizaciones se ven abrumadas por PDFs no estructurados: tablas de tarifas, declaraciones, contratos, políticas, que la búsqueda por palabras clave maneja de forma deficiente. Este proyecto crea un pequeño pipeline que permite buscar en un PDF de manera semántica*: una pregunta en lenguaje natural devuelve los pasajes más relevantes incluso cuando la redacción difiere.

Orquestado como un grafo de estado LangGraph, el pipeline:

Extrae texto con Azure AI Document Intelligence (texto plano y tablas).

Divide el texto en pasajes superpuestos del tamaño adecuado para el modelo.

Incorpora los datos usando Azure OpenAI text-embedding-3-small.

Almacena y busca los vectores en Redis local (o Redis remoto) a través de RediSearch.

Manténgalo deliberadamente pequeño: un puñado de archivos en Python, infraestructura local en la computadora portátil aparte de las dos llamadas a Azure AI.

La prueba de extremo a extremo utilizó un calendario real de tarifas de la tarjeta de gastos de U.S. Bank de una página. Pregunta: “¿cuál es la tarifa por retiro en cajero automático fuera de la red?” — la tarifa de 2,50 dólares obtuvo el primer lugar en similitud.

Por qué Redis como almacén de vectores

Redis es famoso por ser un almacén KV en memoria; RediSearch (en redis-stack) agrega indexación de vectores. Razones por las que es adecuado:

Rapidez — latencia en memoria para búsquedas interactivas
Un sistema, dos funciones — muchas aplicaciones ya utilizan Redis para caché/sesiones
Indexación flexible — FLAT o HNSW
Búsqueda híbrida — vectores además de filtros por etiquetas, numéricos y texto completo

Funciona en cualquier lugar — contenedor para portátiles hasta nube gestionada, con la misma estructura de consultas

Opciones alternativas para búsquedas en Redis

Las bases de datos vectoriales gestionadas y Azure AI Search son actualizaciones adecuadas en etapas posteriores; Redis mantiene el ciclo de aprendizaje local y rápido.

Requisitos previos

Docker; Azure CLI (az login); acceso a Azure OpenAI; Python 3.12 (3.14 rompió esta pila de dependencias en pruebas); uv para la gestión del entorno.

Aprovisionar recursos de Azure

Grupo de recursos rg-pdf-rag (por ejemplo eastus)
Document Intelligence docintel-pdf-rag (FormRecognizer/S0)
Azure OpenAI openai-pdf-rag con una implementación de embeddings

Inicio de Redis localmente

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

Entorno de Python con `uv`

Fije las versiones con cuidado. Las integraciones obsoletas de LangChain con Redis fueron un verdadero problema; prefiera el conjunto de paquetes mantenidos documentado en el repositorio complementario y verifique las importaciones antes de escribir nodos del grafo.

Variables de entorno

Indique el endpoint/clave de Document Intelligence, el endpoint/clave de Azure OpenAI, el nombre del despliegue de los embeddings (no el nombre del modelo base) y la URL de Redis.

Ejemplo de documento y ejecución del pipeline

Pase el PDF con el calendario de tarifas por los pasos extract → chunk → embed → upsert. Confirme que la cantidad de claves en Redis coincida con la cantidad de chunks antes de realizar búsquedas.

Ejecutar una búsqueda semántica

Inserte la pregunta con el mismo despliegue, ejecute KNN contra el índice e imprima los pasajes más relevantes junto con sus puntuaciones. Verifique que la respuesta sobre la tarifa del cajero automático ocupe el primer lugar en la pregunta de referencia.

Lecciones aprendidas

Las diferencias en la versión de Python afectan silenciosamente a los paquetes nativos; fije la versión en 3.12 para esta configuración.
En los clientes Azure OpenAI, el nombre del despliegue no coincide con el nombre del modelo.
La superposición al dividir los datos es importante para las filas de tarifas similares a tablas que trascienden los límites entre bloques.
Redis local es ideal para demostraciones; en entornos de producción se necesita persistencia, autenticación y copias de seguridad.
LangGraph resulta útil cuando se desea poder inspeccionar el estado de los nodos entre la extracción y la actualización.

¿Qué viene después?

Sustituya Redis por Azure AI Search, añada medidas de protección para datos personales en las consultas, el contexto y las respuestas, y convierta la estructura gráfica en una API alojada. Mantenga la pregunta de referencia sobre el calendario de tarifas en los procesos de integración continua para evitar que las migraciones dañen silenciosamente la recuperación de información.

Consejo de fragmentación para tablas de tarifas: prefiera superposiciones ligeramente mayores cuando Document Intelligence devuelva texto orientado a celdas, y almacene los números de página en la metadata de Redis para que las respuestas puedan citar “página 1, tarifas de cajeros automáticos” en lugar de fragmentos sueltos. Esa metadata también permite filtros híbridos cuando varios PDF comparten un mismo índice.

Consejo de fragmentación para tablas de tarifas: prefiera superposiciones ligeramente mayores cuando Document Intelligence devuelva texto orientado a celdas, y almacene los números de página en los metadatos de Redis para que las respuestas puedan citar “página 1, tarifas de cajeros automáticos” en lugar de fragmentos sueltos. Dichos metadatos también permiten filtros híbridos cuando varios PDF comparten un mismo índice.

Consejo para la división en bloques de las tablas de tarifas: prefiera superposiciones ligeramente mayores cuando Document Intelligence devuelva texto orientado a celdas, y almacene los números de página en los metadatos de Redis para que las respuestas puedan citar “página 1, tarifas de cajeros automáticos” en lugar de fragmentos sueltos. Dichos metadatos también permiten filtros híbridos cuando varios PDF comparten un mismo índice.

Lista de verificación operativa antes de considerar que el proceso está “terminado”: el módulo de Redis está cargado (FT._LIST funciona), la dimensión de incrustación coincide con el campo del índice, Document Intelligence devuelve contenido no vacío para el PDF de ejemplo, y la consulta estándar sobre tarifas de cajeros automáticos mantiene el primer lugar en los resultados tras un reinicio en frío del contenedor. Automatice estas cuatro verificaciones para que las actualizaciones de dependencias no dañen silenciosamente la demostración.

La lista de verificación operativa antes de considerar que la pipeline está “listo”: el módulo Redis se ha cargado (FT._LIST funciona), la dimensión de incrustación coincide con el campo del índice, Document Intelligence devuelve contenido no vacío para el PDF de ejemplo, y la consulta de tarifas ATM permanece en el puesto 1 después de un reinicio en frío del contenedor. Automatice esas cuatro verificaciones para que las actualizaciones de dependencias no dañen silenciosamente la demostración.

← Todos los artículos Enviar el brief