أنماط سير العمل في LangGraph: من الحواف الشرطية إلى المقاطعة التفاعلية

هذا هو المقال الذي يجمع ما تحتاجه فعلاً قبل أول سير عمل (workflow) على لانغ غراف (LangGraph): كيف تبني رسماً من الحالة والعقد والحواف، وماذا تعني الخطوة الكبرى (super-step)، ومتى تختار Command أو Send بدل الحواف الشرطية، ثم كيف توقف الرسم بانتظار موافقة إنسان بـinterrupt() — دون أن تكتشف لاحقاً أن الرسم أعاد تنفيذ عملية دفع مرتين.

قبل الدخول في الكود، المصطلحات الثلاثة التي ستقابلها في كل صفحة: الحالة (state) هي بيانات الرسم المشتركة، والعقدة (node) هي دالة تنفّذ العمل، والحافة (edge) هي دالة تقرّر أي عقدة تالية. وسير العمل هو هذا الرسم كله؛ أما الوكيل (agent) فهو الحالة التي يتّخذ فيها النموذج القرارات بنفسه بدل أن تُملى عليه المسارات.

ما الذي يتكوّن منه أي رسم بياني في LangGraph؟

يتكوّن أي رسم في لانغ غراف من ثلاثة عناصر لا رابع لها: الحالة والعقد والحواف، وهي العناصر الثلاثة نفسها التي تبدأ بها صفحة graph-api:

State: A shared data structure that represents the current snapshot of your application. It can be any data type, but is typically defined using a shared state schema. Nodes: Functions that encode the logic of your agents. They receive the current state as input, perform some computation or side-effect, and return an updated state. Edges: Functions that determine which Node to execute next based on the current state. They can be conditional branches or fixed transitions.

والجملة التي تلخّص المكتبة كلّها بنصّها: «nodes do the work, edges tell what to do next» — العقد تنفّذ العمل، والحواف تخبرك ماذا يحدث بعده.

عملياً، هذا يعني أن ما تكتبه داخل العقدة هو بايثون عادي: استدعاء نموذج لغوي، أو قراءة ملف، أو استعلام قاعدة بيانات. المكتبة لا تفرض عليك إطاراً ولا نوعاً خاصاً للدالة.

الحالة: مُخطَّط واحد تتشاركه كل العقد

الحالة تُعرَّف عادةً بـTypedDict، ولكل مفتاح فيها دالة دمج (reducer) تحدّد كيف تُدمج تحديثات عدة عقد في قيمة واحدة:

from operator import add
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

من دون دالة دمج، القاعدة الافتراضية بسيطة: القيمة الجديدة تستبدل القديمة — «The default reducer ignores the left argument and replaces the state value with the right argument». أما مع add فتُدمج القوائم: لو أعادت عقدة {"errors": ["bad sql"]} وأعادت أخرى {"errors": []} تبقى القيمة ["bad sql"]، فالقائمة الفارغة تُدمج ولا تمسح.

ونقطتا START وEND عقدتان افتراضيتان: الأولى تحدّد أي عقدة يبدأ منها الرسم، والثانية تنهي أي طريق. في أمثلة هذا المقال ستبدأ دائماً بـbuilder.add_edge(START, "node") وتنهي بـbuilder.add_edge("node", END).

ما معنى «الخطوة الكبرى» (super-step)؟

معناها أن لانغ غراف ينفّذ البرنامج على دفعات متزامنة تُسمّى الخطوات الكبرى (super-step): كل الحواف المتوازية في الخطوة نفسها تُنفَّذ معاً، والمتسلسلة منها تنتمي إلى خطوات كبرى مختلفة.

المستوحى من نظام Pregel من Google، ويعتمد على تمرير الرسائل: حين تنتهي عقدة ترسل رسائل على حوافها، والعقد التي تستقبل رسائل تصير نشطة وتنفّذ دوالّها، وفي نهاية كل خطوة كبرى تصير العقد التي لا رسائل لها غير نشطة، وينتهي الرسم حين تصير كل العقد غير نشطة ولا رسائل في الطريق.

لهذا التسلسل ثلاث نتائج عملية تفهمها من أول قراءة:

