L’historique de chat, les faits, l’état du flux de travail et les points de contrôle sont quatre types de données distincts.

Partie 9 sur 14 : Historique de chat distinct, faits enregistrés et données de flux de travail reprendables

Dixième article d’une série de quatorze pour construire un service d’assistance qui guide LangChain, depuis sa première appel à un modèle jusqu’à des pratiques de production. Les articles suivants transforment le système final en exercices d’entretien.

L’article précédent a mis en place la recherche dans le manuel de procédures : interroger un corpus sélectionné, conserver des métadonnées sur l’origine des informations, et bloquer les conseils citant des documents que la recherche n’a jamais renvoyés.

Quelqu’un en service demande si le système peut « se souvenir » de l’incident demain. La question est trop vague. Faut-il conserver les échanges de chat ? Les préférences de l’équipe ? Les étiquettes et les extraits récupérés ? Un enregistrement en attente d’approbation ? Chaque champ intermédiaire du graphe ? Les gens regroupent tout cela sous une étiquette floue. Chacun nécessite sa propre clé, son TTL, ses ACL et sa politique de défaillance.

Cette partie sépare quatre idées distinctes :

chat history
  ordered messages for one conversation

saved facts
  selected application data about a user or accountworkflow state
  the current named values for one runcheckpoint
  a saved snapshot of workflow state that can be loaded later

Transférer des messages antérieurs dans un exécutable statique de LangChain ne permet pas magiquement de reprendre l’exécution après une pause. Les threads durables et les snapshots proviennent du modèle d’état et de checkpointer de LangGraph.

Le problème en cours

Reutilisons l’incident connu comme exemple :

Après la mise à jour de 14:05, les appels checkout depuis la région de l’UE échouent. Les journaux de checkout-api indiquent que la base de données a refusé de nouvelles connexions.

Affectez à chaque type de fichier son propre espace de clés :

chat session:    chat:INC-2048
user facts:      user-17
workflow thread: ticket:INC-2048

Considérer l’identifiant de l’incident comme s’il s’agissait d’un identifiant personnel entraîne la fusion de noms d’espace non liés. De même, une seule liste de transcriptions partagées mélange des cas distincts.

Tout d’abord, arrêtez de dire « la mémoire »

Donnez un nom au fichier que vous désignez.

Histoire de chat

Une liste ordonnée telle que :

human: The failure began after 14:05.
assistant: I recorded the start time.
human: The failed requests are only in the EU region.
assistant: I added the affected region to the investigation context.

La séquence est porteuse de charge lorsque vous assemblez ultérieurement des prompts à partir de ces tours.

Faits enregistrés

Champs sélectionnés tels que :

{
  "team": "commerce-platform",
  "timezone": "America/Los_Angeles"
}

Les faits peuvent survivre au-delà d’une seule conversation. Conservez-les uniquement grâce à des règles explicites de l’application — et non en extrayant chaque affirmation inventée par le modèle.

État du flux de travail

Données actuelles pour un processus de ticket :

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

L’état change au fur et à mesure que les étapes s’exécutent.

Point de contrôle

Considérez un point de contrôle comme une image figée du flux de travail ainsi que les données comptables nécessaires pour poursuivre l’exécution. Vous chargez la dernière image d’un thread, reprenez après une pause, examinez ce qu’une étape a vu, et vous récupérez en cas de panne. Les mécanismes d’enregistrement locaux disparaissent à la fermeture ; pour une récupération en production, un mécanisme d’enregistrement basé sur une base de données est nécessaire.

La récupération n’est rien de tout cela

L’index du guide opérationnel curé constitue un corpus de recherche. L’ouvrir lors d’un incident ne transforme pas les résultats en historique de chat. Les extraits ne doivent pas être automatiquement promus en éléments permanents du profil. Les index d’embeddings ne sont pas des bases de données de points de contrôle. Isolez les stockages, même lorsque une seule requête HTTP en touche plusieurs.

Ce que se souvient par défaut une chaîne fixe

Au cours d’appels indépendants, la chaîne ne se souvient de rien à moins que votre application n’injecte ou ne persiste du contexte.

