langgraph azure embeddings rag ephemeral sessions

Búsqueda semántica efímera con LangGraph cuando no se deben conservar vectores

Los embeddings de ámbito de sesión que desaparecen al finalizar la sesión establecen límites claros, y permiten que el posterior cambio a Redis o Azure AI Search sea simplemente una modificación en el adaptador de almacenamiento.

Editor: REACTAPP.TOPAñadido a este sitio 26 sept 20268 min de lectura1939 palabras

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

Qué abarca esta implementación y por qué el enfoque es efímero

A veces, el almacén de vectores adecuado para una sesión es ningún almacén duradero en absoluto: incrustar, buscar, responder y descartar. Esta guía crea un camino de búsqueda semántica simple con LangGraph en embeddings de Azure, manteniendo el estado efímero para que las demostraciones y sesiones de corta duración no dejen índices huérfanos.

Arquitectura

Los nodos de LangGraph se encargan de la ingesta (opcional), incrustación, recuperación y generación. Los vectores se almacenan en la memoria del proceso o en una estructura temporal indexada por sesión. Azure OpenAI proporciona los embeddings y la funcionalidad de chat. No se requiere un índice Redis de larga duración para esta variante.

Configuración

Provea Azure OpenAI (despliegues de embeddings + chat), Python 3.12+ y las dependencias del proyecto. Confirme los nombres de los despliegues antes de programar los clientes.

El código

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

Conecte los nodos de forma que la recuperación no pueda ejecutarse antes de que existan los embeddings para la sesión, y así el proceso de desmontaje elimine los vectores al finalizar la sesión.

Ejecución

Ejecute un documento de ejemplo a través del grafo, haga una pregunta en lenguaje natural y confirme que los fragmentos principales coincidan con las expectativas. Reinicie el proceso y verifique que los vectores de la sesión anterior hayan desaparecido; ese es el propósito del diseño efímero.

Notas finales: por qué esto no escala y qué aporta un almacenamiento permanente

La búsqueda efímera es excelente para talleres, blocs de notas por usuario y sesiones sensibles a la privacidad que no deben conservar embeddings. No sustituye a un índice gestionado cuando muchos usuarios comparten un corpus grande, cuando es importante la latencia de inicio en frío, o cuando se requieren filtros híbridos y búsqueda con múltiples réplicas. Pásela a Redis o Azure AI Search cuando surjan esas necesidades; mantén los límites de los nodos de LangGraph para que el cambio sea principalmente una modificación en get_vector_store().

Valor educativo: el modo efímero impone límites claros entre sesiones y dificulta la conservación accidental de datos. Valor en entornos de producción: úsalo como implementación de referencia para el grafo y luego añade un almacenamiento duradero sin tener que reescribir las instrucciones ni la lógica de particionamiento. Mide el crecimiento de la memoria en sesiones concurrentes; la búsqueda efímera no es “gratuita” solo porque nada se guarda en disco.

Valor educativo: el modo efímero impone límites claros entre sesiones y dificulta la retención accidental de datos. Valor en entornos de producción: úsalo como implementación de referencia para el grafo, y luego añade un almacenamiento duradero sin tener que reescribir las instrucciones ni la lógica de particionamiento. Mide el crecimiento de la memoria en sesiones concurrentes; el modo efímero no es “gratuito” solo porque nada se guarda en disco.

Valor educativo: el modo efímero impone límites claros entre sesiones y dificulta la retención accidental de datos. Valor en entornos de producción: úsalo como implementación de referencia para el grafo, y luego añade un almacenamiento duradero sin tener que reescribir las instrucciones ni la lógica de fragmentación. Mide el crecimiento de la memoria en sesiones concurrentes; el modo efímero no es “gratuito” solo porque nada se guarda en disco.

Valor educativo: el modo efímero impone límites claros entre sesiones y dificulta la retención accidental de datos. Valor en entornos de producción: úsalo como implementación de referencia para el grafo, y luego añade un almacenamiento duradero sin tener que reescribir los comandos ni la lógica de particionamiento. Mide el crecimiento de la memoria en sesiones concurrentes; el modo efímero no es “gratuito” solo porque nada se guarda en disco.

← Todos los artículos Enviar el brief