LangGraph چیست؟ آموزش ساخت Workflow دارای State و Human-in-the-loop با Python

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

در این راهنما به‌جای اتصال فوری به یک مدل ابری، یک پروژه کاملاً محلی می‌سازیم: سامانه تریاژ تیکت که متن درخواست را بررسی می‌کند، اولویت را تشخیص می‌دهد، درخواست‌های پرریسک را برای تأیید اپراتور متوقف می‌کند و سپس نتیجه را ثبت می‌کند. این مثال کمک می‌کند اجزای LangGraph را بدون درگیرشدن با API Key، هزینه مدل یا تفاوت Providerها بفهمید.

LangGraph چیست؟

LangGraph یک Framework سطح پایین برای Orchestration عامل‌ها و Workflowهای طولانی و دارای State است. در آن جریان برنامه به شکل Graph تعریف می‌شود: State داده مشترک اجرای جاری است، Nodeها واحدهای کار هستند و Edgeها تعیین می‌کنند بعد از هر Node چه مسیری اجرا شود. این طراحی برای Durable Execution، Streaming، حافظه، Human-in-the-loop و بازیابی پس از خطا مناسب است.

LangGraph خودِ مدل زبانی نیست و استفاده از آن الزاماً به LangChain وابسته نیست. می‌توانید Nodeها را با توابع معمول Python بسازید، از هر SDK مدل دلخواه استفاده کنید یا حتی مانند پروژه این مقاله هیچ مدلی را فراخوانی نکنید. ارزش اصلی LangGraph در کنترل چرخه اجرا، نگه‌داری State و امکان توقف و ادامه است.

LangGraph چه تفاوتی با LangChain دارد؟

این دو رقیب مستقیم نیستند. در معماری رسمی اکوسیستم، LangChain Framework سطح بالاتر و LangGraph Runtime مربوط به Orchestration است. اگر یک Agent متداول با Tool Calling می‌خواهید، شروع از API سطح بالاتر LangChain سریع‌تر است. اگر مراحل، توقف‌ها، Branchها و سیاست ادامه اجرا باید دقیقاً تحت کنترل شما باشد، LangGraph انتخاب طبیعی‌تری است.

پیش‌نیاز و نصب LangGraph

نسخه فعلی مستندات Python 3.10 یا جدیدتر را توصیه می‌کند. برای جلوگیری از تداخل Dependencyها، یک Virtual Environment بسازید و Package را در همان محیط نصب کنید. پروژه آموزشی ما فقط به بسته اصلی langgraph نیاز دارد.

python -m venv .venv# Windows PowerShell: .venv\Scripts\Activate.ps1# Linux/macOS: source .venv/bin/activatepython -m pip install --upgrade pippython -m pip install -U langgraphpython -m pip show langgraph

در پروژه واقعی نسخه Dependency را پس از آزمایش در فایل Lock ثابت کنید. به‌روزرسانی بدون کنترل Packageهای زیرساختی می‌تواند رفتار Runtime یا Checkpointer را تغییر دهد. همچنین API Key را داخل کد، فایل نمونه یا خروجی Log قرار ندهید؛ پروژه این مقاله اصلاً Secret نیاز ندارد.

مدل ذهنی State، Node و Edge

State مانند یک شیء مشترک است که Snapshot اجرای Workflow را توصیف می‌کند. هر Node بخشی از State را می‌خواند و فقط Update مورد نیاز را برمی‌گرداند. Runtime این Update را روی State اعمال می‌کند. Edge ثابت یا شرطی، Node بعدی را انتخاب می‌کند.

یک نکته طراحی مهم این است که State را کوچک و معنادار نگه دارید. ذخیره کل Responseهای سنگین یا Objectهای غیرقابل Serialize، بازیابی و Debug را دشوار می‌کند. معمولاً شناسه، داده لازم برای تصمیم و خروجی خلاصه هر مرحله کافی است.

ساخت Workflow تریاژ تیکت

ابتدا Schema را با TypedDict تعریف می‌کنیم. سپس سه Node می‌سازیم: classify برای تشخیص اولویت، review برای توقف و دریافت تصمیم انسان و resolve برای تولید نتیجه نهایی. تشخیص اولویت در این آموزش Rule-based است تا رفتار مثال تکرارپذیر بماند.

