Chatverlauf, Fakten, Workflow-Zustand und Kontrollpunkte sind vier verschiedene Speicher.

Teil 9 von 14: Getrennte Chat-Geschichte, gespeicherte Fakten und fortsetzbare Workflow-Daten

Neunte Ausgabe einer Serie von vierzehn Beiträgen zur Entwicklung eines Helpdesks, die LangChain Schritt für Schritt von der ersten Modellaufruf bis zu produktiven Arbeitsweisen führt. Spätere Beiträge wandeln das fertige System in Interviewübungen um.

In der vorherigen Ausgabe wurde die Suche im Runbook implementiert: Abfragen eines sorgfältig ausgewählten Korpus, Beibehaltung von Herkunftsmetadaten sowie Blockierung von Ratschlägen, die auf Dokumente verweisen, die die Suche nie zurückgegeben hat.

Jemand im Bereitschaftsdienst fragt, ob das System den Vorfall auch morgen „merken“ kann. Die Anfrage ist ungenau formuliert: Sollen die Chatgespräche gespeichert werden? Die Teameinstellungen? Etiketten und abgerufte Abschnitte? Eine Schreiboperation, die auf Genehmigung wartet? Jedes Zwischenfeld des Graphen? Die Leute fassen all das unter einem vagen Begriff zusammen. Jedes Element benötigt eigene Schlüssel, TTL-Werte, ACL-Einstellungen sowie Fehlerbehandlungsstrategien.

In diesem Teil werden vier Konzepte voneinander getrennt:

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

Das Übergeben früherer Nachrichten in eine statische LangChain-Ausführung ermöglicht nicht automatisch das Anhalten und Wiederaufnehmen der Verarbeitung. Dauerhafte Threads sowie Snapshots stammen vom Zustands- und Checkpointer-Modell von LangGraph.

Das aktuelle Problem

Nutzen Sie das bekannte Beispiel für Erklärungen:

Nach der Veröffentlichung um 14:05 versagen die Checkout-Aufrufe aus der EU-Region. Die Logs von checkout-api zeigen an, dass die Datenbank neue Verbindungen ablehnt.

Weisen Sie jeder Art von Datensatz ihren eigenen Schlüsselraum zu:

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

Wenn man den Identifikator eines Incidents wie einen Personenidentifikator behandelt, verschmelzen unverwandte Namensräume miteinander. Ebenso vermischen sich bei einer gemeinsamen Transkriptliste getrennte Fälle.

Hören Sie auf, von „dem Speicher“ zu sprechen

Nennen Sie den konkreten Datensatz, den Sie meinen.

Chatverlauf

Eine geordnete Liste wie diese:

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.

Die Sequenz ist tragend, wenn Sie später Anfragen aus diesen Schritten zusammenstellen.

Gespeicherte Fakten

Ausgewählte Felder wie zum Beispiel:

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

Fakten können über einen einzelnen Chat hinaus bestehen bleiben. Speichern Sie sie nur durch explizite Anwendungsregeln – nicht indem Sie jede Aussage, die das Modell erfindet, abrufen.

Ablaufzustand

Aktuelle Daten für einen Ticket-Prozess:

{
  "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"
  ]
}

Der Zustand ändert sich im Laufe der ausgeführten Schritte.

Checkpoint

Betrachten Sie einen Checkpoint als ein eingefrorenes Bild des Ablaufs zusammen mit den Buchhaltungsdaten, die der Laufzeit zum Weitermachen benötigt. Sie laden das neueste Bild eines Threads, setzen nach einer Pause fort, prüfen, was ein Schritt gesehen hat, und können nach einem Absturz wiederherstellen. Prozesslokal gespeicherte Daten verschwinden beim Beenden; für die Wiederherstellung in der Produktion ist ein durch eine Datenbank unterstützter Speicher erforderlich.

Retrieval gehört zu keiner dieser Kategorien

Der sorgfältig zusammengestellte Runbook-Index ist ein Suchkorpus. Seine Öffnung während eines Incidents wandelt die gefundenen Ergebnisse nicht in ein Chat-Verlauf ab. Auszüge sollten nicht automatisch zu dauerhaften Profilinformationen werden. Embedding-Indizes sind keine Checkpoint-Datenbanken. Isolieren Sie die Speicher, selbst wenn eine einzige HTTP-Anfrage mehrere davon betrifft.

Was eine feste Kette standardmäßig speichert

Auf unabhängigen Aufrufen speichert die Kette nichts, es sei denn, Ihre Anwendung injiziert oder speichert Kontext.

Dieser Aufruf:

result = chain.invoke(current_input)

