كيفية منع وكلاء CrewAI من تنفيذ المهام المتكررة: دليل عملي لإزالة التكرار

عندما يبدو أن وكلاء CrewAI ينفذون نفس العمل مرتين، فإن السبب عادةً لا يكون إعداد "مهمة مكررة" واحد. قد ينشأ التكرار من تداخل أوصاف المهام، أو التفويض الهرمي، أو سلوك إعادة المحاولة، أو مُشغّلات التدفق المتعددة، أو عمليات بدء تشغيل الطاقم المتكررة، أو الأدوات ذات التأثيرات الجانبية غير المحمية بمفتاح التكرار.

تمت مراجعة هذا المرجع العملي مع وثائق CrewAI الرسمية بتاريخ 13 سبتمبر 2026. وتشير الوثائق الحالية إلى الإصدار 1.15.14 من CrewAI. ويكمن الفرق الأهم في أن منع الاستدلال المتكرر داخل عملية تشغيل واحدة يختلف عن منع تكرار نفس الإجراء التجاري مرتين في عمليات تشغيل متعددة . يوفر CrewAI سياق المهمة، والمهام المشروطة، وردود الاتصال، وحالة التدفق، والاستمرارية، وتخزين الأدوات مؤقتًا، ولكن لا يزال يتعين عليك تصميم شروط تخطي صريحة للعمل الذي يجب أن يُنفذ مرة واحدة فقط.

تشخيص سريع: لماذا يحدث نفس العمل مرتين؟

المراجع الرسمية: مهام CrewAI ، ووكلاء CrewAI ، وتدفقات CrewAI .

1. ابدأ بمالك واحد لكل وحدة عمل

أبسط قواعد منع التكرار هي أيضاً الأكثر فعالية: يجب أن يكون لكل وحدة عمل مهمة واحدة مسؤول واحد. في عملية CrewAI التسلسلية، تُنفَّذ المهام بالترتيب الذي تم تعريفها به. contextتسمح هذه الخاصية للمهمة اللاحقة باستهلاك مخرجات المهمة السابقة بدلاً من إعادة اكتشاف نفس المعلومات بشكل مستقل.

يبدو النمط المضاد الشائع على النحو التالي من الناحية المفاهيمية:

research_task: "Research the customer and summarize findings"
analysis_task: "Research the customer, analyze findings, and recommend actions"
writer_task:   "Review the customer, research missing details, and write the report"

تتضمن المهام الثلاث جميعها إذنًا بالبحث، لذا فإن تكرار البحث أمر متوقع. يُفضّل اتباع سلسلة بحث أضيق.

from crewai import Agent, Crew, Process, Task

research_task = Task(
    description="Research the customer once and return verified facts and sources.",
    expected_output="Structured research notes with sources.",
    agent=researcher,
)

analysis_task = Task(
    description="Analyze only the research provided in context. Do not perform new research.",
    expected_output="Prioritized findings and recommendations.",
    agent=analyst,
    context=[research_task],
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
)

تدعم وثائق مهام CrewAI الحالية بشكل صريح تبعيات المهام context، وتنفذ العملية التسلسلية المهام بالترتيب المذكور. راجع وثائق تبعيات المهام ووثائق العملية الرسمية .

قائمة مرجعية عملية لحدود المهام

• أعطِ كل مهمة فعلاً مختلفاً عن غيرها: البحث، والتطبيع، والتحليل، والكتابة، والمراجعة.
• حدد ما يجب ألا تفعله المهمة عندما يكون التداخل مكلفًا.
• اجعل الناتج المتوقع ملموساً بما يكفي بحيث يمكن للمهمة التالية استخدامه مباشرة.
• تمرير النتائج السابقة contextبدلاً من إخبار العملاء اللاحقين "بالبحث إذا لزم الأمر".
• قم بتقييد الأدوات على مستوى المهمة أو الوكيل عندما يُسمح لدور واحد فقط بالبحث أو الكتابة إلى قاعدة بيانات أو إرسال رسائل أو استدعاء واجهة برمجة تطبيقات خارجية.

