Notes pratiques : LangGraph Memory – Le guide pratique complet pour la gestion

Les notes suivantes reconstituent une approche pratique pour aborder « LangGraph Memory : Le guide complet et pratique pour gérer ce que votre agent se souvient ». L’accent est mis sur les contrats, les vérifications et les placeholders de code à insérer, plutôt que sur une présentation motivante. Lorsque vous travaillez sur l’aperçu général, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé.

Le problème de la mémoire dont personne ne parle dans les tutoriels

Le problème de mémoire dont personne ne parle dans les tutoriels fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès et refusez toute complétion partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués cachent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Le paysage de la mémoire : quatre choses à comprendre en premier

Le Paysage de la Mémoire : Quatre choses à comprendre en premier fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un enregistrement exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Notez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Gardez l’état du graphe plat et typé ; les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

User sends message
        ↓
Graph loads State from Checkpointer (using thread_id)
        ↓
Memory Management runs (trim / filter / summarize)
        ↓
LLM receives cleaned messages
        ↓
LLM response added to State
        ↓
Checkpointer saves updated State (to memory / SQLite / Postgres)

Module 2 Étendu : L’état conscient de la mémoire

Module 2 Étendu : L’état conscient de la mémoire fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du graphe. Maintenez un état du graphe plat et typé. Les blocs imbriqués masquent l’identité du nœud qui a écrit tel champ et perturbent la reprise après interruption. Module 2 Étendu : L’état conscient de la mémoire fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé.

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

L’état standard conscient de la mémoire

Pour l’état standard conscient de la mémoire, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute finalisation partielle silencieuse. Faites approuver par un humain les cas où de l’argent est dépensé ou où des données de production sont modifiées. La connexion en temps de compilation ne revient pas à une complétude opérationnelle.

# ── MODULE 2: STATE (Memory-Aware Version) ──────────────────
from langgraph.graph import MessagesState
# Option A: The shorthand (recommended for most agents)
# MessagesState is a pre-built TypedDict with just the messages field.
# Use it when you don't need custom fields.
class State(MessagesState):
    summary: str   # The only addition needed for memory management

Les trois stratégies pour gérer la longueur des conversations

Pour les Trois stratégies pour gérer la durée des conversations, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite les factures inattendues lorsque le parcours passe d’un environnement de démonstration à des environnements partagés. Faites approuver par un humain les étapes qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

Stratégie 1 : Filtrage des messages

Pour la Stratégie 1 : Filtrage des messages, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Mettez en place une approbation humaine pour les actions qui entraînent des dépenses ou modifient des données de production. Une connexion effectuée en temps de compilation ne garantit pas la complétude du processus métier. Pour la Stratégie 1 : Filtrage des messages, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un pipeline embrouillé.

Modèle A : Filtrer uniquement ce que voit le LLM (état inchangé)

Lorsque vous travaillez avec le Modèle A : Filtrer uniquement ce que voit le LLM (état inchangé), notez d’abord les exigences : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de maintenir l’honnêteté des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux éléments générés, définez des vérifications de succès et refusez les terminaisons partielles silencieuses. Cachez les instructions système stables ainsi que les schémas des outils ; l’envoi répété d’un préambule identique est une cause fréquente de surconsommation.

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Calls the LLM with only the last 2 messages, but state keeps the full history."""

    # Filter: only send last 2 messages to the LLM
    # State is NOT modified - full history stays in the checkpoint
    recent_messages = state["messages"][-2:]

    response = llm.invoke(recent_messages)
    return {"messages": [response]}

Modèle B : RemoveMessage — Supprimer définitivement de l’état

Lorsque vous travaillez sur le schéma B : RemoveMessage — Supprimer définitivement de l’état, notez d’abord le contrat : entrées requises, signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Faites un point après les étapes coûteuses. La reprise du processus ne doit pas facturer à nouveau la même appel de LLM lorsque l’opérateur réessaie un nœud ultérieur.

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import RemoveMessage

# ── MODULE 4: NODES ──────────────────────────────────────────
def filter_messages_node(state: State) -> dict:
    """A dedicated node that removes all but the last 2 messages from state.

    RemoveMessage works by ID: you pass it the ID of a message you want deleted,
    and the add_messages reducer handles the actual deletion. This is NOT the same
    as just slicing the list - it permanently removes from the checkpoint.
    """

    # Build a list of RemoveMessage objects for every message we want to delete
    # state["messages"][:-2] = all messages EXCEPT the last 2
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {"messages": delete_messages}  # Reducer sees these and deletes them

def chat_model_node(state: State) -> dict:
    """After filtering, the state already has a short message list."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("filter", filter_messages_node)