• نقاط الحفظ (checkpoint) تقع على حدود الخطوات الكبرى، لا في منتصف الدالة. لذلك عند الاستئناف «the affected node runs again from the start of its function»، وكل ما قبل التوقّف يُنفَّذ ثانية.
• عدد الخطوات الكبرى محدود. حدّ التكرار (recursion limit) يحدّ أقصى عدد خطوات كبرى في التشغيل الواحد، وعند بلوغه يُرفع GraphRecursionError؛ والافتراضي «Starting in version 1.0.6, the default recursion limit is set to 1000 steps».
• الحالة قد تُكتب إليها من عدة عقد في خطوة واحدة، ولهذا دوالّ الدمج ليست رفاهية بل ضرورة.

كيف تربط العقد بالحواف العادية والشرطية؟

تربط العقد بحافة عادية add_edge للتحوّل الثابت، وبحافة شرطية add_conditional_edges عندما تقرّر الحالة وجهتها، وتُملأ هذه بدالة تقبل الحالة وتعيد اسم العقدة (أو قائمة أسماء).

وحين تُعيد دالة التوجيه أكثر من اسم، تُنفَّذ كل العقد المذكورة في الخطوة الكبرى التالية معاً.

هذا مثال prompt chaining كامل من الوثائق، بحافة شرطية تقود إلى END أو تعيد المسار:

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END


# Graph state
class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str


# Nodes
def generate_joke(state: State):
    """First LLM call to generate initial joke"""

    msg = llm.invoke(f"Write a short joke about {state['topic']}")
    return {"joke": msg.content}


def check_punchline(state: State):
    """Gate function to check if the joke has a punchline"""

    # Simple check - does the joke contain "?" or "!"
    if "?" in state["joke"] or "!" in state["joke"]:
        return "Pass"
    return "Fail"


def improve_joke(state: State):
    """Second LLM call to improve the joke"""

    msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}")
    return {"improved_joke": msg.content}


def polish_joke(state: State):
    """Third LLM call for final polish"""
    msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}")
    return {"final_joke": msg.content}


# Build workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)

# Add edges to connect nodes
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)

# Compile
chain = workflow.compile()

ومتغيّر llm المستخدَم في الأعلى يأتي من تهيئة بسيطة — استخدمنا Anthropic في أمثلة هذه السلسلة، وموزّعون آخرون قابلون للتبديل في سطر واحد كما شرحنا في مقال الواجهة الموحّدة للنماذج:

pip install langchain_core langchain-anthropic langgraph

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

القاموس {"Fail": "improve_joke", "Pass": END} اختياري لكنه أوضح: يربط كل قيمة تُعيدها دالة التوجيه باسم العقدة المقصودة صراحةً، فيظهر من الكود مباشرةً أي مسار يُتَّخذ ومتى ينتهي.

هل تجميع الرسم بـcompile() إلزامي فعلاً؟

نعم، إلزامي بلا استثناء: الوثائق تقول حرفياً «You MUST compile your graph before you can use it»، وما قبل ذلك بنّاء رسم (builder) لا رسم قابل للتشغيل.

graph = builder.compile()

وcompile() توفّر فحوصاً بنيوية بسيطة (لا عقد معلّقة مثلاً)، وهي أيضاً المكان الذي تمرّر فيه checkpointer إن أردت حفظ الحالة بين التشغيلات.

متى تختار Command بدل الحافة الشرطية؟

تختار Command عندما تريد تحديث الحالة والتوجيه في دالة واحدة، والقاعدة صريحة: «Use Command instead of conditional edges if you want to combine state updates and routing in a single function». أما إذا كنت تريد التوجيه فقط دون تحديث فالحافة الشرطية أنظف.

وCommand يقبل أربعة معاملات بالضبط: update لتطبيق تحديثات الحالة، وgoto للانتقال إلى عقدة بعينها، وgraph لاستهداف الرسم الأب عند التنقّل من رسم فرعي، وresume لتوفير قيمة تُستأنف بها التنفيذ بعد مقاطعة.

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

لاحظ تلميح النوع Command[Literal["my_other_node"]]: هو إلزامي عملياً عند الإرجاع من عقدة، لأنه يوثّق للرسم وأداة التصيير قائمة العقد التي يمكن لهذه العقدة أن تنتقل إليها.

ولتنقّل من عقدة داخل رسم فرعي إلى عقدة في الرسم الأب:

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

وتحذير مهم: Command يضيف حوافاً ديناميكية فقط، والحواف الساكنة التي أضفتها بـadd_edge تبقى نافذة. فإذا كانت عقدة تُرجع Command(goto="a") وأنت أضفت لها أيضاً add_edge نحو b، فإن a وb سيعملان معاً. لكل عقدة استخدم Command أو الحواف الساكنة، لا الاثنتين.