2. تخطي العمل الذي تم إنجازه بالفعل باستخدام ConditionalTask

إذا كانت المهمة مطلوبة فقط عندما تكون نتيجة سابقة غير مكتملة، فلا تجعل العامل يقرر بشكل غير رسمي ما إذا كان سيكرر العمل أم لا. يوفر CrewAI ConditionalTaskشرطًا يستقبل مخرجات المهمة السابقة ويمكنه تخطي التنفيذ عندما يكون الشرط خاطئًا.

يستخدم المثال الرسمي شرطًا يتحقق مما إذا تم إرجاع عدد كافٍ من سجلات الأحداث؛ فإذا كانت البيانات كافية بالفعل، يتم تخطي مهمة الجلب الإضافية. انظر مهام CrewAI الشرطية .

from typing import List
from pydantic import BaseModel
from crewai import Agent, Crew, Task
from crewai.tasks.conditional_task import ConditionalTask
from crewai.tasks.task_output import TaskOutput

class ResearchOutput(BaseModel):
    sources: List[str]
    summary: str

def needs_more_sources(output: TaskOutput) -> bool:
    return len(output.pydantic.sources) < 5

research = Task(
    description="Find up to five authoritative sources about the topic.",
    expected_output="A structured research result.",
    agent=researcher,
    output_pydantic=ResearchOutput,
)

enrich = ConditionalTask(
    description="Find only the missing sources needed to reach five total.",
    expected_output="Additional authoritative sources only.",
    condition=needs_more_sources,
    agent=researcher,
)

هذا النمط أقوى من توجيه الوكيل بـ "تجنب القيام بعمل مكرر" لأن قرار التخطي هو منطق بايثون حتمي وليس حكمًا لنموذج لغة آخر.

3. معرفة متى يؤدي التفويض إلى تكرار واضح

يدعم CrewAI العمليات التسلسلية والهرمية على حد سواء. في نظام الطاقم الهرمي، يقوم المدير بتوزيع المهام، وتفويض العمل، والتحقق من صحة المخرجات، وتحديد ما إذا كان إنجاز المهمة مُرضيًا. تُعد هذه المرونة مفيدة عندما يتطلب توزيع العمل ديناميكية، ولكنها تعني أيضًا أن المسؤولية أقل وضوحًا مقارنةً بنظام الطاقم التسلسلي.

تنص وثائق وكلاء CrewAI حاليًا على أن allow_delegationالإعداد الافتراضي هو False. احتفظ بهذا الإعداد الافتراضي للمتخصصين إلا إذا كان الوكيل بحاجة فعلية لتفويض العمل إلى وكيل آخر. في العمليات الهرمية، يكون المدير نفسه مسؤولاً عن التفويض. راجع دليل العمليات الهرمية .

نقطة البداية الآمنة للطاقم الذي يُجري مكالمات زائدة هي:

researcher = Agent(
    role="Researcher",
    goal="Collect evidence once and return it in structured form.",
    backstory="You gather evidence; you do not write the final report.",
    allow_delegation=False,
    max_iter=8,
    max_retry_limit=1,
)

writer = Agent(
    role="Writer",
    goal="Write from supplied context without doing fresh research.",
    backstory="You synthesize existing evidence into a final answer.",
    allow_delegation=False,
    max_iter=6,
)

ثم أضف التفويض مرة أخرى فقط عندما يُحقق فائدة ملموسة. إذا لم تكن هناك حاجة إلى التنسيق الهرمي، Process.sequentialفسيكون تصحيح الأخطاء أسهل لأن ترتيب المهام وملكية الأفراد يكونان واضحين.

4. لا تخلط بين إعادة المحاولات وجدولة المهام المكررة

