استئناف سير العمل وتصحيح السفر عبر الزمن

الاستئناف بعد الانقطاع

# التشغيل الأولي
config = {"configurable": {"thread_id": "research-task-1"}}
result = app.invoke({"query": "اتجاهات AI 2026"}, config)

# ... النظام يتعطل أو المستخدم يغلق المتصفح ...

# الاستئناف لاحقاً - نفس thread_id
result = app.invoke({"continue": True}, config)

---

get_state: فحص الحالة الحالية

state = app.get_state(config)

print(f"القيم الحالية: {state.values}")
print(f"العقد التالية: {state.next}")
print(f"البيانات الوصفية: {state.metadata}")

---

get_state_history: السفر عبر الزمن

history = list(app.get_state_history(config))

for i, checkpoint in enumerate(history):
    print(f"الخطوة {i}: {checkpoint.metadata.get('step')}")
    print(f"  الحالة: {checkpoint.values}")

# السفر عبر الزمن: استئناف من نقطة تفتيش محددة
old_config = history[3].config
result = app.invoke({"continue": True}, old_config)

---

update_state: تعديل الحالة أثناء سير العمل

state = app.get_state(config)

app.update_state(
    config,
    {
        "documents": state.values["documents"] + ["مستند جديد مهم"],
        "needs_reanalysis": True
    }
)

result = app.invoke({"continue": True}, config)

حالات الاستخدام: تصحيحات الإنسان في الحلقة، حقن البيانات الخارجية.

---

التفرع: فرع من نقطة تفتيش

history = list(app.get_state_history(config))
fork_point = history[5]

new_config = {"configurable": {"thread_id": "fork-experiment-1"}}
app.update_state(new_config, fork_point.values)
result = app.invoke({"alternative_approach": True}, new_config)

حالات الاستخدام: اختبار A/B، تصحيح بدون التأثير على التدفق الرئيسي.

---

أسئلة المقابلة

س: كيف يعمل تصحيح السفر عبر الزمن؟

"يخزن LangGraph الحالة الكاملة في كل نقطة تفتيش. باستخدام get_state_history، يمكنك سرد جميع نقاط التفتيش. يمكنك الاستئناف من أي نقطة تفتيش أو التفرع لخيط جديد للتجريب."

---

النقاط الرئيسية

✅ نفس thread_id = استئناف تلقائي ✅ get_state يفحص الحالة الحالية ✅ get_state_history يمكّن السفر عبر الزمن ✅ update_state لتعديلات الإنسان في الحلقة ✅ التفرع للتجريب بدون التأثير على التدفق الرئيسي

:::