چطور با interrupt در LangGraph از ایجنت تایید بگیریم

برای آنکه یک ایجنت پیش از انتشار، حذف فایل یا پرداخت، از آدم تأیید بگیرد، به چیزی نیاز دارید که واقعا متوقف شود و وضعیتش را نگه دارد: این با یک حلقه‌ی while به دست نمی‌آید، چون حلقه در حافظه‌ی فرایند است و فرایند در خط لوله پاک می‌شود. در این نوشته با تابع interrupt در LangGraph نسخه‌ی ۱٫۲٫۱۲ یک دروازه‌ی تأیید واقعی می‌سازم، آن را اجرا می‌کنم، و سه رفتار را می‌سنجم که هیچ‌کدام در راهنما به این روشنی نیامده‌اند: گره پس از تأیید از سرِ اول اجرا می‌شود، بدون ذخیره‌ساز وضعیت کد بالا می‌رود، و هر تعداد توقف در یک گره به همان تعداد گذر نیاز دارد.

ایده‌ی ساده است. جایی در مسیر، پیش از کار حساس، یک نقطه‌ی توقف می‌گذارید و وضعیت را بیرون از فرایند نگه می‌دارید تا بعد از پاسخ آدم از همان‌جا ادامه دهید. دشواری در همین یک جمله است: وضعیت باید بیرون از فرایند باشد، و ادامه باید دقیقاً از همان نقطه باشد نه از ابتدا.

چرا حلقه‌ی ساده جواب نمی‌دهد

راه اولی که به ذهن می‌رسد، انتظار کشیدن است: تا وقتی آدم جواب نداده، حلقه تکرار شود. این روی کاغذ درست است و در عمل سه مشکل دارد که هر سه در یک استقرار واقعی خودش را نشان می‌دهد.

اول اینکه فرایند در یک سرور وب یا یک کارگر خط لوله زنده نمی‌ماند. یک حلقه‌ی بی‌پایان نگه‌داشتن یک کارگر را به خود مشغول می‌کند و اگر فرایند بازراه‌اندازی شود، وضعیت از بین می‌رود. دوم اینکه زمان انتظار می‌تواند از سقف پرداخت اتصال بیشتر شود. سوم اینکه به آدم باید چیزی برسد تا جواب بدهد، و در حلقه‌ی همگام هیچ چیزی به او نمی‌رسد.

آنچه لازم است سه تکه است: یک نقطه‌ی توقف که وضعیت را ذخیره کند، یک شناسه برای پیدا کردن آن وضعیت بعداً، و یک فراخوانی دوم برای ادامه. هر سه را کتابخانه می‌دهد، و کاری که باید خودتان انجام دهید انتخاب محل توقف است. در سرورهای پروتکلی همین سه تکه هست و مشخصات پروتکل تعریفشان می‌کند.

ساخت دروازه، در بیست خط

سه قطعه لازم است: یک حالت، یک نقطه‌ی توقف، و یک ذخیره‌ساز. حالت فقط وضعیتی است که بین گره‌ها می‌ماند و در این مثال سه فیلد دارد.

<span class="code-comment"># lg_gate.py -- دروازه‌ی تأیید انسان، کامل و اجراشدنی</span>
import importlib.metadata as md
from typing import TypedDict

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

print("langgraph:", md.version("langgraph"))

SIDE_EFFECTS: list[str] = []


class State(TypedDict):
    query: str
    draft: str
    verdict: str


def retrieve(state: State) -> dict:
    SIDE_EFFECTS.append("search:nginx+cdn+cache")
    return {"query": state["query"],
            "draft": "پیش‌نویس اولیه",
            "verdict": "pending"}


def review(state: State) -> dict:
    decision = interrupt({
        "step": "review-before-publish",
        "draft": state["draft"],
        "options": ["approve", "reject", "approve-with-edit"],
    })
    SIDE_EFFECTS.append(f"publish:{decision}")
    return {"verdict": decision}


def notify(state: State) -> dict:
    SIDE_EFFECTS.append(f"notify:author:{state['verdict']}")
    return {}


