بناء سير عمل الذكاء الاصطناعي المستقل مع LLMs

لقد غيرت نماذج اللغات الكبيرة (LLMs) كيفية تفاعلنا مع التكنولوجيا، حيث انتقلت بسرعة من روبوتات المحادثة البسيطة إلى محركات التفكير القادرة على قيادة إجراءات معقدة ومتعددة الخطوات. في حين أن تفاعل الاستجابة السريعة يمكن أن يكون قويًا، فإن القيمة الحقيقية للذكاء الاصطناعي التوليدي في إعدادات المؤسسة تكمن في سير عمل الذكاء الاصطناعي المستقل.

بدلاً من الاعتماد على المشغلين البشريين لتنسيق كل خطوة، يستخدم سير العمل المستقل LLMs كصانعي قرار مركزيين يقومون بتخطيط المهام وتنفيذها وتقييمها وتصحيحها ذاتيًا على مدى فترات طويلة.

يستكشف هذا الغوص العميق كيفية تصميم وبناء ونشر مسارات عمل موثوقة ومستقلة للذكاء الاصطناعي باستخدام أنماط التصميم الحديثة وآلات الحالة وحواجز الحماية القوية.

1. تحول الوكيل: روبوتات الدردشة مقابل سير العمل

يمكن تصنيف تطور تطبيقات LLM إلى أربعة مستويات متميزة من الاستقلالية:

على الرغم من أن عوامل المستوى 4 تتمتع بمرونة عالية، إلا أنه من الصعب التنبؤ بها في بيئات الإنتاج. ولذلك، فإن معظم بنيات المؤسسات مبنية على المستوى 3: سير العمل الموجه، الذي يجمع بين الموثوقية الحتمية لأجهزة حالة البرنامج والتفكير الديناميكي لمدارس LLM.

2. الركائز الأساسية لسير العمل المستقل

لبناء سير عمل مستقل، تحتاج إلى الجمع بين أربعة مكونات أساسية:

أ. التفكير والتخطيط

في قلب سير العمل يوجد نموذج التخطيط. يحاول استدعاء LLM الساذج إخراج الإجابة النهائية على الفور، الأمر الذي يؤدي غالبًا إلى فشل الاستدلال. تستخدم مسارات العمل المستقلة حلقات تخطيط متخصصة:

• ReAct (Reason + Act): يفكر النموذج بشكل متكرر، ويتصرف (يستدعي أداة)، ويلاحظ النتيجة، ويكرر هذه الحلقة حتى يتم تحقيق الهدف.
• سلسلة الأفكار (CoT): إجبار النموذج على إخراج تفكيره خطوة بخطوة قبل الوصول إلى الاستنتاج.
• شجرة الأفكار (ToT): إنشاء مسارات بديلة متعددة وتقييمها، وتتبع الفروع المختلفة، والتراجع عند فشل المسار.

ب. الذاكرة قصيرة المدى وطويلة المدى

يجب أن يحافظ النظام الذاتي على حالته عبر دورات تنفيذ متعددة:

• الذاكرة قصيرة المدى: سياق سلسلة المحادثات ومتغيرات الحالة وسجلات التنفيذ التي تتتبع ما يفعله سير العمل حاليًا.
• الذاكرة طويلة المدى: قواعد بيانات المتجهات وأنظمة الاسترجاع الدلالي التي تسمح لسير العمل باستدعاء عمليات التشغيل التاريخية وتفضيلات المستخدم ووثائق المؤسسة.

ج. الأدوات والتكامل مع الويب

للعمل في العالم المادي أو الرقمي، يجب على LLMs التفاعل مع الخدمات الخارجية. يحتاج النموذج إلى الوصول إلى برامج تشغيل قواعد البيانات وأنظمة الملفات ومتصفحات الويب وواجهات برمجة التطبيقات التابعة لجهات خارجية. تتبنى مسارات العمل الحديثة بشكل متزايد بروتوكول سياق النموذج (MCP)، وتوحيد كيفية اكتشاف LLMs لمصادر البيانات السياقية وآليات التنفيذ الآمنة والاتصال بها بأمان.

3. أنماط التصميم المعماري الرئيسية

عند بناء أنظمة وكيلة معقدة، يعتمد مهندسو البرمجيات على مجموعة من أنماط التصميم المثبتة لإدارة التعقيد والحفاظ على القدرة على التنبؤ:

النموذج 1: نمط جهاز التوجيه

يقرأ جهاز التوجيه المدخلات الواردة ويقرر أي موجه LLM أو قاعدة بيانات أو معالج API متخصص يجب أن يعالجها بعد ذلك. وهذا يمنع موجه LLM واحد متجانس من التحميل الزائد بعدد كبير جدًا من التعليمات.

