استدامة حالة الرسم البياني (نقاط التحقق)

لماذا نحفظ الحالة؟

أداة نقاط التحقق داخل الذاكرة

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

أداة نقاط التحقق في Postgres

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string('postgresql://...')
await checkpointer.setup()   # create tables
app = graph.compile(checkpointer=checkpointer)

أداة نقاط التحقق في Sqlite

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect('agents.db', check_same_thread=False)
checkpointer = SqliteSaver(conn)

معرّفات الخيوط

config = {'configurable': {'thread_id': 'user-42-conversation-1'}}
result = app.invoke({'messages': [...]}, config)

استئناف تنفيذ

# Some hours later:
app.invoke(None, config)

فحص الحالة

snapshot = app.get_state(config)
print(snapshot.values)         # current state dict
print(snapshot.next)           # which node is next
print(snapshot.created_at)

سجل الحالة

for snapshot in app.get_state_history(config):
    print(snapshot.created_at, snapshot.values)

تفريع تنفيذ

app.update_state(config, {'plan': 'a different plan'})
app.invoke(None, config)   # continue with the new plan

التعامل مع الإنسان باستخدام interrupt_before

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=['send_email']
)

result = app.invoke({'messages': [...]}, config)
# Agent paused. Human inspects state.
# When approved:
app.invoke(None, config)

الاستئناف بعد التحرير

app.update_state(config, {'email_draft': 'final version'})
app.invoke(None, config)

تتبع التكلفة عبر نقاط التحقق

class State(TypedDict):
    cost_usd: Annotated[float, lambda a, b: a + b]

# Every LLM-calling node adds to it.

التنظيف

await checkpointer.delete(thread_id='old-thread')

استئناف الخيوط

مراجعة