LangGraph: time travel (replay și fork din checkpoint-uri) în LangChain & LangGraph

history = list(graph.get_state_history(config))        # cel mai recent primul
inainte = next(s for s in history if s.next == ("scrie_gluma",))
graph.invoke(None, inainte.config)                      # replay
fork_config = graph.update_state(inainte.config, {"topic": "găini"})
graph.invoke(None, fork_config)                         # fork

Time travel înseamnă reluarea execuției dintr-un checkpoint anterior și cere un checkpointer. Replay: apelezi invoke(None, config) cu configurația unui checkpoint vechi (care conține checkpoint_id). Nodurile de dinaintea checkpoint-ului nu se reexecută; cele de după rulează din nou, inclusiv apelurile LLM, cererile API și interrupt-urile, deci rezultatul poate fi diferit.

Fork: update_state(config_vechi, values) creează un checkpoint nou, ramificat din punctul ales, cu starea modificată; istoricul original rămâne. Valorile se aplică prin reducerii nodului indicat de as_node, iar execuția continuă cu succesorii acelui nod. Implicit, as_node se deduce din istoricul checkpoint-ului; îl dai explicit la ramuri paralele (altfel apare InvalidUpdateError), pe un fir fără istoric sau când vrei să sari peste noduri.

get_state_history(config) întoarce checkpoint-urile firului în ordine invers cronologică, cu cel mai recent primul. Checkpoint-urile se creează la granița fiecărui super-pas, deci poți relua doar din aceste puncte. Un subgraf fără checkpointer propriu apare la nivelul părintelui ca un singur pas; cu checkpointer=True are checkpoint-uri proprii, accesibile prin get_state(config, subgraphs=True).

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
get_state_history(config)	Checkpoint-urile firului, cel mai recent primul.
get_state(config)	Ultimul checkpoint; .next arată nodurile care urmează.
invoke(None, checkpoint_config)	Replay din checkpoint-ul indicat.
update_state(config, values, as_node=...)	Fork: checkpoint nou cu valorile modificate; nu șterge istoricul.
interrupt_before / interrupt_after	Breakpoint-uri statice, utile ca să oprești graful înainte de un pas și să corectezi starea.

Exemple practice și scenarii reale

Exemplul 1: Replay și fork

După exemplul din documentație.

from typing_extensions import TypedDict, NotRequired
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START

class State(TypedDict):
    topic: NotRequired[str]
    gluma: NotRequired[str]

def alege_topic(state: State):
    return {"topic": "șosete în uscător"}

def scrie_gluma(state: State):
    return {"gluma": f"O glumă despre {state['topic']}"}

graph = (
    StateGraph(State)
    .add_node("alege_topic", alege_topic)
    .add_node("scrie_gluma", scrie_gluma)
    .add_edge(START, "alege_topic")
    .add_edge("alege_topic", "scrie_gluma")
    .compile(checkpointer=InMemorySaver())
)
config = {"configurable": {"thread_id": "1"}}
graph.invoke({}, config)

history = list(graph.get_state_history(config))
inainte = next(s for s in history if s.next == ("scrie_gluma",))

graph.invoke(None, inainte.config)                  # replay
fork = graph.update_state(inainte.config, {"topic": "găini"})
print(graph.invoke(None, fork)["gluma"])            # fork

Exemplul 2: Corectarea stării înainte de un pas sensibil

graph = builder.compile(checkpointer=InMemorySaver(), interrupt_before=["proceseaza_plata"])
config = {"configurable": {"thread_id": "order-101"}}

graph.invoke({"suma": 500, "destinatar": "Necunoscut"}, config)
print(graph.get_state(config).next)
graph.update_state(config, {"destinatar": "Furnizor SRL"})
graph.invoke(None, config)

Greșeli frecvente și capcane

• Să crezi că replay citește rezultatele din cache. Nodurile de după checkpoint rulează din nou, cu apeluri LLM și API reale.
• Să crezi că update_state derulează firul înapoi. Creează un checkpoint nou, ramificat; firul original rămâne.
• Să iei primul element din get_state_history drept cel mai vechi. Lista începe cu cel mai recent checkpoint.

Noțiuni și module înrudite

• langgraph-memory-saver
• langgraph-human-in-the-loop
• langgraph-subgraphs
• langchain-streaming-tools