ماشین وضعیت ایجنت‌ها: گراف‌ها، گره‌ها و چک‌پوینت‌ها در LangGraph

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

راه‌حل این چالش، مدل‌سازی صریح ایجنت به‌عنوان یک ماشین وضعیت (State Machine) بر بستر گراف است. فریم‌ورک LangGraph با تعریف گره‌ها برای اجرای وظایف و یال‌ها برای انتقال داده، بستری فراهم می‌کند که قابلیت‌های کلیدی مانند چک‌پوینت‌گیری (ذخیره‌ی وضعیت گام‌به‌گام)، توقف موقت برای نظارت انسانی (Human-in-the-loop)، استریمینگ دقیق و سفر در زمان برای دیباگ را به ساختار برنامه اضافه می‌کند.

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

یک حلقه‌ی ReAct که دستی نوشته شود، صرفاً یک while True است. اما همین حلقه وقتی در قالب یک گراف صریح پیاده شود، قابلیت چک‌پوینت گرفتن، متوقف شدن، branch زدن و سفر در زمان پیدا می‌کند. ایجنت تغییری نکرده؛ بستری که آن را احاطه کرده عوض شده است.

صورت مسئله

ایجنت مبتنی بر فراخوانی ابزار (function-calling agent) خود را مستقر می‌کنید. سه نوبت گفت‌وگو (turn) به‌درستی کار می‌کند و بعد مشکلی پیش می‌آید: مدل ابزاری را صدا می‌زند که خطای 500 برمی‌گرداند، کاربر وسط کار نظرش عوض می‌شود، یا ایجنت تصمیم می‌گیرد بدون تأیید انسان مبلغ سفارش را پس بدهد. یک حلقه‌ی while True: هیچ قلابی (hook) برای مهار این شرایط ندارد؛ نمی‌توانید موقتاً متوقفش کنید، به عقب برگردانیدش یا مسیری جداگانه باز کنید تا ببینید «اگر مدل آن ابزار دیگر را انتخاب می‌کرد چه می‌شد». به محض اینکه چنین کدی از مرحله‌ی دمو فراتر برود، ایجنت به جعبه‌ی سیاهی تبدیل می‌شود که یا درست کار کرده یا شکست خورده است.

راه‌حل این مشکل بلافاصله پس از دیدن آن روشن می‌شود. ایجنت در ذات خود یک ماشین وضعیت (state machine) است: پرامپت سیستم، تاریخچه‌ی پیام‌ها، فراخوانی‌های در انتظار ابزار و اقدام بعدی. کافی است این ماشین وضعیت را صریح و شفاف تعریف کنید: گره‌هایی برای «تفکر مدل»، «اجرای ابزار»، «تأیید انسان» و یال‌هایی برای انتقال‌های شرطی بین آن‌ها. به محض اینکه گراف صریح شد، بستر اجرایی شما چهار قابلیت حیاتی را به‌رایگان به دست می‌آورد: چک‌پوینت‌گیری (ذخیره‌ی وضعیت بین مراحل)، توقف موقت (وقفه برای تأیید انسان)، استریمینگ (ارسال توکن‌ها و رویدادهای میانی) و سفر در زمان (بازگشت به وضعیت پیشین (prior) و امتحان یک branch دیگر).

پیاده‌سازی مرجع این انتزاع، LangGraph است. این ابزار به معنای متداول در LangChain یک فریم‌ورک ایجنت نیست («این یک AgentExecutor است، موفق باشید»). بلکه یک محیط اجرای (runtime) گراف با پشتیبانی درجه‌یک از وضعیت، ماندگاری داده‌ها و وقفه‌ها است. در اینجا حلقه‌ی ایجنت (agent loop) چیزی است که آن را ترسیم می‌کنید، نه اینکه دستی کدنویسی کنید.

مفهوم

هر StateGraph سه بخش اصلی دارد:

