أساسيات 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
- أبقِ كل طريقة
@listenتركز على عمل واحد - أبقِ التفرع في
@router، تجنب كتل if/else كبيرة في كل مكان - استخدم حالة
BaseModelمكتوبة، وليس dict خام
بهذا الأسلوب، يمكنك العودة إلى الكود بعد ستة أشهر دون الشعور كعالم آثار.
متى يجب استخدام Flow؟
- سير العمل يحتوي أكثر من 3 خطوات
- شروط تفرع واضحة موجودة
- أطقم متعددة تحتاج التنسيق
إذا كان سير عملك مجرد "بحث + كتابة تقرير"، طاقم واحد يكفي.
الخطوة التالية
بعد تنسيق سير العمل، السؤال الشائع التالي هو: "هل سيتذكر الوكلاء السياق؟"
المقالة التالية تغطي الذاكرة ومصادر المعرفة:
👉 الذاكرة والمعرفة