Agent LangGraph avec historique

Dans un vrai agent, on veut souvent accumuler des informations au fil des étapes : un historique de messages, une liste de documents trouves, des logs d'actions. LangGraph gere ca avec les Annotated types et l'operateur add.

Quand tu declares un champ du State avec Annotated[list, operator.add], LangGraph ne remplace plus la valeur mais AJOUTE les nouveaux éléments a la liste existante.

from typing import TypedDict, Annotated
import operator

class State(TypedDict):
messages: Annotated[list, operator.add]
count: int

def noeud(state):
return {
"messages": ["nouveau message"], # s'ajoute a la liste
"count": state["count"] + 1, # remplace la valeur
}

C'est tres utile pour tracer ce que fait l'agent et garder un historique de conversation.

Écris une fonction build_rag_agent() qui crée un graphe avec :

State :
- query : str (la question de l'utilisateur)
- logs : Annotated[list, operator.add] (l'historique des actions)
- documents : Annotated[list, operator.add] (les documents trouves)
- answer : str (la réponse finale)

Noeuds :
- "reformuler" : ajoute au log "Requête reformulee" et ne change rien d'autre
- "chercher" : ajoute au log "Recherche effectuee" et ajoute 2 documents fictifs : "Doc pertinent 1" et "Doc pertinent 2"
- "repondre" : ajoute au log "Réponse générée" et met dans answer le texte "Réponse basee sur X documents" ou X est le nombre de documents

Enchainement : START -> reformuler -> chercher -> repondre -> END

Exemple :

app = build_rag_agent()
r = app.invoke({"query": "test", "logs": [], "documents": [], "answer": ""})
r["logs"] vaut ["Requête reformulee", "Recherche effectuee", "Réponse générée"]
r["documents"] vaut ["Doc pertinent 1", "Doc pertinent 2"]
r["answer"] vaut "Réponse basee sur 2 documents"

Tests (1/1)

Tests

app = build_rag_agent()
r = app.invoke({'query': 'test', 'logs': [], 'documents': [], 'answer': ''})
# Logs accumules
assert r['logs'] == ['Requête reformulee', 'Recherche effectuee', 'Réponse générée'], f'Logs incorrects: {r["logs"]}'
# Documents accumules
assert r['documents'] == ['Doc pertinent 1', 'Doc pertinent 2'], f'Documents incorrects: {r["documents"]}'
# Réponse
assert r['answer'] == 'Réponse basee sur 2 documents', f'Answer incorrecte: {r["answer"]}'
# La query n'est pas modifiee
assert r['query'] == 'test', 'La query ne doit pas changer'

Indices (3 disponibles)

Solution officielle

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    query: str
    logs: Annotated[list, operator.add]
    documents: Annotated[list, operator.add]
    answer: str

def build_rag_agent():
    def reformuler(state):
        return {"logs": ["Requête reformulee"]}

    def chercher(state):
        return {
            "logs": ["Recherche effectuee"],
            "documents": ["Doc pertinent 1", "Doc pertinent 2"],
        }

    def repondre(state):
        nb = len(state["documents"])
        return {
            "logs": ["Réponse générée"],
            "answer": f"Réponse basee sur {nb} documents",
        }

    graph = StateGraph(State)
    graph.add_node("reformuler", reformuler)
    graph.add_node("chercher", chercher)
    graph.add_node("repondre", repondre)
    graph.add_edge(START, "reformuler")
    graph.add_edge("reformuler", "chercher")
    graph.add_edge("chercher", "repondre")
    graph.add_edge("repondre", END)
    return graph.compile()

← Pipeline RAG avec FAISS Evaluer un retriever RAG →

solution.py

Bravo!

Agent LangGraph avec historique

Tests (1/1)

Indices (3 disponibles)

Solution officielle