Accueil / Articles / RAG : des principes fondamentaux aux trois agents de base

rag embeddings bedrock langgraph retrieval aws

RAG : des principes fondamentaux aux trois agents de base

Pourquoi les poids échouent avec des données privées, comment fonctionnent les embeddings et la recherche cosinus, ainsi que trois types d’agents progressifs allant de bases de connaissances gérées à un système d’ingestion auto-géré.

4464 mots

Les modèles de base congelés ne peuvent pas répondre aux questions commerciales privées en se basant uniquement sur leurs poids. Leur entraînement n’a jamais pris en compte les politiques internes de Ressources humaines, les contrats ou les guides opérationnels ; la date de fin de mise à jour fige le temps ; et en l’absence de faits, le générateur produit néanmoins des suppositions fluides. Apprendre à un réseau ce que connaît une entreprise devient donc un problème de système, et non simplement un problème lié à des points de contrôle supplémentaires.

Pourquoi les poids seuls échouent

Trois limites structurelles entravent les systèmes de Q&R d’entreprise :

Les corpus privés n’ont jamais été utilisés pour l’entraînement. Les règles de conservation et les macros de tickets ne correspondent pas à du texte public sur le web.
Le temps s’arrête à la date de fin de mise à jour. Les modifications juridiques du mois dernier restent invisibles pour une version congelée.
Le silence n’est pas la valeur par défaut. La prédiction du token suivant privilégie une continuation plausible plutôt qu’une réponse honnête indiquant l’ignorance.

Trois façons d’intégrer des connaissances — et celles qui fonctionnent

Option 1 : Ajustement fin sur le corpus

Un entraînement supplémentaire peut inculquer un style ou des compétences, mais il dilue les faits dans divers paramètres sans citations, et ne permet pas de contrôler la visibilité par utilisateur car tout le monde partage les mêmes poids. Pour les questions-réponses réglementaires qui exigent des citations et un contrôle d’accès, l’ajustement fin est le mauvais mécanisme.

Option 2 : Coller tout le texte dans les instructions du système

Fournir du texte au moment de la demande transforme le rappel en lecture, ce qui constitue le mécanisme approprié. Cela fonctionne pour de petits ensembles stables. Cependant, cela échoue lorsque la longueur dépasse les limites prévues, lorsque chaque token est facturé à chaque demande, lorsque les réponses se trouvent au milieu d’un prompt et sont de mauvaise qualité, et lorsque un collage statique ne permet pas de filtrer selon le rôle.

Option 3 : Coller uniquement ce qui est important

Conserver le mécanisme de lecture au moment de la demande ; ajuster l’échelle. Récupérer un petit nombre de paragraphes pertinents, puis générer la réponse. C’est en une phrase ce qu’on appelle la génération augmentée par le rappel.

Qu’est-ce que le RAG, du début à la fin

En mode hors ligne : ingestion des fichiers, division en segments, intégration des segments, stockage des vecteurs (et souvent d’index lexicaux). En mode en ligne : intégration de la question, récupération des meilleurs segments, regroupement de ceux-ci avec la question dans une instruction, génération d’une réponse basée sur ces segments.

La partie difficile n’est pas l’appel de génération final — c’est assurer que la récupération des données soit précise, économique et respectueuse des permissions.

Vecteurs et embeddings sans mysticisme

Ici, un vecteur est une liste de nombres qui place un extrait de texte dans un espace où les éléments ayant des significations proches se trouvent à proximité les uns des autres. Un réseau d’embeddings projette le texte dans cet espace. La similarité cosinus compare davantage la direction que la longueur, c’est pourquoi elle domine l’évaluation des résultats de recherche. Les métriques de distance qui ignorent la normalisation donnent de moins bons résultats avec des longueurs de segments inégales.

Le choix de la dimensionnalité et de l’embedder relève d’une décision stratégique : plus de dimensions permettent de distinguer mieux les significations, mais augmentent les coûts de stockage et la latence ; utiliser des embedders différents entre le moment de l’indexation et celui de la requête compromet silencieusement la recherche du voisin le plus proche.

Arithmétique de la similarité

Le cosinus correspond au produit scalaire de vecteurs unitaires. Deux segments concernant les réinitialisations de mot de passe avec des formulations différentes restent proches l’un de l’autre si l’embedder a réussi à capturer l’intention. La recherche par mots-clés rate ce lien ; la récupération dense le détecte — mais peut manquer les tokens SKU exacts à moins qu’une recherche lexicale hybride ne soit utilisée.

Création de trois agents complets sur des services d’inférence cloud gérés

