Praktische Hinweise: LangGraph Memory – Der umfassende Praxisleitfaden zur Verwaltung

Die folgenden Notizen skizzieren einen praktischen Ansatz zu „LangGraph Memory: Der umfassende Praxisleitfaden zur Verwaltung dessen, was Ihr Agent sich merkt“. Der Schwerpunkt liegt auf Verträgen, Überprüfungen und Code-Platzhaltern statt auf motivierenden Formulierungen. Während Sie sich den Überblick ansehen, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufschema.

Das Gedächtnisproblem, über das in den Tutorials nicht gesprochen wird

Das in den Tutorials nicht besprochene Speicherproblem funktioniert am besten, wenn es als messbare Struktur betrachtet wird. Erfassen Sie einen gelungenen Fall, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand der Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

Das Speichermodell: Vier Dinge, die man zuerst verstehen muss

„The Memory Landscape: Four Things to Understand First“ funktioniert am besten, wenn es als messbare Oberfläche betrachtet wird. Erfassen Sie zunächst einen erfolgreichen Fall, einen Fehlerfall sowie die Rollback-Anmerkungen, bevor Sie den Umfang erweitern. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Einsatzbereich von einer Demo auf gemeinsam genutzte Umgebungen verschiebt. Halten Sie den Zustand der Grafik flach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

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)

Modul 2 erweitert: Der memorienerfassende Zustand

Modul 2 Erweitert: Der memory-aware State funktioniert am besten, wenn er als messbare Struktur betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zum Rollback. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne Durchsicht des gesamten Graphen prüfen können. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Ausführung. Modul 2 Erweitert: Der memory-aware State funktioniert am besten, wenn er als messbare Struktur betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zum Rollback. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

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

Der Standard memory-aware State

Für den Standard-Zustand mit Berücksichtigung des Speichers sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Kompilierzeitliche Verbindungen entsprechen nicht der Geschäftsabschlussqualität.

# ── 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

Die drei Strategien zur Steuerung der Gesprächslänge

Zu den Drei Strategien zur Steuerung der Länge von Gesprächen: Definieren Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Tokens oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung verschiebt. Setzen Sie menschliche Freigabe für Schritte ein, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

Strategie 1: Nachrichtenfiltern

Für Strategie 1: Nachrichtenfilterung – definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchlesen zu müssen. Setzen Sie menschliche Freigabe für Schritte ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung. Für Strategie 1: Nachrichtenfilterung – definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

Muster A: Nur das Filtern dessen, was das LLM sieht (Zustand unverändert)

Beim Arbeiten nach Muster A: Nur das Filtern dessen, was das LLM sieht (Zustand unverändert), sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Cachen Sie stabile Systemanweisungen sowie Tool-Schemata – das erneute Senden identischer Vorlagen ist eine häufige Ursache für Ressourcenverschwendung.

# ── 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]}

Muster B: RemoveMessage – Dauerhaft aus dem Zustand löschen

Beim Arbeiten am Muster B: RemoveMessage – Permanente Löschung aus dem Zustand – sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Legen Sie außerdem Zwischenkontrollpunkte nach aufwändigen Schritten fest – das Fortsetzen des Vorgangs sollte keine doppelten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

# ── 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())

Strategie 2: Token-basiertes Kürzen mit trim_messages

Beim Arbeiten mit Strategie 2: Token-basiertes Trimmen mithilfe von trim_messages sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie Konfigurationen außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Systems überprüfen können. Cachen Sie stabile Systemanweisungen sowie Tool-Schemata. Das erneute Senden identischer Vorlagen ist eine häufige Ursache für Ressourcenverschwendung. Beim Arbeiten mit Strategie 2: Token-basiertes Trimmen mithilfe von trim_messages sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes System.

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

Schnellreferenz zu den Schlüsselparametern

Die Schnellreferenz zu den Schlüsselparametern funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie ein optimales Transkript, einen Fehlerfall sowie die Notizen zum Rollback, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand des Graphen einfach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