1. وضعیت (State). یک دیکشنری نوع‌دار (TypedDict یا مدل Pydantic) که در طول گراف جریان می‌یابد. هر گره کل وضعیت را دریافت می‌کند و یک به‌روزرسانی جزئی پس می‌دهد. LangGraph این به‌روزرسانی را با استفاده از یک کاهنده (reducer) به ازای هر فیلد merge می‌کند؛ برای مثال operator.add برای لیست‌هایی که باید تجمیع شوند و حالت پیش‌فرض که بازنویسی مقدار قبلی است.
2. گره‌ها (Nodes). توابع پایتونی به شکل state -> partial_state. هر کدام یک مرحله‌ی مشخص و مستقل‌اند: «فراخوانی مدل»، «اجرای ابزارها»، «خلاصه‌سازی».
3. یال‌ها (Edges). انتقال‌های بین گره‌ها. یال‌های ایستا همیشه به یک مقصد می‌روند. یال‌های شرطی یک تابع مسیریاب (router) به شکل state -> next_node_name دریافت می‌کنند تا گراف بتواند بر اساس خروجی مدل branch بزند.

شما گراف را کامپایل می‌کنید. فرایند کامپایل ساختار گراف را تثبیت می‌کند، یک ثبت‌کننده‌ی چک‌پوینت (checkpointer) به آن متصل می‌سازد (اختیاری است اما برای محیط پروداکشن ضروری است) و یک شیء قابل‌اجرا (runnable) تحویل می‌دهد. سپس آن را با یک وضعیت اولیه و یک thread_id فرامی‌خوانید. در هر مرحله از اجرا، یک چک‌پوینت با کلید (thread_id, checkpoint_id) ذخیره و ماندگار می‌شود.

چهار ابرقدرت

چک‌پوینت‌گیری (Checkpointing). هر جابه‌جایی بین گره‌ها، وضعیت جدید را در یک ذخیره‌ساز می‌نویسد (حافظه‌ی رم برای تست‌ها؛ Postgres/Redis/SQLite برای محیط پروداکشن). برای ادامه‌ی کار، کافی است گراف را مجدداً با همان thread_id فراخوانی کنید. گراف دقیقاً از همان جایی که متوقف شده بود کار را پی می‌گیرد.

توقف موقت (Interrupts). گره‌ای را با interrupt_before=["human_review"] علامت‌گذاری کنید تا فرایند اجرا درست قبل از اجرای آن گره متوقف شود. وضعیت ذخیره می‌ماند. API شما به کاربر پاسخ «در انتظار تأیید» می‌دهد. درخواست بعدی به همان thread_id با استفاده از Command(resume=...) فرایند اجرا را از سر می‌گیرد.

استریمینگ (Streaming). متد graph.stream(state, mode="updates") تغییرات وضعیت را به محض وقوع مخابره می‌کند. حالت mode="messages" توکن‌های LLM را در گره‌های مدل استریم می‌کند. حالت mode="values" اسنپ‌شات‌های کامل وضعیت را ارائه می‌دهد. انتخاب با شماست که کدام را در رابط کاربری خود نمایش دهید.

سفر در زمان (Time-travel). متد graph.get_state_history(thread_id) کل لاگ چک‌پوینت‌ها را برمی‌گرداند. ارسال هر کدام از مقادیر پیشین checkpoint_id به graph.invoke باعث می‌شود فرایند از همان نقطه fork شود. این ویژگی برای دیباگ عالی است («اگر مدل به‌جای این، ابزار B را انتخاب می‌کرد چه می‌شد؟») و برای تست‌های رگرسیون که ردپای اجرای پروداکشن را بازپخش می‌کنند فوق‌العاده کاربرد دارد.

اصل ماجرا کاهنده‌ها (Reducers) هستند