weist automatisch keine früheren Eingaben oder Ausgaben weiter. Sie können eigene vorherige Nachrichten hinzufügen oder ältere History-Wrapper verwenden. In der hier überprüften LangChain-Version warnt RunnableWithMessageHistory und lenkt die neue Arbeit in Richtung der LangGraph-Speicherung.

Mit einer statischen Kette ist es in der Regel einfacher, den Historieverlauf im Anwendungscode zu verwalten:

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

Das ist genau das, was der Begleitcode tut.

Projektstruktur

Der Snapshot von Teil 9 enthält:

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

Installierte Pakete:

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

Das Beispiel führt keine Anrufe bei Providern durch.

Schritt 1: Chatsnachrichten nach Sitzung speichern

Erstellen Sie 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

Die Schlüssel von histories verknüpfen eine Sitzung mit einer geordneten Liste. read erstellt Kopien, sodass Aufrufer nicht durch Nebeneffekte in den Speicher eingreifen können. add_turn speichert ein Paar aus Mensch und Assistenten. prior_turn_count halbiert die Länge, da das Beispiel nur vollständige Paare speichert. Live-Transkripte enthalten außerdem Tool-Nachrichten, unvollendete Sätze sowie Fehler – in der Produktion sollte man nicht von einer ordentlichen Paarung ausgehen.

Die Historie benötigt eine Aufbewahrungsregel

Das Endlose Aufbewahren jeder Nachricht ist keine Produktfunktion. Die Richtlinien müssen angeben, was aufbewahrt werden darf, für wie lange, wer es lesen kann, welche Felder redigiert werden, wie das Löschverfahren funktioniert und wie viele Nachrichten in den nächsten Modellaufruf übergehen. Zu große Historien verschwenden Tokens und Geld. Zusammenfassungen können hilfreich sein, erzeugen aber auch Fehler – betrachten Sie sie als abgeleitete Artefakte mit klaren Herkunftsregeln.

Schritt 2: Gespeicherte Fakten getrennt aufbewahren

Erstellen Sie 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, {}))

Indexieren Sie die Fakten nach user_id, niemals nach Sitzungs- oder Vorkommniss-ID. Akzeptieren Sie nur benannte Felder; speichern Sie niemals den gesamten Transkriptinhalt unter einem Schlüssel. Echte put-Pfade benötigen Allowlists, Validierung, Autorisierung und Audit-Ereignisse. Ein Modellhinweis ist keine Genehmigung zur Dauerhaftigkeitsspeicherung.

Schritt 3: Workflow-Zustände definieren

Wechseln Sie zu einem Zustandsbasierten Workflow. TypedDict in 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 sorgt dafür, dass Felder leer bleiben, bis ein Knoten sie schreibt. Bei Audit-Ereignissen wird ein Reducer verwendet:

Annotated[list[str], add]

Wenn ein Knoten mehr Audit-Spalten zurückgibt, fügt der Reducer diese anstelle von Überschreiben zusammen. Wählen Sie die Reducer absichtlich aus – verwenden Sie „append“ für Ereignisströme und „replace“ für skalare Felder.

Schritt 4: Schreiben Sie kleine, deterministische Knoten

Teaching Graph verwendet reines Python, sodass sich das Verhalten der Checkpoints deutlich erkennen lässt:

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}"],
    }

Jeder Knoten liest den Zustand ein und gibt einen Patch zurück; er ändert das eingehende Dictionary niemals. Der Empfehlungsknoten nutzt die Klassifizierungsergebnisse:

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"],
    }

Es handelt sich dabei um gewöhnliche Python-Funktionen – in diesem Teil konzentriert man sich auf Zustand und Persistenz, nicht auf die Genauigkeit des Klassifikators.

Schritt 5: Erstellen Sie das Graph

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) verknüpft den gemeinsamen Zustand mit dem typisierten Wörterbuch. Knoten und Kanten bestimmen die Reihenfolge; compile überprüft die Struktur und verbindet die Checkpoint-Elemente miteinander. Der Ablauf bleibt linear – das Diagramm ist nur deshalb nützlich, weil Zustand und Checkpoints eine erste Klasse darstellen, nicht weil das Diagramm besonders aufwendig gestaltet ist.

Schritt 6: Geben Sie jedem Workflow eine Thread-ID zu

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

Führen Sie das Ticket aus:

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 teilt die Checkpoint-Geschichte auf. Die Wiederverwendung derselben Thread-ID bei unzusammenhängenden Vorfällen führt zu einem Zustandstransfer zwischen ihnen.

Schritt 7: Lesen Sie den gespeicherten Zustand ein

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

Die Werte enthalten:

{
  "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"
  ]
}

Jene Screenshot-Aufnahme enthält ausschließlich Workflow-Daten – sie ist weder ein Profilspeicher noch Teil des Runbook-Korpus.

