langgraph azure embeddings rag ephemeral sessions

Тымчасовая семантычная паразыскаўня з LangGraph, калі не трэба захоўваць вектары

Эмбедынгі, які існуюць толькі на час сесіі і знікаюць пасля ўсунення сесіі, ствараюць чыстыя межы — і дозволяюць пазнейшую замену на Redis або Azure AI Search выконваць як зміну адаптэра хранення.

Выдавец: REACTAPP.TOPДададзена на сайт 26 вер 20268 хв чытання1939 слоў

Крыніцы: Semantic Search : A Simple, Ephemeral Approach with LangGraph + Azure

Што паводлі гэтага будавання, і чаму такы падход ёсць трымкі

Інодзе прыдатным сховішчам вектараў для сесіі ёст вовсе без трымкага сховішча: уключыць, адшукаць, даўаць адказ, знікнуць. У гэтым кераванні ствараецца просты шлях семантычнай адшукаціі з LangGraph на Azure embeddings, пры чым стан залічваецца трымкім, ўбачліва каб дэманстрацыі і короткакалендарныя сесіі не залішалі без прыгляду індэксаў.

Архітэктура

Вузлы LangGraph адпавядаюць за прыём дадзеных (необов’язкова), уключэнне, выкарыстоўванне і генераванне. Вектары знаходзяцца ў памяці процэса або у тымчасовай структуры, ключаванай па сесіі. Azure OpenAI забезпечвае embeddings і чат-функцыі. Для гэтага варыанта не патрэбны дзейнастыя індэксы Redis.

Настройка

Забезпечыце Azure OpenAI (розгортанні embeddings + чат), 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...

Сконектавайце вузлы так, каб запуск функцыі retrieve не могла выконвацца да тэпер, калі для сесіі ўжо не існуеем эмбедынгі, і каб практыка teardown адчышала вектары пасля завершэння сесіі.

Выконанне

Пройдзіце прыкладны дакумент праз граф, задайце запіт на адмовайшы мове і пераканаўцеся, што найболей значымы фрагменты паспакоўваюць аспектам. Пачніце працэс зноў і пераканаўцеся, што вектары пярэйшай сесіі зніклі — гэта і ёсць мета тымчасовага дызайну.

Заключныя заўважэнні: чаму гэта не магчыма расшырваць, і што даўа стойкія база дадзеных

Тымчасовы пошук ідэальна для рабочых сесій, персональных зоў запісу даных і сесій, якія выказваюць чутлівасць да прыватнасці, калі не паўтараецца зберагачэнне векторных дадзеных. Яна не заменяе кераваны індэкс, калі багато корыстувачаў дзеліцца вялікім корпусам дадзеных, калі мае значэнне затрымка пачатку працы, або калі трэбуюць гібрыдныя фільтры і пошук з калькамі. Калі з’яўляюцца такія патрэбы, перайдзіце на Redis або Azure AI Search — захаваўце межы вузлаў LangGraph, каб замена была пераважна толькі змянай функцыі get_vector_store().

Практычная цэннасць: тымчасовы рэжым выклікае чыстыя межы сесій і ускладнюе віпадковае зберагачэнне даных. Продакшн-цэннасць: выкорыстоўваеце яго як прыкладную рэалізацыю для графа, а пасля прыўяжыце стойкі зберагач, не перапісваючы логіку формулювання запітоў чы розбівкі дадзеных на часткі. Замерайце рост памяці пад адночаснымі сесіямі — тымчасовы рэжым не ўсё ж такі «бесплатны», толькі таму, што нічога не запісваецца на дыск.

Навучальная цэнна: рэжым тымчасовых дадзейбных прытвараў выклікае чыстыя межы сэсій і ускладняе случайнае зберагачанне дадзэйна. Практычная цэнна: выкорыстоўваеце яго як адну з варыянтаў рэалізацыі для графа, а пасля прыўяжыце стойкі хранальнік, не перапісваючы логіку формуляраў чы разбівання дадзэйна на часткі. Замерайце адбыванне памяці пад адночаснымі сэсіямі — рэжым тымчасовых дадзейбных прытвараў не ёсць “бесплатным”, толькі таму, што нічога не запісваецца на дыск.

Навучальная цэнна: рэжым тымчасовых дадзейбных прытвараў выклікае чыстыя межы сэсій і ускладнюе нехтаўны захаванне дадзэйна. Практычная цэнна: выкорыстоўваеце яго як аднаго з прыкладаў рэалізацыі для графа, а пасля прыўяжыце стойкі хранальнік, не перапісваючы логіку формуляраў чы разбівання дадзэйна на часткі. Замерайце адбыванне памяці пад адночаснымі сэсіямі — рэжым тымчасовых дадзейбных прытвараў не ёсць “бесплатным”, толькі таму, што нічога не запісваецца на дыск.

Навучальная цэнна: рэжым тымчасовых дадзейбных прытвараў выклікае чыстыя межы сэсій і ускладнюе случайнае зберагачча дадзэйна. Практычная цэнна: выкорыстоўваеце яго як аднаго з прыкладаў рэалізацыі для графа, а пасля прыўяжыце стойкі хранальнік, не перапісваючы логіку формуляраў чы разбівання дадзэйна на часткі. Замерайце адбыванне памяці пад адночаснымі сэсіямі — рэжым тымчасовых дадзейбных прытвараў не ёсць “бесплатным”, толькі таму, што нічога не запісваецца на дыск.

Навучальная цэнна: рэжым тымчасовых дадзейбных прытвараў выклікае чыстыя межы сэсій і ускладнюе случайнае зберагачанне дадзейнаў. Практычная цэнна: выкорыстоўваеце яго як аднаго з прыкладаў рэалізацыі для графа, а пасля прыўяжыце стойкі хранальнік, не перапісваючы логіку формуляраў чы разбівання дадзейнаў на часткі. Замерайце адбыванне памяці пад адночаснымі сэсіямі — рэжым тымчасовых дадзейбных прытвараў не ёсць „бесплатным“, толькі таму, што нічога не запісваецца на дыск.

Ценнае для выкладчыцтва: рэжым тымчасовых дадзейнаў выклікае чыстыя межы сэсій і ускладняе случайнае зберагачыце дадзейнаў. Ценнае для працы ў рэальных умовах: выкорыстоўваеце яго як аднаго з прыкладаў рэалізацыі для графа, а пасля прыўяжыце стойкі база дадзейнаў, не перапісваючы код для стварэння частак дадзейнаў або логіку ўпорадкавання. Адзначайце зростанне памяці пад адночаснымі сэсіямі — рэжым тымчасовых дадзейнаў не ёсць „бесплатным“, толькі таму, што нічога не запісваецца на дыск.

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