पाठ 15 / 29
State History, Replay और Time Travel
Debugging के लिए पिछली states देखें, पीछे जाएँ और fork करें।
हर चरण एक snapshot है
चूँकि हर चरण के बाद checkpoint सहेजा जाता है, आप thread का state इतिहास (get_state_history) सूचीबद्ध कर सकते हैं, देख सकते हैं कि कोई node चलने से पहले state कैसी थी, कौन-सा node अगला है, और अलग नतीजा खोजने के लिए पुराने checkpoint से फिर शुरू भी कर सकते हैं (state संपादित करके भी)। यह "time travel" agents debug करने (उसने वह शाखा क्यों ली?), बुरे चरण से उबरने, और सब कुछ शुरू से दोबारा चलाए बिना "क्या होता अगर" बदलाव परखने में मदद करती है। यह audit trail भी समर्थित करती है: आप ठीक दिखा सकते हैं कि agent हर चरण पर क्या जानता था।
Thread का इतिहास सूचीबद्ध करना, चलाकर
मैंने यह Python virtual environment में langgraph 1.2.12 और langchain-core 1.6.6 के साथ ऑफ़लाइन चलाया। किसी API key या मॉडल की ज़रूरत नहीं क्योंकि मॉडल की जगह सादे Python functions हैं, इसलिए आउटपुट दोहराने योग्य है। सबसे पुराना पहले: ख़ाली शुरुआत (अगला node __start__ चिह्न), फिर a से पहले n=0, b से पहले n=1, और अंतिम n=2 जिसके बाद चलाने को कुछ नहीं।
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class State(TypedDict):
n: int
def inc(s): return {"n": s["n"] + 1}
g = StateGraph(State)
g.add_node("a", inc); g.add_node("b", inc)
g.add_edge(START, "a"); g.add_edge("a", "b"); g.add_edge("b", END)
app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t"}}
app.invoke({"n": 0}, cfg)
snaps = list(app.get_state_history(cfg)) # newest first
for s in reversed(snaps):
print("n =", s.values.get("n"), "| next:", s.next)
Output:
n = None | next: ('__start__',)
n = 0 | next: ('a',)
n = 1 | next: ('b',)
n = 2 | next: ()त्वरित जाँच: Time travel क्या करने देती है?
- Thread के पुराने checkpoints देखना और उनसे फिर शुरू करना
- असली दुनिया का अतीत बदलना
- सभी nodes छोड़ना
- Checkpoints बंद करना
Answer
Thread के पुराने checkpoints देखना और उनसे फिर शुरू करना — हर checkpoint run का बहाल करने योग्य snapshot है।