langgraph azure embeddings rag ephemeral sessions

Тимчасовий семантичний пошук за допомогою LangGraph, коли не варто зберігати вектори

Ембеддинги, які існують лише протягом сесії та зникають після її закриття, встановлюють чіткі межі — і дозволяють подальшу заміну на Redis чи Azure AI Search здійснити шляхом зміни адаптера сховища.

Видавець: REACTAPP.TOPДодано на сайт 26 вер. 2026 р.8 хв читання1939 слів

Джерела: Semantic Search : A Simple, Ephemeral Approach with LangGraph + Azure

Що охоплює ця версія та чому підхід є тимчасовим

Іноді правильним рішенням для сеансу є зовсім без постійного сховища: вбудовування, пошук, відповідь, видалення. У цьому посібнику створюється простий шлях семантичного пошуку за допомогою LangGraph та ембеддингів від Azure, причому стан зберігається тимчасово, щоб демонстрації та короткочасні сеанси не залишали позаду непотрібні індекси.

Архітектура

Вузли LangGraph обробляють завантаження даних (необов’язково), створення ембеддингів, їх пошук та генерацію відповідей. Ембеддинги зберігаються у пам’яті процесу або у тимчасовій структурі, ключуваній за сеансом. Azure OpenAI забезпечує ембеддинги та функції чату. Для цього варіанту не потрібен довготривалий індекс 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().

Навчальна цінність: тимчасовий режим змушує чітко визначати межі сеансу та ускладнює випадкове збереження даних. Практична цінність: використовуйте його як базову реалізацію для графа, а потім додайте стабільний сховище, не переписуючи логіку обробки даних. Вимірюйте зростання пам’яті під час одночасних сеансів — тимчасовий режим не є „безкоштовним“ лише через те, що дані не зберігаються на диску.

Навчальна цінність: режим тимчасового зберігання змушує чітко визначати межі сеансів та ускладнює випадкове збереження даних. Практична цінність: використовуйте його як базову реалізацію для графа, а потім додайте надійний сховище, не переписуючи логіку обробки даних чи їх розділення на частини. Вимірюйте зростання пам’яті під час одночасних сеансів — режим тимчасового зберігання не є „безкоштовним“ лише через те, що дані не зберігаються на диску.

Навчальна цінність: ефемерний режим змушує чітко визначати межі сеансів та ускладнює випадкове збереження даних. Практична цінність: використовуйте його як базову реалізацію для графа, а потім додайте стабільний сховище, не переписуючи скрипти або логіку розділення даних на частини. Вимірюйте зростання пам’яті під час одночасних сеансів — ефемерність не означає „безкоштовність“, лише тому що дані не зберігаються на диску.

← Усі статті Надіслати бриф