Cette appel :

result = chain.invoke(current_input)

ne transmet pas automatiquement les entrées ou sorties précédentes. Vous pouvez ajouter vous-même les échanges antérieurs, ou utiliser des enveloppes d’historique plus anciennes. Dans la version de LangChain examinée ici, RunnableWithMessageHistory émet des avertissements et oriente le nouveau travail vers la persistance dans LangGraph.

Avec une chaîne statique, il est généralement plus simple de gérer l’historique dans le code de l’application :

read permitted messages
  -> select the messages needed for this request
  -> call the chain
  -> store the new turn under the correct session ID

C’est exactement ce que fait le code d’accompagnement.

Structure du projet

L’aperçu de la partie 9 contient :

langchain-helpdesk/
├── app.py
├── checkpoint_graph.py
├── facts.py
├── history.py
└── tests/
    └── test_state.py

Installation des paquets :

python -m pip install -U langchain-core langgraph pydantic pytest

L’exemple ne effectue aucune appel au fournisseur.

Étape 1 : Stocker les messages de chat par session

Créer history.py :

from dataclasses import dataclass, field

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
)
@dataclass
class ChatHistoryStore:
    histories: dict[str, list[BaseMessage]] = field(
        default_factory=dict
    )    def read(self, session_id: str) -> list[BaseMessage]:
        return list(self.histories.get(session_id, []))    def add_turn(
        self,
        session_id: str,
        user_text: str,
        reply_text: str,
    ) -> None:
        history = self.histories.setdefault(session_id, [])
        history.extend(
            [
                HumanMessage(content=user_text),
                AIMessage(content=reply_text),
            ]
        )    def prior_turn_count(self, session_id: str) -> int:
        return len(self.histories.get(session_id, [])) // 2

Les clés histories associent chaque session à une liste ordonnée. La fonction read crée des clones afin que les appels ne puissent pas modifier le stockage par effet secondaire. La fonction add_turn enregistre une paire utilisateur/assistant. La variable prior_turn_count divise par deux la longueur de la liste, car l’exemple ne stocke que des paires complètes. Les transcriptions en direct contiennent également des messages d’outil, des tours inachevés et des erreurs — ne supposez pas un empareillage parfait en environnement de production.

L’historique nécessite une règle de conservation

Conserver chaque étape indéfiniment n’est pas une fonctionnalité du produit. La politique doit préciser ce qui peut être conservé, pendant combien de temps, qui peut y accéder, quels champs doivent être masqués, comment s’effectue la suppression, et combien d’étapes sont transmises à la prochaine invocation du modèle. Des historiques trop volumineux consomment des tokens et de l’argent. Les résumés peuvent être utiles mais génèrent aussi des erreurs — traitez-les comme des artefacts dérivés soumis à des règles de provenance claires.

Étape 2 : Stocker les faits sélectionnés séparément

Créez facts.py :

from dataclasses import dataclass, field

@dataclass
class UserFactsStore:
    records: dict[str, dict[str, str]] = field(
        default_factory=dict
    )    def put(self, user_id: str, key: str, value: str) -> None:
        self.records.setdefault(user_id, {})[key] = value    def get(self, user_id: str) -> dict[str, str]:
        return dict(self.records.get(user_id, {}))

Indexez les faits par user_id, jamais par identifiant de session ou d’incident. Ne prenez en compte que des champs nommés ; ne stockez jamais l’intégralité d’un enregistrement sous une seule clé. Les vraies chemins put nécessitent des listes d’autorisation, une validation, un mécanisme d’authentification et des événements de contrôle. Un indice fourni par le modèle ne constitue pas une autorisation à le conserver.

Étape 3 : Définir l’état du flux de travail

Passez à un flux de travail à états. Utilisez TypedDict dans checkpoint_graph.py :

from operator import add
from typing import Annotated, TypedDict

class TicketWorkflowState(TypedDict, total=False):
    ticket_id: str
    details: str
    classification: str
    recommendation: str
    audit: Annotated[list[str], add]

total=False permet aux champs de rester manquants jusqu’à ce qu’un nœud les écrive. Les événements d’audit utilisent un réducteur :