كيف تنشئ عدداً مجهولاً مسبقاً من العقد؟ Send وتوزيع العمل

Send هو أداة حين لا تكون الحواف معروفة مسبقاً: حافة شرطية تعيد قائمة من Send("node_name", state_for_that_node)، فتنشئ نسخة عقدة عاملة لكل عنصر، ولكل نسخة حالتها الخاصة، بينما تلتقي كل المخرجات في مفتاح حالة مشترك.

from langgraph.types import Send


# Graph state
class State(TypedDict):
    topic: str  # Report topic
    sections: list[Section]  # List of report sections
    completed_sections: Annotated[
        list, operator.add
    ]  # All workers write to this key in parallel
    final_report: str  # Final report


# Worker state
class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]


def synthesizer(state: State):
    """Synthesize full report from sections"""

    # List of completed sections
    completed_sections = state["completed_sections"]

    # Format completed section to str to use as context for final sections
    completed_report_sections = "\n\n---\n\n".join(completed_sections)

    return {"final_report": completed_report_sections}


# Conditional edge function to create llm_call workers that each write a section of the report
def assign_workers(state: State):
    """Assign a worker to each section in the plan"""

    # Kick off section writing in parallel via Send() API
    return [Send("llm_call", {"section": s}) for s in state["sections"]]


# Build workflow
orchestrator_worker_builder = StateGraph(State)

# Add the nodes
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)

# Add edges to connect nodes
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
    "orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)

# Compile the workflow
orchestrator_worker = orchestrator_worker_builder.compile()

هذا هو نمط مُنسّق وعُمال (orchestrator-worker): المنسّق يقسّم المهمة، والعمال ينفّذون، والمُجمِّع يجمع. لاحظ أن completed_sections له دالة دمج operator.add — لأن كل عامل يكتب في المفتاح نفسه في الخطوة الكبرى ذاتها.

ما أنماط سير العمل الخمسة، ومتى تستخدم كل نمط؟

الوثائق تصنّف سير العمل في خمسة أنماط، وتضع الوكيل حالةً سادسة مستقلّة.

النمط	متى تستخدمه
Prompt chaining — كل استدعاء للنموذج يعالج مخرجات السابق	مهمة محدّدة يمكن تقسيمها إلى خطوات صغيرة قابلة للتحقّق، مثل ترجمة مستند ثم تدقيقه.
Parallelization — عدة مهام فرعية تعمل معاً ثم تجتمع	رفع السرعة بتقسيم المهمة، أو رفع الثقة بتشغيل المهمة نفسها أكثر من مرة بمعيارات تقييم مختلفة.
Routing — مدخل واحد يُوجَّه إلى مسار متخصّص واحد	تصنّف نوع الطلب أولاً ثم تعطيه مساره: أسئلة عن التسعير، الاسترجاع، الخصومات…
Orchestrator-worker — ديناميكية لا تُعرف مسبقاً	حين لا تستطيع تحديد المهام الفرعية مسبقاً، ككتابة تقرير على عدد مجهول من الأقسام.
Evaluator-optimizer — مولّد يولّد، مقيّم يحكم، والمقبوض يعود	حين هناك معيار نجاح واضح لكن التكرار ضروري للوصول إليه، كترجمة تحافظ على المعنى.
الوكيل (agent) — النموذج يقرّر بنفسه	حين المشكلة والحلّ كلاهما غير متوقّعين. «Agents are dynamic and define their own processes and tool usage»، وهم أعلى استقلالية من الأنماط الخمسة.

الفرق الجوهري الذي تعطيه الوثائق: «Workflows have predetermined code paths and are designed to operate in a certain order». فأنت في السير العمل تكتب المسارات، وفي الوكيل تكتب الأدوات والقيود ويتّخذ النموذج الباقي.

متى تكفي الواجهة الوظيفية بدل رسم بياني؟

الواجهة الوظيفية (Functional API) تضيف نفسها إلى كودك الإجرائي بدل أن تفرض عليك رسماً: زخرفة @entrypoint تُعلن نقطة بداية سير عمل، و@task تمثّل وحدة عمل مستقلّة تُنفَّذ داخلها.

