Wskazówki praktyczne: Pamięć LangGraph: Kompletny przewodnik praktyczny po zarządzaniu nią
Krok po kroku instrukcja obsługi Notatki praktyczne: LangGraph Memory – kompletny praktyczny przewodnik po zarządzaniu umowami, weryfikacjami oraz miejscami na kod dostępnymi dla zespołów wykorzystujących ten wzorzec.
Źródła: LangGraph Memory: The Complete Practical Guide to Managing What Your Agent Remembers
Poniższe notatki przedstawiają praktyczny plan działania dotyczący książki „LangGraph Memory: The Complete Practical Guide to Managing What Your Agent Remembers”. Nacisk kładziony jest na kontrakty, sprawdzania oraz miejsca zastępcze dla kodu, a nie na motywacyjne aspekty. Podczas przeglądania ogólnych informacji najpierw zapisz kontrakt: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces.
Problem pamięci, o którym nikt nie mówi w tutorialach
Problem pamięci, o którym nikt nie mówi w tutorialach, funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej typowości. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.
Krajobraz pamięci: cztery rzeczy, które należy najpierw zrozumieć
„Krajobraz pamięci: Cztery rzeczy, które należy najpierw zrozumieć” funkcjonuje najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres badania. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji pracy po przerwach.
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)
Rozszerzenie modułu 2: Stan świadomy pamięci
Moduł 2 Rozszerzony: Stan świadomy pamięci funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcji powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całego grafu. Utrzymuj stan grafu w prostej formie i z określonym typem. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach. Moduł 2 Rozszerzony: Stan świadomy pamięci funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji.
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
Standardowy stan świadomy pamięci
Dla standardowego stanu uwzględniającego pamięć należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymagaj ludzkiej aprobaty w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej.
# ── 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
Trzy strategie zarządzania długością rozmowy
W ramach Trzech strategii zarządzania długością rozmowy należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić tę czynność od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Zatwierdzenie przez człowieka powinno być wymagane w przypadku operacji, które generują koszty lub zmieniają dane produkcyjne. Połączenia realizowane w czasie kompilacji nie gwarantują pełnej kompletności biznesowej.
Strategia 1: Filtrowanie wiadomości
Dla Strategii 1: Filtrowanie wiadomości, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Wprowadź ludzką aprobatę dla operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie gwarantują pełnej kompletności biznesowej. Dla Strategii 1: Filtrowanie wiadomości, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu.
Wzorzec A: Filtruj tylko to, co widzi LLM (stan niezmieniony)
Pracując według Wzorca A: Filtruj tylko to, co widzi LLM (stan niezmieniony), najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadania. Zachowuj w pamięci tymczasowej stabilne instrukcje systemowe oraz schematy narzędzi. Ponowne wysyłanie identycznego wstępu jest częstą przyczyną marnotrawstwa zasobów.
# ── 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]}
Wzorzec B: RemoveMessage — Trwałe usunięcie z stanu
Gdy pracujesz nad wzorem B: RemoveMessage — Usuń trwale ze stanu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
# ── 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())
Strategia 2: Redukcja na podstawie tokenów za pomocą trim_messages
Podczas pracy nad Strategią 2: Odcinanie na podstawie tokenów za pomocą trim_messages, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie.
Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny sekretów oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury.
Zachowuj w pamięci podręcznej stabilne instrukcje systemowe oraz schematy narzędzi. Ponowne wysyłanie identycznego prefiksu jest częstą przyczyną marnotrawstwa zasobów.
Podczas pracy nad Strategią 2: Odcinanie na podstawie tokenów za pomocą trim_messages, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie.
Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na jedną konkretne odpowiedzialność, a nie na skomplikowaną sieć powiązań.
# ── 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]}
Szybki przewodnik po kluczowych parametrach
Szybki przewodnik po kluczowych parametrach działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej typowości. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.
Strategia 3: Streszczanie przy użyciu LLM (standard produkcyjny)
Strategia 3: Streszczanie oparte na LLM (standard produkcyjny) działa najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przepis transkrypcji, jeden przypadek awarii oraz notatkę o cofnięciu działań przed rozszerzaniem zakresu. Zarejestruj czasy wykonywania zadań oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od wersji demonstracyjnej do środowisk współdzielonych. Przydziel budżet tokenów na jeden ruch i na jedną sesję – narzędzia typu agentic intensywnie rozszerzają kontekst; sztywne limity zapobiegają temu, by wersje demonstracyjne przerodziły się w niespodziewane rachunki.
Krok 1: Określenie celu (Moduł 2)
Krok 1: Stan (Moduł 2) funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione bloki danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji działania po przerwach.
class State(MessagesState):
summary: str # Stores the running summary. Empty string by default.
Krok 2: Dwa węzły (Moduł 4)
Krok 2: Dwa węzły (Moduł 4) działają najlepiej, gdy traktuje się je jako mierzalną powierzchnię. Zapisz jeden idealny przepływ operacji, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres badania. Zdokumentuj zarówno pomyślny, jak i alternatywny scenariusz działania. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji działania po zakłóceniach.
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
}
Krok 3: Logika routingu (Moduł 5)
Krok 3: Logika routingu (Moduł 5) funkcjonuje najlepiej, gdy traktowana jest jako mierzalna struktura. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres pracy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwanie kontynuacji po przerwach.
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__"
Krok 4: Budowa grafu (Moduł 6)
Krok 4: Budowa grafu (Moduł 6) działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis, jeden przypadek awarii oraz notatkę o cofnięciu działań przed rozszerzeniem zakresu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Utrzymuj stan grafu w prostej formie i określonej strukturze typów. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach.
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)
Jak to wygląda w praktyce
Metoda „How This Plays Out in Practice” działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię do analizy. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres badania. Zarejestruj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury danych ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji pracy po zakłóceniach.
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+
Czynienie pamięci naprawdę trwałą: zewnętrzne punkty kontrolne
Zapewnienie prawdziwej trwałości pamięci: Zewnętrzne punkty kontrolne działają najlepiej, gdy są traktowane jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres analizy. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w prostym i typizowanym formacie. Wplecione bloki danych utrudniają określenie, który węzeł zapisał dane w danym polu, i powodują przerwę w kontynuacji działania po zakłóceniach. Zapewnienie prawdziwej trwałości pamięci: Zewnętrzne punkty kontrolne działają najlepiej, gdy są traktowane jako mierzalna powierzchnia do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres analizy. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań.
Interfejs punktu kontrolnego
Dla interfejsu Checkpointer należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Konieczna jest ludzka akceptacja w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności procesu biznesowego.
# 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 – szczegółowo
Aby dogłębnie zrozumieć SqliteSaver, należy zdefiniować dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić tę czynność od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie gwarantują pełnej kompletności rozwiązania biznesowego.
# ── 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'."
Kiedy używać SqliteSaver w porównaniu z PostgresSaver
Aby określić, kiedy używać SqliteSaver zamiast PostgresSaver, należy przed zmianą kodu zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Wprowadź ludzką aprobatę dla operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Połączenia skompilowane w czasie kompilacji nie gwarantują pełnej kompletności biznesowej. Aby określić, kiedy używać SqliteSaver zamiast PostgresSaver, należy przed zmianą kodu zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki nad rozbudowanymi skryptami. Gdy dany krok zawiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na całą strukturę.
zawikłaną ścieżkę przetwarzania.Inspekcja pamięci: get_state i get_state_history
Gdy zajmujesz się inspekcją pamięci za pomocą funkcji get_state i get_state_history, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie.
Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i unikaj cichego częściowego ukończenia zadania.
Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie zrealizować późniejszy element.
# ── 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("---")
Łączenie wszystkiego: Kompletny agent zarządzający pamięcią
Gdy pracujesz nad książką „Putting It All Together: The Complete Memory-Managed Agent”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłaty za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
# ============================================================
# 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")
Przewodnik decyzyjny dotyczący strategii pamięci
Gdy pracujesz z Przewodnikiem decyzyjnym dotyczącym strategii pamięci, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Trzymaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. System powinien unikać ponownego pobierania opłat za tę samą operację LLM, gdy operator próbuje ponownie wykonać późniejszy etap. Gdy pracujesz z Przewodnikiem decyzyjnym dotyczącym strategii pamięci, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań.
Zaktualizowana karta referencyjna słów kluczowych
Zaktualizowana karta referencyjna słów kluczowych działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej strukturze typów. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po interwencjach.
Wniosek: Pamięć to architektura, a nie element dodany później
Wniosek: „Pamięć to architektura, a nie dodatek” funkcjonuje najlepiej, gdy traktowana jest jako mierzalna struktura. Zanim rozszerzysz zakres, zapisz jeden idealny przypadek działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwę w kontynuacji działania po zakłóceniach.
Lista kontrolna operacyjna
Dla listy kontrolnej operacyjnej zdefiniuj dane wejściowe, osobę odpowiedzialną za daną czynność oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić daną czynność na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.
Zdokumentuj zarówno idealny przebieg działania, jak i ścieżkę naprawczą. Próby ponownych działań, mechanizmy kontroli ludzkiej oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
Należy uzyskać zatwierdzenie człowieka dla procesów, które wydają pieniądze lub zmieniają dane produkcyjne. Połączenia kompilowane w czasie nie gwarantują kompletności biznesowej.
Napisz krótki przewodnik: jak rotować klucze, jak opróżniać kolejki, jak cofnąć ostatnią operację importu.
Lepiej używać małych, testowalnych jednostek niż rozbudowanych skryptów. Gdy jakiś krok zawiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji.
Należy uzyskać zatwierdzenie człowieka dla procesów, które wydają pieniądze lub zmieniają dane produkcyjne. Połączenia kompilowane w czasie nie gwarantują kompletności biznesowej.
Zanim wdrożysz nową architekturę, zamroź wersje oprogramowania, utwórz dokładny zapis kluczowych operacji dla krytycznej ścieżki i sprawdź kroki cofania zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji uprawnień oraz wyraźnego odpowiedzialnego za rotację haseł. Lepiej mieć nudną, niezawodną architekturę niż genialne, jednorazowe demonstracje.
Uwagi dotyczące d865d505a59e: unikaj umieszczania kluczy dostawcy w repozytorium, ustaw ograniczenie liczby tokenów na sesję oraz przechowuj transkrypcje obok plików testowych, aby późniejsze zmiany modeli pozostały porównywalne.
Literatura pokrewna
- Praktyczne notatki: Budowanie systemu triagowania zapytań podtrzymywania obsługi multi-agent za pomocą LangGraph — Szczegółowy przewodnik po praktycznych notatkach dotyczących budowania takiego systemu: umowy, sprawdzenia oraz gotowe fragmenty kodu dla zespołów wdrażających ten model.