Strategie 3: Zusammenfassung mithilfe von LLMs (Der Produktionsstandard)

Strategie 3: Zusammenfassung auf Basis von LLMs (der Produktionsstandard) funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie ein optimales Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Erfassen Sie außerdem die Laufzeiten sowie die Kosten pro Token oder Abfrage zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatz von Demoumgebungen in gemeinsam genutzte Umgebungen wechselt. Legen Sie Budgets für Tokens pro Turnus und pro Sitzung fest. Agentenbasierte Tools erweitern den Kontext stark; feste Obergrenzen verhindern, dass Demos zu unerwarteten Rechnungen werden.

Schritt 1: Definition (Modul 2)

Schritt 1: Der State (Modul 2) funktioniert am besten, wenn er als messbare Oberfläche behandelt wird. Erfassen Sie einen erfolgreichen Fall, einen Fehlerfall sowie die Rollback-Anmerkung, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne das Durchlesen des gesamten Graphen überprüfen können. Halten Sie den State des Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Wiederaufnehmen der Verarbeitung.

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

Schritt 2: Die beiden Knoten (Modul 4)

Schritt 2: „Die beiden Knoten“ (Modul 4) funktionieren am besten, wenn sie als messbare Oberfläche betrachtet werden. Erfassen Sie vor der Erweiterung des Umfangs einen erfolgreichen Ablauf, einen Fehlerfall sowie die Notizen zur Rücksetzung. Dokumentieren Sie den erfolgreichen Ablauf sowie den Wiederherstellungsablauf gemeinsam. Wiederholversuche, menschliche Kontrollen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen. Halten Sie den Zustand des Graphen einfach und typisiert – verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen des Ablaufs.

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
    }

Schritt 3: Die Routing-Logik (Modul 5)

Schritt 3: Die Routing-Logik (Modul 5) funktioniert am besten, wenn sie als messbare Struktur betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein optimales Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Ziehen Sie kleine, testbare Einheiten vor umfangreichen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Halten Sie den Zustand des Graphen einfach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Wiederaufnehmen der Ausführung.

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__"

Schritt 4: Graphen-Aufbau (Modul 6)

Schritt 4: Graphen-Aufbau (Modul 6) funktioniert am besten, wenn er als messbare Oberfläche betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein erfolgreiches Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand des Graphen einfach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

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)

Wie das in der Praxis aussieht

„How This Plays Out in Practice“ funktioniert am besten, wenn es als messbare Struktur betrachtet wird. Erfassen Sie zunächst ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Einsatzbereich von einer Demo auf gemeinsame Umgebungen verschiebt. Halten Sie den Zustand der Grafik einfach und typisiert – verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

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+

Echte Persistenz der Speicherung schaffen: Externe Checkpointer

Erinnerungen wirklich persistent machen: Externe Checkpointer funktionieren am besten, wenn sie als messbare Struktur betrachtet werden. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreiber ohne das Durchlesen des gesamten Graphen überprüfen können. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Störungen nach Unterbrechungen. Erinnerungen wirklich persistent machen: Externe Checkpointer funktionieren am besten, wenn sie als messbare Struktur betrachtet werden. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

Die Checkpointer-Schnittstelle

Für die Checkpointer-Schnittstelle sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Setzen Sie menschliche Freigabe bei Vorgängen ein, die Geld kosten oder Produktionsdaten ändern. Kompilierzeitliche Verbindungen entsprechen nicht der Geschäftsabschlussqualität.

# 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 im Detail

Für eine detaillierte Nutzung von SqliteSaver sollten Sie die Eingabedaten, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Erfassen Sie die Laufzeiten sowie die Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Setzen Sie menschliche Freigabe für Schritte ein, die Geld kosten oder Produktionsdaten verändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

# ── 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'."

Wann SqliteSaver statt PostgresSaver verwenden?

Für die Entscheidung, wann SqliteSaver statt PostgresSaver verwendet werden soll, sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. Setzen Sie menschliche Freigabe für Kanten ein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung. Für die Entscheidung, wann SqliteSaver statt PostgresSaver verwendet werden soll, sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen anstatt auf mehrere.