من المتوقع حدوث بعض المحاولات المتكررة. يمكن لضوابط مهام CrewAI التحقق من صحة المخرجات وإرسال ملاحظات إلى الوكيل عند فشل التحقق. تشير وثائق المهمة الحالية إلى أن guardrail_max_retriesالقيمة الافتراضية هي 3، وفي حالة فشل أحد الضوابط، تتم إعادة محاولة المهمة حتى هذا الحد.

تُعرّض البرامج أيضًا max_retry_limitأخطاء التنفيذ والحد max_iterالأقصى لعدد مرات تكرارها قبل تقديم أفضل إجابة متاحة. تشير وثائق البرنامج الحالية إلى قيمة افتراضية max_iterقدرها 20، وحد أقصى افتراضي لإعادة المحاولة عند حدوث خطأ قدره 2.

تحل هذه الآليات مشاكل مختلفة:

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

5. تتبع ما تم تنفيذه فعليًا قبل إعادة كتابة التعليمات

يُتيح CrewAI العديد من نقاط المراقبة. على مستوى الطاقم، تتضمن الوثائق الحالية عناصر تحكم التتبع verbose، و step_callback، task_callbackو، output_log_fileو، و. كما يدعم الوكلاء أيضًا step_callback. تُفيد هذه العناصر في الإجابة على أربعة أسئلة:

• هل بدأ المجدول نفس المهمة مرتين؟
• هل قام أحد العملاء بتنفيذ عدة تكرارات داخل مهمة واحدة؟
• هل رفض نظام الحماية المخرجات وأدى إلى إعادة المحاولة؟
• هل تكرر استدعاء أداة ما على الرغم من أن المهمة نفسها تم تشغيلها مرة واحدة؟

لإجراء عملية تشخيصية أولية، قم بتشغيل الإخراج المطول وملف سجل JSON:

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True,
    output_log_file="logs/crew-run.json",
)

يمكنك بعد ذلك إضافة وظائف رد نداء إذا كنت بحاجة إلى عدادات منظمة أو بيانات قياس عن بُعد مخصصة. راجع وثائق سمات الطاقم في CrewAI .

6. منع تكرار مشغلات التدفق

تُضيف التدفقات نوعًا آخر من التكرار. تنص وثائق التدفقات الحالية في CrewAI على أن جميع @start()الطرق المُحققة تُنفذ عند بدء التدفق أو استئنافه. إذا حددتَ عدة بدايات غير مشروطة، وقامت اثنتان منها بتشغيل نفس الطاقم، فإن التكرار يكون في الرسم البياني، وليس داخل الطاقم نفسه.

وبالمثل، or_يمكن تشغيل المستمعين عندما تُصدر أي دالة في المصدر مُخرجات. يُظهر مثال CrewAI تشغيل المستمع مرة واحدة لكل مُخرج في المصدر. استخدم هذه الطريقة and_عندما يجب أن تنتظر العملية اللاحقة حتى تكتمل جميع المتطلبات الأساسية، أو @router()عندما يجب أن يتابع فرع واحد فقط.

قبل إضافة نقطة دخول ثانية @start()، تأكد من أنها تمثل بالفعل نقطة دخول مستقلة. إذا لم تكن كذلك، فاستخدم طريقة بدء واحدة ومستمعين صريحين.

7. أضف مفتاح إكمال لضمان عدم تكرار العمليات عبر التشغيل

هذا هو نمط الإنتاج الأكثر أهمية عندما تؤدي مهمة ما تأثيرًا جانبيًا خارجيًا مثل إرسال بريد إلكتروني، أو تحصيل رسوم من طريقة دفع، أو إنشاء سجل CRM، أو نشر رسالة، أو بدء مهمة.

تدعم CrewAI Flows الحالة المنظمة والديكور @persist. تتيح خاصية الثبات استعادة الحالة بعد إعادة التشغيل. مع ذلك، لا تُحدد الحالة المحفوظة وحدها ما إذا كان ينبغي تخطي إجراء تجاري. لذا، يُنصح بتخزين مفتاح العملية الحتمية الخاص بك والتحقق منه قبل تنفيذ التأثير الجانبي.