Was InMemorySaver kann und nicht kann

Es behält Checkpoints nur während der Laufzeit des Python-Prozesses bei – das eignet sich gut für Unit-Tests und Notebooks. Sie überdauern keine Neustarts, erstrecken sich nicht auf Service-Replicas und erfüllen auch nicht die Anforderungen an Speicherung, Verschlüsselung oder Backup. Die aktuellen Dokumentationen empfehlen, den Speicher des Produktions-Agenten sowie die wiederaufnahmefähigen Threads für eine datenbankbasierte Speicherlösung wie Postgres zu nutzen.

Produktionsform:

from langgraph.checkpoint.postgres import PostgresSaver

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

Verbindungsgeheimnisse, Migrationsprozesse, Pools und Aufräumarbeiten bleiben in der Verantwortung der Anwendung. Halten Sie Datenbank-URIs außerhalb des kommitionierten Quellcodes.

Wo LangChain endet und LangGraph beginnt

Feststehende LangChain-Lösungen reichen aus, wenn

die Abfolge fest ist; eine einzige Anfrage abgeschlossen werden kann, ohne dass ein Mensch eingreifen muss; ein Neustart der gesamten Anfrage kein Problem darstellt; der Zustand zwischen den Schritten nicht dauerhaft gespeichert werden muss; und gewöhnlicher Anwendungscode ausreicht, um die benötigte kurze Historie zu speichern.

LangGraph ist sinnvoller, wenn

Steuerfluss-Branchen oder Schleifen über benannte Zustände; eine Person muss während des Laufs genehmigen; die Arbeit setzt später im selben Thread fort; ein Prozessneustart muss den ausstehenden Zustand beibehalten; Betreiber benötigen überprüfbare Snapshots; die Wiederherstellung sollte von einem gespeicherten Punkt aus fortgesetzt werden anstatt neu zu beginnen.

Der heutige create_agent-Hilfsfunktion gibt bereits einen auf LangGraph platzierten Agenten zurück. Ein Checkpointer wird bereitgestellt, und die Persistenz erfolgt aus diesem Laufzeitumfeld – das ist die erwartete Architektur, kein zufälliger Fehler.

Ein Checkpoint ist kein Prüfprotokoll

Checkpoints existieren, damit der Laufzeitumfeld weiterlaufen kann. Prüfprotokolle existieren, damit Sicherheits- und Geschäftsprüfer Aktionen rekonstruieren können. Manchmal teilen sie Felder, ihre Aufgaben sind jedoch unterschiedlich. Eine Prüfzeile sollte den Anfragenden, das vorgeschlagene Tool, den Genehmiger, die ausgeführten Argumente, das Ergebnis und das Zeitstempel angeben. Behandeln Sie keinen internen serialisierten Snapshot als Prüfverlauf von Compliance-Niveau.

Checkpointe und Nebenwirkungen

Ein persistierender Zustand macht eine externe Schreiboperation nicht idempotent. Wenn ein Prozess ein Ticket aktualisiert und vor dem nächsten Checkpoint abstirbt, kann die Wiederaufnahme der Ausführung das Schreiben wiederholen. Tools benötigen Idempotenzschlüssel oder Prüfungen auf bereits durchgeführte Aktionen. Nebenwirkungen sollten nach der Freigabe erfolgen; stabile Betriebs-IDs müssen vergeben werden; die Semantik von Wiederholungsversuchen muss dokumentiert sein. Die nächste Ausführungspause erfolgt vor dem Schreiben des Tickets, wobei entschieden wird, ob die Aktion genehmigt oder abgelehnt wird.

Prüfung der Trennung

Drei Offline-Tests im begleitenden Snapshot.

Nachrichtenhistorien bleiben getrennt

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

Gespeicherte Fakten sind keine Chatnachrichten

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

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

Checkpointe bleiben pro Ticket-Thread getrennt

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"

Ausführen:

pytest -q

Erwartet:

3 passed

Sie belegen die Grenzen der Namensräume sowie die Isolation von Threads – sie belegen jedoch nicht die Zuverlässigkeit der Datenbank bei Verwendung von InMemorySaver.

Häufige Fehler

Eine globale Historieliste

Dadurch können unzusammenhängende Benutzer oder Vorfälle aufeinandertreffen. Kennzeichnen Sie die Historie stets mit einem authentifizierten, eng gefassten Identifikator.

Speichern jeder Modellanweisung als Fakt

Modelle erfinden fälschlicherweise selbstständig. Speichern Sie nur genehmigte Felder über einen validierten Schreibweg.

Verstecken von Geheimnissen im Zustand

Snapshots werden kopiert, überprüft und aufbewahrt. Legen Sie Geheimnisse in einem Tresor ab und übergeben Sie stattdessen Referenzen.

