پیاده‌سازی دیسپچر فراخوانی ابزارها (Dispatcher) در ایجنت‌های هوش مصنوعی

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

در این درس می‌بینید که چطور با جداسازی وظایف و کپسوله‌سازی منطق اجرای ابزار، مواردی مثل تایم‌اوت به ازای هر فراخوانی، الگوریتم تلاش مجدد با عقب‌گرد نمایی (Exponential Backoff) و حذف درخواست‌های تکراری با کلید یکتایی (Idempotency Key) پیاده‌سازی می‌شوند. این معماری تضمین می‌کند که هیچ خطایی بدون ساختار استاندارد به حلقه‌ی اصلی بازنگردد و منابع سیستمی با اجرای بی‌رویه هدر نروند.

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

اهداف یادگیری

• کپسوله‌سازی هندلر ابزار در تایم‌اوت مجزا به‌ازای هر فراخوانی، به‌طوری که به‌جای قفل کردن حلقه، خطایی با تایپ مشخص برگرداند.
• پیاده‌سازی تلاش مجدد با عقب‌گرد نمایی (Exponential Backoff) همراه با نوسان تصادفی (Jitter) و سقف مشخص برای تعداد تلاش‌ها.
• حذف درخواست‌های تکراری بر اساس کلید یکتایی تراکنش (Idempotency key)، تا تلاش مجددی که با درخواست اولیه‌ی کند هم‌زمان می‌شود، دو بار اجرا نشود.
• نگاشت استثناهای هندلر و خطاهای بستر انتقال (Transport) روی یک ساختار خطای واحد (Error envelope) که حلقه‌ی بستر آزمایشی ساختار آن را می‌شناسد.
• محدود کردن ارسال موازی با سقف هم‌زمانی، تا توزیع موازی (Fan-out) 40 فراخوانی ابزار باعث اتمام منابع حلقه‌ی رویداد نشود.

جایگاه دیسپچر

بین حلقه‌ی بستر آزمایشی (درس 20) و رجیستری ابزار (درس 21). بستر انتقال (درس 22) ورودی را به حلقه می‌رساند. حلقه فراخوانی ابزار (Tool call) را تحویل دیسپچر می‌دهد. دیسپچر سراغ رجیستری می‌رود، هندلر را اجرا می‌کند و در نهایت نتیجه یا یک بسته‌ی خطای شبیه JSON-RPC را برمی‌گرداند.

دیسپچر تنها لایه‌ای است که از تایمرها، تلاش‌های مجدد و بی‌اثر در تکرار بودن (Idempotency) خبر دارد. حلقه از این‌ها بی‌خبر است؛ رجیستری خبر ندارد؛ هندلر هم خبر ندارد. اصل قضیه همین جداسازی وظایف است.

تایم‌اوت‌ها

هر ابزار یک تایم‌اوت پیش‌فرض دارد. رکورد ثبت‌شده در رجیستری، فیلد timeout_ms را در خود دارد. وقتی بستر آزمایشی تایم‌اوت اختصاصی بفرستد، دیسپچر مقدار پیش‌فرض را با آن بازنویسی می‌کند. ما از asyncio.wait_for استفاده می‌کنیم. با اتمام مهلت زمانی، وظیفه‌ی (Task) هندلر لغو می‌شود و دیسپچر خطای DispatchError(kind="timeout") را برمی‌گرداند.

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

تلاش مجدد با عقب‌گرد نمایی

سیاست تلاش مجدد حداکثر 3 بار است. عقب‌گرد به‌صورت نمایی همراه با نوسان تصادفی اعمال می‌شود.

attempt 1  -> delay 0
attempt 2  -> delay 0.1s * (1 + random[0..0.5])
attempt 3  -> delay 0.4s * (1 + random[0..0.5])

تنها خطاهای timeout و transient مشمول تلاش مجدد می‌شوند. خطاهای schema، not_found یا internal دوباره تکرار نمی‌شوند. خطاهای اسکیما قطعی هستند؛ تلاش مجدد نتیجه را تغییر نمی‌دهد و فقط بودجه را می‌سوزاند.

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

حذف درخواست‌های تکراری با کلید یکتایی تراکنش

تلاش مجددی که حین در جریان بودن درخواست اولیه شلیک شود، یک باگ واقعی در محیط عملیاتی (پروداکشن) است. فراخوانی اول در ثانیه‌ی 4.9 گیر می‌کند (درست کمی قبل از تایم‌اوت). تلاش مجدد در ثانیه‌ی 5 شلیک می‌شود. حالا دو درخواست روی بک‌اند یکسان با هم رقابت می‌کنند. اگر ابزار مورد نظر payments.charge باشد، دو بار از کاربر پول کم کرده‌اید.