g = StateGraph(State)
g.add_node("retrieve", retrieve)
g.add_node("review", review)
g.add_node("notify", notify)
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "review")
g.add_edge("review", "notify")
g.add_edge("notify", "__end__")
app = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "t1"}}

# گذر اول: گراف می‌ایستد و سؤالش را برمی‌گرداند
out = app.invoke({"query": "cdn cache invalidation"}, cfg)
print("متوقف شد:", "__interrupt__" in out)
print("سؤال:", out["__interrupt__"][0].value)
print("verdict:", out["verdict"])
print("اثرهای جانبی:", SIDE_EFFECTS)

# گذر دوم: پاسخ آدم از راه Command
final = app.invoke(Command(resume="approve-with-edit"), cfg)
print("هنوز متوقف است:", "__interrupt__" in final)
print("verdict:", final["verdict"])
print("اثرهای جانبی:", SIDE_EFFECTS)

خروجی واقعی همین کد روی همین ماشین، در دو گذر، چنین است. گذر اول متوقف می‌شود و فقط گره‌ی بازیابی اثر جانبی ساخته است؛ هیچ انتشاری انجام نشده. گذر دوم با پاسخ آدم ادامه می‌یابد، توقف از بین می‌رود، و انتشار و اطلاع‌رسانی پس از پاسخ انجام می‌شوند.

<span class="code-comment"># اجرای lg_gate.py</span>
langgraph          : 1.2.12
langgraph-checkpoint: 4.2.0

=== ۱. گذر اول: بدون ورودی آدم ===
  متوقف شد        : True
  چه چیزی پرسید    : {'step': 'review-before-publish',
                'draft': 'پیش‌نویس اولیه',
                'options': ['approve', 'reject', 'approve-with-edit']}
  شناسه‌ی وقفه      : 7a5b5fbf716e0b6e49792f2f78c060e0
  verdict          : pending
  اثرهای جانبی     : ['search:nginx+cdn+cache']

=== ۲. گذر دوم: پاسخ آدم از راه Command ===
  متوقف است؟      : False
  verdict          : approve-with-edit
  اثرهای جانبی     : ['search:nginx+cdn+cache',
                 'publish:approve-with-edit',
                 'notify:author:approve-with-edit']

گذر	ورودی	وضعیت	تصمیم	انتشار
۱	پرسش کاربر	متوقف	pending	انجام نشد
۲	پاسخ آدم	تمام	approve-with-edit	یک بار

این جدول همان چیزی است که در یک سامانه‌ی واقعی لازم دارید: نه فقط اینکه ایجنت متوقف شد، بلکه اینکه در لحظه‌ی توقف چه چیزی منتظر بود و چه چیزی هنوز انجام نشده بود.

سه چیز را از این خروجی بردارید. نخست، هر توقف یک شناسه دارد و همین شناسه کلید ادامه است. دوم، وضعیت برگشتی شامل همان فیلدهای حالت به‌علاوه‌ی کلید وقفه است، پس شما می‌توانید به آدم نشان دهید دقیقاً چه چیزی منتظر تأیید است. سوم، اثرهای جانبی بعد از توقف یعنی انتشار و اطلاع‌رسانی دقیقاً یک بار و پس از پاسخ انجام شده‌اند، نه دوبار.

رفتاری که هزینه می‌سازد: گره از سر اول اجرا می‌شود

مستندات این را می‌گویند و من هم با آزمون تأییدش کردم. وقتی آدم پاسخ می‌دهد، اجرای گراف از ابتدای همان گره دوباره شروع می‌شود، نه از خط بعد از توقف. یعنی هر کاری که در آن گره پیش از فراخوانی توقف انجام داده‌اید، دوباره انجام می‌شود.

این رفتار برای کار پاک بی‌خطر است و برای کار گران فاجعه است. آزمون زیر یک گره می‌سازد که پیش از توقف یک کار فرضی گران انجام می‌دهد، و فراخوانی‌ها را در هر گذر چاپ می‌کند.

<span class="code-comment"># lg_cost.py -- آیا کارِ پیش از توقف دوباره اجرا می‌شود؟</span>
CALLS: list[str] = []


def expensive_first(state) -> dict:
    CALLS.append("expensive-node")      # یک فراخوانی مدل، فرض کنید
    return {"verdict": "pending"}


