أساسيات Flow في CrewAI: نسّق أطقماً متعددة في سير عمل منضبط

2 min read

عندما تصبح عمليتك "بحث → كتابة → مراجعة → نشر"، طاقم واحد لا يكفي.
Flow يعمل كبرج التحكم الذي ينسق أطقماً متعددة.

ما المشاكل التي يحلها Flow؟

  • إدارة الحالة متعددة الخطوات
  • التعامل مع منطق التفرع (نشر عند النجاح، إعادة التشغيل عند الفشل)
  • تحسين الرؤية وقابلية الصيانة

مثال Flow بسيط

from pydantic import BaseModel
from crewai.flow.flow import Flow, start, listen, router
 
class BlogState(BaseModel):
    topic: str = ""
    draft: str = ""
    word_count: int = 0
 
class BlogFlow(Flow[BlogState]):
    @start()
    def init_topic(self) -> str:
        self.state.topic = "CrewAI للمبتدئين"
        return self.state.topic
 
    @listen(init_topic)
    def write_draft(self, topic: str) -> str:
        result = "محتوى المسودة..."
        self.state.draft = result
        self.state.word_count = len(result.split())
        return result
 
    @router(write_draft)
    def quality_gate(self) -> str:
        return "ok" if self.state.word_count > 100 else "retry"

اقتراحات تصميم Flow

  1. أبقِ كل طريقة @listen تركز على عمل واحد
  2. أبقِ التفرع في @router، تجنب كتل if/else كبيرة في كل مكان
  3. استخدم حالة BaseModel مكتوبة، وليس dict خام

بهذا الأسلوب، يمكنك العودة إلى الكود بعد ستة أشهر دون الشعور كعالم آثار.

متى يجب استخدام Flow؟

  • سير العمل يحتوي أكثر من 3 خطوات
  • شروط تفرع واضحة موجودة
  • أطقم متعددة تحتاج التنسيق

إذا كان سير عملك مجرد "بحث + كتابة تقرير"، طاقم واحد يكفي.

الخطوة التالية

بعد تنسيق سير العمل، السؤال الشائع التالي هو: "هل سيتذكر الوكلاء السياق؟"
المقالة التالية تغطي الذاكرة ومصادر المعرفة:
👉 الذاكرة والمعرفة