langgraph azure embeddings rag ephemeral sessions

Tymczasowe wyszukiwanie semantyczne z LangGraph, gdy nie należy przechowywać wektorów

Embeddingi skupione na sesji, które znikają po jej zakończeniu, ustanawiają wyraźne granice — i sprawiają, że późniejsza zmiana na Redis lub Azure AI Search stanowi jedynie modyfikację adaptera magazynu.

Wydawca: REACTAPP.TOPDodane na tej stronie 26 wrz 20268 min czytania1939 słów

Źródła: Semantic Search : A Simple, Ephemeral Approach with LangGraph + Azure

Co obejmuje ta konfiguracja i dlaczego podejście to jest tymczasowe

Czasami odpowiedni magazyn wektorów dla sesji to w ogóle żaden trwały magazyn: wgranie danych, wyszukiwanie, udzielenie odpowiedzi, usunięcie. W tym przewodniku tworzymy prostą ścieżkę wyszukiwania semantycznego za pomocą LangGraph z embeddingami z Azure, zachowując stan tymczasowy, aby demonstracje i krótkotrwałe sesje nie pozostawiały bez opieki indeksów.

Architektura

Węzły LangGraph obsługują procesy takie jak pobieranie danych (opcjonalne), wgranie, odzyskiwanie oraz generowanie. Wektory przechowywane są w pamięci procesu lub w tymczasowej strukturze opartej na identyfikatorze sesji. Azure OpenAI dostarcza embeddingi oraz funkcje do rozmów. Dla tej wersji nie jest wymagany długotrwały indeks Redis.

Ustawienia

Zainstaluj Azure OpenAI (rozwiązania do embeddingów i rozmów), Python 3.12+ oraz zależności projektu. Sprawdź nazwy rozwiązań przed pisaniem kodu klienta.

Kod

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...

Połącz węzły tak, aby funkcja retrieve nie mogła zostać uruchomiona przed utworzeniem embeddingów dla danej sesji, a także tak, aby funkcja teardown usuwała wektory po zakończeniu sesji.

Eksploatacja

Prześlij przykładowy dokument przez graf, zadań pytanie w języku naturalnym i upewnij się, że najważniejsze fragmenty odpowiadają oczekiwaniom. Uruchom proces ponownie i sprawdź, czy wektory z poprzedniej sesji zniknęły – to właśnie ma na celu projekt tymczasowy.

Komentarze końcowe: dlaczego to nie jest skalowalne i co daje przechowywanie trwałe

Szukanie efemeryczne jest doskonałe do warsztatów, notatników indywidualnych oraz sesji wymagających ochrony prywatności, w których nie wolno przechowywać reprezentacji wektorowych. Nie zastępuje ono zarządzanego indeksu, gdy wielu użytkowników dzieli się dużym korpusem danych, gdy istotna jest opóźniona rozpoczęcie działania systemu, lub gdy potrzebne są hybrydowe filtry oraz wyszukiwanie z wieloma replikami. Przełącz się na Redis lub Azure AI Search, gdy pojawią się takie potrzeby – zachowaj granice węzłów LangGraph, aby zmiana polegała głównie na użyciu funkcji get_vector_store().

Wartość edukacyjna: tryb efemeryczny wymusza wyraźne granice sesji i utrudnia przypadkowe przechowywanie danych. Wartość produkcyjna: użyj go jako implementacji referencyjnej dla grafu, a następnie dodaj trwałe miejsce przechowywania bez konieczności przepisywania instrukcji lub logiki dzielenia danych na fragmenty. Pomiierz wzrost zużycia pamięci podczas jednoczesnych sesji – tryb efemeryczny nie jest „bezpłatny” tylko dlatego, że żadne dane nie trafiają na dysk.

Wartość edukacyjna: tryb tymczasowy wymusza wyraźne granice sesji i utrudnia przypadkowe przechowywanie danych. Zastosowanie w produkcji: użyj go jako implementacji referencyjnej dla grafu, a następnie dołącz trwałe miejsce przechowywania bez konieczności przepisywania zapytań lub logiki dzielenia na fragmenty. Pomierz wzrost pamięci podczas jednoczesnych sesji — tryb tymczasowy nie jest „bezpłatny” tylko dlatego, że nic nie trafia na dysk.

Wartość edukacyjna: tryb efemeryczny wymusza wyraźne granice sesji i utrudnia przypadkowe przechowywanie danych. Wartość produkcyjna: użyj go jako implementacji referencyjnej dla grafu, a następnie dołącz trwałe miejsce przechowywania bez przepisywania instrukcji lub logiki dzielenia na fragmenty. Pomiierz wzrost pamięci w warunkach jednoczesnych sesji — tryb efemeryczny nie jest „bezpłatny” tylko dlatego, że nic nie trafia na dysk.

← Wszystkie artykuły Wyślij brief