Agent de support N1 : RAG agentique tracé de bout en bout avec LangFuse

Le support de niveau 1, c’est 80 % de questions déjà répondues quelque part dans la documentation. Le candidat idéal pour un agent — à condition qu’il sache quand chercher, qu’il ne réponde pas hors de sa base, qu’il escalade quand il ne sait pas, et surtout qu’on puisse savoir pourquoi il a répondu ce qu’il a répondu. Sans cette dernière propriété, aucun agent ne devrait toucher un vrai client.

Ce cas construit un tel agent de bout en bout : on part des fichiers Markdown de la documentation, on les vectorise, on bâtit l’agent, on l’escalade, on l’instrumente avec LangFuse, on l’évalue sur un dataset, et on l’expose derrière une API. Chaque étape est donnée en code complet et exécutable.

Architecture : un RAG agentique, pas un RAG figé

Un RAG classique est un pipeline rigide : on récupère, on génère, fin. Il récupère même quand c’est inutile (« bonjour »), ne se rattrape pas quand la récupération est mauvaise, et ne sait pas dire « je ne sais pas ».

Notre agent rend ces étapes décidées : c’est le modèle qui choisit d’appeler l’outil de recherche, qui juge si les documents suffisent, et qui décide de reformuler, de répondre, ou d’escalader. La recherche devient un outil parmi d’autres.

Plutôt que ce pipeline rigide, on adopte la boucle ReAct du premier chapitre, appliquée à la récupération : la recherche et l’escalade deviennent des outils que l’agent appelle — ou non — et peut rappeler après reformulation.

Construire l’agent, étape par étape

pip install langgraph langchain langchain-openai langfuse langchain-community

export OPENAI_API_KEY="sk-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com"

from pathlib import Path
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import MarkdownTextSplitter
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 1) charger tous les .md de la documentation
docs = []
for f in Path("docs/").rglob("*.md"):
    chargeur = TextLoader(str(f), encoding="utf-8")
    for d in chargeur.load():
        d.metadata["source"] = f.name        # on garde la source pour citer
        docs.append(d)

# 2) découper : des fragments ni trop gros (bruit) ni trop petits (perte de contexte)
splitter = MarkdownTextSplitter(chunk_size=1000, chunk_overlap=150)
fragments = splitter.split_documents(docs)

# 3) vectoriser et indexer
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = InMemoryVectorStore.from_documents(fragments, embeddings)
retriever = store.as_retriever(search_kwargs={"k": 4})

print(f"{len(fragments)} fragments indexés depuis {len(docs)} documents.")

from langchain_core.tools import tool

@tool
def chercher_doc(requete: str) -> str:
    """Recherche dans la base de connaissances produit.
    À utiliser pour TOUTE question factuelle sur le fonctionnement, la
    facturation ou la configuration du produit. Renvoie les extraits pertinents
    et leur source."""
    docs = retriever.invoke(requete)
    if not docs:
        return "AUCUN_RESULTAT"
    return "\n\n---\n\n".join(
        f"[source: {d.metadata['source']}]\n{d.page_content}" for d in docs
    )

@tool
def escalader(motif: str, resume: str) -> str:
    """Transfère la demande à un agent humain. À utiliser quand la base ne contient
    pas la réponse, en cas de réclamation, ou pour toute action sur le compte."""
    ticket = helpdesk.creer_ticket(motif=motif, resume=resume, priorite="n2")
    return f"Demande transférée à un conseiller (ticket {ticket.id})."

outils = [chercher_doc, escalader]

SYSTEME = """Tu es l'assistant de support de niveau 1 de DEEP-5.
Règles impératives :
- Réponds UNIQUEMENT à partir des documents renvoyés par `chercher_doc`.
- Cite la source entre crochets à la fin de chaque affirmation factuelle.
- Si la base ne contient pas la réponse, ou si la demande concerne le compte
  ou une réclamation, appelle `escalader` — n'invente jamais.
- N'invente jamais de procédure, de tarif ou de fonctionnalité.
- Réponds en français, de façon concise et actionnable."""

from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model
from langfuse.langchain import CallbackHandler

modele = init_chat_model("openai:gpt-4o-mini")
agent = create_react_agent(modele, outils, prompt=SYSTEME)

langfuse_handler = CallbackHandler()

def repondre(question: str, ticket_id: str, user_id: str) -> str:
    resultat = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config={
            "callbacks": [langfuse_handler],
            "recursion_limit": 8,                 # borne anti-boucle / coût
            "metadata": {
                "langfuse_session_id": ticket_id,  # regroupe la conversation
                "langfuse_user_id": user_id,       # segmente par client
                "langfuse_tags": ["support-n1"],
            },
        },
    )
    return resultat["messages"][-1].content

Un exemple concret, du message à la réponse

Déroulons une vraie requête pour voir l’agent décider.

print(repondre("Comment activer l'authentification à deux facteurs ?",
               ticket_id="T-4821", user_id="cli_902"))

En coulisses, l’agent franchit trois étapes : il décide d’appeler chercher_doc, reçoit deux fragments pertinents (provenant de securite.md), puis rédige une réponse sourcée. Sortie typique :