from typing import Literalfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.types import Command, interrupt class TicketState(TypedDict, total=False):    ticket_id: str    text: str    priority: Literal["normal", "high"]    approved: bool    result: str def classify(state: TicketState) -> dict:    risky_words = {"حذف", "پرداخت", "دسترسی", "امنیت"}    is_high = any(word in state["text"] for word in risky_words)    return {"priority": "high" if is_high else "normal"} def route_after_classify(state: TicketState) -> Literal["review", "resolve"]:    return "review" if state["priority"] == "high" else "resolve" def review(state: TicketState) -> dict:    decision = interrupt({        "ticket_id": state["ticket_id"],        "question": "آیا رسیدگی به این درخواست حساس تأیید می‌شود؟",    })    return {"approved": bool(decision)} def resolve(state: TicketState) -> dict:    if state["priority"] == "high" and not state.get("approved", False):        return {"result": "درخواست حساس رد شد؛ اقدامی انجام نشد."}    return {"result": "تیکت برای رسیدگی ایمن ثبت شد."} builder = StateGraph(TicketState)builder.add_node("classify", classify)builder.add_node("review", review)builder.add_node("resolve", resolve)builder.add_edge(START, "classify")builder.add_conditional_edges("classify", route_after_classify)builder.add_edge("review", "resolve")builder.add_edge("resolve", END)graph = builder.compile(checkpointer=InMemorySaver())

تابع classify یک Update کوچک برمی‌گرداند و State ورودی را درجا تغییر نمی‌دهد. route_after_classify صرفاً نام مسیر بعدی را انتخاب می‌کند. در مسیر high، Node مربوط به review اجرا می‌شود؛ در مسیر normal، Graph مستقیم به resolve می‌رود.

InMemorySaver برای Demo و Test مناسب است، اما با پایان Process داده آن از بین می‌رود و میان چند Instance مشترک نیست. در Production باید Checkpointer پایدار و سازگار با زیرساخت خود انتخاب کنید. ساختار دقیق اتصال و مدیریت Schema را از مستندات همان Checkpointer دنبال کنید و Secret پایگاه‌داده را از Secret Store تزریق کنید.

اجرای مسیر عادی

برای هر جریان مستقل یک thread_id یکتا بسازید. Checkpointer از این شناسه برای یافتن State و Checkpointهای همان جریان استفاده می‌کند. استفاده دوباره از شناسه یک مشتری برای مشتری دیگر، هم نتیجه اشتباه می‌سازد و هم خطر نشت داده دارد.

from ticket_graph import graph config = {"configurable": {"thread_id": "ticket-1001"}}result = graph.invoke(    {"ticket_id": "1001", "text": "راهنمای تغییر ایمیل حساب را می‌خواهم"},    config=config,)print(result["priority"])print(result["result"])

چون متن تیکت هیچ کلمه حساس تعریف‌شده‌ای ندارد، مسیر classify به resolve می‌رود و Graph در END تمام می‌شود. برای مشاهده وضعیت ذخیره‌شده می‌توانید State همان Thread را دریافت کنید؛ این قابلیت در Debug و ساخت صفحه وضعیت عملیات مفید است.

توقف برای Human-in-the-loop و ادامه اجرا

درخواست حساس نباید بی‌درنگ اجرا شود. تابع interrupt یک Payload قابل Serialize به Caller برمی‌گرداند و اجرای Graph را متوقف می‌کند. پس از دریافت تصمیم اپراتور، همان Graph را با همان thread_id و یک Command از نوع resume فراخوانی می‌کنیم.

from langgraph.types import Commandfrom ticket_graph import graph config = {"configurable": {"thread_id": "ticket-2001"}}paused = graph.invoke(    {"ticket_id": "2001", "text": "درخواست حذف دسترسی مدیر را دارم"},    config=config,) if "__interrupt__" in paused:    print(paused["__interrupt__"])    operator_approved = True    completed = graph.invoke(        Command(resume=operator_approved),        config=config,    )    print(completed["result"])