Annotated[list[str], add]

Lorsqu’un nœud renvoie plusieurs lignes d’audit, le réducteur les concatène plutôt que de les écraser. Choisissez les réducteurs en fonction de vos besoins : utilisez append pour les flux d’événements et replace pour les champs scalaires.

Étape 4 : Créer des nœuds petits et déterministes

Teaching graph utilise du Python pur, ce qui rend le comportement des points de contrôle visible :

def classify_node(state: TicketWorkflowState) -> TicketWorkflowState:
    details = state["details"].lower()

    if "database" in details or "connection refused" in details:
        category = "database"
    elif "access" in details or "role" in details:
        category = "access"
    else:
        category = "unknown"    return {
        "classification": category,
        "audit": [f"classified:{category}"],
    }

Chaque nœud lit l’état et renvoie une modification ; il ne modifie jamais le dictionnaire reçu. Le nœud de recommandation utilise les résultats de classification :

def recommend_node(state: TicketWorkflowState) -> TicketWorkflowState:
    category = state["classification"]

    if category == "database":
        recommendation = (
            "Compare database settings with the last good release."
        )
    elif category == "access":
        recommendation = (
            "Confirm the requested role and current access policy."
        )
    else:
        recommendation = "Ask a person to classify the ticket."    return {
        "recommendation": recommendation,
        "audit": ["recommendation_created"],
    }

Il s’agit de fonctions Python ordinaires — cette partie se concentre sur l’état et la persistance, et non sur la précision du classificateur.

Étape 5 : Construire le graphe

from langgraph.graph import END, START, StateGraph

def build_checkpointed_graph(checkpointer=None):
    builder = StateGraph(TicketWorkflowState)
    builder.add_node("classify", classify_node)
    builder.add_node("recommend", recommend_node)
    builder.add_edge(START, "classify")
    builder.add_edge("classify", "recommend")
    builder.add_edge("recommend", END)    return builder.compile(
        checkpointer=checkpointer or InMemorySaver()
    )

StateGraph(TicketWorkflowState) lie l’état partagé au dictionnaire de type défini. Les nœuds et les arêtes déterminent l’ordre ; compile valide la structure et connecte les points de contrôle. Le parcours reste linéaire — le graphique est utile parce que l’état et les points de contrôle sont traités en premier plan, et non parce que le diagramme est sophistiqué.

Étape 6 : Donner un identifiant de thread à chaque flux de travail

def thread_config(thread_id: str) -> dict[str, dict[str, str]]:
    return {"configurable": {"thread_id": thread_id}}

Exécuter la tâche :

config = thread_config("ticket:INC-2048")

result = graph.invoke(
    {
        "ticket_id": "INC-2048",
        "details": (
            "checkout-api reports database connection refused"
        ),
        "audit": ["ticket_received"],
    },
    config,
)

thread_id permet de partitionner l’historique des points de contrôle. Réutiliser un thread pour des incidents non liés entraîne une fuite d’état entre eux.

Étape 7 : Lire l’état enregistré

snapshot = graph.get_state(config)
print(snapshot.values)

Les valeurs contiennent :

{
  "ticket_id": "INC-2048",
  "details": "checkout-api reports database connection refused",
  "classification": "database",
  "recommendation": "Compare database settings with the last good release.",
  "audit": [
    "ticket_received",
    "classified:database",
    "recommendation_created"
  ]
}

Cette capture n’est que des données de flux de travail — ce n’est ni un stockage de profils ni le corpus du guide d’exécution.

Ce que InMemorySaver peut et ne peut pas faire

Il conserve les points de contrôle uniquement pendant la durée de vie du processus Python — ce qui convient bien aux tests unitaires et aux notebooks. Ils ne survivront pas à une redémarrage, ne s’étendront pas aux réplicas du service, et ne répondront pas aux exigences de conservation, de chiffrement ou de sauvegarde. La documentation actuelle oriente la mémoire de l’agent en environnement de production ainsi que les threads reprendables vers un système de sauvegarde basé sur une base de données, comme Postgres.