Pour activer l'A2F : ouvrez Paramètres > Sécurité > Authentification à deux
facteurs, puis scannez le QR code avec votre application d'authentification.
Conservez les codes de secours affichés. [source: securite.md]

Et sur une question hors base — « Pouvez-vous me rembourser ? » — l’agent n’invente pas : il appelle escalader(motif="réclamation", ...) et répond « Demande transférée à un conseiller (ticket …) ». C’est le comportement qu’on veut, et c’est la trace qui va nous le prouver.

Lire la trace : ce que LangFuse révèle

Ouvrez la trace dans LangFuse. Vous voyez l’arbre complet de la décision :

• le span racine : l’invocation de l’agent, durée et coût totaux ;
• une generation : le premier appel LLM, qui décide d’appeler chercher_doc — avec les messages, les tokens consommés et le coût estimé ;
• un span outil : la recherche, avec la requête exacte et les documents rendus ;
• une seconde generation : la rédaction finale à partir des documents.

Mesurer coût et latence

Chaque generation porte son nombre de tokens et son coût. Agrégés dans LangFuse, ils répondent aux questions que tout responsable se pose avant la production :

Évaluer la qualité, en continu

Tracer ne suffit pas : il faut savoir si les réponses sont bonnes. On combine trois sources de score, comme détaillé dans le deep-dive évaluation.

Le LLM-juge de fidélité

Pour le RAG, la métrique reine est la fidélité : la réponse s’appuie-t-elle vraiment sur les documents, sans inventer ? On l’implémente avec un juge à sortie structurée.

from pydantic import BaseModel, Field
from langfuse import Langfuse

langfuse = Langfuse()

class Fidelite(BaseModel):
    fidele: bool = Field(description="La réponse est-elle justifiée par les sources ?")
    raison: str

def juger_fidelite(trace_id: str, sources: str, reponse: str):
    juge = init_chat_model("openai:gpt-4o").with_structured_output(Fidelite)
    v = juge.invoke(
        f"Sources :\n{sources}\n\nRéponse de l'agent :\n{reponse}\n\n"
        f"La réponse est-elle ENTIÈREMENT justifiée par les sources ? Sois strict."
    )
    langfuse.create_score(trace_id=trace_id, name="fidelite",
                          value=int(v.fidele), comment=v.raison)
    return v

Les autres sources complètent le tableau : un score humain (un conseiller note la réponse) et un score implicite (le client a-t-il rouvert le ticket ?).

langfuse.create_score(trace_id=trace_id, name="utile", value=1)        # humain
langfuse.create_score(trace_id=trace_id, name="ticket_rouvert", value=0)  # métier

Le dataset de non-régression

Un test manuel sur trois questions ne prouve rien. On constitue un dataset de cas réels — dont des pièges hors périmètre où l’on attend une escalade — et on rejoue l’agent à chaque changement.

langfuse.create_dataset(name="support-qa")
langfuse.create_dataset_item(
    dataset_name="support-qa",
    input={"question": "Comment activer l'A2F ?"},
    expected_output="Paramètres > Sécurité > A2F, scanner le QR code.",
    metadata={"piege": False},
)
langfuse.create_dataset_item(
    dataset_name="support-qa",
    input={"question": "Remboursez-moi."},
    expected_output="[escalade]",
    metadata={"piege": True},        # on attend une escalade, pas une réponse
)

# rejouer l'agent sur tout le dataset et scorer chaque cas
dataset = langfuse.get_dataset("support-qa")
for item in dataset.items:
    with item.run(run_name="prompt-v4") as root:
        reponse = repondre(item.input["question"], ticket_id=f"eval-{item.id}", user_id="eval")
        a_escalade = "transférée à un conseiller" in reponse
        root.score_trace(name="escalade_correcte",
                         value=int(a_escalade == item.metadata["piege"]))

Exposer l’agent derrière une API

Dernière étape : rendre l’agent accessible. On l’expose en HTTP, en streamant idéalement, ici en version simple.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Demande(BaseModel):
    question: str
    ticket_id: str
    user_id: str

@app.post("/support")
def support(d: Demande):
    return {"reponse": repondre(d.question, d.ticket_id, d.user_id)}

Mise en production : la checklist

Bonnes pratiques & pièges

Ce qu’il faut retenir

• Tout commence par l’ingestion : charger, chunker, vectoriser, indexer — la qualité de l’index plafonne celle de l’agent.
• Un RAG agentique rend la récupération décidée : l’agent cherche quand il faut, reformule, et escalade au lieu d’inventer.
• LangFuse s’attache via un simple callback ; session_id et user_id rendent les traces exploitables.
• La trace transforme le débogage en lecture ; coût, latence et fidélité se mesurent en continu sur un dataset à pièges.
• Une API et une checklist de production referment la boucle, du concept au client.

Cet agent réunit tout le parcours : la boucle ReAct de l’initiation, la rigueur de conception du deep-dive, et l’observabilité qui rend le tout exploitable. Pour aller plus loin, comparez-le à l’assistant de recherche multi-agents et au RAG agentique correctif.