azure redis rag langgraph document-intelligence embeddings

PDF-аўтаматызаваныя запытанні і адказы на аснове сэмантыкі, з дапамогай функцый Document Intelligence, embeddings і Redis

LangGraph выкачвае PDF-файлы з прайс-табламі, дзельнічае іх і вбудоввае ў Azure OpenAI, а таксама адпавядае на запытанні пра платы, заданныя мовай людзей, на аднойчынным векторным індэксе Redis.

Выдавец: REACTAPP.TOPДададзена на сайт 26 вер 202611 хв чытання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 з рэалізацыяй функцыі embeddings

Запуск 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 проты індэксу, вывесты найлепшыя фрагменты з рэйтынгамі. Пераканацца, што адказ пра платах за ATM знаходзіцца на першай пазеці для «золатага» запіту.

Усваятыя урокі

Неравенства версій Python таямніча парадузяюць стандартныя бібліятэкі — для гэтага стаку неабяжна версія 3.12.
У кліентах Azure OpenAI назва развяртання не збігаецца з назвай модэлі.
Перакрыцце пад час разбівання на фрагменты мае значэнне для радкаў з платамі, якія пераходзяць межы фрагментавання.
Локальны Redis ідэальна для дэманстрацый; у працоўных сітках все ж неабяжна наявнасць стойкага зберышча, аўтэнтыфікаціі і рэзервных копій.
LangGraph становіць ся корыстным, калі трэба можлівасць адзіроўваць стан вузлаў межы выкарыстоўванням іх для выкарабківання дадзеных і ўведэння змян.

Што далей?

Заменіць Redis на Azure AI Search, дадаць правіла захавання персональных дадзеных для запытов, кантэксту і адказаў, а таксама пераканацца, што граф будзе працаваць як хоставаная API. Залічыць «золаты» запіт пра графак плат у працэсе CI, каб міграцыі не таямніча не парадузілі працэс выкарыстоўвання дадзеных.

Парадакт па часткаванні для таблыцэ кампансаў: калі Document Intelligence вяртае тэкст, орыентаваны на клеткі, лепш выбіраць адносна большыя перакрыцці, а номеры сторанаках зберагаць у метаданных Redis, каб адпаведзі змаглі цытаваць «сторанка 1, кампансіяў за ATM» замест аддзельных фрагментаў. Гэтыя метаданныя таксама дазволяюць выкарыстоўваць гібрыдныя фільтры, калі калькі PDF-аў дзеляць аднаковы індэкс.

Парадакт па часткаванні для таблыцэ кампансаў: калі Document Intelligence вяртае тэкст, орыѐнтаваны на клеткі, лепш выбіраць адносна большыя перакрыцці, а номеры сторанак хаваць у метаданных Redis, каб адпаведзі змаглі цытаваць «сторанка 1, кампансіяў за ATM» замест аддзельных фрагментаў. Гэтыя метаданныя таксама дазволяюць выкарыстоўваць гібрыдныя фільтры, калі калькі PDF-аў дзеляць аднаковы індэкс.

Парадакт паводле часткавання для табелей з комісіямі: калі Document Intelligence вяртае тэкст, орыентаваны на клеткі, лепш выбіраць адносна большыя перакрыцці, а номеры сторонаках зберагаць у метаданых Redis, каб адказы моглі цытаваць “сторанка 1, комісіі за ATM” замест аддзеяных фрагментаў. Гэтыя метаданы таксама дазволяюць выкарыстоўваць гібрыдныя фільтры, калі калькі PDF-аў дзеляць адной індэксам.

Спіс пераконтрацоў пры готавячыся выклікаць працэс “гатова”: модуль Redis завантажаны (FT._LIST працюе), размернасць умяшчэння падходзіць да поля індэкса, Document Intelligence вяртае непусты тэкст для прыкладнага PDF-а, і запіт пра комісіі за ATM застаецца на першай пазіцыі пасля холаднага запуску кантэйнера. Автаматызуйце гэтыя чатыры пераконтроўкі, каб абнаўленні залежнасцей не моглі таямніча зламаць дэманстрацыю.

Співвідношэнне элементаў прабны гэтапу прыямо перад тым, як называць канал “завершаным”: модуль Redis заваноўваны (FT._LIST працюе), размернасць увярнення падходзіць да поля індэкса, Document Intelligence вяртае непусты тэкст для прыводнага PDF, а запіт пра зношкі ATM застаецца на 1-му месцы пасля холаднага перзапуску кантэйнера. Автаматызуйце гэтыя чатыры пераканрання, каб абнаўленні залежнасцяў не моглі таямніча зламаць дэманстрацыю.

← Усе артыкулы Даслаць брыф