graph_builder.add_node("chat_model", chat_model_node)
# Filter runs FIRST, then chat_model sees the cleaned state
graph_builder.add_edge(START, "filter")
graph_builder.add_edge("filter", "chat_model")
graph_builder.add_edge("chat_model", END)
graph = graph_builder.compile(checkpointer=MemorySaver())

Stratégie 2 : Élagage basé sur des tokens avec trim_messages

Lorsque vous travaillez sur la Stratégie 2 : Élagage basé sur des tokens avec trim_messages, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Mémorisez les instructions stables du système ainsi que les schémas des outils. Envoyer à nouveau un préambule identique est une cause fréquente de gaspillage. Lorsque vous travaillez sur la Stratégie 2 : Élagage basé sur des tokens avec trim_messages, notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un ensemble d’éléments embrouillés.

pipeline.

# ── IMPORTS needed ──────────────────────────────────────────
from langchain_core.messages import trim_messages

# ── MODULE 4: NODES ──────────────────────────────────────────
def chat_model_node(state: State) -> dict:
    """Uses token-based trimming to control exactly how much history the LLM sees."""

    trimmed = trim_messages(
        state["messages"],

        # Maximum number of tokens to allow (tune this to your model and budget)
        max_tokens=1000,

        # "last" = keep the MOST RECENT messages (the ones that fit within max_tokens)
        # This is almost always what you want
        strategy="last",

        # The LLM used to count tokens. Use the same model you're chatting with
        # for accurate counting. This does NOT make an API call - just counts.
        token_counter=ChatOpenAI(model="gpt-4o"),

        # False = never cut a message in half. Either include it fully or drop it.
        # Always set this to False in production to avoid feeding partial tool results.
        allow_partial=False,

        # True = always include the most recent HumanMessage even if it alone
        # exceeds max_tokens. Prevents the agent from receiving an empty context.
        include_system=True,
    )

    response = llm.invoke(trimmed)
    return {"messages": [response]}

Référence rapide des paramètres clés

La référence rapide des paramètres clés fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites des vérifications de succès et refusez toute complétion partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Stratégie 3 : Résumé basé sur des LLM (la norme en production)

Stratégie 3 : La synthèse basée sur des LLM (le standard de production) fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un enregistrement exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque l’on passe d’une démonstration à des environnements partagés. Fixez un budget de tokens par tour et par session. Les outils agents élargissent lourdement le contexte ; des plafonds stricts empêchent que les démonstrations ne se transforment en factures inattendues.

Étape 1 : Définition (Module 2)

Étape 1 : L’état (Module 2) fonctionne le mieux lorsqu’il est traité comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du graphe. Garantissez que l’état du graphe soit plat et typé. Les blocs imbriqués masquent l’identité du nœud qui a écrit tel champ, ce qui perturbe la reprise après interruption.

class State(MessagesState):
    summary: str   # Stores the running summary. Empty string by default.

Étape 2 : Les deux nœuds (Module 4)

Étape 2 : Le concept des Deux Nœuds (Module 4) fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours optimal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Gardez l’état du graphe simple et bien typé ; les blocs imbriqués masquent l’identité du nœud qui a écrit chaque champ et perturbent la reprise après interruption.

def call_model(state: State) -> dict:
    """The main agent node. Uses summary as context if it exists."""

    summary = state.get("summary", "")

    if summary:
        # If we have a summary, prepend it as a system message
        # This gives the LLM the compressed history without the raw messages
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        # No summary yet — just use the raw messages
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}
def summarize_conversation(state: State) -> dict:
    """Triggered when conversation gets long. Compresses history into a summary,
    then deletes the old raw messages to keep state lean."""

    existing_summary = state.get("summary", "")

    # Build the summarization prompt.
    # If a summary already exists, extend it with new messages.
    # This is the "rolling summary" pattern - it never reads all messages from scratch.
    if existing_summary:
        summary_message = (
            f"This is the summary of our conversation so far:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages above. "
            f"Keep it concise - focus on facts and decisions that matter for future turns."
        )
    else:
        summary_message = (
            "Create a concise summary of the conversation above. "
            "Focus on key facts, user intent, and any decisions made. "
            "This summary will replace the full message history."
        )

    # Call the LLM with all current messages + the summarization instruction
    messages = state["messages"] + [HumanMessage(content=summary_message)]
    response = llm.invoke(messages)

    # After summarizing, delete all messages except the last 2.
    # Why keep 2? So the LLM has immediate conversational context
    # (it can see what was just said) even after summarization runs.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,    # Updated summary stored in state
        "messages": delete_messages      # Old messages removed from state
    }