استخدمها حين يكون كودك إجرائياً أصلاً (if/else وحلقات واستدعاءات دوال)، أو حين يكون سير عملك خطياً بسيطاً، أو حين تكون حالتك محصورة داخل دالة واحدة ولا تحتاج أن يشاركها أكثر من عقدة. ولا تستخدمها حين تريد تصييراً بصرياً للرسم: الواجهة الوظيفية لا تدعم التصيير لأن الرسم يُبنى أثناء التشغيل.

وتحمل الواجهة الوظيفية بوابة interrupt() كما في الرسم البياني تماماً — فقواعد المقاطعة التي ستقرأها لاحقاً تنطبق عليها بالضبط:

import time

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt


@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1)  # This is a placeholder for a long-running task.
    return f"An essay about topic: {topic}"


@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt(
        {
            # Any json-serializable payload provided to interrupt as argument.
            "essay": essay,
            "action": "Please approve/reject the essay",
        }
    )

    return {
        "essay": essay,  # The essay that was generated
        "is_approved": is_approved,  # Response from HIL
    }

ولاحظ أن نقطة الحفظ هنا تُبنى بشكل مختلف: «In the Graph API a new checkpoint is generated after every superstep. In the Functional API, when tasks are executed, their results are saved to an existing checkpoint associated with the given entrypoint»؛ أي أن نتيجة @task تُستعاد من النقطة بدل إعادة حسابها.

كيف توقف الرسم عند بوابة موافقة بشرية؟

توقفه بـinterrupt(): «Interrupts allow you to pause graph execution at specific points and wait for external input before continuing»، وعندها يحفظ لانغ غراف الحالة بطبقة الاستمرارية وينتظر «indefinitely until you resume execution».

نحتاج ثلاثة أشياء: مُحافِظ نقاط (checkpointer) يحفظ الحالة، وthread_id في الإعداد يقول للمُحافِظ أي حالة يُستأنف، ونداء interrupt() في المكان الذي تريد التوقّف عنده. والفرق الجوهري: نقاط التوقّف الثابتة تتوقّف قبل عقدة أو بعدها، أما interrupt فهو ديناميكي: «they can be placed anywhere in your code and can be conditional based on your application logic».

وهنا بوابة موافقة كاملة من الوثائق — عقدة تسأل، ثم توجّه إلى proceed أو cancel حسب الجواب:

from typing import Literal, Optional, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]


def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
    # Expose details so the caller can render them in a UI
    decision = interrupt(
        {
            "question": "Approve this action?",
            "details": state["action_details"],
        }
    )

    # Route to the appropriate node after resume
    return Command(goto="proceed" if decision else "cancel")


def proceed_node(state: ApprovalState):
    return {"status": "approved"}


def cancel_node(state: ApprovalState):
    return {"status": "rejected"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# Use a more durable checkpointer in production
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.stream_events(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
    version="v3",
)
_ = initial.output  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'question': ..., 'details': ...}),)

# Resume with the decision; True routes to proceed, False to cancel
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
print(resumed.output["status"])

اخترنا هنا stream_events(..., version="v3") عبر كل الأمثلة لأنها تعرض المقاطعات على stream.interrupts والحالة النهائية على stream.output. الواجهة القديمة graph.invoke(...) ما زالت تعمل وتضع المقاطعات تحت result["__interrupt__"]، لكن لا تخلط الأسلوبين في مشروع واحد.

وملاحظتان عن هذا المثال:

• thread_id هو مؤشّرك الدائم. «The thread_id you choose is effectively your persistent cursor. Reusing it resumes the same checkpoint; using a new value starts a brand-new thread with an empty state.» فاستعمال thread_id جديد يفتح محادثة (thread) جديدة بحالة فارغة.
• InMemorySaver مؤقّت. هو هنا لتسهيل التشغيل المحلي؛ في الإنتاج استبدله بمُحافِظ مدعوم بقاعدة بيانات، وإلا ضاعت كل نقاط الحفظ عند إعادة تشغيل العملية. والأوثائق نفسها تعلّق على السطر: «Use a more durable checkpointer in production».

وإن أردت أن تعرض الواجهة نموذج إدخال منظّماً بدل حقل نصّي حر، مرّر response_schema — «Interrupt response schemas require langgraph>=1.2.12» — فيعرضه لانغ سميث ستوديو (LangSmith Studio) كحقول مُنظَّمة، وتصبح قيمة الاستئناف مُتحقَّقاً منها بنوعها.