Le reste de ce guide concrétise ces idées à travers trois agents progressifs basés sur des modèles gérés de type AWS Bedrock : une base de connaissances curatée, un orchestrateur LangGraph et un pipeline d’ingestion auto-géré. Chaque description constitue un schéma complet d’agent — à considérer comme une architecture exécutable et non comme de simples extraits.

Agent 1 — Chemin vers la base de connaissances gérée

"""
AGENT 1: Managed Knowledge Base. The 30-line version.
Bedrock does the chunking, embedding, indexing and searching. You wire it up.
"""
import boto3
from langchain_aws import AmazonKnowledgeBasesRetriever, ChatBedrockConverse
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

REGION = "eu-west-1"
KB_ID = "ABCD1234EF"

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KB_ID,
    region_name=REGION,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 5,
            "overrideSearchType": "HYBRID",
        }
    },
)

llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
    max_tokens=1024,
)

PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "Answer only from the numbered context. Cite every claim as [n]. "
     "If the context does not contain the answer, say so plainly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def format_docs(docs) -> str:
    return "\n\n".join(
        f"[{i}] source={d.metadata.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{d.page_content}"
        for i, d in enumerate(docs, start=1)
    )


chain = (
    RunnableParallel(context=retriever | format_docs, question=RunnablePassthrough())
    | PROMPT
    | llm
    | StrOutputParser()
)

if __name__ == "__main__":
    print(chain.invoke("How fast must we submit a shortlist for a niche engineering role?"))

Ce chemin repose sur une base de connaissances hébergée : les documents sont stockés, un indexeur géré crée des embeddings, et l’agent répond en utilisant les citations fournies par ce service. Les avantages opérationnels incluent moins de code d’index personnalisé ; les inconvénients, en revanche, sont un contrôle moindre sur les limites des segments et la fusion des données.

Agent 2 — Orchestration via LangGraph

"""
AGENT 2: LangGraph agent that decides WHETHER and WHICH index to search.
Retrieval is a tool. Identity travels in typed context, never in the prompt.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any

import boto3
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
from langchain.tools import ToolRuntime
from langchain_aws import ChatBedrockConverse
from langchain_core.tools import tool

REGION = "eu-west-1"
POLICY_KB = "ABCD1234EF"
ENGINEERING_KB = "WXYZ5678GH"

agent_runtime = boto3.client("bedrock-agent-runtime", region_name=REGION)


@dataclass
class RequestContext:
    """Built from validated JWT claims. The model cannot read or forge it."""
    subject: str
    roles: tuple[str, ...]
    region: str


def _retrieve(kb_id: str, query: str, ctx: RequestContext, k: int = 5) -> str:
    """Bedrock Retrieve API with a server-side metadata filter for tenant + residency."""
    resp = agent_runtime.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": k,
                "overrideSearchType": "HYBRID",
                "filter": {
                    "andAll": [
                        {"in": {"key": "acl_roles", "value": list(ctx.roles)}},
                        {"in": {"key": "region", "value": [ctx.region, "GLOBAL"]}},
                    ]
                },
            }
        },
    )
    results: list[dict[str, Any]] = resp.get("retrievalResults", [])
    if not results:
        return "NO_AUTHORISED_RESULTS: nothing in this index that you may read."
    return "\n\n".join(
        f"[{i}] score={r.get('score'):.3f} "
        f"source={r.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{r['content']['text']}"
        for i, r in enumerate(results, start=1)
    )


@tool
def search_policy_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search HR, compliance and commercial policy.

    Use for shortlist SLAs and service credits, candidate data retention and consent,
    GDPR erasure, statement-of-work approval thresholds, background check rules and
    automated decision governance.
    """
    return _retrieve(POLICY_KB, query, runtime.context)