Payload وقفه باید داده‌ای ساده و JSON-serializable مانند String، Number، Boolean، List یا Dictionary باشد. Objectهای پیچیده، Connection باز یا Exception را در Payload قرار ندهید. در برنامه وب واقعی، Pause را در پایگاه داده نگه می‌دارید، درخواست تأیید را به پنل اپراتور می‌فرستید و پس از احراز هویت و ثبت Audit، فرمان resume را اجرا می‌کنید.

قانون مهم Interrupt: Node از ابتدا اجرا می‌شود

پس از Resume، Node حاوی interrupt از ابتدای تابع دوباره اجرا می‌شود؛ اجرای Python از خط بعد وقفه ادامه پیدا نمی‌کند. بنابراین هر Side Effect پیش از interrupt ممکن است دوباره رخ دهد. ارسال ایمیل، برداشت وجه، حذف رکورد یا فراخوانی API را پیش از وقفه انجام ندهید، مگر عملیات Idempotent طراحی شده باشد.

def safe_review(state: TicketState) -> dict:    approved = interrupt({        "ticket_id": state["ticket_id"],        "action": "revoke-admin-access",    })    return {"approved": bool(approved)} def execute_after_approval(state: TicketState) -> dict:    if not state.get("approved"):        return {"result": "لغو شد"}    # عملیات واقعی را با کلید idempotency یکتا اجرا کنید.    return {"result": "تأیید و اجرا شد"}

interrupt را داخل try/except عمومی نپیچید، زیرا سازوکار توقف از یک Signal ویژه Runtime استفاده می‌کند. همچنین ترتیب چند interrupt در یک Node را میان اجراها تغییر ندهید. ساده‌ترین طراحی قابل اعتماد این است که هر Node یک نقطه تأیید روشن داشته باشد و Side Effect حساس در Node بعدی انجام شود.

Checkpoint و thread_id دقیقاً چه کاری می‌کنند؟

وقتی Graph با Checkpointer Compile شود، State در مرز گام‌های اجرا به شکل Checkpoint ذخیره می‌شود. Checkpointها بر اساس Thread سازمان‌دهی می‌شوند. به همین دلیل Human-in-the-loop، حافظه کوتاه‌مدت، مشاهده تاریخچه State، Time Travel و بازیابی پس از Failure ممکن می‌شود.

thread_id را می‌توان Cursor پایدار یک عملیات دانست، نه شناسه کاربر. یک کاربر ممکن است هم‌زمان چند Workflow داشته باشد و هرکدام باید Thread مستقل بگیرند. نگاشت میان شناسه دامنه مانند ticket_id و thread_id را در لایه Application کنترل کنید و پیش از Resume مجوز کاربر را دوباره بسنجید.

مشاهده State و تاریخچه

config = {"configurable": {"thread_id": "ticket-2001"}}snapshot = graph.get_state(config)print(snapshot.values)print(snapshot.next) history = list(graph.get_state_history(config))for item in history:    print(item.metadata, item.next)

تاریخچه برای Debug بسیار مفید است، اما ممکن است متن ورودی، تصمیم اپراتور یا داده حساس را در خود داشته باشد. Retention، Encryption، دسترسی و حذف داده را از ابتدا طراحی کنید. Log عمومی جای مناسبی برای چاپ کامل State نیست.

Graph API یا Functional API؟

LangGraph دو سبک اصلی دارد. Graph API برای تعریف صریح State، Node و Edge مناسب است و نمودار جریان را خوانا می‌کند. Functional API با Decoratorهای entrypoint و task به کد Imperative نزدیک‌تر است و برای افزودن Persistence و Interrupt به یک Workflow موجود می‌تواند کم‌حجم‌تر باشد.

برای پروژه‌ای با Branchهای متعدد، حلقه Agent و نقاط تأیید مشخص، Graph API خوانایی بیشتری دارد. اگر Workflow فعلی چند تابع و شرط روشن دارد و فقط Durable Execution لازم است، Functional API را نیز ارزیابی کنید.

افزودن مدل زبانی بدون خراب‌کردن معماری

