Ein Language Model ist zustandslos. Jeder Aufruf ist isoliert. Das Modell erhält einen Prompt, generiert eine Antwort und vergisst alles. Keine Erinnerung an vorherige Fragen. Kein Kontext über die Beziehung zum Nutzer. Keine Kontinuität.
Für einen einmaligen API-Aufruf ist das kein Problem. Für eine Konversation ist es fatal.
Stellen wir uns einen Dialog vor: „Wie ist das Wetter in Berlin?” – „Sonnig, 18 Grad.” – „Und morgen?” Das Modell weiß nicht, dass „morgen” sich auf Berlin bezieht. Es hat die erste Frage bereits vergessen. Ohne Memory ist jede Antwort ein Neustart.
Memory ist die Architekturschicht, die dieses Problem löst. Sie verwandelt zustandslose Modellaufrufe in zustandsbehaftete Konversationen.
Memory ist kein magisches Konstrukt. Es ist eine strukturierte Verwaltung von Konversationshistorie und Zustandsinformationen zwischen Modellaufrufen.
Die einfachste Form: Eine Liste von Nachrichten. Jede Nutzeranfrage und jede Modellantwort wird in einer Datenstruktur gespeichert. Bei jedem neuen Aufruf wird diese Historie dem Modell als Kontext übergeben. Das Modell „sieht” die gesamte bisherige Unterhaltung und kann darauf aufbauen.
Das ist keine Erinnerung im menschlichen Sinne. Das Modell lernt nicht. Es erhält nur mehr Input.
Aber dieser Input reicht, um Kontinuität zu erzeugen. Das Modell kann auf vorherige Aussagen verweisen, Kontextinformationen nutzen und eine kohärente Konversation führen.
Memory ist also im Kern: Zustandsverwaltung zwischen Modellaufrufen.
LangChain hat sein Memory-Konzept fundamental überarbeitet. Die alten
ConversationBufferMemory-Objekte und ihre Verwandten sind
deprecated. Das neue Paradigma basiert auf drei Säulen:
State, Checkpointer und
Threads.
Früher gab es dedizierte Memory-Objekte, die man an Chains oder Agents übergab. Diese Objekte verwalteten die Historie intern und boten APIs zum Hinzufügen und Abrufen von Nachrichten. Das war explizit, aber umständlich.
Das neue Konzept integriert Memory in die Graphstruktur selbst. Ein Agent oder Chain hat einen State – eine typisierte Datenstruktur, die alle relevanten Informationen enthält. Dieser State wird bei jeder Ausführung aktualisiert und persistiert. Memory ist nicht mehr ein Add-on, sondern Teil der Architektur.
Das ist ein konzeptioneller Gewinn. State ist transparenter, typensicherer und flexibler als separate Memory-Objekte.
Die zentrale Abstraktion ist AgentState. Das ist ein
TypedDict-Schema, das die Struktur des Agent-Zustands definiert.
Ein minimales AgentState hat genau ein Feld:
messages. Das ist eine Liste von Nachrichten, die die
Konversationshistorie repräsentiert. Jede Nachricht hat einen Typ
(Human, AI, System, Tool) und einen Inhalt.
from langchain.agents import AgentState
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
messages: list[BaseMessage]Dieses Schema ist bewusst schlicht. Die messages-Liste
ist die Konversationshistorie. Mehr braucht ein einfacher Agent
nicht.
Aber AgentState ist erweiterbar. Man kann beliebige
zusätzliche Felder hinzufügen: Nutzerpräferenzen, Session-Metadaten,
berechnete Zwischenergebnisse. Alles, was zwischen Aufrufen erhalten
bleiben soll, gehört in den State.
Das ist der entscheidende Unterschied zu alten Memory-Objekten: State ist nicht auf Konversationshistorie beschränkt. Er kann jede Art von Zustandsinformation verwalten.
Ein State existiert zur Laufzeit im Speicher. Sobald das Programm endet, ist er verloren. Für echte Konversationen brauchen wir Persistierung.
Das übernimmt der Checkpointer. Er serialisiert den State nach jedem Schritt und speichert ihn in einem Backend – entweder im Memory (für Tests) oder in einer Datenbank (für Production).
Der einfachste Checkpointer ist InMemorySaver. Er hält
den State im RAM. Perfekt für Entwicklung und Unit-Tests. Nicht geeignet
für Production, da der State beim Neustart verloren geht.
Für echte Anwendungen nutzt man einen datenbankgestützten
Checkpointer wie PostgresSaver. Er speichert jeden
State-Checkpoint in PostgreSQL. Konversationen überleben Neustarts.
Mehrere Agent-Instanzen können auf denselben State zugreifen.
Die Checkpoint-Strategie ist pluggable. Man kann eigene Implementierungen schreiben, die auf Redis, MongoDB oder S3 basieren. Die Schnittstelle ist definiert, das Backend ist austauschbar.
Ein Problem: Wenn alle Konversationen denselben State teilen, vermischen sich die Historien. Nutzer A sieht plötzlich die Nachrichten von Nutzer B. Das ist inakzeptabel.
Die Lösung sind Threads. Ein Thread ist eine isolierte Konversation mit eigenem State. Jeder Thread hat eine eindeutige ID. Der Checkpointer speichert States pro Thread.
Beim Aufruf eines Agents übergibt man eine
thread_id:
agent.invoke(
{"messages": [{"role": "user", "content": "Hallo"}]},
{"configurable": {"thread_id": "user_123_session_456"}}
)Der Agent lädt den State für
thread_id = "user_123_session_456". Alle Nachrichten werden
in diesem Thread gespeichert. Ein anderer Aufruf mit einer anderen
thread_id erhält einen komplett unabhängigen State.
Threads sind das Mittel zur Mandantentrennung. Eine Anwendung kann tausende parallele Konversationen verwalten, jede mit eigenem Kontext.
Memory in LangChain fokussiert sich auf Short-Term Memory – Informationen, die innerhalb einer Konversation relevant sind. Die Message-Historie, temporäre Variablen, Session-Daten.
Long-Term Memory ist ein anderes Problem. Es geht um Informationen, die über Konversationen hinweg persistieren: Nutzerpräferenzen, historische Interaktionen, gelernte Fakten. Das ist kein Feature des State-Mechanismus, sondern erfordert separate Datenspeicherung (Datenbanken, Vector Stores).
Die Grenze ist nicht scharf. Man kann Custom State nutzen, um langfristige Informationen zu speichern, die dann thread-übergreifend geladen werden. Aber das ist nicht der primäre Use Case.
LangChain-Memory löst das Problem der Konversationskontinuität. Für echten Langzeitspeicher braucht man zusätzliche Infrastruktur.
Eine naive Memory-Implementierung führt zu einem fundamentalen Problem: Die Message-Historie wächst unbegrenzt. Nach hundert Fragen und Antworten enthält der State 200+ Nachrichten. Das sprengt das Kontextfenster jedes LLMs.
Selbst wenn das Modell technisch 100k Tokens unterstützt, wird die Performance leiden. LLMs sind bei langen Kontexten ineffizient. Sie werden langsam, teuer und „abgelenkt” – wichtige Informationen gehen in der Masse unter.
Memory ohne Begrenzung ist keine Lösung. Es ist ein Ressourcenleck.
Daher braucht jede ernsthafte Memory-Implementierung eine Strategie zur Kontextreduktion. Die gängigen Ansätze: Trimming, Deletion und Summarization.
Trimming behält nur die letzten N Nachrichten. Die ältesten werden verworfen. Einfach, aber brutal: Informationen gehen verloren.
Deletion entfernt gezielt bestimmte Nachrichten aus dem State. Flexibler als Trimming, aber man muss wissen, was gelöscht werden soll.
Summarization fasst ältere Nachrichten zusammen und ersetzt sie durch einen kompakten Text. Die teuerste Strategie (erfordert LLM-Aufrufe), aber sie erhält mehr Informationen.
Keine Strategie ist perfekt. Jede ist ein Kompromiss zwischen Kontextreduktion und Informationsverlust.
Ein technisches Detail, das entscheidend ist: Wie wird der State aktualisiert?
Bei jedem Schritt gibt der Agent ein Update zurück. Das könnte eine neue Nachricht sein, ein verändertes Feld im Custom State oder ein Löschbefehl. Dieses Update muss mit dem existierenden State zusammengeführt werden.
Das übernehmen Reducer. Ein Reducer ist eine Funktion, die den alten State und das Update nimmt und einen neuen State produziert.
Für das messages-Feld nutzt LangChain den
add_messages-Reducer. Er fügt neue Nachrichten an die Liste
an. Aber er unterstützt auch RemoveMessage: Ein spezielles
Objekt, das angibt, welche Nachricht gelöscht werden soll.
Das ist der Mechanismus hinter Message-Deletion. Man gibt kein neues
State-Dictionary zurück, sondern ein Update mit
RemoveMessage-Objekten. Der Reducer versteht diese und
entfernt die entsprechenden Nachrichten aus der Historie.
Custom State-Felder können eigene Reducer haben. Ein Zähler könnte inkrementiert statt überschrieben werden. Eine Liste könnte dedupliziert werden. Reducer geben volle Kontrolle über die State-Update-Semantik.
LangChain v1.0 führt eine Unterscheidung ein, die subtil, aber wichtig ist: State und Context.
State ist veränderlich. Er wird bei jedem Schritt aktualisiert. Die Message-Historie wächst, Custom Fields ändern sich.
Context ist unveränderlich. Er wird zu Beginn der Agent-Ausführung gesetzt und bleibt konstant. Typische Anwendung: Nutzer-ID, Session-Metadaten, Read-Only-Konfiguration.
Warum diese Trennung? Performance und Semantik. Context muss nicht persistiert werden, da er sich nie ändert. Er wird von außen übergeben und steht dem Agent zur Verfügung, aber er ist nicht Teil des Checkpoint-Zustands.
Tools können auf Context zugreifen, aber ihn nicht verändern. Das erzwingt eine klare Grenze zwischen veränderlichen und unveränderlichen Informationen.
In der Praxis: State für Konversationshistorie und dynamische Daten. Context für statische Metadaten.
Beginnen wir mit dem Minimum. Ein Agent mit Memory braucht genau zwei Dinge: einen Checkpointer und eine Thread-ID.
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
# Tool-Definition (einfaches Beispiel)
def get_weather(city: str) -> str:
"""Gibt das Wetter für eine Stadt zurück."""
return f"Das Wetter in {city} ist sonnig, 18 Grad."
# Agent mit Memory erstellen
agent = create_agent(
model="openai:gpt-4",
tools=[get_weather],
checkpointer=InMemorySaver() # Memory aktiviert
)
# Erste Nachricht
response1 = agent.invoke(
{"messages": [{"role": "user", "content": "Wie ist das Wetter in Berlin?"}]},
{"configurable": {"thread_id": "session_001"}}
)
# Zweite Nachricht - im selben Thread
response2 = agent.invoke(
{"messages": [{"role": "user", "content": "Und morgen?"}]},
{"configurable": {"thread_id": "session_001"}}
)Was passiert hier?
Beim ersten invoke() ist der State leer. Der Agent
erhält nur die Nutzeranfrage. Er ruft das Weather-Tool auf, erhält eine
Antwort und speichert alles im State. Der Checkpointer serialisiert
diesen State und assoziiert ihn mit
thread_id = "session_001".
Beim zweiten invoke() lädt der Agent den State für
thread_id = "session_001". Er sieht die gesamte Historie:
Die erste Frage, den Tool-Aufruf, die Antwort. Die neue Frage „Und
morgen?” erhält dadurch Kontext. Das Modell weiß, dass „morgen” sich auf
Berlin bezieht.
Das ist Memory in ihrer einfachsten Form. Keine explizite Konfiguration. Nur ein Checkpointer und Thread-IDs.
InMemorySaver ist für Tests. In Production brauchen wir
echte Persistierung.
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:password@localhost:5432/langchain_memory"
# Checkpointer mit PostgreSQL-Backend
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# Tabellen automatisch erstellen (einmalig)
checkpointer.setup()
agent = create_agent(
model="openai:gpt-4",
tools=[get_weather],
checkpointer=checkpointer
)
# Agent nutzen - State wird in PostgreSQL gespeichert
response = agent.invoke(
{"messages": [{"role": "user", "content": "Hallo"}]},
{"configurable": {"thread_id": "user_456"}}
)Der Code ist nahezu identisch. Nur der Checkpointer hat sich geändert. Das ist das Design-Prinzip: Die Persistierungs-Strategie ist ein Implementierungsdetail.
PostgresSaver speichert jeden Checkpoint in einer
Tabelle. Die Schema-Definition ist intern. Man muss sich nicht um
SQL-Statements kümmern. setup() erstellt die nötigen
Tabellen automatisch.
Jeder Checkpoint enthält: Thread-ID, Timestamp, serialisierten State, Metadaten. Man kann auf die Checkpoint-Historie zugreifen und ältere Zustände wiederherstellen.
Das ist die Basis für Features wie Undo, Branching oder Debugging. Der gesamte State-Verlauf ist nachvollziehbar.
Die Default-State-Struktur mit messages reicht für
einfache Chatbots. Sobald man mehr Kontext braucht, erweitert man das
Schema.
Angenommen, unser Agent soll sich Nutzerpräferenzen merken. Wir definieren ein Custom State-Schema:
from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict
# Custom State-Schema durch Vererbung
class CustomAgentState(AgentState):
user_id: str
preferences: dict
interaction_count: int
agent = create_agent(
model="openai:gpt-4",
tools=[get_weather],
state_schema=CustomAgentState, # Custom Schema verwenden
checkpointer=InMemorySaver()
)
# State mit zusätzlichen Feldern initialisieren
response = agent.invoke(
{
"messages": [{"role": "user", "content": "Zeig mir das Wetter"}],
"user_id": "user_789",
"preferences": {"temperature_unit": "celsius"},
"interaction_count": 0
},
{"configurable": {"thread_id": "session_xyz"}}
)Das Custom Schema definiert drei neue Felder: user_id,
preferences und interaction_count. Diese
werden Teil des Agent-States und bei jedem Checkpoint gespeichert.
Tools und Middleware können auf diese Felder zugreifen. Ein Tool
könnte die Temperatur basierend auf
preferences["temperature_unit"] formatieren. Middleware
könnte interaction_count bei jedem Aufruf erhöhen.
Das ist der Kern der Erweiterbarkeit: State ist kein festes Konstrukt. Es ist ein typisiertes Schema, das man an die eigenen Bedürfnisse anpasst.
Nach zwanzig Fragen enthält der State vierzig Nachrichten. Nach hundert Fragen zweihundert. Das ist nicht nachhaltig.
Die Lösung: Middleware, die vor jedem Modellaufruf aktiv wird und den State modifiziert.
LangChain v1.0 führt Middleware-Dekoratoren ein. Der wichtigste für
Memory-Management ist @before_model. Er wird ausgeführt,
bevor das LLM aufgerufen wird, und kann die Message-Liste kürzen.
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from typing import Any
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Behält nur die ersten und letzten Nachrichten."""
messages = state["messages"]
# Wenn weniger als 4 Nachrichten, nichts tun
if len(messages) <= 3:
return None
# Erste Nachricht (oft System-Prompt) behalten
first_msg = messages[0]
# Die letzten 3 Nachrichten behalten
recent_messages = messages[-3:]
# Neue Message-Liste: Erste + Letzte
new_messages = [first_msg] + recent_messages
# State-Update mit RemoveMessage zurückgeben
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES), # Alle löschen
*new_messages # Neue Liste setzen
]
}
agent = create_agent(
model="openai:gpt-4",
tools=[get_weather],
middleware=[trim_messages], # Middleware registrieren
checkpointer=InMemorySaver()
)Diese Middleware wird vor jedem Modellaufruf ausgeführt. Sie prüft die Message-Anzahl. Wenn mehr als drei vorhanden sind, behält sie nur die erste und die letzten drei Nachrichten.
Der Mechanismus: RemoveMessage(id=REMOVE_ALL_MESSAGES)
löscht alle Nachrichten aus dem State. Dann werden die gefilterten
Nachrichten neu hinzugefügt. Das ist die idiomatische Methode in
LangChain v1.0.
Das Ergebnis: Der Agent behält Kontext, aber die Historie wächst nicht unbegrenzt. Die ältesten Nachrichten verschwinden automatisch.
Ein wichtiger Hinweis: Diese Strategie ist naiv. Sie löscht möglicherweise wichtigen Kontext. In echten Anwendungen braucht man intelligenteres Trimming: Nachrichten nach Relevanz filtern, Tool-Aufrufe bevorzugt behalten, semantische Ähnlichkeit nutzen. Aber das Prinzip bleibt gleich.
Manchmal will man nicht pauschal trimmen, sondern gezielt Nachrichten entfernen. Etwa weil sie sensible Informationen enthalten oder veraltet sind.
Dafür gibt es RemoveMessage mit spezifischen
Message-IDs:
from langchain.messages import RemoveMessage
def delete_old_tool_calls(state: AgentState) -> dict:
"""Entfernt alte Tool-Call-Nachrichten."""
messages = state["messages"]
# Finde alle Tool-Messages, die älter als die letzten 5 sind
tool_messages = [m for m in messages if m.type == "tool"]
if len(tool_messages) > 5:
old_tool_messages = tool_messages[:-5]
return {
"messages": [RemoveMessage(id=m.id) for m in old_tool_messages]
}
return {}Diese Funktion identifiziert alte Tool-Messages und gibt ein Update
zurück, das sie entfernt. Der add_messages-Reducer versteht
RemoveMessage und löscht die entsprechenden Einträge.
Das ist präziser als Trimming. Man kann gezielt bestimmte Nachrichtentypen filtern, während andere erhalten bleiben.
In der Praxis kombiniert man oft mehrere Strategien: Trimming für die Gesamtlänge, selektives Löschen für spezifische Nachrichtentypen.
Die eleganteste, aber teuerste Strategie: Alte Nachrichten zusammenfassen.
Statt die ersten fünfzig Nachrichten zu löschen, fasst man sie in einem kurzen Text zusammen: „Der Nutzer fragte nach Wetterdaten für verschiedene Städte und erhielt Antworten.” Dieser Summary ersetzt die Originalnachrichten.
Das erhält mehr Information als Deletion, kostet aber zusätzliche LLM-Aufrufe.
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
@before_model
def summarize_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
"""Fasst alte Nachrichten zusammen, wenn die Historie zu lang wird."""
messages = state["messages"]
# Schwellwert: mehr als 10 Nachrichten
if len(messages) <= 10:
return None
# Die ersten 6 Nachrichten (ohne System-Prompt) zusammenfassen
first_msg = messages[0]
to_summarize = messages[1:7]
recent = messages[7:]
# LLM für Zusammenfassung nutzen
summarizer = ChatOpenAI(model="gpt-4", temperature=0)
summary_prompt = "Fasse die folgende Konversation in 2-3 Sätzen zusammen:\n\n"
summary_prompt += "\n".join([f"{m.type}: {m.content}" for m in to_summarize])
summary = summarizer.invoke([{"role": "user", "content": summary_prompt}])
summary_message = SystemMessage(content=f"Frühere Konversation: {summary.content}")
# State-Update: Alte Nachrichten durch Summary ersetzen
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
first_msg,
summary_message,
*recent
]
}Diese Middleware ruft ein LLM auf, um alte Nachrichten zusammenzufassen. Der Summary wird als System-Nachricht in den State eingefügt. Die Originalnachrichten werden gelöscht.
Das ist teuer: Jedes Mal, wenn die Schwelle überschritten wird, kostet es einen zusätzlichen LLM-Call. Aber es bewahrt Kontext effektiver als reines Löschen.
In Production würde man Caching nutzen: Summaries für bestimmte Message-Ranges speichern und wiederverwenden.
Nicht alles gehört in den State. Manche Informationen ändern sich nie und müssen nicht persistiert werden.
Dafür gibt es das Context-Schema. Es definiert unveränderliche Daten, die dem Agent zur Verfügung stehen.
from typing import TypedDict
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
class CustomContext(TypedDict):
user_id: str
api_key: str
agent = create_agent(
model="openai:gpt-4",
tools=[],
context_schema=CustomContext,
checkpointer=InMemorySaver()
)
# Context beim Aufruf übergeben
response = agent.invoke(
{"messages": [{"role": "user", "content": "Hallo"}]},
context=CustomContext(user_id="user_123", api_key="secret"),
config={"configurable": {"thread_id": "session_1"}}
)Der Context wird bei jedem invoke() explizit übergeben.
Er ist nicht Teil des Checkpoints. Tools können über das
ToolRuntime-Objekt darauf zugreifen:
from langchain.agents import tool
from langgraph.runtime import ToolRuntime
@tool
def fetch_user_data(runtime: ToolRuntime[CustomContext, AgentState]) -> str:
"""Lädt Nutzerdaten basierend auf der User-ID aus dem Context."""
user_id = runtime.context["user_id"]
# API-Call mit user_id...
return f"Daten für User {user_id} geladen."Der Unterschied zu State-Feldern: Context muss bei jedem Aufruf neu übergeben werden. Er persistiert nicht. Das ist ideal für sensible Daten (API-Keys, Tokens) oder für Informationen, die sich pro Request ändern (Request-IDs, Trace-Metadaten).
Memory ermöglicht nicht nur Konversationskontinuität, sondern auch dynamische Prompt-Generierung.
Das @dynamic_prompt-Middleware-Dekorator erzeugt
System-Prompts basierend auf State oder Context:
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from typing import TypedDict
class UserContext(TypedDict):
user_name: str
language: str
@dynamic_prompt
def personalized_system_prompt(request: ModelRequest) -> str:
"""Generiert einen personalisierten System-Prompt basierend auf Context."""
user_name = request.runtime.context["user_name"]
language = request.runtime.context["language"]
return f"""Du bist ein hilfreicher Assistent.
Der Nutzer heißt {user_name}.
Antworte auf {language}."""
agent = create_agent(
model="openai:gpt-4",
tools=[],
middleware=[personalized_system_prompt],
context_schema=UserContext
)
response = agent.invoke(
{"messages": [{"role": "user", "content": "Wie spät ist es?"}]},
context=UserContext(user_name="Anna", language="Deutsch")
)Der Prompt wird bei jedem Aufruf neu generiert, basierend auf dem aktuellen Context. Das ermöglicht Personalisierung ohne Hard-Coding.
Man kann auch auf State zugreifen:
@dynamic_prompt
def context_aware_prompt(request: ModelRequest) -> str:
"""Prompt basierend auf Interaktionszahl anpassen."""
state = request.runtime.state
count = state.get("interaction_count", 0)
if count == 0:
return "Du bist ein freundlicher Assistent. Dies ist die erste Interaktion."
else:
return f"Du bist ein freundlicher Assistent. Dies ist Interaktion {count}."Dynamic Prompts sind mächtig: Sie ermöglichen adaptive System-Instruktionen, die auf Historie und Kontext reagieren.
Middleware kann auch nach dem Modellaufruf aktiv werden. Das ist nützlich für Validierung, Filterung oder Post-Processing.
from langchain.agents.middleware import after_model
@after_model
def filter_sensitive_content(state: AgentState, runtime: Runtime) -> dict | None:
"""Entfernt Modellantworten, die sensible Wörter enthalten."""
BLACKLIST = ["password", "secret", "token"]
last_message = state["messages"][-1]
content = last_message.content.lower()
# Prüfen, ob Blacklist-Wörter vorkommen
if any(word in content for word in BLACKLIST):
# Nachricht entfernen
return {"messages": [RemoveMessage(id=last_message.id)]}
return None
agent = create_agent(
model="openai:gpt-4",
tools=[],
middleware=[filter_sensitive_content],
checkpointer=InMemorySaver()
)Diese Middleware läuft nach jedem Modellaufruf. Wenn die Antwort sensible Wörter enthält, wird sie aus dem State gelöscht. Der Agent generiert dann eine neue Antwort.
Das ist defensives Engineering: Man kann nicht verhindern, dass das Modell gelegentlich problematische Inhalte generiert. Aber man kann sicherstellen, dass diese nicht persistiert werden.
Bringen wir alles zusammen. Ein Agent, der Wetter-Informationen liefert, sich Nutzer-Präferenzen merkt und automatisch die Historie verwaltet.
from langchain.agents import create_agent, AgentState, tool
from langchain.agents.middleware import before_model, dynamic_prompt, ModelRequest
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime, ToolRuntime
from typing import TypedDict, Any
# Custom State: Nutzer-Präferenzen speichern
class WeatherAgentState(AgentState):
temperature_unit: str # "celsius" oder "fahrenheit"
location_history: list[str] # Besuchte Städte
# Custom Context: Unveränderliche User-ID
class UserContext(TypedDict):
user_id: str
# Tool-Definition
@tool
def get_weather(
city: str,
runtime: ToolRuntime[UserContext, WeatherAgentState]
) -> str:
"""Gibt das Wetter für eine Stadt zurück."""
# Auf State zugreifen für Temperatureinheit
unit = runtime.state.get("temperature_unit", "celsius")
# Simulierte Wetterdaten
temp_c = 18
temp_f = int(temp_c * 9/5 + 32)
temp_display = f"{temp_c}°C" if unit == "celsius" else f"{temp_f}°F"
# Location History aktualisieren (wird Teil des State-Updates)
current_history = runtime.state.get("location_history", [])
if city not in current_history:
runtime.state["location_history"] = current_history + [city]
return f"Das Wetter in {city}: Sonnig, {temp_display}"
# Middleware: Message-Trimming
@before_model
def trim_old_messages(state: WeatherAgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Behält nur die letzten 4 Nachrichten."""
messages = state["messages"]
if len(messages) <= 4:
return None
recent_messages = messages[-4:]
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
*recent_messages
]
}
# Middleware: Dynamic Prompt mit Präferenzen
@dynamic_prompt
def weather_system_prompt(request: ModelRequest) -> str:
"""Generiert System-Prompt basierend auf User-Präferenzen."""
state = request.runtime.state
unit = state.get("temperature_unit", "celsius")
visited = state.get("location_history", [])
prompt = f"Du bist ein Wetter-Assistent. Temperaturen in {unit}."
if visited:
prompt += f" Der Nutzer hat bereits Wetter für folgende Städte abgefragt: {', '.join(visited)}."
return prompt
# Agent erstellen
agent = create_agent(
model="openai:gpt-4",
tools=[get_weather],
state_schema=WeatherAgentState,
context_schema=UserContext,
middleware=[trim_old_messages, weather_system_prompt],
checkpointer=InMemorySaver()
)
# Verwendung
config = {"configurable": {"thread_id": "user_001_session"}}
context = UserContext(user_id="user_001")
# Erste Anfrage mit Präferenz
response1 = agent.invoke(
{
"messages": [{"role": "user", "content": "Wie ist das Wetter in Berlin?"}],
"temperature_unit": "celsius",
"location_history": []
},
context=context,
config=config
)
print(response1["messages"][-1].content)
# Zweite Anfrage - State wird geladen
response2 = agent.invoke(
{"messages": [{"role": "user", "content": "Und in München?"}]},
context=context,
config=config
)
print(response2["messages"][-1].content)
# Dritte Anfrage - zeigt Location History
response3 = agent.invoke(
{"messages": [{"role": "user", "content": "Welche Städte habe ich abgefragt?"}]},
context=context,
config=config
)
print(response3["messages"][-1].content)
# Ausgabe: "Du hast das Wetter für Berlin und München abgefragt."Dieser Agent demonstriert alle Konzepte:
Custom State speichert Temperatureinheit und besuchte Städte. Diese Information persistiert über Aufrufe hinweg.
Context enthält die User-ID, die bei jedem Request übergeben wird.
Tools greifen über ToolRuntime auf
State und Context zu. Sie können State-Felder lesen und
modifizieren.
Middleware verwaltet die Message-Historie automatisch (Trimming) und generiert dynamische Prompts basierend auf State.
Checkpointer speichert alles zwischen Aufrufen. Der zweite und dritte Request laden den existierenden State und bauen darauf auf.
Das Resultat: Ein Agent, der sich an Präferenzen erinnert, Kontext versteht und dabei effizient mit Ressourcen umgeht.
Memory ist mächtig, aber nicht ohne Fallstricke.
Kontextlänge bleibt limitiert. Auch mit Trimming und Summarization gibt es Grenzen. Sehr lange Konversationen (hunderte Nachrichten) sind schwer zu handhaben. Für solche Fälle braucht man hybride Ansätze: Kurz-Term-Memory im State, Long-Term-Memory in Vector Stores.
State-Updates sind nicht transaktional. Wenn ein Agent abstürzt, kann der State inkonsistent sein. Der Checkpointer speichert nach jedem Schritt, aber Race Conditions sind möglich bei parallelen Aufrufen auf denselben Thread.
Middleware-Reihenfolge ist wichtig.
@before_model läuft vor dem LLM, @after_model
danach. Mehrere Middleware werden in der Reihenfolge ihrer Registrierung
ausgeführt. Die Ordnung kann Ergebnisse beeinflussen.
Performance-Kosten. Jeder Checkpoint ist I/O. Bei hoher Load kann das zum Bottleneck werden. Caching und Batching helfen, aber man muss die Persistierungs-Strategie bewusst wählen.
Best Practices:
Nutze Custom State nur für Daten, die wirklich persistiert werden müssen. Unnötige Felder blähen den Checkpoint auf.
Implementiere aggressive Trimming-Strategien. Lieber zu früh kürzen als Kontextfenster sprengen.
Verwende Context für sensible oder temporäre Daten. Sie gehören nicht in Checkpoints.
Teste Memory-Verhalten explizit. Schreibe Tests, die mehrere Aufrufe durchlaufen und State-Persistierung verifizieren.
Monitore Checkpoint-Größen. Wenn sie stark wachsen, stimmt etwas mit der Trimming-Strategie nicht.
Memory in LangChain ist kein Magie-Feature. Es ist sorgfältig konstruierte Zustandsverwaltung mit klaren Mechanismen für Persistierung, Updates und Reduktion. Verstehen wir diese Mechanik, können wir Agenten bauen, die echte Konversationskontinuität bieten – ohne in Ressourcen-Fallen zu tappen.
Die Zukunft liegt nicht in unbegrenztem Memory, sondern in intelligentem Management: Welche Informationen sind wichtig? Was kann vergessen werden? Wie komprimieren wir Kontext effizient?
Ein guter Agent erinnert sich. Ein exzellenter Agent weiß, was er vergessen darf.