Trwałość LangGraph, część 2: PostgresSavers, HITL i serwery tła w produkcji
Przenieś LangGraph z InMemorySaver do PostgresSaver z użyciem zasobów PostgreSQL oraz asynchronicznych mechanizmów zapisu, porównaj SQLite z Redis, a także uruchom funkcje przerwania i wznowienia w środowisku produkcyjnym.
Od punktów kontrolnych w pamięci RAM do trwałych backendów
Część pierwsza omówiła, dlaczego istnieje mechanizm trwałości w LangGraph – stan aplikacji, punkty kontrolne, narzędzia do ich tworzenia, wątki oraz InMemorySaver. Ta część przekształca te same koncepcje w rozwiązania gotowe do użycia w produkcji. Narzędzia do zapisu danych w pamięci znikają po ponownym uruchomieniu procesu; prawdziwe agenty wymagają magazynowania danych, które przetrwają dłużej niż sam proces w Pythonie.
PostgresSaver dla trwałości w produkcji
Baza danych znajduje się poza procesem aplikacji. Awarie, ponowne wdrożenia oraz zmiany sprzętowe nie wpływają na integralność historii punktów kontrolnych. PostgresSaver zapisuje te punkty kontrolne do PostgreSQL zamiast do pamięci RAM.
PostgreSQL jest popularnym wyborem, ponieważ zapewnia trwałość danych, jest szeroko dostępny jako zarządzane usługi chmurowe, bezpieczny dla równoczesnie działających serwerów aplikacji oraz zdolny do rozszerzania się wraz ze wzrostem ilości punktów kontrolnych i ruchu.
Zainstaluj odpowiedni pakiet:
pip install langgraph-checkpoint-postgres
Bazowe ustawienia:
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# This creates the required tables the first time you use it.
# Only needs to run once per database.
checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "session-99"}}
result = graph.invoke(
{"messages": [{"role": "user", "content": "Hello"}]},
config=config,
)
from_conn_string otwiera połączenie zarządzane w kontekście. Funkcja setup() tworzy takie tabele jak checkpoints i checkpoint_blobs tylko raz – należy ją traktować jak migrację, a nie jako procedurę wykonywaną przy każdej prośbie. Funkcja compile(checkpointer=...) korzysta z tej samej interfejsu co metoda działająca w pamięci; zmiana backendów nie powoduje przepisywania struktury grafu.
Zbiory połączeń
Otwieranie nowego połączenia przy każdej prośbie marnuje czas. Lepiej użyć zbiórka gotowych połączeń:
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://user:password@localhost:5432/mydb"with ConnectionPool(DB_URI) as pool:
checkpointer = PostgresSaver(pool)
checkpointer.setup() # one-time setup
graph = builder.compile(checkpointer=checkpointer)
Wariant asynchroniczny
Aplikacje w stylu FastAPI powinny korzystać z AsyncPostgresSaver:
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
await checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)
Jeśli ręcznie tworzysz połączenia z psycopg, ustaw autocommit=True oraz row_factory=dict_row; w przeciwnym razie funkcja setup() może nie zapewnić trwałości tabel. Zachowaj, aby thread_id miał mniej niż 255 znaków – UUID to bezpieczny standard.
Inne backendy
SQLite
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
Jeden lokalny plik przetrwa restarty i nadaje się do prototypów lub skryptów działających w jednym procesie. Blokowanie plików utrudnia wykorzystanie go w implementacjach webowych z wieloma pracownikami.
Redis
Szybki, oparty na pamięci z opcjonalną przechowywalnością na dysku – przydatny dla produktów do czatowania o wysokiej przepustowości z wieloma jednoczesnymi wątkami.
Zarządzany Postgres w chmurze
RDS, Cloud SQL, Supabase, Neon oraz podobne rozwiązania używają standardowego PostgreSQL, więc PostgresSaver funkcjonuje poprawnie, podczas gdy dostawca zajmuje się kopiami zapasowymi i ciągłością działania.
Sprawdzone narzędzia do kontrolowania stanu
Klasyfikuj podklasę BaseCheckpointSaver i zaimplementuj metody put, get_tuple, list (oraz ich wersje asynchroniczne), gdy przechowywanie wewnętrzne jest obowiązkowe.
Szybki wybór
Lokalny, trwały skrypt → SQLite. Środowisko produkcyjne z wieloma pracownikami → Postgres/AsyncPostgres. Ekstremalny ruch punktów kontrolnych w warunkach konkurowania → Redis. Istniejący zarządzany Postgres → Postgres hostowany w chmurze. Przymusowe przechowywanie wewnętrzne → niestandardowy mechanizm zapisu.
Pięć przykładów stopniowego rozwoju
Przykład 1 — brak trwałości danych (amnezja)
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
count: intdef increment(state: State) -> dict:
return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)graph = builder.compile() # No checkpointer passed in!print(graph.invoke({"count": 0})) # {'count': 1}
print(graph.invoke({"count": 0})) # {'count': 1} again — no memory of the last call!
{'count': 1}
{'count': 1}
Druga próba wywołania nie może uzyskać dostępu do pierwszej: brak punktu kontrolnego, brak identyfikatora wątku.
Przykład 2 — ciągłość dzięki InMemorySaver
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class State(TypedDict):
count: intdef increment(state: State) -> dict:
return {"count": state["count"] + 1}builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "counter-1"}}print(graph.invoke({"count": 0}, config)) # {'count': 1}
print(graph.invoke({"count": 0}, config)) # {'count': 2}
{'count': 1}
{'count': 2}
Te same dane wejściowe są teraz kontynuowane od poprzedniego punktu kontrolnego zamiast być resetowane.
Przykład 3 — wznowienie po symulowanym awarii
from langgraph.checkpoint.memory import InMemorySaver
# Shared storage dictionary simulates a database that survives across "restarts"
shared_storage = {}checkpointer_1 = InMemorySaver()
checkpointer_1.storage = shared_storage # pretend this is durable storagegraph = builder.compile(checkpointer=checkpointer_1)
config = {"configurable": {"thread_id": "resume-demo"}}graph.invoke({"count": 0}, config) # count becomes 1
print("Process 'crashes' here...")# Simulate a restart: a brand new checkpointer object, same underlying storage
checkpointer_2 = InMemorySaver()
checkpointer_2.storage = shared_storage
graph_after_restart = builder.compile(checkpointer=checkpointer_2)# Resume using the SAME thread_id
result = graph_after_restart.invoke({"count": 0}, config)
print(result) # {'count': 2} — continued, did NOT reset to 0
Dzielenie się słownikiem w ramach tego samego procesu symuluje ponowny start; w środowisku produkcyjnym po prostu nowy proces odwołuje się do tej samej bazy danych Postgres i kontynuuje działanie za pomocą thread_id.
Przykład 4 — udział człowieka z użyciem interrupt()
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class State(TypedDict):
task: str
approved: booldef request_approval(state: State) -> dict:
decision = interrupt(f"Approve this task? '{state['task']}' (yes/no)")
return {"approved": decision == "yes"}builder = StateGraph(State)
builder.add_node("request_approval", request_approval)
builder.add_edge(START, "request_approval")
builder.add_edge("request_approval", END)checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)config = {"configurable": {"thread_id": "approval-1"}}# Step 1: Start the graph — it will pause at interrupt()
result = graph.invoke({"task": "Send $500 refund", "approved": False}, config)
print(result) # Contains an '__interrupt__' key with our question# ... hours later, a human reviews and approves ...# Step 2: Resume with the human's answer
final_result = graph.invoke(Command(resume="yes"), config)
print(final_result) # {'task': 'Send $500 refund', 'approved': True}
interrupt(...) wstrzymuje działanie i tworzy punkty kontrolne. Wywołujący widzi dane w polu __interrupt__. Później Command(resume=...) kontynuuje proces. Podczas wznowienia LangGraph ponownie uruchamia cały węzeł od początku — efekty uboczne powinny być zachowane po przerwaniu, a nie przed nim.
Przykład 5 — chatbot oparty na Postgres
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
class State(TypedDict):
messages: listdef chatbot_node(state: State) -> dict:
last_message = state["messages"][-1]
reply = f"You said: {last_message}"
return {"messages": state["messages"] + [reply]}builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)DB_URI = "postgresql://user:password@localhost:5432/mydb"with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # run once
graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-42"}}
result = graph.invoke({"messages": ["Hi there"]}, config)
print(result)
{'messages': ['Hi there', 'You said: Hi there']}
Zatrzymaj skrypt, uruchom go jutro z tym samym identyfikatorem wątku, a rozmowa będzie kontynuowana, ponieważ baza danych przechowuje punkty kontrolne.
Dlaczego HITL zależy od trwałości danych
Trzy elementy: interrupt() do pauzowania z danymi, nieskończone czekanie bez wykorzystania mocy obliczeniowej, gdy stan jest przechowywany, oraz Command(resume=...) do kontynuacji. Bez wskaźnika stanu pauzowanie i czekanie nie mają gdzie przechować zamarłej struktury danych.
Rzeczywiste wzorce zatwierdzania – decyzje o pożyczkach powyżej określonego progu, zalecenia kliniczne przed sporządzaniem raportów, weryfikacja klauzul prawnych, przekroczenia budżetu – często wymagają oczekiwania od kilku minut do tygodni. Persistencja sprawia, że to oczekiwanie jest tanie i bezpieczne.
Persistencja między możliwościami agentów
Długie rozmowy, plany wieloetapowe, które przetrwają awarie, pętle narzędziowe, które nie mogą zapomnieć o wcześniejszych wynikach, oraz procesy typu „rozwidlenie i ponowne połączenie” wszystkie polegają na trwałych wskaźnikach stanu. Identyfikatory wątków to elementy, których używają operatorzy i interfejsy użytkownika do wznowienia właściwego procesu.
Błędy, które ciągle się pojawiają
- Wywoływanie
setup()przy każdej prośbie.
thread_id.interrupt().thread_id podczas kontynuacji działania.Praktyki, które dobrze się sprawdzają
Uruchamiaj konfigurację schematu jako migrację. Woląć zbiory węzłów. Wykorzystywać asynchroniczne mechanizmy zapisu w aplikacjach asynchronicznych. Utrzymywać krótkie i stałe identyfikatory wątków. Dokumentować, które środowiska używają danego backendu. Dodawać testy dymowe, które powodują awarię i jej ponowne uruchomienie. Monitorować wzrost i przechowywanie tabel z punktami kontrolnymi.
Podsumowanie
Persistentne przechowywanie danych w LangGraph polega głównie na „tej samej API do grafów, ale innym trwałym magazynie”. Postgres jest standardowym wyborem dla przedsiębiorstw; SQLite i Redis zajmują określone nisze, natomiast niestandardowe rozwiązania służą do obsługi wymaganej infrastruktury. Ludzki nadzór oraz przywracanie po awariach to nie opcjonalne funkcje – to właśnie dlatego istnieją narzędzia do tworzenia punktów kontrolnych. Podłącz backend jak najszybciej, aby możliwe było przerwanie i wznowienie działania przed tym, jak agent nawiąże kontakt z użytkownikami.
Przewodnik operacyjny dla backendów z punktami kontrolnymi
Traktuj przechowywanie punktów kontrolnych tak samo jak główne dane aplikacji. Harmonogramy tworzenia kopii zapasowych, ćwiczenia przywracania danych oraz zasady przechowywania powinny być omawiane w tych samych rozmowach co informacje o klientach. Produkt czatowy, który nie potrafi przywrócić rozmów po awarii regionalnej, tak naprawdę nie dostarczył pamięci – dostarczył tylko nadzieję.
Planowanie pojemności powinno obejmować monitorowanie punktów kontrolnych dla każdego aktywnego wątku dziennie, średniej wielkości obiektów w stanach o dużej ilości wiadomości oraz wzrostu obciążenia przy maksymalnej jednoczesności. Duże obiekty przechowywane w stanie bazy danych znacząco zwiększą jej rozmiar na długo przed tym, jak liczba wierszy stanie się alarmująca. Lepiej przechowywać duże pliki w systemach magazynowania obiektów, a w stanie zachowywać jedynie wskaźniki.
Kontrola dostępu jest ważna: użytkownik bazy danych odpowiedzialny za punkty kontrolne nie powinien pełnić tej samej roli co osoba, która za pomocą uprawnień DDL migruje niespowiązane schematy. Należy regularnie zmieniać dane uwierzytelniające, preferować autoryzację typu IAM tam, gdzie dostawcy chmury to umożliwiają, oraz sprawdzać, kto może wykonywać polecenie SELECT na stanie rozmowy. Identyfikatory wątków często odpowiadają użytkownikom; pliki wyeksportowane należy traktować jako dane osobowe.
Gdy migruje się z InMemorySaver na Postgres w systemie produkcyjnym, podwójne zapisywanie jest rzadko konieczne, jeśli zaakceptuje się, że wątki RAM są resetowane tylko raz. Należy zaplanować okno konserwacji, wdrożyć nowy mechanizm kontrolny oraz monitorować wskaźniki błędów po wznowieniu działania. W przypadku przeniesień typu blue/green upewnij się, że oba środowiska odnoszą się do tej samej bazy danych, w przeciwnym razie wystąpi rozdźwięk w historii wątków.
Wskazówki dotyczące testowania: uruchom Postgres w środowisku CI (lub w tymczasowym kontenerze), wywołaj funkcję setup() raz, przetestuj pięć przykładowych scenariuszy jako testy integracyjne i upewnij się, że drugi proces może wznowić wątek utworzony przez pierwszy. Ten jeden test międzyprocesowy wykrywa więcej błędów w środowisku produkcyjnym niż jakikolwiek test jednostkowy działający w pamięci operacyjnej.
Na koniec należy nauczyć personel dyżurny różnicy pomiędzy „graph failed” a „checkpointer unreachable”. Objawy są podobne – pojawia się błąd – ale rozwiązania różnią się. Panel sterowania, które oddzielają wyjątki aplikacji od przeciążeń bazy danych, oszczędzają godziny podczas incydentów.
Tożsamość wątku i użytkownicze doświadczenie produktu
ID wątków to nie tylko klucze bazy danych; są to również obiekty produktu. Celowo powiązujcie je z sesjami użytkowników, zgłoszeniami obsługi klienta lub ID dokumentów. Umożliwcie funkcję „kontynuuj rozmowę” w interfejsie poprzez ponowne użycie tego ID. Funkcję „rozpocznij od nowa” uruchomcie poprzez utworzenie nowego ID. Mieszanie tych funkcji w jednym ID powoduje problemy, gdy stare wyniki narzędzi ponownie się pojawiają.
Dla użytkowników korzystających z kilku urządzeń, trwałość ID wątków między różnymi klientami wymaga wyszukiwań uwzględniających informacje o autoryzacji. Nigdy nie akceptujcie ID wątku dostarczonego przez klienta bez sprawdzenia jego właściciela. Checkpointery przechowują wszystko, co im podasz; nie kontrolują przynależności do konkretnego użytkownika.
Ewolucja schematu dla stanu z punktami kontrolnymi
TypedDicty stanu ulegają zmianom. Dodawanie pól opcjonalnych jest proste; przemiana nazw lub usuwanie pól niszczy stare punkty kontrolne. Należy preferować zmiany addytywne, mechanizmy domyślnego uzupełniania brakujących kluczy oraz wyraźne migracje przy zmianach destruktywnych. Przechowuj schema_version w stanie i używaj go do sterowania logiką węzłów, gdy jest to konieczne.
Długie listy wiadomości wymagają zasad przycinania: podsumuj starsze wpisy w jedno pole, zachowaj najnowsze wiadomości w formie surowej i przetestuj możliwość kontynuacji po przycięciu. Nieograniczona historia jest najczęstą przyczyną rozrastania się tabel punktów kontrolnych w Postgresie.
Wzorce przerwania interfejsu użytkownika
Interfejsy użytkownika służące do zatwierdzania powinny pokazywać treść przerwy, odpowiedni fragment stanu oraz terminy. Tożsamość osoby dokonującej zatwierdzenia należy przechowywać w stanie po wznowieniu działania. Jeśli terminy zatwierdzeń wygasną, należy uruchomić mechanizm nadzoru, który anuluje proces lub przenosi go na wyższy poziom, zamiast pozostawiać wątki w stanie zawieszenia na zawsze. Możliwość przechowywania danych umożliwia długie oczekiwanie; polityka produktu musi je ograniczyć.
Ćwiczenia przywracania po katastrofie
Kwartalnie należy przywrócić kopię zapasową do środowiska testowego i wznowić działanie znanego identyfikatora wątku. Należy potwierdzić, że komunikaty, status przerwy oraz pozostałości narzędzi są zgodne. Należy udokumentować wartości RPO/RTO dla bazy danych punktów kontrolnych obok głównej bazy danych aplikacji. Agenty, które pamiętają użytkowników, muszą spełniać te same wymagania ciągłości działania co systemy e-mail.
Uwagi dotyczące wydajności
Bazy połączeń powinny być dostosowywane pod liczbę jednocześnie działających procesorów grafu, a nie tylko obsługujących żądania webowe. Uważaj na wzorce N+1, w których każdy węzeł niepotrzebnie zapisuje punkt kontrolny — dostosuj użycie mechanizmów kontroli zgodnie z domyślnymi ustawieniami wersji LangGraph. Kompresuj lub przenoś na zewnątrz duże pliki wyjściowe narzędzi binarnych.
Uwagi dotyczące bezpieczeństwa
Szyfruj dyski, używaj TLS do komunikacji z Postgresem, ogranicz ścieżki sieciowe od procesorów do magazynu danych oraz usuwaj poufne informacje z pamięci stanu przed zapisaniem punktu kontrolnego, jeśli narzędzia mogą uzyskać dostęp do surowych kluczy. Rozważ użycie narzędzi do redagowania danych na poziomie poszczególnych pól. Podczas audytów zgodności będą zadawane pytania — przygotuj odpowiedzi wraz z diagramami przepływu danych od użytkownika → graf → mechanizm kontroli → dysk.
To właśnie te warstwy operacyjne przekształcają przykłady kodu w system, którym można spokojnie zarządzać. API są proste; to odpowiednia dyscyplina w ich implementacji stanowi prawdziwą zaletę w środowisku produkcyjnym.
Koordynacja mechanizmów kontroli z horyzontalnym skalowaniem automatycznym
Gdy Kubernetes skaluje pracowniki grafowi z dwóch podsów do dwudziestu, każdy pod musi komunikować się z tym samym trwałym punktem kontrolnym. Sesje przyklejone są niepotrzebne, jeśli identyfikatory wątków są przekazywane z każdym żądaniem. Zbiory pracowników „warm pools” nie powinny zakładać istnienia lokalnych plików SQLite. Sprawdzania stanu powinny weryfikować połączenie z bazą danych, a nie tylko aktywność procesu. Podczas wdrażania pracownicy powinni zostać wyłączeni lub zrobić punkt kontrolny w trakcie działania, aby przerwy można było wznowić. Przeprowadź test chaosu, zamykając podsy podczas oczekiwania na interrupt() i sprawdzając możliwość wznowienia działania na innym podzie. Ten jeden eksperyment potwierdza architekturę lepiej niż tuzin testów jednostkowych.
Tłok w przetwarzaniu dokumentów: jeśli przechowywanie punktów kontrolnych zwalnia, czy roboty powinny zmniejszyć obciążenie lub czekać w kolejce? Lepiej wybrać jawne rozwiązanie problemu niż ciche rozwiązania w pamięci, które różnią się między poszczególnymi kontenerami. Operatorzy nigdy nie powinni „tymczasowo” przechodzić na InMemorySaver w środowisku produkcyjnym, aby „naprawić opóźnienia” — taka zmiana niszczy obietnicę dotyczącą funkcjonowania produktu.
List kontrolny przed uznaniem persistencji za zakończoną
Potwierdzone, że setup() jest wykonywany jako migracja; skonfigurowane zasoby; tryb asynchroniczny i synchroniczny dostosowane do frameworka webowego; egzekwowane przypisanie identyfikatorów wątków; sprawdzone przywracanie przerw pomiędzy procesami; jednorazowe udane przywrócenie kopii zapasowych; udokumentowany proces przechowywania danych; przeanalizowane sposoby obsługi PII; panele kontrolne odróżniają błędy bazodanowe od błędów graficznych; podręczniki obsługi łączą identyfikatory wątków z narzędziami wsparcia użytkowników. Gdy każda pozycja zostanie zaznaczona, PostgresSaver (lub wybrany przez Ciebie backend) przestaje być zwykłym fragmentem kodu – staje się częścią gwarancji pamięci produktu. Do tego czasu traktuj demonstracje jako takie i wyłącz flagi funkcjonalne w przypadku procesów HITL o długim czasie trwania.
Krótka historia nieudanego przywracania
Jeden z inżynierów na zmianie kiedyś zaobserwował, że akceptacje utknęły na godziny. Graf funkcjonował prawidłowo; hasło checkpointer było rotowane, a pracownicy przechowywali w pamięci tymczasowej stare dane uwierzytelniające. Nowe przerwy w działaniu nie powodziły się, podczas gdy stare wątki nadal odczytywały dane. Rozwiązaniem było odświeżenie danych uwierzytelniających oraz dodanie instrukcji w procedurze: sprawdzenie autoryzacji checkpointer przed ponownym uruchomieniem kontenerów grafu. Zachowaj tę lekcję obok swoich fragmentów PostgresSaver, aby kolejna rotacja nie powtórzyła awarii. Trwała pamięć obejmuje również pamięć operacyjną – ludzką wiedzę zapisaną w podręcznikach obsługi.
To zdarzenie wyjaśniło również kwestię monitoringu: powiadamianie o błędach autoryzacji checkpointer jako osobna strona, a nie w ramach ogólnej kategorii 500. Gdy pamięć jest produktem, ścieżka do bazy danych stanowi zależność najwyższego priorytetu. Dzięki takiemu podejściu zamiana InMemorySaver na PostgresSaver przestaje być straszną migracją i staje się rutynowym krokiem w procesie przekształcania notatnika w usługę.
Powiązane materiały
- Historia rozmowy, fakty, stan pracy i punkty kontrolne to cztery różne zbiory danych — Przestań nazywać wszystko pamięcią. Oddziel transkrypcje sesji, trwałe fakty, stan pracy ticketów oraz punkty kontrolne LangGraph — dla każdego z nich ustal zasady przechowywania i autoryzacji.
- Wzmocnienie agencyjnych grafów: umowy stanu, izolacja i bramy HITL — Niezawodność w produkcji dla LangGraph: prosty stan typu TypedDict, węzły do jednego celu, krawędzie powtórek/błędów, ludzka weryfikacja oraz możliwości obserwacji.