ما الفخاخ التي ستُكلّفك مالاً حقيقياً في المقاطعة التفاعلية؟

أخطر ما في interrupt خمس قواعد: العقدة تُعاد من بدايتها عند الاستئناف لا من سطر النداء، وحلقة while True مع interrupt() تعيد تنفيذاً أُسيّاً يُضاعف الفاتورة، ولا يجوز تلفيف النداء بـtry/except، والمطابقة بين قيم الاستئناف والاستدعاءات بالفهرس حصراً، وCommand(resume=...) هو المدخل الوحيد الصالح إلى invoke().

⚠️ القاعدة الجامعة: كل سطر تكتبه قبل interrupt() في نفس العقدة سيُنفَّذ ثانية عند الاستئناف. اجعل ما قبل المقاطعة قابلاً للتكرار (idempotent)، أو انقل الأثر الجانبي إلى ما بعدها أو إلى عقدة منفصلة.

العقدة تُعاد من بدايتها، لا من سطر interrupt

«When execution resumes (after you provide the requested input), the runtime restarts the entire node from the beginning—it does not resume from the exact line where interrupt was called.»

بمعنى عملي: كل سطر كتبته قبل interrupt في تلك العقدة سيُنفَّذ مرة أخرى عند الاستئناف. لذلك:

# ✅ Good: using upsert operation which is idempotent
db.upsert_user(
    user_id=state["user_id"],
    status="pending_approval"
)

approved = interrupt("Approve this change?")

return {"approved": approved}

# ❌ Bad: creating a new record before interrupt
# This will create duplicate records on each resume
audit_id = db.create_audit_log({
    "user_id": state["user_id"],
    "action": "pending_approval",
    "timestamp": datetime.now()
})

approved = interrupt("Approve this change?")

return {"approved": approved}

القاعدة العملية: إما اجعل ما قبل المقاطعة قابلية للتكرار (idempotent) بمعنى أنها تعطي النتيجة نفسها لو نُفّذت مرات، أو انقل الأثر الجانبي إلى ما بعد المقاطعة أو إلى عقدة منفصلة.

while True مع interrupt() = إعادة تنفيذ أُسّية

هذه أخطر فخّة في هذه المقالة كلها: «a loop that calls interrupt() multiple times causes each resume to replay all previous iterations: the first resume replays 1 iteration, the second replays 2, and so on. The result is exponential re-execution of any code inside the loop body.»

عملياً، أول استئناف يعيد تكراراً واحداً، والثاني يعيد اثنين، وهكذا — فكل استئناف يضاعف عمل العقدة.

❌ لا تكتب هذا أبداً:

while True:
    answer = interrupt("Enter a value")
    if validate(answer):
        break

البديل الصحيح هو نمط التحقّق من مُدخل بشري: استدعِ interrupt() مرة واحدة فقط لكل استدعاء للعقدة، واحفظ السؤال التالي في الحالة، ثم عُد إلى العقدة بحافة شرطية:

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt


class FormState(TypedDict):
    age: int | None
    pending_question: str | None


def get_age_node(state: FormState):
    question = state.get("pending_question") or "What is your age?"
    answer = interrupt(question)  # called exactly once per node invocation
    if isinstance(answer, int) and answer > 0:
        return {"age": answer, "pending_question": None}
    return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}


def route(state: FormState):
    # Loop back to collect_age until we have a valid age
    return END if state.get("age") is not None else "collect_age"


builder = StateGraph(FormState)
builder.add_node("collect_age", get_age_node)
builder.add_edge(START, "collect_age")
builder.add_conditional_edges("collect_age", route)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "form-1"}}
first = graph.stream_events({"age": None, "pending_question": None}, config=config, version="v3")
_ = first.output  # drive the stream to completion
print(first.interrupts)  # -> (Interrupt(value='What is your age?', ...),)

# Provide invalid data; the node re-prompts via the conditional edge
retry = graph.stream_events(Command(resume="thirty"), config=config, version="v3")
_ = retry.output
print(retry.interrupts)  # -> (Interrupt(value="'thirty' is not a valid age...", ...),)

# Provide valid data; route() returns END and the graph finishes
final = graph.stream_events(Command(resume=30), config=config, version="v3")
print(final.output["age"])  # -> 30

لا تلتفّ try/except حول interrupt