def gate(state) -> dict:
    CALLS.append("pre-interrupt-work")  # کارِ قبل از توقف
    decision = interrupt({"ask": "approve?"})
    CALLS.append(f"post:{decision}")     # بعد از توقف
    return {"verdict": decision}


app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "replay"}}

app.invoke({"verdict": "?"}, cfg)
print("گذر اول :", CALLS)

CALLS.clear()
app.invoke(Command(resume="yes"), cfg)
print("گذر دوم :", CALLS)

<span class="code-comment"># اجرای lg_cost.py</span>
=== گذر اول: مدل می‌ایستد ===
  فراخوانی‌ها: ['expensive-node', 'pre-interrupt-work']

=== گذر دوم: آدم جواب می‌دهد ===
  فراخوانی‌ها: ['pre-interrupt-work', 'post:yes']

  گره دوباره از سر اول اجرا شد؟ خیر

جواب روشن است. گره دوم اجرا نشد، اما کارِ داخل گره دروازه دوباره اجرا شد، چون از سر گره شروع شده‌ایم. اگر آن فراخوانیِ گران داخل گره دروازه می‌بود، دو بار پرداخت می‌کردید: یک بار پیش از توقف و یک بار بعد از تأیید آدم.

قاعده‌ی عملی یک خط است: فراخوانی توقف باید نخستین دستور آن گره باشد و هر کاری پس از آن بیاید. اگر مجبورید قبل از توقف چیزی در همان گره انجام دهید، آن را به گره‌ی جداگانه قبلی ببرید، دقیقاً مثل گره‌ی بازیابی در نمونه‌ی اول که پیش از دروازه است و در گذر دوم تکرار نشد.

نکته‌ی دوم که با همین آزمون روشن شد، ترتیب است. اگر در یک گره دو نقطه‌ی توقف داشته باشید، هر کدام یک گذر جدا می‌خواهد و پاسخ‌ها به ترتیب تطبیق داده می‌شوند، نه با کلید.

<span class="code-comment"># دو توقف در یک گره = سه گذر</span>
def two_interrupts(state) -> dict:
    a = interrupt("اول")
    b = interrupt("دوم")
    return {"log": f"{a}|{b}"}

out1 = app2.invoke({"log": ""}, c2)   # می‌پرسد: اول
out2 = app2.invoke(Command(resume="A"), c2)  # می‌پرسد: دوم
out3 = app2.invoke(Command(resume="B"), c2)  # تمام: {'log': 'A|B'}

این رفتار درست است اما یک تله دارد. اگر در گذر دوم به‌جای پاسخ دوم، پاسخ اول را بفرستید، بین پاسخ و سؤال جابه‌جایی می‌افتد و بی‌صدا اشتباه می‌شود. اگر چند نقطه‌ی تأیید دارید، هر کدام را در گره‌ی خودش بگذارید و وضعیت مرحله را در حالت نگه دارید تا بدانید کدام پاسخ را باید فرستاد.

سه حالت شکست که باید جلویشان را بگیرید

دو آزمون منفی روی همین کد، دو خطای واقعی نشان دادند که در استقرار ناگهانی ظاهر می‌شوند.

آزمون نخست: حذف ذخیره‌ساز وضعیت. اگر گراف را بدون ذخیره‌ساز کامپایل کنید، فراخوانی اول هیچ خطایی نمی‌دهد و کلید توقف حتی در خروجی هست. یعنی نشانه‌ی توقف سر جایش است و شما فکر می‌کنید همه‌چیز درست کار می‌کند. خطا فقط در فراخوانی دوم می‌آید، آن هم با پیامی که درباره‌ی کار شما نیست. راهنمای مدیریت خطا همین جدایی را در سرورها استاندارد کرده است.

<span class="code-comment"># بدون ذخیره‌ساز: گذر اول ساکت، گذر دوم خطا</span>
app2 = g.compile()          # InMemorySaver جا افتاده
out2 = app2.invoke({"query": "x"}, {"configurable": {"thread_id": "t9"}})
print("__interrupt__ حاضر است؟", "__interrupt__" in out2)   # True