from hashlib import sha256
import json
from pydantic import BaseModel
from crewai.flow.flow import Flow, start
from crewai.flow.persistence import persist

class JobState(BaseModel):
    completed_keys: list[str] = []
    report: str = ""

def operation_key(customer_id: str, period: str) -> str:
    payload = {"customer_id": customer_id, "period": period}
    raw = json.dumps(payload, sort_keys=True).encode()
    return sha256(raw).hexdigest()

@persist
class ReportFlow(Flow[JobState]):

    @start()
    def run_report(self):
        key = operation_key("customer-123", "2026-09")

        if key in self.state.completed_keys:
            return self.state.report

        result = reporting_crew.kickoff(
            inputs={"customer_id": "customer-123", "period": "2026-09"}
        )

        self.state.report = result.raw
        self.state.completed_keys.append(key)
        return self.state.report

تُتيح وثائق CrewAI @persistإمكانية تخزين حالة Flow عبر عمليات إعادة التشغيل، كما أن استئناف التشغيل باستخدام مُعرّف الحالة نفسه يُعيد تحميل أحدث لقطة. راجع وثائق CrewAI الخاصة باستمرارية Flow .

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

8. قم بتخزين استدعاءات الأدوات المتكررة مؤقتًا، ولكن لا تخلط بين التخزين المؤقت وتكرار المهمة.

تُتيح وكلاء CrewAI وفرق العمل الوصول إلى البيانات cache، وتصفها الوثائق الرسمية بأنها تخزين مؤقت لنتائج تنفيذ الأداة. تُظهر وثائق الوكيل الحالية أن التخزين المؤقت مُفعّل افتراضيًا، وتوصي إرشادات الأداء بإبقائه مُفعّلاً عند استخدام الأداة بشكل متكرر.

يُفيد ذلك عندما يُجري العميل نفس عملية البحث المكلفة أو استدعاء أداة حتمية أكثر من مرة. لكن هذا لا يعني أن الاستدعاء crew.kickoff()مرتين سيتجاوز مهام الطاقم تلقائيًا، فالمهمة لا تزال ضمن مخطط التنفيذ.

استخدم ذاكرة التخزين المؤقت للقراءة المتكررة. استخدم مفتاح التكرار للكتابة المتكررة.

9. تقلل الذاكرة من الاكتشاف المتكرر، لكنها لا تلغي المهام

يخزن نظام الذاكرة الموحد في CrewAI المعلومات بعد إنجاز المهام، ويسترجع السياق ذي الصلة قبل إنجازها. وتشير الوثائق الحالية إلى أنه عند تفعيل ذاكرة الطاقم، تُستخرج المعلومات المحددة من مخرجات المهام، وتُضاف الذكريات ذات الصلة إلى مطالبات المهام اللاحقة.

يُمكن لهذا أن يُقلل من إعادة الاكتشاف غير الضرورية، خاصةً عندما يكون من المفترض أن يعرف الكاتب ما توصل إليه الباحث بالفعل. لكن الذاكرة هي سياق الاسترجاع، وليست علامة تخطي. ستظل المهمة المُجدولة مُنفذة ما لم يُقرر نظام إدارة المهام أو منطق سير العمل خلاف ذلك.

استخدم الذاكرة لسؤال "ما الذي نعرفه بالفعل؟" استخدم الحالة أو مهمة مشروطة لسؤال "هل يجب تشغيل هذه العملية؟" انظر ذاكرة CrewAI .

10. استخدم مخرجات منظمة لجعل قرارات التخطي موثوقة

يصعب استخدام مخرجات اللغة الطبيعية للتحكم في سير العمل بشكل حتمي. يمكن لمهام CrewAI إرجاع مخرجات Pydantic أو JSON من خلال output_pydantic. output_jsonتُسهّل النتيجة المنظمة تحديد ما إذا كانت هناك حاجة إلى إثراء أو مراجعة أو تصعيد أو مهمة أخرى.