@tool
def search_engineering_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search platform runbooks and integration documentation.

    Use for the requisition sync service, Workday and SuccessFactors webhooks, Kafka
    topics and event schemas, idempotency and retry behaviour, and vector index refresh.
    """
    return _retrieve(ENGINEERING_KB, query, runtime.context)


agent = create_agent(
    model=ChatBedrockConverse(
        model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name=REGION,
        temperature=0,
    ),
    tools=[search_policy_docs, search_engineering_docs],
    system_prompt=(
        "You are an enterprise knowledge assistant.\n"
        "- Answer only from tool results and cite the [n] markers they return.\n"
        "- If a tool returns NO_AUTHORISED_RESULTS, say you cannot access a source and "
        "name the document owner. Never fill the gap from memory.\n"
        "- For questions spanning policy and platform, call both tools before answering."
    ),
    context_schema=RequestContext,
    middleware=[
        ModelCallLimitMiddleware(thread_limit=8, run_limit=6, exit_behavior="end"),
        ToolCallLimitMiddleware(thread_limit=6, run_limit=5, exit_behavior="end"),
    ],
)

if __name__ == "__main__":
    claims = {"sub": "asha@example.com", "roles": ["recruiter"], "region": "EU"}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "How is the vector index refreshed?"}]},
        context=RequestContext(
            subject=claims["sub"],
            roles=tuple(claims["roles"]),
            region=claims["region"],
        ),
    )
    print(result["messages"][-1].content)

Ici, le flux de contrôle est représenté par un graphe explicite : on classe l’intention, on appelle des outils de récupération, on peut solliciter un humain si nécessaire, puis on répond. Les points de contrôle et les interruptions transforment les threads d’assistance à long terme en sessions durables plutôt que en dictionnaires en mémoire.

Agent 3 — Ingestion et récupération gérées par l’agent lui-même

"""
AGENT 3: Self-managed pipeline: your chunking, your index, your fusion, your rerank.
Chosen when Bedrock KB's fixed options are not enough (custom fusion, per-stage timing,
an abstention gate, or a chunking rule the managed service will not express).
"""
from __future__ import annotations
import time
from typing import Annotated, Any, TypedDict

import boto3
from langchain_aws import BedrockEmbeddings, ChatBedrockConverse
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langgraph.graph import END, START, StateGraph

REGION = "eu-west-1"
CONFIDENCE_FLOOR = 0.34
CANDIDATE_K, CONTEXT_K = 20, 4

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name=REGION,
    model_kwargs={"dimensions": 1024, "normalize": True},
)
llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
)
rerank_client = boto3.client("bedrock-agent-runtime", region_name=REGION)


# ---------------------------------------------------------------- ingest
def chunk_policy_document(markdown: str, source: str, acl: list[str], region: str):
    """Structure-aware first, size-capped second. Headings are the natural unit for
    policy text; the recursive splitter only intervenes on oversized sections."""
    by_heading = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "title"), ("##", "section"), ("###", "clause")]
    ).split_text(markdown)

    capper = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
    chunks: list[Document] = []
    for part in capper.split_documents(by_heading):
        heading = " > ".join(
            v for k, v in part.metadata.items() if k in ("title", "section", "clause")
        )
        chunks.append(
            Document(
                # contextual retrieval: the chunk carries its own scope
                page_content=f"{heading}\n{part.page_content}",
                metadata={**part.metadata, "source": source, "acl": acl, "region": region},
            )
        )
    return chunks


# ---------------------------------------------------------------- state
class State(TypedDict, total=False):
    question: str
    roles: list[str]
    region: str
    candidates: list[Document]
    context_docs: list[Document]
    confidence: float
    answer: str
    timings: Annotated[dict[str, float], lambda a, b: {**(a or {}), **(b or {})}]


def _ms(name: str, t0: float) -> dict[str, float]:
    return {name: round((time.perf_counter() - t0) * 1000, 1)}


# ---------------------------------------------------------------- nodes
def retrieve(state: State, *, store) -> State:
    """Hybrid search with an ACL pre-filter pushed into the index query."""
    t0 = time.perf_counter()
    docs = store.similarity_search(
        state["question"],
        k=CANDIDATE_K,
        filter={"acl": {"$in": state["roles"]}, "region": {"$in": [state["region"], "GLOBAL"]}},
    )
    return {"candidates": docs, "timings": _ms("retrieve", t0)}


def rerank(state: State) -> State:
    """Bedrock Rerank: a cross-encoder scoring each (query, chunk) pair jointly."""
    t0 = time.perf_counter()
    cands = state["candidates"]
    if not cands:
        return {"context_docs": [], "confidence": 0.0, "timings": _ms("rerank", t0)}

    resp = rerank_client.rerank(
        queries=[{"type": "TEXT", "textQuery": {"text": state["question"]}}],
        sources=[
            {"type": "INLINE", "inlineDocumentSource": {
                "type": "TEXT", "textDocument": {"text": d.page_content}}}
            for d in cands
        ],
        rerankingConfiguration={
            "type": "BEDROCK_RERANKING_MODEL",
            "bedrockRerankingConfiguration": {
                "numberOfResults": CONTEXT_K,
                "modelConfiguration": {
                    "modelArn": f"arn:aws:bedrock:{REGION}::foundation-model/amazon.rerank-v1:0"
                },
            },
        },
    )
    ranked = resp["results"]
    top = [cands[r["index"]] for r in ranked]
    return {
        "context_docs": top,
        "confidence": round(float(ranked[0]["relevanceScore"]), 3),
        "timings": _ms("rerank", t0),
    }


def generate(state: State) -> State:
    """Abstain below the floor rather than answering from weak context."""
    t0 = time.perf_counter()
    if not state["context_docs"] or state["confidence"] < CONFIDENCE_FLOOR:
        return {
            "answer": "I cannot ground an answer for that in a source you are permitted "
                      "to read. Routing to the document owner.",
            "timings": _ms("generate", t0),
        }
    context = "\n\n".join(
        f"[{i}] {d.metadata.get('source')}\n{d.page_content}"
        for i, d in enumerate(state["context_docs"], start=1)
    )
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer only from the numbered context and cite each claim as [n]."),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])
    reply = (prompt | llm).invoke({"context": context, "question": state["question"]})
    return {"answer": reply.content, "timings": _ms("generate", t0)}


def build_graph(store):
    g = StateGraph(State)
    g.add_node("retrieve", lambda s: retrieve(s, store=store))
    g.add_node("rerank", rerank)
    g.add_node("generate", generate)
    g.add_edge(START, "retrieve")
    g.add_edge("retrieve", "rerank")
    g.add_edge("rerank", "generate")
    g.add_edge("generate", END)
    return g.compile()

Cette variante gère le découpage en blocs, les appels d’embedding, le stockage vectoriel ainsi que l’assemblage des prompts. Les équipes l choisissent lorsque les politiques de segmentation, la fusion hybride lexicale et dense ou les filtres de location doivent faire partie intégrante du code plutôt que de simples paramètres à activer depuis la console.

Sélectionner parmi les trois options

Privilégiez la solution gérée lorsque la rapidité de mise en œuvre est cruciale et que les corpus sont modestes. Optez pour LangGraph lorsque la prise de décision par branches, l’examen humain et la mémoire entre les interactions jouent un rôle prépondérant. Prenez en charge tout le pipeline lorsque la qualité de la récupération des identifiants, des ACL ou de la segmentation par domaine devient l’élément clé du produit.

Une évaluation qui améliore réellement la qualité

Suivez le Recall@k et la Précision@k sur un ensemble de questions figé, ainsi que le degré de fidélité des réponses par rapport aux segments récupérés. Associez chaque modification concernant la taille des segments ou l’embedder aux scores avant/après. Refusez de générer lorsque les meilleurs segments sont faibles — un silence calibré vaut mieux qu’une invention fluide.

Quand ne pas utiliser RAG

Omettez la récupération d’informations pour les faits que le réseau de base connaît déjà, pour des tâches purement créatives, pour des données en constante évolution mieux gérées par des API, ainsi que pour des parcours à très faible latence qui ne peuvent pas se permettre une opération de récupération. Parfois, une instruction améliorée ou un léger ajustement du format suffit.

Conclusion

RAG est comme le système de plomberie : un apport propre, une récupération honnête, des instructions de génération précises et des critères mesurables. Les trois agents illustrent la même idée à trois niveaux de contrôle différents. Choisissez le niveau qui correspond à vos besoins en termes de contrôle — et non celui qui semble le plus avancé sur une diapositive.

Checklist opérationnelle : (1) un embedder pour l’indexation et les requêtes, (2) des métadonnées pour les ACL et les dates dès le premier jour, (3) une recherche hybride lorsque des identifiants apparaissent, (4) un réclassement lorsque la précision est importante, (5) des questions clés dans le CI, (6) des déclencheurs de reconstruction lorsque les sources changent. Traitez le plan de connaissances comme un produit avec des responsables, des SLO et des mécanismes de rollback – et non comme un index créé le week-end que personne ne surveille. Lorsque les graphes d’agents s’agrandissent, mesurez le coût en tokens en même temps que la qualité des réponses ; une orchestration qui consomme quinze fois plus de ressources qu’un simple chat sans préserver la fidélité est une régression déguisée en architecture. Conservez des guides opérationnels pour les refus dus à des résultats vides, les incidents liés à un index obsolète ou les incohérences des embedders, afin que l’équipe de veille ne doive pas improviser à partir des journaux de chat.

Décrivez chaque source de corpus avec un responsable et une fréquence de mise à jour ; les buckets orphelins deviennent une source de dégradation silencieuse.

Préférez un chevauchement des segments précoce ; n’adoptez les séparateurs sémantiques que lorsque les évaluations montrent que les fenêtres fixes échouent.

Enregistrez les identifiants de récupération des données avec les réponses afin de pouvoir diagnostiquer les mauvaises réponses comme des échecs de récupération ou des déviations de génération.

Séparez, si possible, les juges d’évaluation du réseau de réponse pour réduire le biais d’autopréférence.

Budgetnez séparément les dépenses API liées à l’indexation de celles relatives au service interactif, afin que les services financiers puissent suivre les deux postes de dépenses.

Mettez à jour les modèles de prompts avec la version correspondante aux mises à jour d’index ; une simple modification des prompts sans réévaluation est la cause principale des dégradations de performance.

Pour les environnements multi-locataires, vérifiez que les filtres ne peuvent pas être contournés par des injections de prompts prétendant jouer le rôle d’administrateur.

Maintenez un interrupteur de sécurité qui désactive la génération et ne renvoie que des citations lorsque les moniteurs de fidélité détectent des anomalies.

Revoyez les poids lexicaux hybrides chaque fois que les codes de produit dominent la composition du trafic.

Formez le personnel d’assistance aux trains à lire les panneaux de citation avant d’élever des tickets concernant des « bots incorrects ».

Dokumentez chaque source de corpus en indiquant son responsable et la fréquence de mise à jour ; les conteneurs orphelins deviennent une source de problèmes silencieux.

Préférez d’abord les chevauchements de segments ; n’utilisez des séparateurs sémantiques que lorsque les évaluations montrent que les fenêtres fixes ne fonctionnent pas.

Enregistrez les identifiants de récupération ainsi que les réponses afin de pouvoir diagnostiquer si de mauvaises réponses sont dues à des échecs de récupération ou à une dérive de génération.

Séparez autant que possible les juges d’évaluation du réseau de réponse pour réduire le biais lié à la préférence personnelle.

Budgettez séparément les dépenses API liées à l’indexation de celles relatives au service interactif, afin que le service financier puisse suivre les deux postes de dépenses.

Mettez à jour les modèles de prompts en même temps que les versions d’index ; une modification des prompts sans réévaluation est la cause principale des dégradations de performance.

Dans les environnements multi-locataires, vérifiez que les filtres ne peuvent pas être contournés par des injections de prompts prétendant représenter des rôles administratifs.

Conservez un interrupteur de sécurité qui désactive la génération et ne restitue que les citations lorsque les moniteurs de fidélité baissent.

Revoyez les pondérations lexicales hybrides chaque fois que les codes de produit dominent la composition du trafic.

Formez le personnel d’assistance à lire les panneaux de citations avant d’élever des tickets concernant un « bot incorrect ».

Dokumentez chaque source de corpus avec son responsable et la fréquence de mise à jour ; les conteneurs orphelins deviennent une source de problèmes silencieux.

Préférez d’abord les chevauchements de segments ; n’adoptez des séparateurs sémantiques que lorsque les évaluations montrent que les fenêtres fixes ne fonctionnent pas.

Enregistrez les identifiants de récupération ainsi que les réponses afin de pouvoir diagnostiquer les mauvaises réponses comme des échecs de récupération plutôt que des déviations de génération.

Séparez autant que possible les juges d’évaluation du réseau de réponse pour réduire le biais d’auto-préférence.

Budgettez séparément les dépenses API liées à l’indexation de celles relatives au service interactif, afin que le service financier puisse suivre les deux postes de dépenses.

Les modèles de prompts de version sont affichés à côté des builds d’index ; une mise à jour des prompts sans réévaluation est la manière dont les dérives se propagent.

Pour les environnements multi-locataires, vérifiez que les filtres ne peuvent pas être contournés par des injections de prompts prétendant jouer le rôle d’administrateur.

Maintenez un interrupteur de sécurité qui désactive la génération et ne renvoie que des citations lorsque les moniteurs de fidélité chutent.

Revoyez les poids lexicaux hybrides chaque fois que les codes de produit dominent la composition du trafic.

Formez le personnel d’assistance à lire les panneaux de citations avant d’élever des tickets concernant un « mauvais bot ».

Maintenez les évaluations associées et les métriques de refus à côté de la latence afin que les coûts et la qualité soient présentés sur le même tableau.

Conservez les évaluations associées et les métriques de refus à côté de la latence afin que le coût et la qualité soient présentés sur la même feuille.

Conservez les évaluations associées et les métriques de refus à côté de la latence afin que le coût et la qualité soient présentés sur une même feuille.

Tous les articles Envoyer le brief