langgraph azure embeddings rag ephemeral sessions

在无需保留向量的情况下，使用 LangGraph 进行短暂性语义搜索

会在会话结束时消失的会话级嵌入能够明确界定边界——这样一来，后续切换到 Redis 或 Azure AI Search 时只需修改存储适配器即可。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 8 分钟1939 词

来源: Semantic Search : A Simple, Ephemeral Approach with LangGraph + Azure

本次构建的内容及为何采用临时方案

有时，针对会话而言最合适的向量存储方式根本不需要持久化存储：嵌入、搜索、回答、丢弃。本示例利用 Azure 的嵌入功能结合 LangGraph 构建了一个简单的语义搜索路径，通过保持状态临时化，避免演示或短期会话留下无主索引。

架构

LangGraph 节点负责数据导入（可选）、嵌入、检索和生成操作。向量存储在进程内存中或以会话为键的临时结构中。该方案无需使用长期存在的 Redis 索引。

环境准备

需配置 Azure OpenAI（嵌入功能及聊天服务）、Python 3.12 及以上版本以及项目依赖项。在编写客户端代码之前，请先确认部署名称。

代码实现

cd infra
./deploy.sh

az cognitiveservices account create --kind FormRecognizer --sku S0 ...   # extraction
az cognitiveservices account create --kind AIServices --sku S0 ...      # Foundry
az cognitiveservices account deployment create \
  --model-name text-embedding-3-small --model-format OpenAI ...          # embeddings
az cognitiveservices account deployment create \
  --model-name gpt-5-mini --model-format OpenAI ...                      # chat

DOCUMENTINTELLIGENCE_ENDPOINT=https://adi-....cognitiveservices.azure.com/
DOCUMENTINTELLIGENCE_KEY=...
AZURE_OPENAI_ENDPOINT=https://aif-....cognitiveservices.azure.com/
AZURE_OPENAI_API_KEY=...
AZURE_OPENAI_API_VERSION=2024-10-21
AZURE_OPENAI_EMBEDDING_DEPLOYMENT=text-embedding-3-small
AZURE_OPENAI_CHAT_DEPLOYMENT=gpt-5-mini

uv add langgraph langchain-core langchain-openai langchain-text-splitters \
       azure-ai-documentintelligence azure-core python-dotenv

import os
from dotenv import load_dotenv

load_dotenv()