بعد از اینکه Graph محلی درست کار کرد، می‌توانید Node مربوط به classify یا تولید پاسخ را با فراخوانی یک مدل جایگزین کنید. قرارداد Node را ثابت نگه دارید: ورودی از State، خروجی یک Update معتبر. نتیجه مدل را پیش از ورود به State اعتبارسنجی کنید و برای مسیرهای حساس هرگز صرفاً به متن آزاد مدل اعتماد نکنید.

• خروجی مدل را با Schema ساخت‌یافته محدود و Validate کنید.
• Timeout، Retry محدود و Circuit Breaker برای Provider در نظر بگیرید.
• Prompt و نسخه مدل را برای Audit ثبت کنید، اما Secret و داده حساس را Log نکنید.
• عملیات حساس را پشت Policy قطعی و تأیید انسانی نگه دارید.
• Nodeهای دارای Side Effect را Idempotent طراحی کنید.

اگر ابتدا می‌خواهید مدل را روی سیستم خود اجرا کنید، راهنمای نصب Ollama در ویندوز نقطه شروع مناسبی است. برای کار با محیط‌های آزمایشی ابری و نگه‌داری امن کلید نیز راهنمای Google AI Studio و API Key را ببینید.

خطاهای رایج در LangGraph

چک‌لیست Production

1. برای هر Workflow شناسه Thread یکتا و قابل ردیابی تعریف کنید.
2. Checkpointer پایدار، Encryption، Backup و Retention مناسب داشته باشد.
3. دسترسی به State و Resume با Authorization سمت Server کنترل شود.
4. Nodeهای Side Effect دارای Idempotency Key باشند.
5. ورودی و خروجی مدل با Schema معتبر شوند.
6. Timeout و Retry محدود باشد و Failure بی‌نهایت تکرار نشود.
7. Interrupt Payload فقط داده لازم و غیرحساس را حمل کند.
8. Metrics شامل زمان هر Node، تعداد Pause و نرخ Failure باشد.
9. مسیر رد تأیید به اندازه مسیر قبول آزمایش شود.
10. سناریوی Restart Process و ادامه از Checkpoint در تست Integration پوشش داده شود.

چه زمانی LangGraph انتخاب مناسبی نیست؟

برای یک Chat ساده با یک فراخوانی مدل، افزودن Graph و Persistence احتمالاً پیچیدگی بی‌دلیل است. برای Jobهای کاملاً قطعی و زمان‌بندی‌شده نیز یک Queue یا Workflow Engine عمومی ممکن است انتخاب مناسب‌تری باشد. LangGraph زمانی ارزش بیشتری دارد که State عامل، Branch پویا، Loop، Streaming، توقف انسانی یا ادامه طولانی‌مدت بخشی واقعی از مسئله باشد.

قبل از انتخاب، یک Diagram کوچک از مراحل بکشید. اگر چند Branch، بازگشت، Pause یا نیاز به بازیابی مشاهده می‌کنید، Graph مدل ذهنی مفیدی فراهم می‌کند. اگر جریان فقط دریافت ورودی و تولید خروجی است، از ساده‌ترین معماری شروع کنید.

جمع‌بندی

LangGraph کنترل زیرساختی لازم برای Workflowهای دارای State و عامل‌های طولانی‌مدت را فراهم می‌کند. State داده مشترک را تعریف می‌کند، Nodeها واحدهای کارند، Edgeها مسیر را می‌سازند، Checkpointer نقطه‌های بازیابی را نگه می‌دارد و interrupt امکان دخالت امن انسان را ایجاد می‌کند.

پروژه تریاژ این مقاله عمداً بدون LLM ساخته شد تا معماری اصلی شفاف بماند. ابتدا جریان قطعی، توقف، Resume و شکست را درست کنید؛ سپس مدل و ابزارها را پشت قراردادهای معتبر به Nodeها اضافه کنید.

همین حالا مثال تریاژ را اجرا کن، یک تیکت عادی و یک تیکت حساس بساز و تفاوت مسیر مستقیم با توقف Human-in-the-loop را در State ببین.

منابع رسمی

• LangGraph overview
• Install LangGraph
• Graph API overview
• Use the Graph API
• Persistence
• Interrupts
• Functional API overview