langgraph azure embeddings rag ephemeral sessions

Ephemeral semantic search with LangGraph when you should not keep vectors

Session-scoped embeddings that disappear on teardown teach clear boundaries—and make the later swap to Redis or Azure AI Search a store-adapter change.

Publisher: REACTAPP.TOPAdded to this site Sep 26, 20268 min read1939 words

This article is published in English.

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

What this build covers, and why the approach is ephemeral

Sometimes the right vector store for a session is no durable store at all: embed, search, answer, discard. This walkthrough builds a simple semantic search path with LangGraph on Azure embeddings, keeping state ephemeral so demos and short-lived sessions do not leave orphaned indexes behind.

Architecture

LangGraph nodes handle ingest (optional), embed, retrieve, and generate. Vectors live in process memory or a throwaway structure keyed by session. Azure OpenAI supplies embeddings and chat. No long-lived Redis index is required for this variant.

Setup

Provision Azure OpenAI (embeddings + chat deployments), Python 3.12+, and project dependencies. Confirm deployment names before coding clients.

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

Wire nodes so retrieve cannot run before embeddings exist for the session, and so teardown clears vectors when the session ends.

Execution

Run a sample document through the graph, ask a natural-language question, and confirm top chunks match expectations. Restart the process and verify prior session vectors are gone—that is the point of the ephemeral design.

Closing notes: why this doesn’t scale, and what a permanent store adds

Ephemeral search is excellent for workshops, per-user scratchpads, and privacy-sensitive sessions that must not retain embeddings. It does not replace a managed index when many users share a large corpus, when cold-start latency matters, or when hybrid filters and multi-replica search are required. Promote to Redis or Azure AI Search when those needs appear—keep the LangGraph node boundaries so the swap is mostly a get_vector_store() change.

Teaching value: ephemeral mode forces clear session boundaries and makes accidental data retention harder. Production value: use it as a reference implementation for the graph, then attach a durable store without rewriting prompts or chunking logic. Measure memory growth under concurrent sessions—ephemeral is not “free” just because nothing hits disk.

← All articles Send the brief