سير العمل متعدد الخطوات

تطبيقات الوكلاء في العالم الحقيقي تجمع أنماطاً متعددة في سير عمل متماسك. يُظهر هذا الدرس كيفية ربط الإجراءات وإدارة الحالة عبر العمليات المعقدة.

هندسة سير العمل

سير العمل المصمم جيداً يربط الأنماط في خط معالجة:

class AgentWorkflow:
    def __init__(self):
        self.state = {}
        self.history = []

    def run(self, task):
        # 1. التخطيط للنهج
        plan = self.plan(task)

        # 2. التنفيذ مع الأدوات
        for step in plan.steps:
            result = self.execute_step(step)
            self.state[step.name] = result
            self.history.append({"step": step, "result": result})

            # 3. التأمل والتعديل
            if self.should_reflect(result):
                improved = self.reflect(result)
                self.state[step.name] = improved

        # 4. تجميع المخرج النهائي
        return self.synthesize(self.state)

إدارة الحالة

الحفاظ على الحالة عبر الخطوات أمر حاسم لسير العمل المعقد:

نوع الحالة	الغرض	مثال
الذاكرة العاملة	سياق الخطوة الحالية	المتغيرات، النتائج الوسيطة
تاريخ المحادثة	ما حدث قبلاً	الرسائل السابقة، القرارات
تتبع التنفيذ	التصحيح والتعلم	الإجراءات المتخذة، النتائج
الحالة الخارجية	تغييرات العالم	تحديثات قاعدة البيانات، تغييرات الملفات

أنماط سير العمل الشائعة

خط الأنابيب المتسلسل

المدخل ← الخطوة 1 ← الخطوة 2 ← الخطوة 3 ← المخرج

الأفضل لـ: معالجة البيانات، توليد المحتوى

المنطق التفرعي

المدخل ← القرار ← [المسار أ] أو [المسار ب] ← المخرج

الأفضل لـ: دعم العملاء، المعالجة الشرطية

التنفيذ المتوازي

المدخل ← [الخطوة 1] + [الخطوة 2] + [الخطوة 3] ← الدمج ← المخرج

الأفضل لـ: البحث، جمع مصادر بيانات متعددة

الحلقة مع الخروج

المدخل ← المعالجة ← الفحص ← (التكرار أو الخروج) ← المخرج

الأفضل لـ: التحسين التكراري، مهام البحث

مثال عملي: وكيل بحث المحتوى

async def research_workflow(topic):
    # الخطوة 1: تخطيط نهج البحث
    research_plan = await plan_research(topic)

    # الخطوة 2: جمع المصادر (متوازي)
    sources = await asyncio.gather(
        search_academic(topic),
        search_news(topic),
        search_industry(topic)
    )

    # الخطوة 3: التحليل والتجميع
    analysis = await analyze_sources(sources)

    # الخطوة 4: التأمل في الاكتمال
    gaps = await identify_gaps(analysis, research_plan)

    # الخطوة 5: سد الفجوات إذا لزم الأمر
    if gaps:
        additional = await targeted_research(gaps)
        analysis = await merge_findings(analysis, additional)

    # الخطوة 6: إنشاء التقرير النهائي
    return await generate_report(analysis)

النقاط الرئيسية

• اجمع الأنماط بعناية بناءً على متطلبات المهمة
• أدِر الحالة لتمكين التفكير المعقد متعدد الخطوات
• ابنِ نقاط تفتيش للتصحيح والاستعادة
• صمم للفشل مع بدائل في كل خطوة

الآن اختبر فهمك مع اختبار الوحدة! :::