هر فیلد وضعیت یک کاهنده دارد. رفتار پیش‌فرض برای بیشتر موارد مناسب است: مقدار جدید جایگزین مقدار قدیمی می‌شود. اما لیست پیام‌ها به operator.add نیاز دارند تا پیام‌های جدید به‌جای جایگزین شدن، اضافه شوند. یال‌های موازی به‌روزرسانی‌های خود را از طریق کاهنده merge می‌کنند. اگر دو گره هر دو فیلد messages را به‌روزرسانی کنند و شما Annotated[list, add_messages] را فراموش کرده باشید، گره دوم بدون هیچ خطایی برنده می‌شود و نیمی از نوبت گفت‌وگو از دست می‌رود. کاهنده تنها بخش ظریف و حساس این کتابخانه است؛ اگر آن را درست پیاده کنید، باقی قطعات به‌خوبی کنار هم می‌نشینند.

گراف ReAct در چهار گره

یک ایجنت ReAct واقعی برای محیط پروداکشن از چهار گره و دو یال تشکیل می‌شود:

1. گره agent — مدل زبانی را با تاریخچه‌ی پیام‌های فعلی فراخوانی می‌کند و پیام دستیار (assistant message) را پس می‌دهد (که ممکن است حاوی tool_calls باشد).
2. گره tools — هرگونه tool_calls موجود در آخرین پیام دستیار را اجرا کرده و نتایج ابزارها را در قالب پیام‌های ابزار (tool messages) اضافه می‌کند.
3. یک یال شرطی از agent که اگر آخرین پیام حاوی tool_calls باشد مسیر را به tools هدایت می‌کند و در غیر این صورت به END می‌رود.
4. یک یال ایستا از tools که به agent برمی‌گردد.

همین و بس. شما کل حلقه‌ی ReAct (شامل تفکر ← اقدام ← مشاهده ← تفکر ← ...) را با پشتیبانی از چک‌پوینت، وقفه و استریمینگ، تنها در حدود 40 خط کد در اختیار خواهید داشت.

مقایسه‌ی StateGraph و Send (توزیع موازی یا Fanout)

دستور Send(node_name, state) به یک گره اجازه می‌دهد زیرگراف‌های موازی را اجرا کند. برای مثال: ایجنت تصمیم می‌گیرد به‌طور هم‌زمان از سه بازیاب داده (retriever) پرس‌وجو کند. هر Send یک اجرای موازی از گره مقصد را کلید می‌زند؛ خروجی آن‌ها از طریق کاهنده‌ی وضعیت با یکدیگر merge می‌شوند. LangGraph به این شیوه الگوی ارکستریتور-کارگر (orchestrator-workers) را بدون نیاز به ساختارهای چندریسمانی (threading primitives) پیاده‌سازی می‌کند.

زیرگراف‌ها (Subgraphs)

یک گراف کامپایل‌شده می‌تواند به‌عنوان یک گره در دل گرافی دیگر قرار بگیرد. گراف بیرونی آن را یک تک‌گره می‌بیند؛ در حالی که گراف درونی وضعیت و چک‌پوینت‌های مستقل خودش را دارد. تیم‌ها به همین شکل ایجنت‌های ناظر-کارگر (supervisor-worker) را می‌سازند: گراف ناظر نیت (intent) کاربر را به زیرگراف کارگر مربوط به هر دامنه (scope) هدایت می‌کند.

پیاده‌سازی عملی

مرحله‌ی 1: وضعیت و گره‌ها

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def agent_node(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

tool_node = ToolNode(tools=[search_web, read_file])

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile(checkpointer=MemorySaver())

تابع add_messages کاهنده‌ای است که باعث می‌شود لیست پیام‌ها به‌جای رونویسی، تجمیع شوند. فراموش کردن آن رایج‌ترین باگ در LangGraph است.

مرحله‌ی 2: اجرا با یک Thread

config = {"configurable": {"thread_id": "user-42"}}
for event in app.stream(
    {"messages": [HumanMessage("find the Anthropic headquarters address")]},
    config,
    stream_mode="updates",
):
    print(event)

هر به‌روزرسانی یک دیکشنری به شکل {node_name: state_delta} است. فرانت‌اند شما می‌تواند این موارد را به رابط کاربری استریم کند تا کاربران ببینند: «ایجنت در حال فکر کردن است... فراخوانی search_web... دریافت نتیجه... پاسخ‌دهی».

مرحله‌ی 3: افزودن وقفه‌ی مبتنی بر انسان در چرخه (Human-in-the-loop)

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

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # pause before every tool call
)

