ترکیب Pydantic AI و chanx امنیت استریمِ عامل‌های هوش مصنوعی را تضمین کرد

اگر امروز در حال توسعه‌ی عامل‌هایی هستید که باید عملیات حساس را روی دیتابیس اجرا کنند، احتمالاً با کابوسِ مدیریتِ پیام‌های نامشخص در وب‌ساکت‌ها دست‌وپنجه نرم می‌کنید. اینجاست که یک معماری جدید وارد می‌شود تا قراردادهای ارتباطی بین فرانت‌اند و بک‌اند را از حالت «حدس و گمان» به «تضمین فنی» تبدیل کند. یک پیاده‌سازی جدید نشان می‌دهد که چگونه می‌توان Pydantic AI و chanx را برای ایجاد یک خط لوله (Pipeline) وب‌ساکت کاملاً تایپ‌شده برای عامل‌هایی که برای اقدامات تخریبی به تایید انسانی نیاز دارند، ادغام کرد.

بسیاری از توسعه‌دهندگان در حال حاضر برای استریمِ پاسخ‌های هوش مصنوعی از SSE (Server-Sent Events) استفاده می‌کنند که ذاتاً یک‌طرفه است. طبق گزارش‌های فنی، SSE برای چت‌های ساده عالی است، اما وقتی یک عامل (Agent) — شبیه به کارمندی که برای انجام هر مرحله از کار نیاز به تایید مدیر دارد — باید متوقف شود، اجازه حذف یک رکورد را از کاربر بگیرد و سپس ادامه دهد، شکست می‌خورد. این چرخه «توقف-پرسش-ادامه»، نقطه قوت وب‌ساکت‌هاست، اما اکثر فریم‌ورک‌های عامل، راهی تایپ‌شده برای پیاده‌سازی آن ندارند و توسعه‌دهنده را در تله‌ی زنجیره‌های بی‌پایان if/else و اعتبارسنجی دستی JSON می‌اندازند.

معماری عامل‌های تایپ‌شده

همان‌طور که در تحلیل‌های قبلی ما درباره‌ی امنیت مدل‌های بازمتن اشاره کردیم، حذف نقاط کور در ارتباطات سیستمی، اولین قدم برای استقرار در مقیاس صنعتی است. در این معماری، رویکرد «اول قرارداد» (Contract-first) حاکم است. به جای برخورد با پیام‌های وب‌ساکت به عنوان رشته‌های متنی خام، سرور هر پیام احتمالی را به عنوان یک مدل Pydantic تعریف می‌کند. سپس chanx این مدل‌ها را به هندلرهای مسیریابی و یک اسکیمای AsyncAPI 3.0 تبدیل می‌کند که کلاینت React از آن برای تولید تایپ‌های TypeScript خود استفاده می‌کند.

این سازوکار، باگ‌های رایج «تایپ رشته‌ای» (Stringly-typed) را که در اپلیکیشن‌های بلادرنگ (Real-time) شایع است، حذف می‌کند. اگر توسعه‌دهنده نام یک فیلد را در مدل پydantic پایتون تغییر دهد، کامپایلر TypeScript بلافاصله تمام خطوط خراب در فرانت‌اند را علامت‌گذاری می‌کند. سیستم از یک فیلد تشخیص‌دهنده (Discriminator) به نام action برای مسیریابی پیام‌هایی مثل chat ،tool_decision و approval_request استفاده می‌کند، بدون اینکه نیاز به منطق مسیریابی دستی باشد.

جزئیات پروتکل

بر اساس مستندات این پیاده‌سازی، پروتکل ارتباطی کامل، کم‌حجم اما جامع است. جریان کلاینت-به-سرور محدود به chat و tool_decision است. با این حال، جریان سرور-به-کلاینت برای مدیریت حالت‌های مختلف چرخه حیات یک عامل، پیچیده‌تر است:

• چرخه حیات استریم: شامل stream_start ،text_delta و stream_end.
• فعالیت ابزارها: شامل tool_call و tool_result.
• حلقه انسانی: پیام approval_request برای تایید عملیات حساس.
• وضعیت و متادیتا: شامل suggestions ،history ،tasks_updated و notification.
• مدیریت خطا: پیام agent_error.

با استفاده از BaseConsumer در chanx، هندلرها دقیقاً اعلام می‌کنند چه چیزی می‌پذیرند و چه پاسخی می‌دهند. این اعلان به عنوان مستندات زنده برای کل سیستم عمل می‌کند و از طریق مسیر /asyncapi.json قابل دسترسی است. این دقیقاً همان معادل وب‌ساکتی است که drf-spectacular برای APIهای REST فراهم می‌کرد.

پیاده‌سازی حلقه انسانی (Human-in-the-Loop)

یکی از حیاتی‌ترین بخش‌ها، استفاده از پرچم requires_approval=True در Pydantic AI برای ابزارهای خطرناک است. در عامل دمو («Tasklet») که یک لیست وظایف را در SQLite مدیریت می‌کند، ابزارهای ایمن آزادانه اجرا می‌شوند، اما ابزارهای تخریبی — مانند delete_task یا clear_all_tasks — برای تایید متوقف می‌شوند.

وقتی عامل سعی می‌کند ابزاری تخریبی را فراخوانی کند، اجرا (Run) ابزار را اجرا نمی‌کند. در عوض، زودتر به پایان می‌رسد و یک شیء DeferredToolRequests برمی‌گرداند. به همین دلیل است که output_type عامل به صورت یک Union تعریف شده است: TextAnswer | DeferredToolRequests. این سیستم تایپ، توسعه‌دهنده را مجبور می‌کند هر دو حالت (پاسخ نهایی یا درخواست اجازه) را مدیریت کند. این رویکرد در واقع پیاده‌سازی فنی از حلقه‌های بازبینی انسانی است که در مقابل مهندسی پرامپت برای تضمین کیفیت و صحت خروجی‌ها در محیط‌های عملیاتی به کار می‌رود.

سپس سیستم یک پیام approval_request را برای کاربر پخش (Broadcast) می‌کند. سرور به سادگی متوقف شده و منتظر می‌ماند. در اینجا هیچ Polling یا درخواست HTTP معلقی وجود ندارد. پس از اینکه کاربر در رابط کاربری روی «تایید» یا «رد» کلیک کرد، کلاینت یک پیام tool_decision را به سرور می‌فرستد. سپس سیستم DeferredToolResults را می‌سازد و اجرا را با استفاده از تاریخچه ذخیره‌شده از سر می‌گیرد.

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

استریم خروجی‌های ساختاریافته

استریم کردن متن ساده آسان است، اما استریم یک شیء ساختاریافته (مانند پاسخ نهایی همراه با چیپ‌های پیشنهادی برای ادامه گفتگو) دشوارتر است. این پیاده‌سازی از run_stream() و stream_output() برای ارائه اسنپ‌شات‌های نیمه‌اعتبارسنجی شده از شیء نهایی استفاده می‌کند.

به جای یک استریم متنی خام، سیستم از یک مدل TextAnswer استفاده می‌کند که شامل رشته content و لیستی از follow_ups است. با مقایسه (Diffing) مقادیر متوالی محتوا، سیستم دلتاهای متنی را استخراج کرده و در لحظه به کلاینت می‌فرستد. این به UI اجازه می‌دهد افکت تایپ را برای پاسخ اصلی رندر کند و هم‌زمان چیپ‌های ساختاریافته «دنبال‌کننده» را آماده کند که فقط پس از پایان استریم ظاهر می‌شوند.

رویدادهای ابزارها از طریق یک event_stream_handler مدیریت می‌شوند. سیستم FunctionToolCallEvent و FunctionToolResultEvent را مستقیماً روی قرارداد پیام‌ها نگاشت می‌کند تا کاربر کارت‌های زنده فعالیت ابزار را ببیند، در حالی که حلقه اصلی مدل از اینکه چه ابزارهای خاصی وجود دارند بی‌اطلاع می‌ماند. یک نکته ظریف این است که فراخوانی‌های ابزار defer شده هرگز به event_stream_handler نمی‌رسند زیرا اجرا نشده‌اند؛ بنابراین سیستم باید این فراخوانی‌های متوقف شده را خودش قبل از approval_request اعلام کند.