Étape 3 : La logique de routage (Module 5)

Étape 3 : La logique de routage (Module 5) fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

from typing import Literal
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """After the main agent responds, decide: keep going or trigger summarization?

    The threshold (6 messages here) is your tuning knob.
    Too low = you summarize frequently, losing some nuance.
    Too high = you let the context grow before compressing.
    For most production agents, 6–10 messages is a good starting range.
    """

    if len(state["messages"]) > 6:
        return "summarize_conversation"

    return "__end__"

Étape 4 : Assemblage du graphe (Module 6)

Étape 4 : Assemblage du graphe (Module 6) fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites des vérifications de succès et refusez toute complétion partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

graph_builder = StateGraph(State)

graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
# After summarization, the conversation ends for this turn
graph_builder.add_edge("summarize_conversation", END)
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

Comment cela se traduit en pratique

« Comment cela se traduit en pratique » fonctionne le mieux lorsqu’il est considéré comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Gardez l’état du graphe simple et typé ; les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Turn 1–3:   Messages accumulate. should_continue → "__end__" each time.
Turn 4:     call_model runs. len(messages) > 6 → should_continue → "summarize_conversation"
            summarize_conversation:
              - LLM reads all 7 messages, generates a 2-sentence summary
              - state["summary"] = "User is troubleshooting a Python import error..."
              - Deletes messages[:-2] (keeps only last 2 raw messages)
Turn 5:     call_model prepends summary as SystemMessage + 2 recent messages
            LLM sees: [SystemMessage(summary), HumanMessage, AIMessage, HumanMessage(latest)]
            Token cost: 4 messages instead of potentially 20+

Faire en sorte que la mémoire soit véritablement persistante : des pointeurs de vérification externes

Rendre la mémoire véritablement persistante : les pointeurs de checkpoint externes fonctionnent le mieux lorsqu’ils sont considérés comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et une note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Gardez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ, ce qui perturbe la reprise après interruption. Rendre la mémoire véritablement persistante : les pointeurs de checkpoint externes fonctionnent le mieux lorsqu’ils sont considérés comme une surface mesurable. Capturez un enregistrement idéal, un cas d’échec et une note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé.

L’interface du pointeur de checkpoint

Pour l’interface Checkpointer, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute exécution partielle silencieuse. Faites approuver par un humain les cas où de l’argent est dépensé ou des données de production sont modifiées. La connexion en temps de compilation ne revient pas à une complétude opérationnelle.

# Development: everything in RAM (lost on restart)
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

# Local / Single-machine production: persists to a .db file on disk
from langgraph_checkpoint_sqlite import SqliteSaver
import sqlite3
conn = sqlite3.connect("chat_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
# Cloud / Distributed production: persists to a Postgres database
from langgraph_checkpoint_postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@host:5432/dbname"
)
# This one line is the ONLY thing that changes between environments:
graph = graph_builder.compile(checkpointer=checkpointer)

SqliteSaver en détail

Pour une utilisation approfondie de SqliteSaver, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite des factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Faites approuver par un humain les étapes qui entraînent des dépenses ou modifient des données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

# ── Full SqliteSaver Setup ───────────────────────────────────
import sqlite3
from langgraph_checkpoint_sqlite import SqliteSaver

# File-based: state persists across restarts
conn = sqlite3.connect(
    "agent_memory.db",
    check_same_thread=False  # Required for LangGraph's multi-threaded execution
)
checkpointer = SqliteSaver(conn)
graph = graph_builder.compile(checkpointer=checkpointer)
# Usage is IDENTICAL to MemorySaver:
config = {"configurable": {"thread_id": "user-123"}}
response = graph.invoke({"messages": [HumanMessage(content="Hello")]}, config)
# Stop and restart your app. Run this again with the SAME thread_id:
# The agent picks up exactly where it left off.
response = graph.invoke({"messages": [HumanMessage(content="What did I just say?")]}, config)
# Agent: "You said 'Hello'."