دیسپچر یک آرگومان اختیاری به نام idempotency_key می‌گیرد. اگر هنگام رسیدن یک فراخوانی، کلیدی مشابه همچنان در حال پردازش باشد، دیسپچر منتظر همان فیوچرِ در جریان می‌ماند و نتیجه‌ی آن را برمی‌گرداند. کشْ کلیدها را تا 60 ثانیه پس از اتمام نگه می‌دارد تا تلاش‌های مجدد دیرهنگام را خنثی کند.

تولید کلید بر عهده‌ی فراخواننده است. بستر آزمایشی آن را از پلنر می‌گیرد: f"{step_id}:{tool_name}:{hash(args)}". دیسپچر خودش کلید نمی‌سازد؛ چون مشتق کردن کلید صرفاً از روی آرگومان‌ها باعث می‌شود دو فراخوانی با معنای کاملاً متفاوت، یکسان به نظر برسند.

قالب خطا

فراخوانی ناموفق فقط در قالب یک ساختار واحد خروجی می‌دهد.

DispatchError
  kind        : "timeout" | "transient" | "schema" | "not_found" | "internal" | "budget_exceeded"
  message     : str
  attempts    : int
  jsonrpc_code: int   (one of -32601, -32602, -32603)

حلقه‌ی بستر آزمایشی نوع فیلد kind را به وضعیت بعدی نگاشت می‌کند. خطاهای schema و not_found به on_error می‌روند و برنامه‌ریزی مجدد (Replan) را تحریک می‌کنند. خطاهای timeout و transient به on_error فرستاده می‌شوند و بسته به تعداد تلاش‌ها ممکن است به برنامه‌ریزی مجدد ختم شوند یا نشوند. خطای budget_exceeded هم وضعیت on_budget_exceeded را فعال می‌کند.

سقف هم‌زمانی در توزیع موازی

دستور gather(*calls) همه‌ی کوروتین‌ها را هم‌زمان اجرا می‌کند. با 40 فراخوانی ابزار، یعنی 40 سوکت باز یا 40 پایپِ ساب‌پراسس. بیشتر بک‌اندها روی خوشی به 40 اتصال موازی از سمت یک کلاینت نشان نمی‌دهند.

دیسپچر دستور gather را داخل یک سمافور (Semaphore) قرار می‌دهد. سقف هم‌زمانی پیش‌فرض 8 است. هر فراخوانی قبل از ارسال، سمافور را دریافت می‌کند و پس از اتمام آن را آزاد می‌سازد. فراخواننده خروجی‌ای درست مشابه gather دریافت می‌کند، اما زمان‌بندی واقعی فراخوانی‌ها دارای محدودیت است.

فلو برای یک فراخوانی

راهنمای خواندن کد

فایل code/main.py کلاس‌های Dispatcher، DispatchError و TransientError را تعریف می‌کند. دیسپچر در زمان ساخت، رجیستری را تحویل می‌گیرد. متد ناهمگام dispatch(name, args, ...) تنها نقطه‌ی ورود به سیستم است. تایم‌اوت‌های هر تلاش به‌شکل درون‌خطی در تابع _run_with_retries و با استفاده از asyncio.wait_for اعمال می‌شوند. متد gather_bounded(calls) چندین فراخوانی را با رعایت سقف هم‌زمانی اجرا می‌کند.

تست‌های code/tests/test_dispatcher.py مواردی مثل فعال شدن تایم‌اوت، تلاش مجدد در خطاهای گذرا، عدم تلاش مجدد در خطای اسکیما، حذف فراخوانی‌های تکراری با کلید یکتایی (کاهش دو فراخوانی هم‌زمان با یک کلید به اجرای یک هندلر) و محدودسازی هم‌زمانی (کارکرد عملی سمافور) را پوشش می‌دهند.

تست‌ها از asyncio.sleep(0) و هندلرهای قطعی مبتنی بر Counter استفاده می‌کنند، بنابراین ظرف چند میلی‌ثانیه تمام می‌شوند و به زمان ساعت سیستم وابسته نیستند.

گام‌های بعدی

دو قابلیت تکمیلی که دیسپچرهای پروداکشن به سیستم اضافه می‌کنند: اول، لاگ‌گیری ساختاریافته در هر تغییر وضعیت (که جریان رویدادهای حلقه پیشاپیش در اختیارتان می‌گذارد، اما دیسپچر هم باید رویدادهای dispatch.attempt و dispatch.retry را ارسال کند). دوم، مدارشکن‌ها (Circuit Breakers): بعد از N بار شکست در یک بازه‌ی زمانی مشخص، ابزار وارد دوره‌ی استراحت (Cool-down) می‌شود و در این مدت، فراخوانی‌ها بدون تلاش برای اجرای هندلر، بلافاصله مقدار مستقیم (بی‌واسطه) با وضعیت kind="circuit_open" را برمی‌گردانند. هر دو قابلیت را می‌توان بدون دستکاری قرارداد این دیسپچر، روی آن پیاده کرد.

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