گروه‌های Broadcast در مقابل ساکت‌های مستقیم

برای جلوگیری از قطع شدن استریم هنگام رفرش صفحه، معماری از نوشتن مستقیم در یک ساکت اجتناب می‌کند. در عوض، از یک الگوی Broadcast استفاده می‌شود که در آن عامل به عنوان یک Task پس‌زمینه اجرا شده و رویدادها را به یک گروه در Redis (مثلاً conversation.{id}) می‌فرستد.

این طراحی سه مزیت فوری دارد:

• استریم‌های ضد-رفرش: اجرای مدل طولانی‌تر از اتصال ساکت است. کاربر می‌تواند صفحه را رفرش کند و دوباره به گروه متصل شود تا رویدادهای زنده را دریافت کند.
• همگام‌سازی چند-تب: تمام تب‌های باز برای یک گفتگو، دلتاها، کارت‌های ابزار و درخواست‌های تایید را به‌طور هم‌زمان رندر می‌کنند. حتی پرامپت خود کاربر از طریق Broadcast به عنوان یک user_message بازتاب داده می‌شود.
• تایید بین-دستگاهی: درخواست‌های تایید بر اساس گفتگو کلیدگذاری شده‌اند، نه اتصال. کاربر می‌تواند درخواست را روی دسکتاپ دریافت کرده و آن را از طریق دستگاه موبایل تایید کند.

فراتر از ساکت: HTTP و Workerها

از آنجایی که broadcast_event یک متد کلاس است، سیستم می‌تواند از خارج از وب‌ساکت هم تحریک شود. دمو یک Endpoint از نوع POST /conversations/{id}/notify را برای ارسال توست‌های اعلان به تمام تب‌های متصل فراهم می‌کند. این همچنین امکان اجرای کارهای پس‌زمینه غیرهمزمان را فراهم می‌کند؛ مثلاً ابزار schedule_reminder می‌تواند تسکی را زمان‌بندی کند که مدت‌ها پس از پایان استریم اصلی، یک نوتیفیکیشن به گروه ارسال کند.

برای محیط‌های تولیدی با بار زیاد، لایه کانال اجازه می‌دهد عامل به یک صف تسک مانند Celery, Taskiq یا ARQ منتقل شود. مصرف‌کننده (Consumer) می‌تواند با استفاده از self.channel_name به عنوان یک آدرس پاسخ تایپ‌شده، یک جاب را در صف قرار دهد. این به پروسه Worker اجازه می‌دهد رویدادهای تایپ‌شده را به آن اتصال خاص بازگرداند بدون اینکه مصرف‌کننده مسدود (Block) شود. این بهینه‌سازی در لایه انتقال، یادآور راهکارهای حذف تأخیر در پاسخ‌دهی لحظه‌ای است که برای رسیدن به تجربه کاربری بدون وقفه در مقیاس بالا ضروری است.

تست بدون LLM

به دلیل تایپ سخت‌گیرانه پروتکل، کل جریان — از استریم و اجرای ابزار تا تاییدات — بدون نیاز به API Key قابل تست است. با استفاده از FunctionModel در Pydantic AI، توسعه‌دهندگان می‌توانند دقیقاً اسکریپت کنند که «LLM» چه چیزی برگرداند و بررسی کنند که مصرف‌کننده وب‌ساکت با توالی درستی از پیام‌ها پاسخ می‌دهد.

برای مثال، یک تست می‌تواند درخواست حذف کاربر را شبیه‌سازی کند، تایید کند که سرور یک approval_request ارسال کرده (بدون حذف واقعی داده‌ها) و سپس بررسی کند که داده‌ها تنها پس از ارسال ToolDecisionMessage حذف شده‌اند. مجموعه تست‌ها موارد زیر را پوشش می‌دهد:

• مسیر رد کردن (Denial) برای ابزارهای تخریبی.
• فیدهای Broadcast در چند تب.
• اتصال مجدد به یک نشست در میانه تایید.
• محافظ‌های اجرای همزمان (Concurrent-run guards) با استفاده از یک دیکشنری RUNNING برای رد درخواست‌های هم‌پوشان.
• تست‌های قرارداد که تضمین می‌کنند /asyncapi.json شامل تمام انواع پیام‌های مورد نیاز است.

یک تله فنی: Agent.override(model=...) از contextvars استفاده می‌کند که در تسک‌های مصرف‌کننده تحت ارتباط‌گر تست منتقل نمی‌شوند. الگوی Factory (build_agent) با اجازه دادن به تست‌ها برای پاس دادن مستقیم یک FunctionModel این مشکل را دور می‌زند.

ملاحظات تولیدی (Production)

برای مقیاس‌پذیری، سیستم می‌تواند مدیریت وب‌ساکت را از منطق LLM جدا کند. یک سرویس Django می‌تواند نشست‌های کاربر و وب‌ساکت‌ها را مدیریت کند (به عنوان یک لایه عبور نازک)، در حالی که یک سرویس FastAPI مجزا، عامل‌های Pydantic AI را اجرا کند و هر دو از طریق همان لایه کانال Redis ارتباط برقرار کنند. این تضمین می‌کند که ساکتِ رو به کاربر هرگز در انتظار پاسخ کند مدل مسدود نشود.

یک نکته حیاتی در مهندسی پرامپت این است که مدل باید صراحتاً بداند ایمنی توسط زیرساخت مدیریت می‌شود. اگر در پرامپت صرفاً گفته شود «عملیات تخریبی نیاز به تایید دارد»، مدل ممکن است سعی کند در متن چت بپرسد «آیا اجازه دارید؟» و جریان تایپ‌شده تایید را کاملاً دور بزند. دستورالعمل باید صریح باشد: «وقتی کاربر عملیات تخریبی می‌خواهد، مستقیماً ابزار را فراخوانی کن — هرگز در چت درخواست تایید نکن».

سایر بهبودهای تولیدی شامل موارد زیر است:

• احراز هویت: استفاده از هوک‌های post_authentication برای اعتبارسنجی نشست‌ها یا توکن‌ها قبل از پیوستن به گروه‌ها.
• پایداری (Durability): ذخیره DeferredToolRequests در دیتابیس تا تاییدات پس از ری‌استارت سرور باقی بمانند.
• امنیت: اجرای تاریخچه پیام‌های پذیرفته شده از طریق sanitize_messages در pydantic-ai.
• ماندگاری: استفاده از ModelMessagesTypeAdapter برای ذخیره وضعیت گفتگو به عنوان یک Blob اعتبارسنجی شده، که بازپخش تاریخچه و بازسازی ترانسکریپت‌ها را آسان می‌کند.

این چرخش به سمت طراحی «اول قرارداد»، توسعه عامل‌ها را از «هک کردن پرامپت» به سمت دیسیپلین‌های استاندارد مهندسی نرم‌افزار می‌برد. با تبدیل تعاملات عامل به یک API مستند، تیم‌ها می‌توانند جریان‌های کاری پیچیده با همان اطمینانی مقیاس کنند که در REST APIها دارند.

گام بعدی شما

• اگر از WebSockets برای AI استفاده می‌کنید، مدل‌های Pydantic را جایگزین JSONهای خام کنید تا خطاهای Runtime کاهش یابد.
• برای ابزارهای حساس، الگوی DeferredToolRequests را پیاده کنید تا کنترل نهایی در دست انسان بماند.
• لایه Redis را برای جداسازی اجرای مدل از اتصال ساکت اضافه کنید تا تجربه کاربری در هنگام رفرش صفحه بهبود یابد.

اما داستان سخت‌افزاری این تحول حتی شگفت‌انگیزتر است — به تحلیل ما درباره‌ی تراشه‌های Blackwell مراجعه کنید.