Quand utiliser SqliteSaver plutôt que PostgresSaver

Pour déterminer quand utiliser SqliteSaver plutôt que PostgresSaver, il convient de définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Mettez en place une approbation humaine pour les actions qui entraînent des dépenses ou modifient des données de production. Une connexion effectuée en temps de compilation ne garantit pas la complétude du processus métier. Pour déterminer quand utiliser SqliteSaver plutôt que PostgresSaver, il convient de définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt que…

un pipeline enchevêtré.

Inspection de la mémoire : get_state et get_state_history

Lorsque vous travaillez sur l’inspection de la mémoire via get_state et get_state_history, notez d’abord les exigences : entrées requises, signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux éléments générés, définez des vérifications de succès, et refusez toute exécution partielle silencieuse. Faites un point après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel du LLM lorsque l’opérateur réessaie un nœud ultérieur.

# ── Inspecting Current State ─────────────────────────────────
config = {"configurable": {"thread_id": "user-123"}}
# Get the current state snapshot for this thread
state_snapshot = graph.get_state(config)
# What's in the snapshot?
print(state_snapshot.values["messages"])  # Current message list
print(state_snapshot.values.get("summary", ""))  # Summary, if any
print(state_snapshot.next)  # Which nodes would run next (empty list = graph is at END)
# ── Inspecting State History ─────────────────────────────────
# get_state_history gives you every checkpoint ever saved for this thread.
# Each checkpoint is a moment in time after a node ran.
for state in graph.get_state_history(config):
    print(f"Step: {state.metadata['step']}")
    print(f"Messages: {len(state.values['messages'])}")
    print(f"Next: {state.next}")
    print("---")

Assemblage final : l’agent entièrement géré en mémoire

Lorsque vous travaillez sur « Putting It All Together: The Complete Memory-Managed Agent », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Créez un point de contrôle après les étapes coûteuses. La reprise du processus ne doit pas facturer à nouveau la même appel de LLM lorsque l’opérateur réessaie un nœud ultérieur.

# ============================================================
# LANGGRAPH MEMORY-MANAGED AGENT — COMPLETE TEMPLATE
# Combines: Summarization + External Persistence (SqliteSaver)
# ============================================================

# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import os
import sqlite3
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage, AIMessage, SystemMessage,
    BaseMessage, RemoveMessage
)
from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState
from langgraph.checkpoint.memory import MemorySaver
# For production, swap MemorySaver for one of these:
# from langgraph_checkpoint_sqlite import SqliteSaver
# from langgraph_checkpoint_postgres import PostgresSaver
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── MODULE 2: STATE ─────────────────────────────────────────
class State(MessagesState):
    # summary stores the rolling compressed history.
    # It starts empty and gets written by summarize_conversation.
    summary: str

# ── MODULE 4: NODES ─────────────────────────────────────────
def call_model(state: State) -> dict:
    """Main reasoning node. Uses rolling summary for long-context efficiency."""

    summary = state.get("summary", "")

    if summary:
        system_message = SystemMessage(
            content=f"Summary of the conversation so far:\n\n{summary}"
        )
        messages = [system_message] + state["messages"]
    else:
        messages = state["messages"]

    response = llm.invoke(messages)
    return {"messages": [response]}

def summarize_conversation(state: State) -> dict:
    """Compression node. Summarizes history and prunes old messages from state."""

    existing_summary = state.get("summary", "")

    if existing_summary:
        summary_prompt = (
            f"Current conversation summary:\n{existing_summary}\n\n"
            f"Extend this summary with the new messages. "
            f"Be concise. Preserve facts, user intent, and key decisions."
        )
    else:
        summary_prompt = (
            "Summarize this conversation concisely. "
            "Focus on facts, user intent, and decisions made."
        )

    messages = state["messages"] + [HumanMessage(content=summary_prompt)]
    response = llm.invoke(messages)

    # Keep only the 2 most recent raw messages in state.
    # Everything older is now captured in the summary.
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]

    return {
        "summary": response.content,
        "messages": delete_messages
    }

