Recherchez les guides d’opération, ne les collez pas : une petite approche RAG citée dans LangChain
Créez un chemin de récupération en deux étapes dans LangChain : chargez les runbooks, divisez-les en chunks, effectuez une recherche avec une requête ciblée, conservez les métadonnées, exigez des IDs de sources citées et testez-le sans utiliser d’embeddings payants.
Recherchez dans les guides de procédure plutôt que de remplir le champ de demande
À ce stade, l’application du service client se divise en fonction du type de ticket. Un ticket lié à une base de données rassemble d’un seul coup le catalogue des services, les incidents récents ainsi que les déploiements récents, tandis qu’un ticket d’accès suit une autre voie. Les étapes détaillées permettant de mettre en œuvre une recommandation manquent encore.
Vous pourriez coller tous les guides de procédure dans le champ de demande. Cela fonctionne pour trois fichiers courts, mais cela cesse de marcher à mesure que la collection s’agrandit. Les demandes volumineuses coûtent plus cher, fonctionnent plus lentement et rendent difficile l’identification du passage qui a servi de base à une réponse. Elles peuvent également fournir au modèle des documents auxquels l’utilisateur actuel n’a jamais eu accès. La solution, c’est la recherche.
Le processus mis en place comporte deux phases. Lors de l’indexation, les guides de procédure au format Markdown deviennent des objets Document contenant des métadonnées sources, qui sont ensuite divisés en plus petits fragments :
indexing time
Markdown runbooks
-> Document objects with source metadata
-> smaller chunks
-> embeddings
-> vector store
Lors de la demande, la classification du ticket génère une requête de recherche ciblée, qui est envoyée au système de récupération :
request time
ticket classification
-> focused search query
-> retriever
-> a few runbook passages with source IDs
-> recommendation chain
-> source-ID check
Les gens appellent souvent cela une génération améliorée par récupération en deux étapes, ou RAG à deux étapes. L’étiquette importe moins que la séquence : la recherche a lieu en premier, et la génération ne reçoit que les extraits qui en résultent. Le modèle ne décide jamais s’une recherche doit avoir lieu, et il ne peut pas choisir arbitrairement une source de données. Aucun agent n’est nécessaire.
Le ticket utilisé tout au long
L’exemple présenté est un ticket indiquant que les demandes de paiement en UE échouent après le déploiement à 14:05, et que les journaux de checkout-api affichent "database connection refused." Sa classification structurée se présente comme suit :
{
"ticket_id": "INC-2048",
"summary": "EU checkout fails after a deployment.",
"category": "database",
"severity": "SEV2",
"service": "checkout-api",
"confidence": 0.91,
"evidence": ["database connection refused"],
"missing_information": ["Affected request percentage"]
}
Une bonne recherche renvoie le manuel de fonctionnement de la base de données de paiement, et non un guide sur la latence des paiements ou un guide d’accès aux identités. Elle doit également conserver suffisamment de métadonnées pour répondre à quatre questions : de quel fichier provient le texte, quel service en est propriétaire, où dans le fichier commence le fragment, et quels IDs de source la recommandation cite. Un texte sans métadonnées de source est difficile à gérer.
Les six étapes du retraitement
Chaque étape nécessite une définition avant d’être utilisée.
Document. Considérez un Document comme un bloc de texte accompagné d’étiquettes. Le texte peut provenir d’un fichier Markdown, d’un PDF, d’une page web, d’un wiki ou même d’une ligne dans un tableau, et les étiquettes peuvent indiquer l’ID de source, le titre, le service, le chemin, le propriétaire, la version ou les personnes autorisées à le consulter.
Chunk. Les documents sont divisés en parties pouvant être consultées indépendamment. Un seul manuel de procédures peut décrire à la fois le diagnostic, la récupération et le retrait de modifications, mais une question concernant les identifiants de base de données n’a pas besoin de restituer à chaque fois le matériel nécessaire au retrait de modifications.
Embedding. Un modèle d’embedding transforme du texte en une liste de nombres, de sorte que les textes ayant un sens similaire se retrouvent proches l’un de l’autre selon la méthode de comparaison choisie. Il ne s’agit pas d’une vérification des faits. Des textes similaires peuvent tout de même être erronés, obsolètes ou interdits d’accès.
Stockage de vecteurs. Un stockage de vecteurs conserve des vecteurs de documents et permet de trouver ceux qui leur sont proches. Certains existent uniquement en mémoire, d’autres sont stockés dans une base de données ou un service hébergé. Le lieu où les vecteurs sont stockés ne remplace pas le contrôle d’accès.
Récupération. Si l’on donne à un système de récupération un texte de requête libre, il renvoie des objets Document. Il s’agit d’une approche plus large que la recherche vectorielle : derrière cette même interface, on peut intégrer une correspondance par mots-clés, un stockage vectoriel, un service de recherche hybride ou une API de recherche interne.
Génération. La dernière étape prend les extraits choisis et produit une réponse structurée. Quoi qu’elle écrive, elle ne peut pas prétendre que la recherche a retourné quelque chose qui n’existait pas.
La récupération n’est pas de la mémoire
Un ensemble de guides opérationnels constitue un corpus de connaissances, et interroger ce dernier relève de la récupération. L’historique des conversations est une séquence de messages, les données utilisateur enregistrées sont des données d’application ordinaires, et les points de contrôle des workflows enregistrent l’état d’exécution. Leurs règles de conservation et de permission diffèrent toutes, c’est pourquoi l’étape suivante les sépare dans le code.
Structure et configuration du projet
Le snapshot de cette étape a cette structure :
langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│ ├── checkout-database.md
│ ├── identity-access.md
│ └── payment-latency.md
└── tests/
└── test_retrieval.py
Installez les paquets :
python -m pip install -U \
langchain \
langchain-openai \
langchain-text-splitters \
pydantic \
python-dotenv \
pytest
L’application fait appel aux embeddings du fournisseur. Les tests substituent un petit remplaçant local, de sorte qu’aucun appel au fournisseur n’a lieu.
Étape 1 : rédiger des guides de fonctionnement identifiables
Le guide de fonctionnement pour le processus de paiement commence par une ligne de titre :
# Checkout database connection failures
Viennent ensuite le service, le propriétaire et la carrosserie :
Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
previous successful deployment.
2. Confirm that the application secret points to the expected
credential version.
3. Test network reachability from the application environment without
printing the credential.
4. Compare connection-pool limits with the database connection limit.
Le titre, le service et le propriétaire sont du texte brut, et le chargeur copie également les valeurs sélectionnées dans les métadonnées. Dans un système d’entreprise, il ne faut jamais laisser un en-tête définir l’accès. Les étiquettes d’accès doivent provenir d’un système fiable et être vérifiées à chaque exécution de la récupération.
Étape 2 : charger chaque fichier dans un Document
Insérez load_runbooks dans retrieval.py. Il a d’abord besoin de l’aide-mémoire de chemins :
from pathlib import Path
Puis le chargeur lui-même :
from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
documents: list[Document] = [] for path in sorted(runbook_dir.glob("*.md")):
text = path.read_text(encoding="utf-8")
title = text.splitlines()[0].removeprefix("# ").strip()
service_line = next(
line for line in text.splitlines()
if line.startswith("Service:")
)
service = service_line.split(":", 1)[1].strip() documents.append(
Document(
page_content=text,
metadata={
"source_id": path.stem,
"title": title,
"service": service,
"path": str(path),
},
)
) if not documents:
raise ValueError(f"No Markdown runbooks found in {runbook_dir}") return documents
Examen détaillé :
glob("*.md")sélectionne les fichiers Markdown situés dans un répertoire connu.sorted(...)assure que l’ordre de l’indexation locale reste reproductible. La qualité de la recherche ne devrait pas dépendre de l’ordre du système de fichiers, mais un entrée reproductible facilite le débogage.- Le fichier est décodé en UTF-8.
- L’extraction du titre et du service est volontairement simpliste, car les fichiers d’exemple suivent un format fixe. Un chargeur réel devrait valider les métadonnées et signaler les fichiers mal formatés au lieu de deviner.
page_contentcontient le texte pouvant être recherché.metadatacontient les champs qui doivent accompagner chaque morceau de données ultérieur.- Vérification finale pour éviter que un index vide ne soit considéré comme réussi.
Étape 3 : diviser les documents
Utilisez RecursiveCharacterTextSplitter :
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_runbooks(documents: list[Document]) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=420,
chunk_overlap=60,
add_start_index=True,
)
return splitter.split_documents(documents)
chunk_size=420 demande des blocs d’environ 420 caractères, ce qui convient bien à cet exemple de petite taille. chunk_overlap=60 répète une petite partie du texte entre les blocs voisins, ce qui permet de conserver une phrase à côté de son explication lorsque la limite se trouve au milieu. Un surcroît d’overlap n’est pas automatiquement mieux : cela signifie plus de texte stocké, plus de travail pour les embeddings et davantage de résultats proches des doublons.
Avec add_start_index=True, chaque bloc enregistre l’offset caractéristique où il a commencé, ce qui vous permet de revenir au fichier original. split_documents renvoie des objets Document frais, et chaque bloc hérite des métadonnées de son parent.
Choix de la taille du bloc
Copier un numéro provenant d’un tutoriel n’est pas une décision à prendre à la légère. Examinez d’abord comment vos documents sont organisés et ce que les gens demandent réellement. Une étape de dépannage peut nécessiter son titre, ses conditions et ses avertissements regroupés en une seule section, tandis qu’une longue table de politiques peut exiger un outil de séparation tenant compte de la structure. Testez ensuite la récupération en fonction de questions étiquetées.
Étape 4 : créer le récupérateur de stockage vectoriel
Afin de garder l’exemple simple, l’application utilise InMemoryVectorStore :
from langchain_core.vectorstores import InMemoryVectorStore
def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
chunks = split_runbooks(load_runbooks(runbook_dir))
store = InMemoryVectorStore.from_documents(
chunks,
embedding=embeddings,
)
return store.as_retriever(search_kwargs={"k": k})
from_documents incorpore et stocke les fragments, tandis que as_retriever expose l’interface standard de récupération. La valeur k=3 demande trois documents. Cela ne signifie pas que les trois sont pertinents, et cela ne garantit pas non plus la présence du passage souhaité parmi eux.
L’application crée des embeddings pour les fournisseurs :
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
Path(__file__).parent / "runbooks",
embeddings,
)
Les modèles disponibles et leurs noms dépendent du compte de votre fournisseur, il est donc conseillé de conserver le nom du modèle dans la configuration. De plus, InMemoryVectorStore perd son index lorsque le processus s’arrête. Cela convient pour des exemples pédagogiques et de petits tests, mais ce n’est pas un index adapté à une utilisation en production partagée.
Étape 5 : créer une requête ciblée à partir de champs fiables
Évitez de rechercher en utilisant l’ensemble complet des données de l’application. Sélectionnez uniquement les champs de classification pertinents :
def build_query(classification: TicketClassification) -> str:
values = [
classification.service or "unknown service",
classification.category,
classification.summary,
*classification.evidence,
]
return " | ".join(values)
Pour le billet continu, la requête ressemble à peu près à ceci :
checkout-api | database | EU checkout fails after a deployment. |
database connection refused
La chaîne de caractères associe le service probable, la catégorie du problème, un bref résumé ainsi que les éléments prélevés sur le ticket. Les secrets, les instructions cachées et l’historique complet des messages ne sont pas inclus.
Étape 6 : transmettre les métadonnées lors de la conversion
Ce que le récupérateur renvoie, ce sont des objets Document, et chacun d’eux doit devenir un passage structuré :
class RetrievedPassage(BaseModel):
source_id: str
title: str
service: str
start_index: int
text: str
Puis associez le Document à celui-ci :
def to_passage(document: Document) -> RetrievedPassage:
return RetrievedPassage(
source_id=str(document.metadata["source_id"]),
title=str(document.metadata["title"]),
service=str(document.metadata["service"]),
start_index=int(document.metadata.get("start_index", 0)),
text=document.page_content.strip(),
)
Les champs obligatoires sont indiqués entre crochets ; par conséquent, un morceau manquant de source_id, title ou service provoque immédiatement une erreur. start_index reçoit une valeur par défaut, car les morceaux enregistrés précédemment ne l’ont pas nécessairement, et c’est à la migration en production de déterminer si cette valeur par défaut est acceptable.
Étape 7 : créer une entrée de recommandation ciblée
L’outil destiné au modèle reçoit ce contrat :
class RecommendationInput(BaseModel):
ticket_id: str
ticket_summary: str
category: Category
severity: Severity
service: str | None
evidence: list[str]
passages: list[RetrievedPassage] = Field(
min_length=1,
max_length=4,
)
Lorsque aucune référence n’est retournée, il n’y a pas de recommandation fiable :
def prepare_recommendation_input(
classification: TicketClassification,
passages: list[RetrievedPassage],
) -> RecommendationInput:
if not passages:
raise ValueError("At least one runbook passage is required")
return RecommendationInput(
ticket_id=classification.ticket_id,
ticket_summary=classification.summary,
category=classification.category,
severity=classification.severity,
service=classification.service,
evidence=classification.evidence,
passages=passages[:4],
)
Si la recherche ne trouve rien, le modèle ne doit pas être laissé à improviser en se basant uniquement sur ses connaissances générales. Dirigez le ticket vers une personne ou utilisez une solution de secours qui a été approuvée séparément.
Étape 8 : demander des identifiants de source dans un format structuré
Le schéma des résultats inclut des citations :
class GroundedRecommendation(BaseModel):
explanation: str = Field(min_length=20, max_length=500)
next_checks: list[str] = Field(min_length=1, max_length=4)
source_ids: list[str] = Field(min_length=1, max_length=4)
needs_human_review: bool
Le texte d’instruction est le suivant :
Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.
Les instructions de promptage sont utiles, mais les vérifications au niveau de l’application restent importantes.
Étape 9 : rejeter les citations qui n’ont pas été retrouvées
Lorsque l’auteur a effectué la recherche, comparez ses identifiants de source avec les extraits retrouvés :
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError(
"Recommendation cited sources that retrieval did not return: "
+ ", ".join(sorted(unknown))
)
Cela ne détecte qu’un ID de source fabriqué, et rien de plus. Un modèle peut citer un document qui a bien été retourné et malgré cela interpréter son contenu de manière erronée ; par conséquent, une évaluation et une révision humaine restent nécessaires.
Étape 10 : connecter le flux de travail corrigé
Le code en temps de demande est simple :
def run_retrieval_workflow(
classification: TicketClassification,
retriever,
recommendation_chain,
) -> RetrievalResult:
query = build_query(classification)
passages = retrieve_passages(retriever, query)
values = prepare_recommendation_input(classification, passages)
recommendation = recommendation_chain.invoke(values)
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError("Recommendation cited an unavailable source") return RetrievalResult(
query=query,
recommendation=recommendation,
passages=passages,
)
Il n’y a pas de boucle cachée. La recherche s’exécute une fois, l’écriture s’exécute une fois, et la vérification de la source s’exécute une fois.
Résultat
Un résultat ressemble à ceci :
{
"query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
"recommendation": {
"explanation": "The error matches the checkout database connection runbook.",
"next_checks": [
"Compare the active database host, port and database name with the previous successful deployment.",
"Confirm that the application secret points to the expected credential version."
],
"source_ids": ["checkout-database"],
"needs_human_review": false
},
"passages": [
{
"source_id": "checkout-database",
"title": "Checkout database connection failures",
"service": "checkout-api",
"start_index": 0
}
]
}
L’API peut transformer chaque ID de source en un lien, mais uniquement après avoir vérifié que l’appelant actuel a le droit de l’ouvrir.
Test de la récupération sans payer pour les embeddings
Le snapshot d’accompagnement inclut KeywordEmbeddings, un petit substitut de test qui compte un vocabulaire fixe :
VOCABULARY = (
"database",
"connection",
"checkout",
"deployment",
"credential",
"pool",
"payment",
"latency",
"timeout",
"identity",
"access",
"role",
"claim",
)
Ne le confondez pas avec un embedder de production. Son unique but est de garantir que les tests se déroulent rapidement, hors ligne et de manière déterministe. Le premier test vérifie le ticket en cours :
passages = retrieve_passages(
retriever,
build_query(classification()),
)
assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0
Les deuxièmes mesures concernent deux des trois requêtes étiquetées :
score = hit_at_k(
retriever,
[
("checkout database connection refused", "checkout-database"),
("payment provider latency timeout", "payment-latency"),
("identity role claim access denied", "identity-access"),
],
)
assert score == 1.0
Ces deux exemples soulèvent une petite question : la source attendue se trouve-t-elle parmi les deux premiers résultats ? Un véritable ensemble de test nécessite de nombreuses questions, des négatifs difficiles, des documents similaires ainsi que des règles d’accès actuelles. Un score parfait sur trois exemples prouve uniquement que ces trois ont réussi l’épreuve.
Le troisième test introduit une recommandation fictive indiquant not-returned, et le flux de travail doit la rejeter. Exécutez tout cela avec :
pytest -q
Résultat attendu :
3 passed
Dénomination des échecs de récupération
Chaque type d’échec mérite son propre nom, afin de savoir où chercher.
- Échec d’indexation : un fichier n’a pas pu être lu, les métadonnées manquaient ou une demande d’incorporation a échoué. Ne publiez jamais un index inachevé comme s’il était complet.
- Échec de fraîcheur : une politique a changé mais la recherche continue de fournir le guide précédent. Enregistrez les versions et les timestamps, et déterminez comment les modifications et suppressions se propagent dans l’index.
- Échec de permission : l’index a affiché un passage auquel cet utilisateur n’a pas le droit de lire. Appliquez des filtres d’accès avant que tout texte n’atteigne le modèle, car cacher le lien par la suite est trop tardif.
- Échec de classement : le document correct est indexé mais absent parmi les résultats principaux. Ajustez la requête, les filtres, le découpage, les embeddings ou le classement, puis mesurez à nouveau.
Le texte récupéré n’est pas fiable
Les fiches de procédure contiennent des erreurs, et parfois des instructions qui ressemblent à des ordres adressés au modèle. Traitez les passages récupérés comme du matériel source cité, et jamais comme des règles d’application ayant une priorité plus élevée. Gardez un comportement stable dans le prompt du système, délimitez le texte récupéré et restreignez les outils dans le code d’application. Aucune fiche de procédure ne devrait jamais pouvoir s’autoriser à déclencher un outil d’écriture.
Lorsque la recherche vectorielle n’est pas l’outil adapté
Certaines recherches n’ont pas besoin de vecteurs. Un ID de ticket exact peut être obtenu via une simple requête de base de données, et un nom de service connu correspond directement à une entrée dans le catalogue. Les codes d’erreur exacts et les identifiants rares se prêtent mieux à une recherche par mots-clés. Si l’entreprise dispose déjà d’un service de recherche qui indexe les documents appropriés et gère les permissions, il convient d’en utiliser un. La recherche vectorielle n’est qu’un outil parmi plusieurs.
Qu’il manque encore pour la mise en production
Plusieurs éléments sont encore absents avant qu’une entreprise ne puisse compter sur cette approche :
- un stockage persistant plutôt que de la mémoire
- une personne clairement responsable de chaque document
- des filtres d’accès appliqués au moment de la requête
- un mécanisme pour gérer les versions et les suppressions de documents
- un plan en cas de modification du modèle d’embedding
- une limite quant à la quantité de texte retournée
- des tests étiquetés pour l’extraction des informations
Le cadre relie les différentes composantes, et l’application prend les décisions correspondantes.
La position de l’application de support
Elle peut désormais rechercher dans un ensemble de manuels de procédures contrôlés tout en conservant ses sources :
classification
-> focused query
-> retriever
-> typed passages with metadata
-> recommendation
-> source-ID check
Ce qui vient ensuite, c’est la séparation de l’historique des messages, des données utilisateur enregistrées, de l’état du flux de travail en temps réel et des points de contrôle en éléments distincts. Tracer cette frontière permet également de déterminer si une simple chaîne LangChain suffit ou si LangGraph est plus adapté.
Lectures complémentaires :
- https://docs.langchain.com/oss/python/langchain/retrieval
- https://docs.langchain.com/oss/python/langchain/knowledge-base
- https://docs.langchain.com/oss/python/integrations/splitters/index
- https://docs.langchain.com/oss/python/integrations/retrievers