على سبيل المثال، الإرجاع:

{
  "status": "complete",
  "sources_found": 7,
  "missing_fields": [],
  "needs_review": false
}

ثم يتم توجيه الطلبات بناءً على الحقول بدلاً من تكليف وكيل آخر بتفسير فقرة نثرية. هذا عادةً ما يقلل من تكرار العمل والغموض في التعليمات.

التكوين الموصى به لمكافحة التكرار

بالنسبة لخط سير البحث النموذجي الذي يبدأ من البحث وحتى إعداد التقرير، ابدأ بشكل متحفظ:

researcher = Agent(
    role="Researcher",
    goal="Collect evidence once.",
    backstory="Owns external research.",
    allow_delegation=False,
    max_iter=8,
    max_retry_limit=1,
    cache=True,
)

analyst = Agent(
    role="Analyst",
    goal="Analyze supplied evidence only.",
    backstory="Does not repeat research.",
    allow_delegation=False,
    max_iter=6,
)

crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential,
    verbose=True,
    cache=True,
    output_log_file="logs/run.json",
)

ثم أضف التعقيد فقط عندما يتطلب الأمر ذلك:

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

قائمة التحقق النهائية لحل المشكلات

• هل تم ذكر نفس الشيء Taskمرتين في قائمة مهام الطاقم؟
• هل يسمح وصفان للمهام بنفس البحث أو استدعاء الأداة؟
• هل يمكن لمهمة لاحقة أن تستهلك المهمة السابقة contextبدلاً من ذلك؟
• هل ينبغي أن تكون المهمة المتكررة ConditionalTask؟
• هل تستخدم التنسيق الهرمي في حين أن التنسيق التسلسلي سيكون كافياً؟
• هل allow_delegationيتم تفعيل هذه الميزة على الأجهزة التي لا تحتاجها؟
• هل يؤدي رفض نظام الحماية إلى إعادة المحاولة؟
• هل max_iterحجمها كبير بما يكفي بحيث تقوم مهمة واحدة بتنفيذ العديد من استدعاءات الأدوات المتشابهة؟
• هل تقوم عدة @start()طرق أو or_أجهزة استماع بتشغيل نفس الطاقم في اتجاه المصب؟
• هل تمثل العملية الثانية kickoff()تشغيلاً جديداً متعمداً أم استدعاءً مكرراً عرضياً؟
• هل تعتمد على الذاكرة أو ذاكرة التخزين المؤقت كما لو كانت عناصر تحكم في عدم تكرار العمليات على مستوى المهمة؟
• هل تحتوي عمليات الكتابة الخارجية على مفتاح تطابق حتمي؟
• هل يمكن للسجلات أن تثبت ما إذا كان التكرار قد حدث على مستوى المهمة، أو خطوة الوكيل، أو الأداة، أو مستوى التدفق؟

خلاصة القول

يُعدّ إيقاف مهام CrewAI المتكررة في الغالب مشكلة تنسيق. اجعل ملكية المهام واضحة، واربط المهام ببعضها context، وتخطَّ المهام المنجزة بالفعل بشكل مشروط، واحرص على أن يكون التفويض محدودًا، وافهم متى تكون إعادة المحاولة مقصودة، وتتبّع التنفيذ قبل تغيير المطالبات. بالنسبة للتدفقات والآثار الجانبية للإنتاج، اذهب خطوة أبعد: احتفظ بمفتاح إكمال حتمي أو استخدم آلية خارجية للتكرار.

النموذج الذهني المفيد بسيط: السياق يمنع إعادة الاكتشاف، والشروط تمنع المهام غير الضرورية، والذاكرة المؤقتة تمنع تكرار حسابات الأدوات، والتكرارية تمنع الآثار الجانبية المتكررة . إنها تحل مشاكل مترابطة، لكنها ليست قابلة للتبادل.