# ── MODULE 5: ROUTING ────────────────────────────────────────
def should_continue(state: State) -> Literal["summarize_conversation", "__end__"]:
    """Trigger summarization when conversation exceeds 6 messages.

    Tune this threshold based on your average message length and token budget.
    For GPT-4o at ~$0.005/1K tokens, 6 messages is a practical starting point.
    """
    if len(state["messages"]) > 6:
        return "summarize_conversation"
    return "__end__"

# ── MODULE 6: GRAPH ASSEMBLY ─────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("summarize_conversation", summarize_conversation)
graph_builder.add_edge(START, "call_model")
graph_builder.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "summarize_conversation": "summarize_conversation",
        "__end__": END
    }
)
graph_builder.add_edge("summarize_conversation", END)
# ── CHECKPOINTER SELECTION ───────────────────────────────────
# Development (in-memory, lost on restart):
checkpointer = MemorySaver()
# Uncomment for local persistence (survives restarts):
# conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
# checkpointer = SqliteSaver(conn)
# Uncomment for cloud/production (requires PostgreSQL):
# checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
graph = graph_builder.compile(checkpointer=checkpointer)

# ── MODULE 7: ENTRYPOINT ─────────────────────────────────────
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "user-001"}}

    print("Memory-managed agent ready. Type 'exit' to quit, 'state' to inspect memory.\n")

    while True:
        user_text = input("You: ").strip()

        if not user_text or user_text.lower() == "exit":
            break

        # Debug command: inspect what the agent currently remembers
        if user_text.lower() == "state":
            snapshot = graph.get_state(config)
            print(f"\n[DEBUG] Messages in state: {len(snapshot.values['messages'])}")
            print(f"[DEBUG] Summary: {snapshot.values.get('summary', '(none)')}\n")
            continue

        response = graph.invoke(
            {"messages": [HumanMessage(content=user_text)]},
            config=config
        )

        print(f"Agent: {response['messages'][-1].content}\n")

Le guide de décision pour la stratégie mémoire

Lorsque vous travaillez avec le Guide de décision pour la stratégie de mémoire, notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Créez des points de contrôle après les étapes coûteuses. Le système de reprise ne doit pas facturer à nouveau la même appel d’LLM lorsque l’opérateur réessaie un nœud ultérieur. Lorsque vous travaillez avec le Guide de décision pour la stratégie de mémoire, notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’intégrité des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une responsabilité précise plutôt qu’un processus embrouillé.

La carte de référence des mots-clés mise à jour

La carte de référence des mots-clés mise à jour fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des critères de succès et refusez toute complétion partielle silencieuse. Maintenez l’état du graphe plat et typé. Les blocs imbriqués masquent le fait que tel nœud a écrit telle champ et perturbent la reprise après interruption.

Conclusion : La mémoire fait partie de l’architecture, ce n’est pas une considération secondaire

Conclusion : « La mémoire est une architecture, pas un élément ajouté ultérieurement » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Gardez l’état du graphe simple et typé ; les blocs imbriqués masquent le fait que tel nœud a modifié tel champ et perturbent la reprise après interruption.

Liste de contrôle opérationnelle

Pour la liste de contrôle opérationnelle, définissez les entrées, le responsable de chaque étape ainsi que les critères d’achèvement avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez en même temps le parcours normal et le parcours de récupération. Les tentatives de répétition, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure.

Obtenez l’approbation humaine pour les étapes qui engagent des dépenses ou modifient les données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

Rédigez un petit manuel d’utilisation : comment rotationner les clés, comment vider la file d’attente, comment annuler la dernière ingestion.

Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité et non un processus embrouillé.

Obtenez l’approbation humaine pour les étapes qui engagent des dépenses ou modifient les données de production. La connexion en temps de compilation ne garantit pas la complétude du processus métier.

Au préalable de promouvoir l’ensemble du système, figez les versions, conservez une version référence pour le parcours critique et vérifiez les étapes de réversion. Les environnements partagés nécessitent des limites de fréquence, des contrôles d’attribution et un responsable clair pour la rotation des secrets. Préférez une fiabilité solide à de brillantes démonstrations ponctuelles.

Remarque de lot pour d865d505a59e : ne pas inclure les clés du fournisseur dans le répertoire, fixer une limite pour les tokens par session, et stocker les transcriptions à côté des fichiers d’évaluation afin que les remplacements ultérieurs de modèles restent comparables.