state = app.invoke({"messages": [HumanMessage("delete the production database")]}, config)
# مقدار state["__interrupt__"] تنظیم شده است. فراخوانی‌های پیشنهادی ابزار را بررسی کنید.
# در صورت تأیید:
from langgraph.types import Command
app.invoke(Command(resume=True), config)
# در صورت رد: پیام رد را بنویسید و ادامه دهید
app.update_state(config, {"messages": [AIMessage("Blocked by human reviewer.")]})

وضعیت، چک‌پوینت و thread همگی در طول وقفه ذخیره می‌مانند. هیچ داده‌ای در حافظه باقی نمی‌ماند مگر در حین اجرا.

مرحله‌ی 4: سفر در زمان برای دیباگ

history = list(app.get_state_history(config))
for snapshot in history:
    print(snapshot.values["messages"][-1].content[:80], snapshot.config)

# ایجاد fork از یک چک‌پوینت پیشین
target = history[3].config  # three steps back
for event in app.stream(None, target, stream_mode="values"):
    pass  # replay from that point forward

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

مرحله‌ی 5: تعویض ثبت‌کننده‌ی چک‌پوینت برای محیط پروداکشن

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

پشتیبانی از SQLite، Redis و Postgres به‌صورت آماده ارائه شده است. MemorySaver فقط مخصوص تست است. هر چیزی که قرار است با ری‌استارت شدن برنامه حفظ شود، به یک ذخیره‌ساز واقعی نیاز دارد.

مهارت عملی

ایجنت‌ها را در قالب گراف بسازید، نه در قالب حلقه‌های while True.

پیش از اینکه سراغ LangGraph بروید، این طراحی 60 ثانیه‌ای را انجام دهید:

1. گره‌ها را نام‌گذاری کنید. هر تصمیم مستقل یا عملیاتی که اثر جانبی (side-effect) دارد، یک گره است: «ایجنت فکر می‌کند»، «ابزار اجرا می‌شود»، «ناظر تأیید می‌کند»، «پاسخ استریم می‌شود». اگر نتوانید این مراحل را فهرست کنید، مسئله هنوز به شکل یک تسک مناسب برای ایجنت درنیامده است.
2. وضعیت را تعریف کنید. یک TypedDict حداقلی به همراه کاهنده برای هر فیلدی که از نوع لیست است بسازید. همه‌چیز را داخل messages نریزید؛ فیلدهای اختصاصی تسک (یک plan کاری، یک شمارنده‌ی budget، یک لیست retrieved_docs) را به بالاترین سطح ببرید.
3. یال‌ها را رسم کنید. یال‌ها باید ایستا باشند مگر اینکه مرحله‌ی بعد به خروجی مدل وابسته باشد. هر یال شرطی به یک تابع مسیریاب (router) با شاخه‌های نام‌گذاری‌شده نیاز دارد.
4. از همان ابتدا ثبت‌کننده‌ی چک‌پوینت را انتخاب کنید. MemorySaver برای تست‌ها و Postgres/Redis/SQLite برای هر کار دیگری. سیستم را بدون آن به پروداکشن نبرید؛ نبود ثبت‌کننده‌ی چک‌پوینت یعنی نداشتن قابلیت ادامه، وقفه و سفر در زمان.
5. وقفه‌ها را قبل از اجرای ابزارها بگذارید، نه بعد از آن‌ها. دریافت تأییدیه‌ها باید روی یال ورودی به گره‌های دارای اثر جانبی باشد تا بتوانید جلوی آسیب را بگیرید؛ اعتبارسنجی را هم روی یال خروجی از مدل بگذارید تا بتوانید فراخوانی‌های نادرست را با کمترین هزینه رد کنید.
6. پیش‌فرض خود را روی استریمینگ بگذارید. استفاده از mode="updates" برای رابط کاربری، mode="messages" برای استریم در سطح توکن در گره‌های مدل، و mode="values" برای دریافت اسنپ‌شات‌های کامل در زمان ارزیابی (eval).

