آموزش LangGraph با API پاستا؛ ساخت ایجنت با حافظه و ادامه اجرا

برنامه کامل: پیش‌نویس، توقف و تأیید

کد زیر را در فایل agent.py ذخیره کنید. در حالت شروع، شناسه و متن پرسش را می‌گیرد؛ در حالت تأیید یا رد، همان شناسه را باز می‌کند. فرمان وضعیت هم فقط داده ذخیره‌شده را می‌خواند. نام فایل دیتابیس عمداً ثابت و محلی است؛ برای استفاده چندکاربره، این برنامه خط فرمان را بدون لایه احراز هویت به یک API عمومی تبدیل نکنید. کلید، متن درخواست و پاسخ خام خطای ارائه‌دهنده نیز نباید وارد لاگ عمومی شوند.

import json
import os
import sys
import urllib.error
import urllib.request
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command, interrupt

class State(TypedDict, total=False):
    question: str
    draft: str
    status: str

def draft(state: State):
    payload = {
        "model": os.environ["PAASTA_MODEL"],
        "messages": [
            {"role": "system", "content": "Write a short Persian support draft. Never promise a refund or an action."},
            {"role": "user", "content": state["question"]},
        ],
        "max_tokens": 512,
    }
    request = urllib.request.Request(
        "https://console.paasta.cloud/api/ai/v1/chat/completions",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json",
                 "Authorization": "Bearer " + os.environ["PAASTA_API_KEY"]},
    )
    try:
        with urllib.request.urlopen(request, timeout=45) as response:
            result = json.load(response)
    except urllib.error.HTTPError as exc:
        raise RuntimeError(f"Model request failed: HTTP {exc.code}") from None
    text = result["choices"][0]["message"]["content"]
    if not isinstance(text, str) or not text.strip():
        raise ValueError("Expected a non-empty text response")
    return {"draft": text, "status": "waiting"}

def review(state: State):
    approved = interrupt({"draft": state["draft"]})
    if not isinstance(approved, bool):
        raise ValueError("Approval must be a boolean")
    return {"status": "approved" if approved else "rejected"}

def build(saver):
    builder = StateGraph(State)
    builder.add_node("draft", draft)
    builder.add_node("review", review)
    builder.add_edge(START, "draft")
    builder.add_edge("draft", "review")
    builder.add_edge("review", END)
    return builder.compile(checkpointer=saver)

if __name__ == "__main__":
    mode, thread_id = sys.argv[1:3]
    config = {"configurable": {"thread_id": thread_id}}
    with SqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
        graph = build(saver)
        if mode == "start":
            if graph.get_state(config).values:
                raise ValueError("Use a new thread ID for a new request")
            graph.invoke({"question": sys.argv[3]}, config)
        elif mode in ("approve", "reject"):
            snapshot = graph.get_state(config)
            if snapshot.next != ("review",):
                raise ValueError("No review is awaiting approval")
            graph.invoke(Command(resume=mode == "approve"), config)
        elif mode != "status":
            raise ValueError("Expected start, approve, reject or status")
        print(json.dumps(graph.get_state(config).values, ensure_ascii=False))

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

آزمایش قطع فرایند و ادامه با همان شناسه

python agent.py start demo-001 'کاربر درباره زمان آماده‌شدن سفارش سؤال دارد.'
python agent.py status demo-001
# Close the shell or restart the process, then:
python agent.py approve demo-001
python agent.py status demo-001

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

انتقال به PostgreSQL برای سرویس چندکاربره

برای چند نمونه برنامه یا چند Worker، نگه‌داشتن SQLite داخل هر کانتینر انتخاب مناسبی برای وضعیت مشترک نیست. از Checkpointer پایدار متصل به دیتابیس مشترک استفاده کنید و نیازهای اتصال، ظرفیت و نسخه را بررسی کنید. مرجع رسمی PostgreSQL در LangGraph درباره راه‌اندازی جدول‌ها و محدودسازی deserialization توضیح می‌دهد. ساخت جدول و مهاجرت را با دسترسی لازم، کنترل‌شده انجام دهید؛ نقش اجرای عادی برنامه را صرفاً برای راحتی، مالک تمام دیتابیس نکنید.

# Install langgraph-checkpoint-postgres and a suitable psycopg build.
from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as saver:
    # Run saver.setup() during a controlled database initialization step.
    graph = build(saver)
    # Use the same authorized thread_id when resuming a run.

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

استقرار روی پاستا: وب، Worker و داده را جدا ببینید

برنامه خط فرمان بالا خودش وب‌سرور نیست. برای محصول واقعی، یک API احراز هویت‌شده درخواست را ثبت می‌کند و یک Worker اجرای طولانی را پیش می‌برد؛ رابط کاربری وضعیت را می‌خواند. در پلتفرم ابری پاستا، سرویس‌ها را با فرمان اجرای درست، متغیرهای محیطی و منابع متناسب منتشر کنید. اگر از هاست Next.js برای رابط استفاده می‌کنید، کلید مدل و دسترسی Checkpoint را در مرورگر نگذارید. قطع‌شدن اتصال SSE نباید به معنی حذف رکورد کار باشد.

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

مجوز، Retry و تأیید انسانی

شناسه Thread رمز عبور نیست. در API محصول، هر بار خواندن، ادامه یا تأیید باید مالکیت حساب و دسترسی کاربر را بررسی کند؛ دانستن شناسه نباید راه مشاهده پاسخ دیگران باشد. متن تأیید را به همان نسخه پیش‌نویس و همان عملیات متصل کنید. اگر مقدار سفارش یا مقصد پس از تأیید عوض شد، تأیید قبلی را معتبر فرض نکنید. همچنین اطلاعات Checkpoint ممکن است حاوی داده خصوصی باشد؛ برای دسترسی تیم پشتیبانی، نگهداری، حذف و بازیابی آن سیاست روشن لازم است.

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

چک‌لیست پذیرش و قدم بعدی

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

مسیر پیشنهادی این است: ابتدا گراف کوچک بالا را بفهمید، بعد وضعیت را به PostgreSQL منتقل کنید و در پایان لایه API و Worker را به آن اضافه کنید. برای شروع بخش مدل، کلید مستقل API پاستا بسازید و مدل موجود را با همان متن آزمون مقایسه کنید. اگر هنوز مرحله‌ای نیاز به وضعیت پایدار ندارد، آن را پیچیده نکنید؛ هدف از LangGraph قابل اعتمادترشدن مسیر کار است، نه اضافه‌کردن یک ابزار دیگر به پروژه.