Format pour l’environnement de production :

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = build_checkpointed_graph(checkpointer)

Les secrets de connexion, les migrations, la gestion des ressources en commun et le nettoyage restent des responsabilités de l’application. Évitez d’inclure les URI de la base de données dans le code source commité.

Où LangChain s’arrête et où commence LangGraph

LangChain suffit lorsqu’

la séquence est fixe ; une seule requête peut être terminée sans pause humaine ; il est possible de redémarrer toute la requête ; l’état à mi-étape n’a pas besoin d’être persistant ; du code d’application ordinaire peut conserver l’historique limité dont vous avez besoin.

LangGraph est plus adapté lorsque

Les branches ou boucles de flux de contrôle s’effectuent sur des états nommés ; une personne doit approuver en cours d’exécution ; le travail se poursuit ultérieurement sur la même thread ; un redémarrage du processus doit conserver l’état en attente ; les opérateurs ont besoin de captures d’écran inspectables ; la récupération doit reprendre à partir d’un point enregistré plutôt que de recommencer depuis zéro.

L’aide-mémoire create_agent actuel renvoie déjà un agent installé sur LangGraph. Un point de contrôle est fourni, et la persistance s’effectue à partir de ce runtime — c’est l’architecture prévue, et non une fuite accidentelle.

Un point de contrôle n’est pas un journal d’audit

Les points de contrôle existent afin que le runtime puisse se poursuivre. Les journaux d’audit existent pour permettre aux responsables de la sécurité et aux auditeurs commerciaux de reconstituer les actions. Ils partagent parfois des champs, mais leurs finalités diffèrent. Une ligne de journal d’audit doit indiquer le demandeur, l’outil proposé, l’approuveur, les arguments exécutés, le résultat et l’heure. Ne considérez pas une capture d’écran serialisée interne comme un journal d’audit de niveau conformité.

Points de contrôle et effets secondaires

L’état persistant ne rend pas une écriture externe idempotente. Si le processus met à jour un ticket puis s’arrête avant le prochain point de contrôle, la reprise peut répéter l’écriture. Les outils ont besoin de clés d’idempotence ou de vérifications « déjà appliqué ». Placez les effets secondaires après l’approbation ; ajoutez des étiquettes indiquant une opération stable ; documentez les règles de tentative répétée. La prochaine étape consiste à placer le processus en attente avant l’écriture du ticket et à déterminer si elle doit être approuvée ou rejetée.

Tester la séparation

Trois tests hors ligne dans le snapshot associé.

Les historiques de messages restent séparés

history.add_turn("chat:first", "First note", "First reply")
history.add_turn("chat:first", "Second note", "Second reply")
history.add_turn("chat:second", "Other ticket", "Other reply")

assert history.prior_turn_count("chat:first") == 2
assert history.prior_turn_count("chat:second") == 1

Les faits enregistrés ne sont pas des messages de chat

facts.put("user-17", "team", "commerce-platform")

assert facts.get("user-17") == {
    "team": "commerce-platform"
}
assert history.read("user-17") == []

Les points de contrôle restent séparés par fil de ticket

first, first_config = run_ticket(
    graph,
    "INC-2048",
    "checkout-api reports database connection refused",
)
second, second_config = run_ticket(
    graph,
    "INC-2050",
    "identity-api denied an access role request",
)

assert graph.get_state(first_config).values["ticket_id"] == "INC-2048"
assert graph.get_state(second_config).values["ticket_id"] == "INC-2050"

Exécuter :

pytest -q

Résultat attendu :

3 passed

Ils prouvent les limites des espaces de noms et l’isolation des threads — ils ne prouvent pas la durabilité de la base de données lors de l’utilisation de InMemorySaver.

Erreurs courantes

Une seule liste d’historique global

Cela entraîne des collisions entre des utilisateurs ou des incidents non liés. Identifiez toujours l’historique à l’aide d’un identifiant authentifié et au champ d’application restreint.

Enregistrement de chaque instruction de modèle en tant que fait

Les modèles inventent avec confiance. Ne persistez que les champs autorisés via un chemin d’écriture validé.

Stockage de secrets dans l’état

