LangChain Konversationsgedächtnis -- Checkpointer

Standardmäßig ist jeder agent.invoke() unabhängig, und der Agent erinnert sich nicht daran, was zuvor gesagt wurde.

Der Checkpointer (Checkpoint-Speicherer) ermöglicht es dem Agenten, sich an den Gesprächsverlauf zu erinnern, was echte Mehrrunden-Konversationen ermöglicht.

Probleme ohne Checkpointer

Schauen wir uns zunächst die Situation ohne Checkpointer an:

Beispiel

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)

# Runde 1
result1 = agent.invoke({
    "messages": [HumanMessage(content=Mein Name ist Xiaoming)]
})
print(fRunde 1: {result1['messages'][-1].content})

# Runde 2 — Agent erinnert sich nicht an den Inhalt der ersten Runde!
result2 = agent.invoke({
    "messages": [HumanMessage(content=Wie heiße ich?)]
})
print(fRunde 2: {result2['messages'][-1].content})

Ausgabe:

Round 1: Hello Xiaoming! Nice to meet you.
Round 2: Sorry, I don't have your information, and I don't know what your name is.

Checkpointer verwenden, um Konversationen zu merken

Nach dem Hinzufügen eines Checkpointers werden Konversationen unter derselben thread_id automatisch verknüpft:

Beispiel

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# Einen In-Memory-Checkpointer erstellen
checkpointer = InMemorySaver()

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    checkpointer=checkpointer,  # Den Checkpointer übergeben
    system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)

# thread_id verwenden, um den Konversationsthread zu identifizieren
config = {"configurable": {"thread_id": "user-001"}}

# Runde 1
result1 = agent.invoke(
    {"messages": [HumanMessage(content="Mein Name ist Xiaoming, ich lerne Python")]},
    config=config,
)
print(fRunde 1: {result1['messages'][-1].content})

# Zweite Runde — mit derselben thread_id erinnerte sich der Agent!
result2 = agent.invoke(
    {"messages": [HumanMessage(content="Wie heiße ich? Was lerne ich?")]},
    config=config,
)
print(fRunde 2: {result2['messages'][-1].content})

Ausgabe:

Round 1: Hello Xiaoming! Python is a great language to start with. How can I help you?
Round 2: Your name is Xiaoming and you are learning Python. Is there anything specific I can help with?

thread_id ist der Schlüssel. Konversationen unter derselben thread_id sind fortlaufend, während Konversationen unter unterschiedlichen thread_id-Werten vollständig isoliert sind. Dadurch können Sie mehrere Benutzer mit einer einzigen Agent-Instanz bedienen.

Wie der Checkpointer funktioniert

Der Checkpointer speichert nach jeder Agent-Ausführung automatisch einen Zustands-Snapshot (Checkpoint). Wenn dieselbe thread_id das nächste Mal verwendet wird, wird der Zustand automatisch aus dem neuesten Checkpoint wiederhergestellt.

Detaillierter Workflow:

  1. Rufen Sie agent.invoke() auf und übergeben Sie config (enthält thread_id)
  2. Der Agent prüft, ob ein Checkpoint für diese thread_id existiert
  3. Wenn er existiert, historische Nachrichten laden, neue Nachrichten anhängen und fortfahren
  4. Nach der Ausführung automatisch einen neuen Checkpoint speichern

Beispiel

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

checkpointer = InMemorySaver()
agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    checkpointer=checkpointer,
    system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
)

config = {"configurable": {"thread_id": "demo-001"}}

# Simuliere eine Multi-Turn-Konversation
questions = [
    Mein Name ist Xiaoming,
    "Ich lerne Python",
    "Hilf mir, Informationen über mich zusammenzufassen",
]

for i, q in enumerate(questions, 1):
    result = agent.invoke(
        {"messages": [HumanMessage(content=q)]},
        config=config,
    )

    # Checkpoint-Status prüfen
    state = .get_state(config)
    print(f"\nNach Runde {i}:")
    print(f" Nachrichtenanzahl: {len(state.values.get('messages', []))}")
    print(f" Nächster Schritt: {state.next}")
    print(f" Antwort: {result['messages'][-1].content[:80]}...")

Ausgabe:

After round 1:
  Message count: 2
  Next step: ()
  Reply: Hello Xiaoming! Nice to meet you.

After round 2:
  Message count: 4
  Next step: ()
  Reply: Python is a very popular programming language, simple and easy to learn.

After round 3:
  Message count: 6
  Next step: ()
  Reply: Based on our conversation, here is your info: your name is Xiaoming, and you are learning Python.

Checkpointer-Typen