Nutzung von thread_id als Autorisierung

Eine Thread-ID findet den Zustand – sie belegt jedoch nie, dass der Aufrufer ihn lesen darf. Autorisieren Sie separat.

Einen Vektor-Speicher als „langfristiges Gedächtnis“ bezeichnen

Der Slogan verschleiert das Eigentumsverhältnis und die Löschung. Nennen Sie die Datensätze, Autoren, Abfragen sowie die Löschrichtlinien.

Erwartung eines Checkpoints zur Behebung eines Fehlers

Snapshots bewahren alles, was geschrieben wurde – einschließlich Fehler. Validierung und Tests bleiben weiterhin erforderlich.

Ergebnis von Teil 9

Vier benannte Speichergrenzen:

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

Eine statische Kette eignet sich weiterhin für Ein-Pass-Aufgaben. LangGraph bietet eine klarere Struktur, sobald dauerhafter Zustand, Pause, Wiederaufnahme oder Wiederherstellung erforderlich sind. Danach folgt die erste echte Entscheidung des Agents – Tools mit nur Lesezugriff können automatisch ausgeführt werden; Änderungen an Tickets warten auf eine menschliche Überprüfung.

Dokumentationsüberprüfung: Vergleich mit den LangChain-Dokumenten zur Kurzzeitgedächtnisfunktion sowie den LangGraph-Dokumenten zur Persistenz am 26. August 2026. Die API-Strukturen der Pakete haben sich geändert.

Weitere Literatur: LangChain Kurzzeitgedächtnis, LangChain Agents, LangGraph Persistenz.

Produktions-Helpdesk-Systeme benötigen in der Regel alle vier Speicherformen gleichzeitig: einen für die aktuelle Ingenieurin/einen aktuellen Ingenieur gesicherten Chat-Puffer, einen für den Benutzer bestimmten Faktspeicher für dauerhafte Einstellungen, einen für den Thread bestimmten Workflow-Zustand für das Ticket-Graph sowie einen suchbaren Runbook-Index, der niemals gleichzeitig als Historie oder Checkpoint dient. Die Benennung dieser Grenzen in Code-Reviews verhindert den klassischen Umweg, alles in eine einzige Redis-Liste mit dem Namen „Memory“ zu packen. Beim Einführen eines neuen Teammitglieds bitten Sie ihn/sie, die vier Kästchen zu zeichnen und die Schlüssel zu beschriften – wenn das nicht möglich ist, ist das Design noch nicht bereit für Unterbrechung/Wiederaufnahme.

Wenn später die menschliche Freigabe eingeführt wird (Teil 10), wird der Checkpoint zum Ort, an dem der Workflow wartend pausiert. Die Chat-Historie läuft unabhängig weiter, sodass der Ingenieur klärende Fragen stellen kann, ohne die ausstehende Schreiboperation zu verändern. Fakten bleiben außerhalb des Unterbrechungspfades, es sei denn, eine explizite Regel kopiert ein Feld. Genau diese Trennung verhindert, dass „Nach der Mittagspause fortsetzen“ zu „Die gesamte Konversation in einem Ticket-Update abspielen“ wird.

Mischung von Speicherrichtlinien in verschiedenen Systemen

Chat-Transkripte, dauerhafte Fakten, Workflow-Checkpoints sowie Runbook-Einbettungen teilen fast nie denselben Speicherrhythmus. Ihr Abstimmen „aus Gründen der Einfachheit“ verstößt in der Regel entweder gegen Anfragen zur Löschung aus Gründen des Datenschutzes oder gegen die Notwendigkeit, Vorfälle nachzuspielen. Dokumentieren Sie vier Zeitpunkte, vier Verantwortliche sowie vier Löschpunkte – selbst wenn zwei davon derzeit auf dieselbe Redis-Instanz verweisen.

Auslaufwarnungen als optional behandeln

Wenn die Bibliothek warnt, dass History-Wrapper auf LangGraph-Persistenz umgestellt werden, sollte man dies als Designsignal betrachten. Die Einführung einer neuen Helpdesk-Funktion über den veralteten Weg führt später zu einer Neuimplementierung innerhalb der Frist. Wählen Sie das Checkpointer-Modell für jeden Ablauf, der möglicherweise pausieren könnte.

Vergessen, dass Reducer Teil des Schemas sind

Teams diskutieren stundenlang über Feldnamen und fügen anschließend beiläufig einen Append-Reducer zu einem Feld hinzu, das eigentlich ersetzt werden sollte. Der Fehler tritt Wochen später in Form von doppelten Klassifizierungen oder gelöschten Prüfungszeilen auf. Überprüfen Sie Reducer im selben PR wie den TypedDict.