interrupt() से Human-in-the-Loop

रुकें, पूछें, फिर शुरू करें

कुछ कार्रवाइयाँ बिना व्यक्ति के नहीं चलनी चाहिए: refund जारी करना, बाहरी ईमेल भेजना, डेटा हटाना। Node के भीतर interrupt(value) बुलाने से run रुकता है, state checkpointer में सहेजी जाती है और value (सवाल या प्रस्तावित कार्रवाई) caller तक पहुँचता है। बाद में आप उसी thread को Command(resume=answer) से फिर शुरू करते हैं: node interrupt call से उत्तर को लौटे मान के रूप में लेकर आगे बढ़ता है। इसके लिए checkpointer और thread_id चाहिए। चूँकि ठहराव restarts में बचता है, अनुमोदन web UI, Slack संदेश या ईमेल से मिनटों या दिनों बाद आ सकता है। मनुष्य को ठीक दिखाएँ कि क्या होगा, और दर्ज करें किसने अनुमोदित किया।

ठहरें, retry करें, stream करें

Interrupts लोगों के लिए रुकते हैं, retry policies गड़बड़ियाँ सोखती हैं, और streaming प्रगति दिखाती है।

Refund अनुमोदित करना, चलाकर

मैंने यह Python virtual environment में langgraph 1.2.12 और langchain-core 1.6.6 के साथ ऑफ़लाइन चलाया। किसी API key या मॉडल की ज़रूरत नहीं क्योंकि मॉडल की जगह सादे Python functions हैं, इसलिए आउटपुट दोहराने योग्य है। पहला invoke interrupt पर रुकता है और __interrupt__ में सवाल लौटाता है; get_state(...).next दिखाता है कि approve node प्रतीक्षा में है। Command(resume="yes") के साथ दूसरा invoke आगे बढ़ता है और अंतिम status refunded हो जाती है।

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

class State(TypedDict):
    amount: int
    status: str

def propose(state): return {"status": "waiting for approval"}

def approve(state):
    decision = interrupt({"question": f"Approve refund of {state['amount']}?"})   # pauses here
    return {"status": "refunded" if decision == "yes" else "rejected"}

g = StateGraph(State)
g.add_node("propose", propose); g.add_node("approve", approve)
g.add_edge(START, "propose"); g.add_edge("propose", "approve"); g.add_edge("approve", END)
app = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "refund-1"}}
first = app.invoke({"amount": 250, "status": "new"}, cfg)
print("paused with:", first["__interrupt__"][0].value)
print("next node:", app.get_state(cfg).next)
final = app.invoke(Command(resume="yes"), cfg)          # a human answers; the graph continues
print("final:", final)

Output:

paused with: {'question': 'Approve refund of 250?'}
next node: ('approve',)
final: {'amount': 250, 'status': 'refunded'}

दुष्प्रभाव interrupt के बाद रखें

Node फिर शुरू हो तो interrupt से पहले का कोड फिर चलता है। अपरिवर्तनीय कार्रवाइयाँ अनुमोदन के बाद रखें और उससे पहले का कोड दुष्प्रभाव-मुक्त रखें।

त्वरित जाँच: interrupt() के काम करने के लिए क्या चाहिए?

• Checkpointer और thread_id
• GPU
• दूसरा graph
• एक का recursion limit