app2.invoke(Command(resume="approve"), {"configurable": {"thread_id": "t9"}})
# RuntimeError: Cannot use Command(resume=...) without checkpointer

آزمون دوم: پاسخی که شکل درست اما محتوای غلط دارد. اگر به‌جای رشته یک واژه از کلیدهای ناشناخته بفرستید، هیچ اعتبارسنجی انجام نمی‌شود. همان واژه عیناً در وضعیت می‌نشیند و بعد به‌عنوان تصمیم به گره‌ی بعدی می‌رود و همان‌جا هم به‌عنوان تصمیم ثبت می‌شود.

<span class="code-comment"># آزمون منفی: پاسخ بدشکل بی‌صدا قبول می‌شود</span>
app3.invoke({"query": "y"}, c3)
bad = app3.invoke(Command(resume={"unexpected": "object"}), c3)
print("verdict:", bad["verdict"])
# {'unexpected': 'object'}
# و در ادامه: publish:{'unexpected': 'object'}

پس سه حالت را جدا نگه دارید، چون درجه‌ی خطرشان یکسان نیست:

نبودِ ذخیره‌ساز وضعیت: خرابی خاموش در گذر اول و خطای گمراه‌کننده در گذر دوم.
پاسخ نادرست: خرابی خاموش در هر دو گذر، بدون هیچ خطایی.
چند توقف در یک گره: درست کار می‌کند اما جابه‌جایی پاسخ را بی‌صدا ممکن می‌کند.

این بدترین یافته‌ی این بررسی است، چون هیچ نشانه‌ای ندارد. ایجنت با اطمینان کامل یک تصمیم نامعتبر را منتشر می‌کند. پس اعتبارسنجی را داخل خود گره بگذارید، نه در کد فراخواننده.

<span class="code-comment"># اعتبارسنجی داخل گره، پیش از هر کار حساس</span>
def review(state: State) -> dict:
    decision = interrupt({
        "step": "review-before-publish",
        "options": ["approve", "reject", "approve-with-edit"],
    })
    if decision not in ("approve", "reject", "approve-with-edit"):
        return {"verdict": f"rejected:bad-input:{decision!r}"}
    if decision == "approve":
        SIDE_EFFECTS.append("publish:approved")
    return {"verdict": decision}

همین الگوی سه‌مرحله‌ای را می‌توانید برای هر عمل حساس به کار ببرید: گره‌ی آماده‌سازی که کار گران را یک بار انجام می‌دهد، گره‌ی دروازه که فقط توقف و اعتبارسنجی دارد، و گره‌ی اجرا که پس از تأیید کار می‌کند. اگر هر سه را در یک گره بریزید، هر بار که آدم پاسخ می‌دهد کار گران را دوباره پرداخت می‌کنید. در پیش‌پرواز پیش از تولید اسکیل همین ترتیب روی اسکیل‌ها اجرا شده است.

دو چیز را هم اینجا صریح بگویم. این نوشته از ذخیره‌ساز درون‌حافظه‌ای استفاده می‌کند که با هر راه‌اندازی دوباره پاک می‌شود؛ برای کاری که چند دقیقه در انتظار آدم می‌ماند کافی است، اما برای کاری که تا فردا منتظر است باید ذخیره‌ساز پایدار بگذارید. و پاسخ آدم از راه فراخوانی دوم می‌آید، پس اگر کار شما در یک فرایند وب است، شناسه‌ی رشته را در پایگاه‌داده‌ی خودتان نگه دارید تا بتوانید بعد از هر راه‌اندازی دوباره آن را پیدا کنید.

منابع

مدیریت خطا در سرور
مشخصات پروتکل
مستندات کلاینت — الگوی توقف و ادامه
بسته‌ی mcp
نسخه‌های پروتکل
بسته‌ی mcp-types
مخزن guizang-yingzao-skill — دروازه‌ی پیش‌پرواز

چرا حلقه‌ی ساده جواب نمی‌دهد

ساخت دروازه، در بیست خط

رفتاری که هزینه می‌سازد: گره از سر اول اجرا می‌شود

سه حالت شکست که باید جلویشان را بگیرید

منابع

دیدگاه‌ها