Les captures d’écran sont copiées, examinées et conservées. Conservez les secrets dans un coffre-fort et transmettez plutôt des références.

Utilisation de thread_id comme autorisation

Dénomination d’un stockage vectoriel « mémoire à long terme »

Ce slogan cache la question de la propriété et du effacement des données. Il faut indiquer les enregistrements, les auteurs, le chemin d’accès et la politique de suppression.

On espère qu’un point de contrôle permettra de corriger une erreur

Les captures d’écran conservent tout ce qui a été écrit, y compris les erreurs. La validation et les tests restent obligatoires.

Résultat de la partie 9

Quatre limites de stockage nommées :

session ID -> ordered chat messages
user ID    -> selected saved facts
thread ID  -> current workflow state
checkpoint -> persisted workflow snapshot

Une chaîne statique convient encore aux tâches à un seul passage. LangGraph offre une structure plus claire lorsque l’on a besoin d’un état durable, de la possibilité de pause, de reprise ou de récupération. Ensuite vient la première véritable décision de l’agent : les outils en lecture seule peuvent s’exécuter automatiquement ; les mutations des tickets nécessitent une intervention humaine.

Vérification de la documentation : examinée par rapport aux documents sur la mémoire à court terme de LangChain et à la persistance de LangGraph le 26 août 2026. Les API des packages changent.

Lectures complémentaires : mémoire à court terme de LangChain, agents de LangChain, persistance de LangGraph.

Les systèmes de helpdesk de production ont généralement besoin des quatre types de bases de données en même temps : un buffer de chat dédié à la session pour l’ingénieur actif, une base de données de préférences persistantes dédiée à l’utilisateur, un état de flux de travail dédié au fil de discussion pour représenter le graphe des tickets, ainsi qu’un index de manuel opérationnel consultable qui ne sert jamais à conserver l’historique ni les points de contrôle. Définir clairement ces limites lors des revues de code empêche l’erreur courante consistant à tout stocker dans une seule liste Redis intitulée « memory ». Lors de l’intégration d’un nouveau collègue, demandez-lui de dessiner ces quatre cases et d’étiqueter les clés associées ; s’il n’y parvient pas, le système n’est pas prêt pour la fonction de reprise après interruption.

Lorsque vous introduisez par la suite l’approbation humaine (Partie 10), le point de contrôle devient l’endroit où le flux de travail est mis en pause en attendant. L’historique des conversations se poursuit indépendamment, permettant ainsi à l’ingénieur de poser des questions pour obtenir plus d’éclaircissements sans modifier l’écriture en attente. Les faits ne font pas partie du chemin d’interruption, sauf si une règle explicite copie un champ. C’est cette séparation qui empêche que le « reprise après déjeuner » ne se transforme en « rejouer toute la conversation dans une mise à jour de ticket ».

Mélanger les politiques de conservation entre différents stockages

Les transcriptions de conversations, les faits durables, les points de contrôle du flux de travail et les embeddings des manuels opérationnels n’ont presque jamais le même mécanisme de conservation. Les aligner « par simplicité » viole généralement soit les demandes de suppression pour des raisons de confidentialité, soit les besoins de rejouement d’incidents. Documentez quatre mécanismes de conservation, quatre responsables et quatre points d’entrée pour la suppression — même si deux d’entre eux pointent actuellement vers la même instance Redis.

Considérer les avertissements de dépréciation comme optionnels

Lorsque la bibliothèque avertit que les enveloppes historiques sont transférées vers la persistance LangGraph, interprétez cela comme un signal de conception. Intégrer une nouvelle fonctionnalité d’assistance sur le chemin déprécié entraînera une réécriture sous deadline plus tard. Préférez le modèle checkpointer pour tout flux susceptible de s’arrêter.

Oublier que les reducers font partie du schéma

Les équipes débattent des noms de champs pendant des heures, puis ajoutent négligemment un reducer d’ajout à un champ qui devrait être remplacé. La panne apparaît des semaines plus tard sous forme de classifications dupliquées ou de lignes d’audit effacées. Vérifiez les reducers dans la même PR que le TypedDict.