Type Speicherort Persistenz Installation Geeignete Szenarien
InMemorySaver Gedächtnis Nein (nach Programmende verloren) Integriert Entwicklungs-Debugging, Unit-Testing
SqliteSaver SQLite-Datenbank Ja langgraph-checkpoint-sqlite Standalone-Deployment, kleine Anwendungen
PostgresSaver PostgreSQL Ja langgraph-checkpoint-postgres Produktionsumgebung, Multi-Instanz-Sharing

Wenn der Agentainvoke()asynchrone Aufrufe verwendet, müssen Sie zur entsprechenden Async-Version wechselnAsyncSqliteSaver / AsyncPostgresSaver, die Verwendung ähnelt der Sync-Version, außer dass sie mitasync withkombiniert werden muss, um sie zu verwenden.

SqliteSaver-Beispiel

InMemorySaverNur im Speicher gespeichert, Daten gehen nach dem Beenden des Programms verloren. Wenn Sie möchten, dass Konversationen persistiert werden, können Sie darauf zurückgreifenSqliteSaver。

Bei der erstmaligen Verwendung muss der SQLite Checkpointer installiert werden:

pip install langgraph-checkpoint-sqlite

Beispiel

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
)

# Zum Eintreten muss die with-Anweisung verwendet werden; beim Verlassen wird die Datenbankverbindung automatisch geschlossen
with SqliteSaver.from_conn_string("conversations.db") as checkpointer:
    agent = create_agent(
        model=model,
        checkpointer=checkpointer,
        system_prompt="Du bist der Assistent von EXAMPLE Tutorial.",
    )

    config = {"configurable": {"thread_id": "user-001"}}

    result = agent.invoke(
        {
            "messages": [
                HumanMessage(content=Hallo)
            ]
        },
        config=config,
    )

    print(result["messages"][-1].content)

SqliteSaver speichert Checkpoints, die bei jeder Konversation in der SQLite-Datenbank erzeugt werden. Selbst wenn das Programm beendet oder neu gestartet wird, solange Sie dieselbe Datenbankdatei weiterverwenden undthread_id, kann der Agent den zuvor gespeicherten Konversationszustand wiederherstellen. Hinweis:withDie Datenbankverbindung wird geschlossen, nachdem der Codeblock endet, daher sollte Logik mit mehreren Agent-Aufrufenwithinnerhalb des Blocks platziert werden; für Dienste mit langer Laufzeit (wie Webanwendungen) wird empfohlen, die Verbindung während der Lebensdauer der Anwendung aufrechtzuerhalten oder die asynchrone Version vonAsyncSqliteSaver。

Konversationsthreads verwalten

Konversationszustand anzeigen

Beispiel

# Konversationsstatus anzeigen
state = .get_state(config)
print(f"Nächster Schritt: {state.next}")  # () zeigt Leerlauf an
print(f"Nachrichtenanzahl: {len(state.values.get('messages', []))}")

# Konversationsverlauf anzeigen
for msg in state.values.get("messages", []):
    print(f" [{msg.type}] {str(msg.content)[:60]}")

Einen neuen Thread erstellen

Beispiel

# Unterschiedliche thread_id = unterschiedliche unabhängige Konversationen
config_alice = {"configurable": {"thread_id": "alice"}}
config_bob = {"configurable": {"thread_id": "bob"}}

# Alices Konversation
agent.invoke(
    {"messages": [HumanMessage(content="Ich bin Alice")]},
    config=config_alice,
)

# Bobs Konversation – völlig unabhängig, weiß nicht, was Alice gesagt hat
agent.invoke(
    {"messages": [HumanMessage(content=# Isolation überprüfen)]},
    config=config_bob,
)

"Nachrichtenanzahl in Alices Konversation: {len(alice_state.values['messages'])}"
alice_state = agent.get_state(config_alice)
bob_state = agent.get_state(config_bob)
print(f"Nachrichtenanzahl in Bobs Konversation: {len(bob_state.values['messages'])}")
print(fZustand aktualisieren — Konversationen manuell ändern)

Zustand aktualisieren — Konversationen manuell ändern

Beispiel

Beispiel

from langchain.messages import SystemMessage

"(Benutzer wurde auf VIP-Mitgliedschaft hochgestuft)"
agent.update_state(
    config,
    {
        "messages": [
            SystemMessage(content=# Nachfolgende Konversationen enthalten diese eingefügte Nachricht)
        ]
    }
)

Die Parameter von update_state() werden durch den add_messages Reducer (für das Feld messages) verarbeitet, sodass neue Nachrichten angehängt statt überschrieben werden. Wenn Sie den Verlauf löschen und neu beginnen möchten, ist der einfachste Weg, eine neue

; wenn Sie wirklich einige historische Nachrichten löschen müssen, können Sie die entsprechendethread_id(fromRemoveMessage), um die angegebenen Nachrichten präzise zu entfernen.langchain.messagesAndere Erweiterungen

KI denkt nach...