النمط 2: العمال المنسقون

يقوم المنسق المركزي LLM بتقسيم مهمة كبيرة ومعقدة إلى مهام فرعية مستقلة. ثم يقوم بعد ذلك بتفويض هذه المهام إلى العقد العاملة (والتي يمكن أن تكون LLMs متخصصة أو خدمات صغيرة قياسية) ويقوم بتجميع النتائج.

النموذج 3: حلقة المقيم والمحسن

يقوم المحسن بإنشاء مسودة استجابة أو تنفيذ مهمة، ويقوم المقيِّم بفحصها وفقًا للمعايير الرسمية (مثل اختبارات الوحدة، أو الماسحات الضوئية الأمنية، أو مطالبة تقييم منفصلة). إذا فشل الفحص، فسيتم تمرير الملاحظات مرة أخرى إلى المحسن لإعادة إنشاء الاستجابة.

4. بناء وكيل معتمد على الدولة في لغة بايثون

دعونا نلقي نظرة على التنفيذ الملموس للوكيل المستقل باستخدام آلة الحالة البسيطة. سنقوم بتحديد وكيل يقوم بمعالجة طلبات استرداد الأموال. يتحقق الوكيل من سجل الشراء الخاص بالعميل، ويتحقق من صحة الطلب وفقًا للسياسات، ويصوغ ردًا عبر البريد الإلكتروني، ويطلب موافقة بشرية إذا تجاوز المبلغ المسترد 100 دولار.

import json
from typing import Dict, Any

# Mock databases and tools
PURCHASE_DB = {
    "user_123": {"item": "Premium Subscription", "price": 149.00, "days_ago": 12},
    "user_456": {"item": "Basic License", "price": 49.00, "days_ago": 45}
}

class AutonomousRefundAgent:
    def __init__(self):
        self.state: Dict[str, Any] = {
            "step": "INIT",
            "user_id": None,
            "refund_amount": 0.0,
            "policy_passed": False,
            "requires_approval": False,
            "approved": False,
            "response_draft": "",
            "log": []
        }

    def run(self, user_id: str, request_text: str):
        self.state["user_id"] = user_id
        self.state["log"].append(f"Started workflow for user {user_id} with request: '{request_text}'")
        
        while self.state["step"] != "COMPLETE":
            current_step = self.state["step"]
            if current_step == "INIT":
                self._fetch_user_data()
            elif current_step == "VALIDATE_POLICY":
                self._validate_policy()
            elif current_step == "CHECK_APPROVAL":
                self._check_approval_requirements()
            elif current_step == "WAITING_FOR_HUMAN":
                # Pause execution and yield control back to the orchestrator
                self.state["log"].append("Execution paused: waiting for human operator approval.")
                break
            elif current_step == "EXECUTE_REFUND":
                self._execute_refund()
            elif current_step == "DRAFT_RESPONSE":
                self._draft_response()
                
        return self.state

    def _fetch_user_data(self):
        user_id = self.state["user_id"]
        purchase = PURCHASE_DB.get(user_id)
        if not purchase:
            self.state["response_draft"] = "No purchase history found for this user."
            self.state["step"] = "DRAFT_RESPONSE"
            self.state["log"].append("Fetch failed: User not found in database.")
            return

        self.state["refund_amount"] = purchase["price"]
        self.state["purchase_age_days"] = purchase["days_ago"]
        self.state["step"] = "VALIDATE_POLICY"
        self.state["log"].append(f"Fetched purchase data: {purchase}")

    def _validate_policy(self):
        # Business rule: Refunds only allowed within 30 days
        age = self.state["purchase_age_days"]
        if age <= 30:
            self.state["policy_passed"] = True
            self.state["step"] = "CHECK_APPROVAL"
            self.state["log"].append("Policy validation passed (within 30-day window).")
        else:
            self.state["policy_passed"] = False
            self.state["response_draft"] = "Sorry, our policy only allows refunds within 30 days of purchase."
            self.state["step"] = "DRAFT_RESPONSE"
            self.state["log"].append("Policy validation failed: Purchase older than 30 days.")

    def _check_approval_requirements(self):
        # Business rule: Refunds over $100 require human review
        amount = self.state["refund_amount"]
        if amount > 100.0:
            self.state["requires_approval"] = True
            self.state["step"] = "WAITING_FOR_HUMAN"
            self.state["log"].append(f"Refund of ${amount} exceeds limit. Moving to human approval state.")
        else:
            self.state["requires_approval"] = False
            self.state["step"] = "EXECUTE_REFUND"
            self.state["log"].append(f"Refund of ${amount} is within limits. Proceeding to execution.")

    def resume_with_human_decision(self, approved: bool):
        if self.state["step"] != "WAITING_FOR_HUMAN":
            raise ValueError("Agent is not currently waiting for approval.")
        
        self.state["approved"] = approved
        self.state["log"].append(f"Human manager decision received: Approved = {approved}")
        
        if approved:
            self.state["step"] = "EXECUTE_REFUND"
        else:
            self.state["response_draft"] = "Your refund request has been reviewed and declined by a customer service manager."
            self.state["step"] = "DRAFT_RESPONSE"
            
        # Resume the workflow loop
        return self.run(self.state["user_id"], "")

    def _execute_refund(self):
        amount = self.state["refund_amount"]
        # Trigger actual external API call/Stripe integration here
        self.state["log"].append(f"Successfully processed stripe refund for ${amount}.")
        self.state["response_draft"] = f"Your refund request for ${amount} has been successfully processed."
        self.state["step"] = "DRAFT_RESPONSE"

    def _draft_response(self):
        # Prompt LLM to draft a polite, personalized message incorporating response_draft
        self.state["final_message"] = f"Dear Customer,\n\n{self.state['response_draft']}\n\nBest regards,\nGhaznix Support Agent"
        self.state["step"] = "COMPLETE"
        self.state["log"].append("Customer email drafted successfully. Workflow complete.")

