Par: Raj Tulluri
Choisir Lakebase plutôt que Vector Search pour le téléversement de documents par clavardage
Le problème
L’une des exigences les plus courantes pour un agent conversationnel (chatbot) est la réponse aux questions sur documents: téléverser un fichier, poser des questions à son sujet et recevoir des réponses ancrées dans ce fichier. C’est le même modèle popularisé par ChatGPT et Claude, et c’est devenu une attente de base pour toute interface de clavardage.
Ce type de réponse aux questions sur documents est presque toujours limité à une seule session. Un contrat est téléversé, trois questions sont posées sur une clause de résiliation, puis la conversation se termine. Le document n’a aucune vie au-delà de cette session. Personne ne s’attend à ce qu’il demeure interrogeable par la suite, et personne ne veut qu’il ressurgisse dans le clavardage de quelqu’un d’autre.
La réponse standard de Databricks à une exigence de RAG (génération augmentée par récupération) est Vector Search: analyser les documents, les découper en segments (chunks), les vectoriser (embedding), synchroniser le résultat dans un index géré, puis l’interroger. Ce modèle convient à une base de connaissances partagée et durable, destinée à exister pendant des mois ou des années. Il convient mal lorsque le contenu à indexer n’existe que pour la durée d’une seule conversation. Provisionner un index Vector Search par session de clavardage ne correspond pas à la façon dont ce service est conçu pour fonctionner, et ne correspond pas non plus au besoin réel : quelque chose qui peut être créé en quelques secondes, interrogé immédiatement, et laissé inactif à un coût presque nul.
Ce décalage est le sujet de cet article. L’objectif était un pipeline documentaire confiné à une session, rapide à ingérer, rapide à interroger, isolé par session par construction, et peu coûteux lorsqu’inactif. L’approche testée ici utilise Lakebase, la base Postgres gérée et à mise à l’échelle automatique de Databricks, comme magasin vectoriel par session plutôt que d’acheminer le téléversement de documents par Vector Search.
L’architecture de départ
Avant d’aborder le pipeline documentaire, il vaut la peine de définir les deux éléments d’infrastructure sur lesquels il repose: l’agent lui-même, et Lakebase.
L’agent en tant qu’application Databricks
Une application Databricks (Databricks App) exécute du code applicatif personnalisé directement sur le calcul Databricks, exposé via l’authentification et le réseau propres à la plateforme. Il s’agit d’une voie de déploiement différente de Model Serving, qui enregistre un modèle et le sert derrière une interface de prédiction fixe. Une App est un véritable processus de longue durée : elle peut maintenir des connexions ouvertes, exécuter des tâches en arrière-plan et desservir plus d’une route. Le fait que l’agent soit une application Databricks plutôt qu’un point de terminaison Model Serving le rend avec état (stateful), ce qui facilite grandement l’ajout d’une fonctionnalité multi-route comme le téléversement de documents par session.
@invoke()
async def invoke_handler(request: ResponsesAgentRequest) -> ResponsesAgentResponse:
# exécuter votre agent ici et retourner la réponse complète
...
@stream()
async def stream_handler(request: ResponsesAgentRequest) -> AsyncGenerator[ResponsesAgentStreamEvent, None]:
# exécuter votre agent ici et produire les événements au fur et à mesure
...
agent_server = AgentServer("ResponsesAgent", enable_chat_proxy=True)
app = agent_server.app
Dans notre exemple de RAG jetable, l’agent lui-même est une application LangGraph, enveloppée par le cadre agent_server de MLflow. Plutôt que de sous-classer directement l’interface ResponsesAgent de MLflow, agent_server l’expose via deux décorateurs, @invoke et @stream, appliqués à de simples fonctions asynchrones. La figure 1 montre le trajet résultant d’une requête : une requête client arrive sur le calcul de l’application, est captée par le MLflow Agent Server, et atteint le code de l’agent via l’interface ResponsesAgent. À partir de là, l’agent fait appel à un point de terminaison de service de modèle pour l’inférence, ainsi qu’à toute autre ressource dont il a besoin, traitées de manière uniforme comme des outils : espaces Genie, expériences MLflow, actifs Unity Catalog, serveurs MCP, et ainsi de suite.
C’est la couche dans laquelle s’intègre le pipeline documentaire décrit dans cet article. Le gestionnaire de téléversement et l’outil de récupération s’exécutent tous deux sur le même calcul que l’agent conversationnel, plutôt que comme des services distincts.
Lakebase pour la mémoire
Lakebase est l’offre Postgres gérée et à mise à l’échelle automatique de Databricks. Elle se comporte comme une instance Postgres ordinaire du point de vue des connexions, tandis que le calcul sous-jacent se suspend en cas d’inactivité et reprend à la demande.
Avant même que le RAG documentaire limité à une session n’existe comme fonctionnalité, Lakebase abritait déjà la mémoire conversationnelle propre de l’agent. Un point de contrôle (checkpointer) LangGraph stocke l’historique à court terme d’un fil de discussion, de sorte qu’une conversation puisse reprendre en cours de route. Un magasin (store) LangGraph conserve des faits à long terme sur un utilisateur qui persistent d’une conversation distincte à l’autre. Les deux fonctionnent sur la même instance Postgres à mise à l’échelle automatique, via le même regroupement de connexions (connection pooling) et la même rotation des identifiants.
from databricks_langchain import AsyncCheckpointSaver, AsyncDatabricksStore
lakebase_kwargs = {
"project": "my-lakebase-project",
"branch": "production",
"schema": "agent_memory",
}
# Mémoire à court terme : historique de conversation, indexé par id de fil
checkpointer = AsyncCheckpointSaver(**lakebase_kwargs)
await checkpointer.__aenter__()
await checkpointer.setup()
# Mémoire à long terme : faits durables sur un utilisateur, avec recherche sémantique
store = AsyncDatabricksStore(
**lakebase_kwargs,
embedding_endpoint=”databricks-gte-large-en",
embedding_dims=1024
)
await store.__aenter__()
await store.setup()
La configuration du RAG
La figure 2 montre le trajet complet emprunté. Un message de clavardage et un téléversement de fichier constituent deux points d’entrée distincts depuis le client, l’un aboutissant sur l’agent lui-même, l’autre sur un gestionnaire de téléversement dédié, mais les deux voies finissent par lire et écrire dans la même instance Lakebase.
La sélection d’un fichier démarre le téléversement immédiatement, indépendamment de ce qui est en train d’être tapé dans la boîte de message à ce moment-là.
Ingestion
Le pipeline suit une forme familière une fois qu’un fichier est sélectionné :
- Le fichier est analysé et découpé en segments. La taille des segments et leur chevauchement dépendent du cas d’usage — un contrat juridique et une transcription de soutien n’ont pas besoin de la même stratégie de découpage — cette étape est donc laissée flexible plutôt que fixée à une seule méthode.
- Chaque segment est vectorisé à l’aide d’un modèle de vectorisation hébergé par Databricks.
- Les vecteurs sont écrits dans Lakebase, dans une table Postgres avec l’extension pgvector activée, aux côtés du texte du segment et de l’id de la session de clavardage qui l’a produit.
La table elle-même est une table Postgres ordinaire, pgvector fournissant le type de colonne vectorielle:
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS session_documents (
id BIGSERIAL PRIMARY KEY,
session_id TEXT NOT NULL,
filename TEXT NOT NULL,
chunk_text TEXT NOT NULL,
embedding VECTOR(1024) NOT NULL,
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS session_documents_session_id_idx
ON session_documents (session_id);
Et l’étape d’ingestion elle-même tient en une courte fonction: découper, vectoriser, insérer.
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
embeddings = DatabricksEmbeddings(endpoint="databricks-gte-large-en")
async def ingest_document(session_id: str, filename: str, text: str, conn):
chunks = splitter.split_text(text)
vectors = await embeddings.aembed_documents(chunks)
async with conn.cursor() as cur:
for chunk, vector in zip(chunks, vectors):
await cur.execute(
"""
INSERT INTO session_documents (session_id, filename, chunk_text, embedding)
VALUES (%s, %s, %s, %s)
""",
(session_id, filename, chunk, vector),
)
La colonne id de session constitue à elle seule tout le mécanisme d’isolation. Rien d’autre ne distingue les documents d’un clavardage de ceux d’un autre, et aucune étape de configuration par session n’est requise avant qu’un document devienne interrogeable.
Récupération
La récupération est un simple outil LangChain que l’agent peut appeler de sa propre initiative:
@tool
async def search_uploaded_documents(query: str, session_id: str, conn) -> str:
"""Rechercher dans les documents téléversés plus tôt dans cette session de clavardage."""
[query_vector] = await embeddings.aembed_documents([query])
async with conn.cursor() as cur:
await cur.execute(
"""
SELECT chunk_text
FROM session_documents
WHERE session_id = %s
ORDER BY embedding <=> %s
LIMIT 5
""",
(session_id, query_vector),
)
rows = await cur.fetchall()
if not rows:
return "Aucun contenu correspondant trouvé pour cette session."
return "\n\n".join(row["chunk_text"] for row in rows)
L’opérateur <=> est la distance cosinus de pgvector, donc trier selon celui-ci en ordre croissant retourne d’abord les correspondances les plus proches. La clause WHERE session_id = %s effectue tout le travail d’isolation décrit ci-dessus, en SQL pur.
Le modèle décide lui-même du moment où appeler cet outil : si la question ressemble à quelque chose qu’un document téléversé pourrait répondre, l’outil s’exécute; si rien n’a été téléversé dans cette session, il ne retourne rien, et l’agent répond à partir de la conversation seule. Comme pgvector effectue la recherche directement sur cette table Postgres, un segment nouvellement inséré devient interrogeable dès qu’il est validé (committed), sans étape d’indexation ou de synchronisation distincte entre l’ingestion et la récupération.
Suppression
Une table limitée à une session ne reste pas petite d’elle-même. Sans stratégie de purge en place, la table session_documents croît sans limite, et à un certain point, une grande table en croissance continue sur du Postgres à mise à l’échelle automatique cesse d’être l’option rapide et économique visée par cette conception au départ. Décider comment et quand supprimer les anciennes données fait autant partie du pipeline que l’ingestion et la récupération. Deux considérations sont à retenir: l’une concernant le moment où un clavardage cesse d’être interrogeable, et l’autre concernant ce qu’il advient de ses données une fois que c’est le cas.
La première est une rétention basée sur la dernière activité plutôt que sur la date de téléversement. Une échéance fixe au calendrier purge un clavardage sept jours (ou toute autre période requise) après son premier téléversement, peu importe si quelqu’un y est retourné depuis. Une rétention basée sur le dernier accès réinitialise plutôt le compteur à chaque ouverture du clavardage, de sorte qu’un clavardage auquel quelqu’un revient constamment demeure interrogeable indéfiniment dans les faits, tandis qu’un clavardage abandonné après une seule question expire selon l’échéancier prévu.
La seconde consiste à déplacer les segments expirés vers Unity Catalog plutôt que de les supprimer. Lakebase est le bon endroit pour des données devant être interrogées en direct, mais pas pour une archive permanente qui s’accumule. Une table Delta dans Unity Catalog, appuyée sur un simple stockage d’objets, constitue un endroit beaucoup plus économique pour ces mêmes données une fois que personne ne les interroge activement, et elle peut aisément croître à une taille qui rendrait une table Lakebase de plus en plus coûteuse à conserver.
CREATE TABLE IF NOT EXISTS main.rag_archive.session_documents (
session_id STRING,
filename STRING,
chunk_text STRING,
embedding ARRAY<FLOAT>,
archived_at TIMESTAMP
) USING DELTA;
Réactiver une session est une recherche ponctuelle et légère plutôt qu’une réexécution du pipeline d’ingestion : les segments sont déjà analysés et vectorisés, donc les restaurer consiste simplement à les copier depuis Delta vers Lakebase, avec le vecteur reconverti au type vector de pgvector au passage.
def rehydrate_session(session_id: str, pg_conn, uc_connection):
rows = uc_connection.execute(
"""
SELECT filename, chunk_text, embedding
FROM main.rag_archive.session_documents
WHERE session_id = ?
""",
(session_id,),
).fetchall()
with pg_conn.cursor() as cur:
for filename, chunk_text, embedding in rows:
cur.execute(
"""
INSERT INTO session_documents (session_id, filename, chunk_text, embedding)
VALUES (%s, %s, %s, %s::vector)
""",
(session_id, filename, chunk_text, embedding),
)
Lakebase c. Databricks Vector Search
C’est la décision de conception centrale derrière ce projet, elle mérite donc d’être abordée directement : pourquoi ne pas utiliser Vector Search pour ceci également?
Vector Search est un excellent produit pour un corpus vaste, durable et partagé, interrogé par de nombreux utilisateurs sur une longue période. Il fournit une indexation gérée des plus proches voisins approximatifs (ANN) et s’adapte bien au-delà de ce qu’une table Postgres avec pgvector peut raisonnablement gérer. La question pertinente ici est plus étroite : cette forme correspond-elle à un document qui n’existe que pour la durée d’une seule conversation?
Deux propriétés de Vector Search en font un mauvais choix dans ce cas.
- Un point de terminaison Vector Search fonctionne comme une infrastructure de service toujours active et est facturé en continu, que la conversation soit active ou non. Pour un document dont l’utilité se mesure en minutes, c’est un mauvais compromis.
- L’unité naturelle dans Vector Search est l’index, et créer un index par session de clavardage ne correspond pas à ce modèle. Construire un index et y synchroniser des données prend un temps réel, typiquement des minutes plutôt que des secondes — le mauvais budget de latence pour un document qui doit être interrogeable immédiatement après le téléversement.
Lakebase convient à ce type de problème pour les raisons inverses. Il fonctionne comme du Postgres à mise à l’échelle automatique, ce qui lui permet de se suspendre complètement en cas d’inactivité et de reprendre à la connexion suivante, ce qui signifie que le coût suit l’utilisation réelle plutôt qu’une capacité permanente calculée pour le pire cas. Partitionner par session ne nécessite rien de plus qu’une colonne id de session : aucune étape de provisionnement par session n’existe, et les données d’une session ne sont séparées de celles de toutes les autres que par une seule clause WHERE. Avec pgvector fonctionnant directement sur cette table, une rangée devient interrogeable dès qu’elle est validée (committed).
Rien de tout cela ne rend Lakebase supérieur à Vector Search en tant que produit général. Il convient mieux à cette tâche particulière. Une table pgvector sur Postgres ne peut pas rivaliser avec une indexation ANN conçue sur mesure à très grande échelle. Vector Search demeure le bon choix pour un corpus vaste et permanent. Lakebase convient à un corpus petit, jetable, et qui doit sembler instantané.