القاعدة مكتوبة بصيغة أمر: «Do not wrap interrupt calls in try/except». السبب أن المقاطعة تتوقّف برمي استثناء خاص، فلو التقطته أنت لبقي التوقّف بلا أثر: لم تعد الحالة محفوظة ولن ينتظرك أحد.

# ❌ Bad: wrapping interrupt in bare try/except
# will catch the interrupt exception
try:
    interrupt("What's your name?")
except Exception as e:
    print(e)

# ✅ Good: catching specific exception types
# will not catch the interrupt exception
try:
    name = interrupt("What's your name?")
    fetch_data()  # This can fail
except NetworkException as e:
    print(e)

القاعدة العملية: افصل نداء interrupt عن الكود القابل للخطأ، والتقط أنواع استثناء محدّدة فقط.

المطابقة بالفهرس حصرياً، والترتيب مهم

«Matching is strictly index-based, so the order of interrupt calls within the node is important.» أي أن لانغ غراف يطابق كل استدعاء interrupt مع قيمة الاستئناف بموضعه، لا بالسياق.

# ❌ Bad: conditionally skipping interrupts changes the order
name = interrupt("What's your name?")

if state.get("needs_age"):
    age = interrupt("What's your age?")

city = interrupt("What's your city?")

return {"name": name, "city": city}

إن كان أول تشغيل تخطّى نداءً وثاني تشغيل لم يتخطّاه، انزاح كل ما بعده وذهب كل جواب إلى السؤال الخطأ. لذلك: عدد استدعاءات interrupt وترتيبها داخل العقدة يجب أن يكونا ثابتين في كل تشغيل.

Command(resume=...) هو المدخل الوحيد الصالح، وما عداه يبدو «عالقاً»

«Command(resume=...) is the only Command pattern intended as input to invoke()/stream()» — أما المعاملات الأخرى (update وgoto وgraph) فمصمَّمة للإرجاع من العقد.

والأخطر: «passing any Command as input resumes from the latest checkpoint … the graph will appear stuck if it already finished». أي أن تمرير Command(update=...) كمدخل يستأنف من آخر نقطة حفظ بدل أن يبدأ من __start__، فيبدو الرسم عالقاً إن كان قد انتهى أصلاً.

# WRONG - graph resumes from the latest checkpoint
# (last step that ran), appears stuck
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# CORRECT - plain dict restarts from __start__
graph.invoke( {
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

وثمة قاعدة أخيرة: القيمة التي تمرّرها إلى interrupt() يجب أن تكون قابلة للتسلسل (JSON). مرّر دالة أو نسخة من صنف وستفشل عند التنفيذ.

ماذا ستبني عملياً؟

أنشئ ملفاً واحداً وضع فيه ثلاث عقد: عقدة draft تولّد نصاً، وعقدة review تستدعي interrupt() وتطلب موافقة، وعقدة ship تنفّذ الفعل الحقيقي (بريد، أو دفعة، أو كتابة في قاعدة بيانات) — بعد الموافقة لا قبلها. ثم اربطها بحافة شرطية تُخرج review إلى END عند الرفض، واجمع الرسم بـcompile(checkpointer=...)، وشغّله بمدخل يحمل thread_id ثابتاً. عند أول استئناف اطبع stream.interrupts، ثم استأنف بـCommand(resume=True).

إن نفّذت ذلك على عقدة فيها عمل حقيقي — إرسال بريد أو دفعة أو كتابة في قاعدة بيانات — فستفهم فوراً لماذا تشترط الوثائق قابلية التكرار: العقدة تُعيد نفسها من بدايتها عند كل استئناف، وهذا سلوك مقصود لا عارض.

نصيحة أخيرة قبل الانتقال: ابدأ بالوضع المحلي على InMemorySaver، ولتشخّص الرسم عقدةً عقدة استخدم نقاط التوقّف الثابتة عبر interrupt_before وinterrupt_after لا بوابة المقاطعة. الوثائق صريحة: «Static interrupts are not recommended for human-in-the-loop workflows. Use the interrupt function instead.»

---

تحتاج وكيلاً يعمل لفريقك؟

نصمّم ونبني أنظمة وكلاء للشركات والأفراد فوق لانغ تشين (LangChain) ولانغ غراف (LangGraph) — من إثبات المفهوم في أسبوع إلى نظام إنتاجي يُشغَّل يومياً. ابدأ بطلبك من صفحة التواصل.