an einem verwickelten Pipeline.

Überprüfung des Speichers: get_state und get_state_history

Beim Arbeiten mit der Überprüfung des Speichers: get_state und get_state_history sollte man zunächst einen Vertrag aufstellen: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Erstellen Sie Zwischenpunkte nach aufwändigen Schritten. Die Wiederaufnahme sollte keine doppelten LLM-Aufrufe tätigen, wenn ein Operator einen späteren Knoten erneut versucht.

# ── 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("---")

Alles zusammenfassen: Der vollständige memory-gesteuerte Agent

Beim Arbeiten an „Putting It All Together: The Complete Memory-Managed Agent“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Legen Sie nach aufwändigen Schritten einen Checkpoint an. Das Wiederaufnehmen der Ausführung sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

# ============================================================
# 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")

The Memory Strategy Decision Guide

Beim Arbeiten mit dem „The Memory Strategy Decision Guide“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne das Durchlesen des gesamten Systems überprüfen können. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Betreuer einen späteren Schritt erneut ausführt. Beim Arbeiten mit dem „The Memory Strategy Decision Guide“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Ziehen Sie kleine, testbare Einheiten vor großen Skripten vor. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich verweisen und nicht auf ein verworrenes Ablaufschema.

Die aktualisierte Schlüsselwort-Referenzkarte

Die aktualisierte Schlüsselwort-Referenzkarte funktioniert am besten, wenn sie als messbarer Rahmen betrachtet wird. Erfassen Sie eine gelungene Transkription, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Halten Sie den Zustand der Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung.

Fazit: Speicher ist Architektur – kein Nachgedanke

Fazit: „Memory Is Architecture, Not an Afterthought“ funktioniert am besten, wenn es als messbare Größe betrachtet wird. Erfassen Sie eine „goldene“ Transkription, einen Fehlerfall sowie die Rollback-Anmerkungen, bevor Sie den Umfang erweitern. Notieren Sie die Laufzeiten sowie die Kosten für Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsam genutzte Umgebungen übergeht. Halten Sie den Zustand der Graphen einfach und typisiert – verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Wiederaufnehmen des Ablaufs.

Operative Checkliste

Für die operative Checkliste sollten Sie die Eingaben, den Verantwortlichen für jeden Schritt sowie die Beendigungskriterien vor dem Codeändern definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zustände schließen zu müssen.

Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Versuche, menschliche Kontrollen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Setzen Sie menschliche Freigabe für Schritte ein, bei denen Geld ausgegeben wird oder Produktionsdaten geändert werden. Kompilierzeitbasierte Verbindungen bedeuten noch nicht vollständige Geschäftsabläufe.

Schreiben Sie ein kurzes Handbuch: Wie man Schlüssel rotiert, wie man die Warteschlange leert und wie man den letzten Eingang rückgängig macht.

Ziehen Sie kleine, testbare Einheiten vor großen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf.

Setzen Sie menschliche Freigabe für Schritte ein, bei denen Geld ausgegeben wird oder Produktionsdaten geändert werden. Kompilierzeitbasierte Verbindungen bedeuten noch nicht vollständige Geschäftsabläufe.

Vor der Einführung des Stack-Systems sollten Versionen eingefroren werden, ein „goldener Transkript“ für den kritischen Ablauf erstellt und die Schritte zum Rückschalten überprüft werden. Gemeinsame Umgebungen benötigen Geschwindigkeitsbeschränkungen, Überprüfungen der Zuordnung sowie einen klaren Verantwortlichen für die Rotation von Geheimnissen. Ziehen Sie langweilige Zuverlässigkeit vor cleveren, einmaligen Demonstrationen.

Batch-Hinweis für d865d505a59e: Halten Sie die Anbieter-Schlüssel außerhalb des Repositories, legen Sie eine Obergrenze für Tokens pro Sitzung fest und speichern Sie die Transkripte neben den Evaluierungs-Beispielen, damit spätere Modellwechsel vergleichbar bleiben.