azure redis rag langgraph document-intelligence embeddings

结合文档智能、嵌入技术及Redis的PDF语义问答系统

LangGraph提取费用表PDF文件，将其分割并嵌入Azure OpenAI中，同时通过本地的Redis向量索引回答自然语言形式的费用相关问题。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 11 分钟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模型对段落进行向量化处理。
通过 RediSearch 将向量存储在本地 Redis（或远程 Redis）中并支持搜索。

刻意保持规模小巧：仅几个Python文件，除了两次Azure AI调用外没有其他本地基础设施。

端到端测试使用了真实的美国银行单页费用表。问题为：“在网络外使用ATM取款需要支付多少手续费？”——相似度最高的答案是2.50美元。

为何选择Redis作为向量存储

Redis以其内存中的键值存储功能而闻名；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）
文档智能服务docintel-pdf-rag（FormRecognizer/S0版本）
带有嵌入模型部署功能的Azure OpenAIopenai-pdf-rag

在本地启动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.
...

使用`uv`的Python环境

请谨慎指定版本号。已废弃的LangChain Redis集成确实存在很大问题——建议使用配套仓库中记录的受维护包集，并在编写图节点之前验证导入情况。

环境变量

需指定文档智能处理端点/密钥、Azure OpenAI端点/密钥、嵌入模型部署名称（而非基础模型名称），以及Redis地址。

示例文档与流程运行

将费用明细PDF依次经过提取→分块→嵌入→更新操作。搜索前需确认Redis键的数量与分块数量一致。

执行语义搜索

使用相同的部署方式嵌入问题，对索引运行KNN算法，输出得分最高的段落。需验证关于ATM手续费的答案在黄金问题中的排名是否为第一。

经验总结

Python版本不一致会悄悄导致原生包无法正常使用——为此类架构请固定使用3.12版本。
Azure OpenAI客户端中部署名称与模型名称并不相同。
对于跨越分块边界的类似表格结构的费用行，分块方式的选择很重要。
本地Redis非常适合演示用途；实际生产环境仍需具备数据持久化、身份验证及备份功能。
当需要在信息提取与更新操作之间查看节点状态时，LangGraph能发挥重要作用。

下一步计划？

用Azure AI Search替代Redis，在查询、上下文及答案中添加个人信息保护机制，并将该图结构升级为托管API。需在持续集成流程中保留黄金费用表问题，以避免迁移过程中悄无声息地破坏信息检索功能。

费用表的分块技巧：当 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手续费查询仍保持排名第一。应自动执行这四项检查，以避免依赖升级悄无声息地破坏演示功能。

← 全部文章发送简报