def _require(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        raise RuntimeError(
            f"Missing required environment variable: {name}. "
            f"Did you run infra/deploy.sh and fill in app/.env?"
        )
    return val


# --- Azure AI Foundry (embedding + chat model deployments) ---
AZURE_OPENAI_ENDPOINT = _require("AZURE_OPENAI_ENDPOINT")
AZURE_OPENAI_API_KEY = _require("AZURE_OPENAI_API_KEY")
AZURE_OPENAI_API_VERSION = os.environ.get("AZURE_OPENAI_API_VERSION", "2024-10-21")
AZURE_OPENAI_EMBEDDING_DEPLOYMENT = os.environ.get(
    "AZURE_OPENAI_EMBEDDING_DEPLOYMENT", "text-embedding-3-small"
)
AZURE_OPENAI_CHAT_DEPLOYMENT = os.environ.get("AZURE_OPENAI_CHAT_DEPLOYMENT", "gpt-5-mini")

# --- Azure Document Intelligence (used for all file text extraction) ---
DOCUMENTINTELLIGENCE_ENDPOINT = _require("DOCUMENTINTELLIGENCE_ENDPOINT")
DOCUMENTINTELLIGENCE_KEY = _require("DOCUMENTINTELLIGENCE_KEY")

# --- App tuning ---
CHUNK_SIZE = int(os.environ.get("CHUNK_SIZE", "800"))
CHUNK_OVERLAP = int(os.environ.get("CHUNK_OVERLAP", "120"))
TOP_K = int(os.environ.get("TOP_K", "4"))

MAX_FILES = 3
MAX_FILE_SIZE_MB = 5

from pathlib import Path
from typing import List

from azure.ai.documentintelligence import DocumentIntelligenceClient
from azure.ai.documentintelligence.models import DocumentContentFormat
from azure.core.credentials import AzureKeyCredential
from langchain_text_splitters import RecursiveCharacterTextSplitter

from . import config

# Formats Document Intelligence's prebuilt-layout model accepts.
DOCINTEL_EXTENSIONS = {
    ".pdf", ".docx", ".xlsx", ".pptx", ".html",
    ".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".heif",
}
# Formats with no layout to analyze - read as-is, no API call.
PLAIN_TEXT_EXTENSIONS = {".txt", ".md", ".csv"}
SUPPORTED_EXTENSIONS = DOCINTEL_EXTENSIONS | PLAIN_TEXT_EXTENSIONS

_client: DocumentIntelligenceClient | None = None


def _get_client() -> DocumentIntelligenceClient:
    global _client
    if _client is None:
        _client = DocumentIntelligenceClient(
            endpoint=config.DOCUMENTINTELLIGENCE_ENDPOINT,
            credential=AzureKeyCredential(config.DOCUMENTINTELLIGENCE_KEY),
        )
    return _client


def validate_file(file_path: str) -> None:
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(file_path)
    if path.suffix.lower() not in SUPPORTED_EXTENSIONS:
        raise ValueError(
            f"Unsupported file type '{path.suffix}'. Supported: {sorted(SUPPORTED_EXTENSIONS)}"
        )
    size_mb = path.stat().st_size / (1024 * 1024)
    if size_mb > config.MAX_FILE_SIZE_MB:
        raise ValueError(
            f"'{path.name}' is {size_mb:.1f}MB, over the {config.MAX_FILE_SIZE_MB}MB limit"
        )


def extract_text(file_path: str) -> str:
    """Pull text (and table structure, as markdown) out of a single uploaded file."""
    path = Path(file_path)
    ext = path.suffix.lower()

    if ext in PLAIN_TEXT_EXTENSIONS:
        return path.read_text(errors="ignore")

    if ext in DOCINTEL_EXTENSIONS:
        client = _get_client()
        with open(file_path, "rb") as f:
            # Passing a file handle directly (vs. wrapping in
            # AnalyzeDocumentRequest) makes the SDK set
            # content-type: application/octet-stream automatically.
            poller = client.begin_analyze_document(
                "prebuilt-layout",
                body=f,
                output_content_format=DocumentContentFormat.MARKDOWN,
            )
        result = poller.result()
        return result.content or ""

    raise ValueError(f"Unsupported file type: {ext}")


def chunk_text(text: str) -> List[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=config.CHUNK_SIZE,
        chunk_overlap=config.CHUNK_OVERLAP,
    )
    return [c for c in splitter.split_text(text) if c.strip()]


def load_and_chunk_files(file_paths: List[str]) -> List[dict]:
    """Returns a flat list of {text, file_name, chunk_index} dicts, ready to embed."""
    if len(file_paths) > config.MAX_FILES:
        raise ValueError(f"Max {config.MAX_FILES} files per session, got {len(file_paths)}")

    for fp in file_paths:
        validate_file(fp)

    chunks = []
    for fp in file_paths:
        text = extract_text(fp)
        for i, chunk in enumerate(chunk_text(text)):
            chunks.append(
                {
                    "text": chunk,
                    "file_name": Path(fp).name,
                    "chunk_index": i,
                }
            )
    return chunks

embeddings = AzureOpenAIEmbeddings(
    azure_endpoint=config.AZURE_OPENAI_ENDPOINT,
    api_key=config.AZURE_OPENAI_API_KEY,
    api_version=config.AZURE_OPENAI_API_VERSION,
    azure_deployment=config.AZURE_OPENAI_EMBEDDING_DEPLOYMENT,
)
llm = AzureChatOpenAI(
    azure_endpoint=config.AZURE_OPENAI_ENDPOINT,
    api_key=config.AZURE_OPENAI_API_KEY,
    api_version=config.AZURE_OPENAI_API_VERSION,
    azure_deployment=config.AZURE_OPENAI_CHAT_DEPLOYMENT,
    reasoning_effort="minimal",
)

# One InMemoryVectorStore per chat session. Lives only as long as this
# process does - no persistence, no TTL, nothing to provision.
_SESSION_STORES: Dict[str, InMemoryVectorStore] = {}


def get_vector_store(session_id: str) -> InMemoryVectorStore:
    if session_id not in _SESSION_STORES:
        _SESSION_STORES[session_id] = InMemoryVectorStore(embeddings)
    return _SESSION_STORES[session_id]


def clear_session(session_id: str) -> None:
    """Explicit cleanup, since nothing expires on its own in memory."""
    _SESSION_STORES.pop(session_id, None)


class ChatFileSearchState(TypedDict):
    session_id: str
    file_paths: List[str]
    query: Optional[str]
    retrieved: List[Document]
    answer: Optional[str]


def ingest_node(state: ChatFileSearchState) -> ChatFileSearchState:
    if not state.get("file_paths"):
        return state

    chunks = load_and_chunk_files(state["file_paths"])
    if not chunks:
        return state

    docs = [
        Document(
            page_content=c["text"],
            metadata={"file_name": c["file_name"], "chunk_index": c["chunk_index"]},
        )
        for c in chunks
    ]

    vs = get_vector_store(state["session_id"])
    vs.add_documents(docs)
    return state


def retrieve_node(state: ChatFileSearchState) -> ChatFileSearchState:
    if not state.get("query"):
        state["retrieved"] = []
        return state

    vs = get_vector_store(state["session_id"])
    # No filter needed: this store only ever contains this session's docs.
    state["retrieved"] = vs.similarity_search(state["query"], k=config.TOP_K)
    return state


def generate_node(state: ChatFileSearchState) -> ChatFileSearchState:
    if not state.get("query"):
        state["answer"] = None
        return state

    if not state["retrieved"]:
        context = "No relevant content was found in the uploaded files."
    else:
        context = "\n\n---\n\n".join(
            f"[{doc.metadata.get('file_name', 'unknown')}] {doc.page_content}"
            for doc in state["retrieved"]
        )

    prompt = (
        "Answer the question using only the context below, which comes from "
        "files the user uploaded to this chat. If the answer isn't in the "
        "context, say you don't know - don't make anything up.\n\n"
        f"Context:\n{context}\n\n"
        f"Question: {state['query']}"
    )
    response = llm.invoke(prompt)
    state["answer"] = response.content
    return state

uv run python -m manualvectorization.main --session demo1 --files ./source/sample_loan_agreement.pdf --interactive
Ingested 1 file(s) into session 'demo1' (in-memory, this process only).

Interactive mode - type a question, or 'quit' to exit.

> What's the overdraft protection fee?
A: A $10.00 overdraft protection transfer fee applies per occurrence...
Sources:
  - sample_bank_statement.pdf (chunk 7): If a transaction is presented that exceeds...

> What's the monthly payment on the loan?
A: The monthly payment is $454.37...
Sources:
  - sample_loan_agreement.pdf (chunk 2): Monthly Payment | $454.37 | Origination Fee...

通过连接节点确保在会话的嵌入向量生成之前无法执行检索操作，同时让会话结束时能清除相关向量。

执行流程

将示例文档传递给图结构，提出自然语言问题，并确认最相关的文本片段符合预期。重新启动该流程后，验证之前的会话向量是否已消失——这正是临时存储设计的意义所在。

总结：为何此方案无法扩展，以及永久存储能带来什么

临时搜索非常适合用于工作坊、个人草稿板以及那些不能保留嵌入向量的隐私敏感场景。当多名用户共享大量数据集、冷启动延迟至关重要，或需要混合过滤器与多副本搜索时，它无法替代托管索引。出现这些需求时应升级到 Redis 或 Azure AI Search——保持 LangGraph 节点结构不变，这样转换只需修改 get_vector_store() 函数即可。

教学价值：临时模式能明确界定会话边界，降低数据意外保留的风险。实际应用价值：可将其作为图结构的参考实现，之后再添加持久化存储，而无需重写提示词或分块逻辑。需监测并发会话下的内存增长情况——即便数据未写入磁盘，临时模式也并非真正“免费”的。

教学价值：临时模式能够明确界定会话边界，降低数据意外保留的概率。生产环境应用价值：可将其作为图的参考实现，之后再接入持久化存储，而无需重写提示逻辑或分块逻辑。需监测并发会话下的内存增长情况——即便数据没有写入磁盘，临时模式也并非真正“免费”的。

教学价值：临时模式能够明确界定会话边界，降低数据意外保留的概率。生产环境应用价值：可将其作为图的参考实现，之后再接入持久化存储，而无需重写相关提示或分块逻辑。需监测并发会话下的内存增长情况——即便数据没有写入磁盘，临时模式也并非真正“免费”的。

教学价值：临时模式能够明确界定会话边界，降低数据意外保留的概率。生产环境应用价值：可将其作为图的参考实现，之后再接入持久化存储，而无需重写相关提示或分块逻辑。需检测并发会话下的内存增长情况——即便数据没有写入磁盘，临时模式也并非真正“免费”的。

教学价值：临时模式能够明确界定会话边界，降低数据意外保留的风险。实际应用价值：可将其作为图的参考实现，之后再接入持久化存储，而无需重写提示词处理或分块逻辑。需测试并发会话下的内存增长情况——即便数据未写入磁盘，临时模式也并非真正“免费”。

← 全部文章发送简报