langgraph azure embeddings rag ephemeral sessions

Ephemerer semantischer Suchalgorithmus mit LangGraph, wenn Vektoren nicht aufbewahrt werden sollten

Sitzungsspezifische Embeddings, die bei Beendigung der Sitzung verschwinden, schaffen klare Grenzen – und machen den späteren Wechsel zu Redis oder Azure AI Search zu einer Änderung des Store-Adapters.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 26. Sept. 20268 Min. Lesezeit1939 Wörter

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

Was diese Implementierung umfasst und warum der Ansatz vorübergehend ist

Mannchmal ist der richtige Vektorlagers für eine Sitzung kein dauerhafter Lager überhaupt: Einbetten, Suchen, Beantworten, Wegwerfen. In dieser Anleitung wird ein einfacher semantischer Suchweg mit LangGraph auf Azure Embeddings erstellt, wobei der Zustand vorübergehend gehalten wird, damit Demonstrationen und kurzlebige Sitzungen keine verwaisten Indizes hinterlassen.

Architektur

Die LangGraph-Node kümmern sich um das Einlesen (optional), Einbetten, Abrufen und Erstellen. Die Vektoren befinden sich im Prozessspeicher oder in einer temporären Struktur, die nach Sitzung gekennzeichnet ist. Azure OpenAI stellt die Embeddings sowie das Chat-Funktionen bereit. Für diese Variante ist kein langlebiger Redis-Index erforderlich.

Einrichtung

Stellen Sie Azure OpenAI (Embeddings + Chat-Implementierungen), Python 3.12+ sowie die Projektabhängigkeiten bereit. Überprüfen Sie vor dem Programmieren der Clients die Namen der Implementierungen.

Der Code

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

Vernetzen Sie die Knoten so, dass das Abrufen nicht vor dem Vorhandensein von Embeddings für die Sitzung ausgeführt werden kann, und sodass bei Beendigung der Sitzung die Vektoren gelöscht werden.

Ausführung

Lassen Sie ein Beispieldokument durch das Graphen laufen, stellen Sie eine Frage in natürlicher Sprache und überprüfen Sie, ob die wichtigsten Abschnitte den Erwartungen entsprechen. Starten Sie den Prozess erneut und stellen Sie sicher, dass die Vektoren der vorherigen Sitzung verschwunden sind – das ist der Sinn des vorübergehenden Designs.

Zusammenfassende Anmerkungen: Warum dies nicht skalierbar ist und was ein dauerhafter Speicher hinzufügt

Ephemerer Suchmodus eignet sich hervorragend für Workshops, persönliche Notizbücher pro Benutzer sowie datenschutzkritische Sitzungen, bei denen Embeddings nicht gespeichert werden dürfen. Er ersetzt jedoch keinen verwalteten Index, wenn viele Benutzer ein großes Korpus teilen, wenn Latenzzeiten bei der Initialisierung eine Rolle spielen oder hybride Filter sowie Suchfunktionen mit mehreren Replikaten erforderlich sind. Wechseln Sie auf Redis oder Azure AI Search, sobald diese Anforderungen auftreten – halten Sie die Grenzen der LangGraph-Node bei, sodass der Wechsel hauptsächlich durch eine Änderung in get_vector_store() erfolgt.

Bildungsaspekt: Der ephemere Modus erzwingt klare Sitzungsgrenzen und erschwert das versehentliche Speichern von Daten. Produktionsaspekt: Nutzen Sie ihn als Referenzimplementierung für das Graphenmodell und fügen Sie anschließend einen dauerhaften Speicher hinzu, ohne die Prompt-Strukturen oder die Logik zur Aufteilung in Blöcke umschreiben zu müssen. Messen Sie das Speichwachstum bei gleichzeitigen Sitzungen – der ephemere Modus ist nicht „kostenlos“, nur weil keine Daten auf die Festplatte geschrieben werden.

Bildungswert: Der vorübergehende Modus erzwingt klare Sitzungsgrenzen und erschwert das versehentliche Verbleiben von Daten. Produktivnutzung: Nutzen Sie ihn als Referenzimplementierung für das Graphenmodell und fügen Sie anschließend einen dauerhaften Speicher hinzu, ohne die Anfragen oder die Logik zur Aufteilung in Blöcke umschreiben zu müssen. Messen Sie das Speichwachstum bei gleichzeitigen Sitzungen – der vorübergehende Modus ist nicht „kostenlos“, nur weil keine Daten auf die Festplatte geschrieben werden.

Bildungswert: Der vorübergehende Modus erzwingt klare Sitzungsgrenzen und erschwert das versehentliche Behalten von Daten. Produktivnutzung: Verwenden Sie ihn als Referenzimplementierung für das Graphenmodell und fügen Sie anschließend einen dauerhaften Speicher hinzu, ohne die Anfragen oder die Logik zur Aufteilung in Blöcke umschreiben zu müssen. Messen Sie das Speichwachstum bei gleichzeitigen Sitzungen – der vorübergehende Modus ist nicht „kostenlos“, nur weil keine Daten auf die Festplatte geschrieben werden.

Bildungswert: Der ephemere Modus erzwingt klare Sitzungsgrenzen und erschwert das versehentliche Speichern von Daten. Produktivitätswert: Verwenden Sie ihn als Referenzimplementierung für das Graphen-System, fügen Sie anschließend einen dauerhaften Speicher hinzu, ohne die Anfragen oder die Logik zur Aufteilung in Blöcke umschreiben zu müssen. Messen Sie das Speichwachstum bei gleichzeitigen Sitzungen – ephemere Daten sind nicht automatisch „kostenlos“, nur weil nichts auf die Festplatte geschrieben wird.

← Alle Artikel Brief senden