هرگز ایجنت مبتنی بر LangGraph را بدون ثبت‌کننده‌ی چک‌پوینت تحویل ندهید. هرگز ایجنتی نسازید که وقفه‌ی آن پس از اعمال اثر جانبی رخ می‌دهد. و هرگز فیلد messages را بدون مشخص کردن add_messages به‌عنوان کاهنده‌اش رها نکنید.

تمرین‌ها

1. ساده. گراف ReAct چهارگره‌ای بالا را با یک ابزار ماشین‌حساب و یک ابزار جست‌وجوی وب پیاده‌سازی کنید. بررسی کنید که دستور list(app.get_state_history(config)) برای یک مکالمه‌ی دو نوبتی حداقل چهار چک‌پوینت برگرداند.
2. متوسط. یک گره planner اضافه کنید که پیش از agent اجرا شود و ساختاری به شکل plan: list[str] را در وضعیت بنویسد. کاری کنید که agent مراحل این پلن را به عنوان انجام‌شده علامت بزند. تست را طوری بنویسید که اگر plan در زمان ادامه‌ی اجرا از یک چک‌پوینت از بین رفت (به‌دلیل کاهنده‌ی اشتباه)، تست رد شود.
3. سخت. یک گراف ناظر بسازید که با استفاده از Send بین سه زیرگراف (researcher، writer و reviewer) مسیریابی کند. هر زیرگراف باید وضعیت و ثبت‌کننده‌ی چک‌پوینت اختصاصی خود را داشته باشد. یک interrupt_before=["writer"] روی گراف بیرونی قرار دهید تا انسان بتواند خلاصه پژوهش را تأیید کند. تأیید کنید که سفر در زمان از یک چک‌پوینت قبلی، فقط همان branch جداشده را دوباره اجرا می‌کند.

اصطلاحات کلیدی

برای مطالعه‌ی بیشتر

• مستندات LangGraph — مرجع اصلی برای StateGraph، کاهنده‌ها، ثبت‌کننده‌های چک‌پوینت و وقفه‌ها.
• مفاهیم LangGraph: وضعیت، کاهنده‌ها و ثبت‌کننده‌های چک‌پوینت — مدل ذهنی استفاده‌شده در این درس، مستقیماً از منبع اصلی.
• ماندگاری داده‌ها و چک‌پوینت‌ها در LangGraph — جزئیات ذخیره‌سازهای Postgres/SQLite/Redis، فضاهای نام چک‌پوینت و شناسه‌های thread.
• انسان در چرخه (Human-in-the-loop) در LangGraph — بررسی interrupt_before، interrupt_after، دستور Command(resume=...) و الگوی ویرایش وضعیت.
• مقاله یائو و همکاران: "ReAct: Synergizing Reasoning and Acting in Language Models" (ICLR 2023) — الگویی که در هسته‌ی تمام ایجنت‌های LangGraph پیاده می‌شود؛ برای درک منطق ردپای استدلال آن را مطالعه کنید.
• Anthropic — ساخت ایجنت‌های اثربخش (دسامبر 2024) — ساختارهای مختلف گراف (زنجیره، مسیریاب، ارکستریتور-کارگر، ارزیاب-بهینه‌ساز) و زمان کاربرد هرکدام.
• فاز 11 · 09 (فراخوانی توابع یا Function Calling) — سازوکار اولیه‌ی فراخوانی ابزار که در گره‌های ایجنت در LangGraph استفاده می‌شود.
• فاز 11 · 14 (Model Context Protocol) — کشف ابزارهای خارجی که از طریق آداپتور MCP به ToolNode در LangGraph متصل می‌شوند.
• فاز 11 · 17 (مقایسه‌ی فریم‌ورک‌های ایجنت) — چه زمانی LangGraph را به CrewAI، AutoGen یا Agno ترجیح دهیم.