5. موثوقية الإنتاج والحواجز الأمنية

يتطلب نشر الأنظمة المستقلة تحولاً في طريقة تفكيرنا في الاختبار ومعالجة الأخطاء. فيما يلي حواجز الحماية المهمة التي يجب عليك بنائها في أي نظام إنتاج:

أ. منع حلقات التنفيذ الهاربة

قد يقوم الوكيل المستقل الذي يواجه خطأً أو حالة حافة بالاستعلام عن نفس الأداة بشكل متكرر، مما يؤدي إلى إنفاق آلاف الدولارات في تكاليف واجهة برمجة التطبيقات (API) في غضون دقائق.

• الحل: تنفيذ الحدود القصوى للتنفيذ. حدد دائمًا حدًا أقصى لعدد الخطوات أو إجمالي الرموز المميزة للنموذج المسموح بها لكل مثيل سير عمل (على سبيل المثال، الحد الأقصى 10 تكرارات).

ب. التحقق من صحة المخطط الهيكلي

تعد LLMs احتمالية ولا تضمن بشكل طبيعي مخرجات منظمة مثل JSON الصالح أو المخططات المطابقة.

• الحل: استخدم مكتبات التحقق من الصحة مثل Pydantic أو Instructor أو Outlines لفرض البنية على مستوى الاستدلال. إذا قام النموذج بإخراج مخططات غير صالحة، فارفضها مبكرًا واطلب من النموذج الذي يحتوي على خطأ التحليل إصلاح نفسه (جزء من حلقة Evaluator-Optimizer).

ج. تنفيذ التعليمات البرمجية في وضع الحماية

إذا كان وكيلك يكتب التعليمات البرمجية ويقوم بتشغيلها (مثل تحليل البيانات أو تحويلات قاعدة البيانات)، فإن تشغيلها مباشرة على خادم التطبيق الخاص بك يمثل ثغرة أمنية كبيرة.

• الحل: استخدم بيئات Micro-VM الآمنة والسريعة الزوال (مثل حاويات Docker أو gVisor أو أوقات تشغيل WASM) لتشغيل البرامج النصية التي ينشئها المستخدم أو ينشئها الوكيل بأمان.

خاتمة

يتطلب إنشاء مسارات عمل مستقلة للذكاء الاصطناعي باستخدام LLM سد الفجوة بين التفكير المرن للذكاء الاصطناعي والانضباط الهندسي المنظم. من خلال استبدال الوكلاء ذوي النهايات المفتوحة بسير عمل موجه من قبل الآلة، وتنظيم المهام عبر أنماط مثل Orchestrator-Workers، وتغليف كل شيء في تنفيذ صارم وحواجز حماية أمنية، يمكن للمطورين بناء أنظمة ذكية للغاية وجاهزة للمؤسسات.

إن مستقبل هندسة البرمجيات لا يتعلق باستبدال التعليمات البرمجية بالمطالبات، بل يتعلق بتنسيق الوكلاء والأنظمة الحتمية لبناء مسارات عمل تعمل بشكل مستقل، وتتعلم ديناميكيًا، وتقدم النتائج بشكل موثوق.

استكشف المزيد من المقالات المتعلقة بهندسة الذكاء الاصطناعي على مدونة غازنيكس →