إنشاء وكلاء CrewAI متعددي النماذج باستخدام CometAPI:

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

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

هندسة أنظف هي أن تدع CrewAI يدير الوكلاء وسير العمل بينما CometAPI يدير الوصول إلى النماذج.

يوفّر CometAPI نقطة نهاية متوافقة مع OpenAI على الرابط https://api.cometapi.com/v1، بحيث يمكن للتطبيقات توجيه الطلبات إلى نماذج من مزوّدين متعدّدين عبر واجهة API مشتركة. تدعم توثيق البدء السريع الحالي أيضاً استخدام حزمة OpenAI Python SDK القياسية عبر تغيير مفتاح الـAPI وعنوان base URL.

في هذا الدليل، ستبني سير عمل CrewAI بثلاثة وكلاء باستخدام:

• Gemini 3.7 Flash للبحث
• Claude Opus 5 للتحليل
• GPT-5.6 للكتابة النهائية
• مفتاح CometAPI واحد
• عنوان API base URL واحد
• ضبط نموذج لكل وكيل
• تراجع محدود للحالات العابرة
• تخزين نقاط التحقق (checkpointing) في CrewAI لاستعادة الإنتاج
• تتبّع استخدام الرموز والتنفيذ
• تحقق من النماذج على الخادم

الحدّ المعماري المهم بسيط:

CrewAI يتولّى تنظيم الوكلاء. CometAPI يتولّى الوصول إلى النماذج. معرّفات النماذج تحدّد التوجيه.

---

ما هو توجيه النماذج متعدد الوكلاء في CrewAI؟

CrewAI هو إطار Python لإنشاء وكلاء ومهام وفرق وسير عمل متعدد الوكلاء. يمكن لكل وكيل الحصول على إعداد LLM خاص به، بينما يقوم Crew بتنسيق كيفية تنفيذ هؤلاء الوكلاء للمهام وتبادل السياق.

يدعم إعداد LLM الحالي في CrewAI تحديد إعدادات model وapi_key وbase_url بشكل صريح، بما في ذلك نقاط النهاية المتوافقة مع OpenAI.

وهذا يجعل هندسة النماذج المتعددة مباشرة:

                         CometAPI                            │              https://api.cometapi.com/v1                            │        ┌───────────────────┼───────────────────┐        │                   │                   │   Researcher            Analyst             Writer        │                   │                   │ Gemini 3.7 Flash      Claude Opus 5         GPT-5.6

يبقى الوكلاء منفصلين من منظور منطقي، لكن يتم توحيد وصولهم إلى النماذج.

هذا يختلف عن القول إن جميع النماذج قابلة للتبديل. توفر واجهة API المتوافقة مع OpenAI واجهة طلب مشتركة؛ لكنها لا تضمن حدود سياق متطابقة، أو دعم أدوات، أو عناصر تحكم في الاستدلال، أو سلوك إخراج، أو زمناً للاستجابة، أو تسعيراً متطابقاً.

هذا الفارق مهم عند تصميم التوجيه للإنتاج.

---

لماذا تستخدم CometAPI مع CrewAI؟

الميزة الأساسية ليست أن CrewAI يصبح فجأة إطاراً متعدد المزوّدين. CrewAI يدعم بالفعل مزوّدين متعددين.

الميزة هي أن الوصول إلى النماذج يمكن توحيده خلف طبقة API واحدة.

بدون طبقة API موحّدة، قد يبدو سير عمل بثلاثة وكلاء كما يلي:

مع CometAPI:

تصف توثيقات البدء السريع الحالية لـ CometAPI نقطة النهاية كبديل مباشر لعنوان OpenAI API base URL وتعرض نماذج من مزوّدين متعددين عبر الخدمة نفسها.

هذا يمنح التطبيق فصلاً مفيداً:

CrewAI

• يحدّد أدوار الوكلاء
• يحدّد المهام
• يمرّر السياق
• يتحكّم بالتنفيذ
• يدير تكرارات الوكلاء
• يتعامل مع تنظيم الطاقم على مستوى الفريق

CometAPI

• يوفّر طبقة وصول موحّدة للنماذج
• يركّز مصادقة API
• يوفّر توجيه النماذج عبر معرّفات النماذج
• يمنح التطبيق نقطة نهاية API واحدة
• يوفّر رؤية مركزية للاستخدام والفوترة

---

ماذا سينتج هذا سير عمل CrewAI؟

ينشئ المثال ثلاثة وكلاء متسلسلين.

هذا سياسة توجيه مثالية كمثال، وليست ترتيباً معيارياً.

يعتمد النموذج المناسب لوكيلك على:

• تعقيد المهمة
• طول السياق المطلوب
• استخدام الأدوات
• متطلبات الإخراج المنظم
• الكمون (latency)
• الموثوقية
• تكلفة الرموز
• جودة المخرجات
• نتائج التقييم الخاصة بالتطبيق

قاعدة مفيدة:

اختر النموذج وفقاً لعمل الوكيل، وليس لمجرّد المزوّد الذي يأتي منه.

---

أي نموذج يجب أن يستخدمه كل وكيل في CrewAI؟

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

الباحث: Gemini 3.7 Flash

غالباً ما يتضمن البحث معالجة كميات كبيرة نسبياً من المعلومات وإنتاج نتيجة وسيطة مضغوطة.

لذلك يمكن أن يكون النموذج السريع مفيداً لمهام البحث ذات الحجم الكبير.

"researcher": "gemini-3.7-flash"

المحلل: Claude Opus 5

للمحلل دور أضيق ولكنه أكثر كثافة في الاستدلال. يتلقى مخرجات البحث ويحوّلها إلى توصية.

"analyst": "claude-opus-5"

الكاتب: GPT-5.6

يحوّل الوكيل الأخير البحث والتحليل إلى مذكرة قرار موجّهة للمطورين.

"writer": "gpt-5.6"

الجزء المهم ليس هذه التعيينات الثلاثة بالتحديد. يجب على تطبيقك تقييم النماذج المرشحة مقابل مهام تمثيلية قبل تثبيت سياسة التوجيه.

---

ما الذي تحتاجه قبل البدء؟

تحتاج إلى:

• Python 3.10+
• CrewAI
• توافق مع OpenAI Python SDK
• python-dotenv
• مفتاح CometAPI API
• معرّفات النماذج التي تنوي استخدامها

يدعم تكامل Python الحالي لـ CometAPI واجهة API المتوافقة مع OpenAI، وتوثّق حزمة CometAPI Python الرسمية المتغيرات COMETAPI_KEY وCOMETAPI_BASE_URL كخيارات ضبط عبر البيئة.

نقطة النهاية القياسية هي:

https://api.cometapi.com/v1

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

---

كيف تثبّت CrewAI والاعتمادات؟

أنشئ بيئة Python جديدة:

python -m venv .venv

قم بتفعيلها:

source .venv/bin/activate

على Windows:

.venv\Scripts\Activate.ps1

ثم ثبت الاعتمادات:

pip install "crewai[openai]" openai python-dotenv

استخدام openai صراحةً مقصود لأن تنفيذ التراجع أدناه يستورد أصناف استثناءات OpenAI SDK مباشرة.

للإنتاج، قم بتثبيت إصدارات محدّدة تم اختبارها بدلاً من الاعتماد إلى أجل غير مسمى على أحدث الإصدارات الطافية.

على سبيل المثال:

crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION

طبقة LLM في CrewAI تتطور بنشاط، لذا يجب التحقق من الباني (constructor) الدقيق وإعداد المزوّد مقابل نسخة CrewAI المستخدمة في تطبيقك. توثيق CrewAI الحالي يدعم ضبط LLM مع base_url مخصص ومفتاح API.

---

كيف تضبط مفتاح CometAPI API؟

أنشئ ملف .env:

COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1

حمّل هذه القيم في Python:

import osfrom dotenv import load_dotenvload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv(    "COMETAPI_BASE_URL",    "https://api.cometapi.com/v1",)

لا تقم أبداً بضمّ ملف .env إلى Git.

أضفه إلى .gitignore:

.env.venv/__pycache__/

يجب أن يبقى مفتاح الـAPI كبيانات اعتماد على جهة الخادم. توجيهات البدء السريع الحالية لـ CometAPI توصي أيضاً بتخزين المفتاح في متغيرات البيئة بدلاً من الشفرة المصدرية.

---

كيف توصل CrewAI بـ CometAPI؟

يمكن أن يتلقّى كائن LLM في CrewAI اسم نموذج ومفتاح API وعنوان base URL مخصص.

أنشئ أداة مساعدة:

from crewai import LLMdef cometapi_llm(model_id: str) -> LLM:    return LLM(        model=model_id,        base_url=COMETAPI_BASE_URL,        api_key=COMETAPI_KEY,        timeout=60.0,        max_retries=0,    )

هذا أفضل من تضمين الإعداد نفسه في كل وكيل على حدة.

كل وكيل يحتاج الآن فقط إلى معرّف نموذج:

research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")

لماذا نحدّد max_retries=0؟

السبب هو التحكم في التراجع (fallback).

إذا كان عميل LLM الأساسي يعيد المحاولة تلقائياً وتطبيقك يطبّق أيضاً تراجعاً، فقد يتحوّل فشل واحد إلى عدة طلبات مخفية قبل أن تنفذ منطق التراجع.

لأغراض دليل تعليمي مع توجيه صريح، من الأنظف أن تدع التطبيق يقرر متى يعيد المحاولة أو يبدّل النموذج.

---

كيف تعرّف سياسة توجيه النماذج؟

أبقِ التوجيه خارج مطالباتك:

PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",}FALLBACK_MODELS = {    "researcher": "gpt-5.6",    "analyst": "gpt-5.6",    "writer": "gemini-3.7-flash",}

هذا يخلق حدود ضبط واضحة.

يمكنك لاحقاً نقل الخريطة ذاتها إلى:

• ضبط عبر البيئة
• YAML
• JSON
• قاعدة بيانات
• أعلام ميزات
• خدمة داخلية لتوجيه النماذج

دون إعادة كتابة مطالبات الوكلاء.

---

كيف تبني وكلاء CrewAI الثلاثة؟

أنشئ كائناً واحداً من LLM لكل وكيل.

from crewai import Agentdef build_agents(model_map: dict[str, str]):    researcher = Agent(        role="Market Researcher",        goal="Collect the facts needed to answer the topic",        backstory=(            "You create concise, source-aware research briefs "            "and clearly separate facts from assumptions."        ),        llm=cometapi_llm(model_map["researcher"]),        max_iter=3,        allow_delegation=False,    )    analyst = Agent(        role="Product Analyst",        goal="Turn research into a defensible recommendation",        backstory=(            "You identify evidence, assumptions, risks, "            "and trade-offs before making recommendations."        ),        llm=cometapi_llm(model_map["analyst"]),        max_iter=3,        allow_delegation=False,    )    writer = Agent(        role="Technical Writer",        goal="Produce a concise technical decision memo",        backstory=(            "You write clear technical explanations "            "without unnecessary marketing language."        ),        llm=cometapi_llm(model_map["writer"]),        max_iter=3,        allow_delegation=False,    )    return researcher, analyst, writer

أصبح تعيين النموذج الآن مستقلاً تماماً عن تعريف دور الوكيل.

وهذا ما يجعل توجيه النماذج عملياً.

---

كيف تربط الوكلاء بمهام متسلسلة؟

أنشئ ثلاث مهام:

from crewai import Taskdef build_tasks(researcher, analyst, writer):    research_task = Task(        description=(            "Research this topic: {topic}. "            "Return the key facts, uncertainties, "            "and relevant sources that the analyst should consider."        ),        expected_output=(            "A compact research brief containing facts, "            "uncertainties, and source references."        ),        agent=researcher,    )    analysis_task = Task(        description=(            "Using the research brief, analyze {topic}. "            "Identify the strongest conclusion and explain "            "the major trade-offs."        ),        expected_output=(            "A decision outline with evidence, "            "assumptions, risks, and trade-offs."        ),        agent=analyst,        context=[research_task],    )    writing_task = Task(        description=(            "Write a concise technical decision memo about {topic}. "            "State the recommendation early and preserve "            "important caveats."        ),        expected_output="A polished technical decision memo in Markdown.",        agent=writer,        context=[research_task, analysis_task],    )    return research_task, analysis_task, writing_task

سلسلة الاعتمادية هي:

Topic  ↓Research  ↓Analysis  ↓Final memo

يتلقى المحلل مخرجات مهمة البحث، بينما يتلقى الكاتب سياق البحث والتحليل كلاهما.

---

كيف تبني الطاقم؟

اجمع الوكلاء والمهام:

from crewai import Crew, Processdef build_crew(model_map: dict[str, str]) -> Crew:    researcher, analyst, writer = build_agents(model_map)    research_task, analysis_task, writing_task = build_tasks(        researcher,        analyst,        writer,    )    return Crew(        agents=[researcher, analyst, writer],        tasks=[            research_task,            analysis_task,            writing_task,        ],        process=Process.sequential,        verbose=True,    )

الآن أصبح توجيه النماذج مدفوعاً بالكامل عبر الضبط.

تغيير:

"researcher": "gemini-3.7-flash"

إلى نموذج مدعوم آخر لا يتطلب تغيير مطالبة البحث أو تعريف المهمة.

---

كيف يجب أن يعمل التراجع عن نموذج CrewAI؟

هنا يحتاج تنفيذ موجّه للإنتاج إلى مزيد من العناية.

خطأ شائع هو:

Any error   ↓Switch model

هذا عدواني جداً.

على سبيل المثال، لا ينبغي لهذه الأخطاء عادةً أن تفعّل التراجع عن النموذج:

400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error

لن يؤدي تبديل النماذج إلى إصلاح مفتاح API غير صالح أو طلب غير مُشكّل بشكل صحيح.

التراجع أنسب للإخفاقات المؤقتة مثل:

408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout

يجب أن تكون سياسة التراجع بالتالي:

أعد المحاولة أو بدّل النماذج فقط للإخفاقات العابرة المحدودة وفقط عندما يدعم نموذج التراجع عقد الطلب نفسه.

---

كيف تكتشف الأخطاء القابلة لإعادة المحاولة؟

يمكنك استخدام أصناف أخطاء OpenAI SDK:

from collections.abc import Iteratorfrom openai import (    APIConnectionError,    APIStatusError,    APITimeoutError,)def exception_chain(error: BaseException) -> Iterator[BaseException]:    current: BaseException | None = error    seen: set[int] = set()    while current is not None and id(current) not in seen:        seen.add(id(current))        yield current        current = (            current.__cause__            or current.__context__        )def should_fallback(error: BaseException) -> bool:    for current in exception_chain(error):        if isinstance(            current,            (APIConnectionError, APITimeoutError),        ):            return True        if isinstance(current, APIStatusError):            return (                current.status_code in {408, 429}                or current.status_code >= 500            )    return False

هذا يستبعد عمداً أخطاء 400 المتعلّقة بالضبط باستثناء 408 و429.

---

هل تعيد محاولة الطاقم بأكمله أم الوكيل الذي فشل فقط؟

هناك استراتيجيتان مختلفتان للتراجع.

تراجع على مستوى الطاقم

أسهل تنفيذ هو:

Start crew   ↓failure   ↓change routing   ↓run crew again

هذا سهل الفهم، لكنه قد يكرر المهام المنجزة.

على سبيل المثال:

Research → completedAnalysis → completedWriter → failed

إعادة محاولة kickoff() كاملة قد تنفّذ:

Research → againAnalysis → againWriter → fallback

هذا يزيد:

• استخدام الرموز
• الكمون
• تكلفة API
• الآثار الجانبية المحتملة

استرداد على مستوى المهمة

بدلاً من ذلك، ينبغي أن يخزّن سير العمل الإنتاجي نقاط تحقق للعمل المكتمل:

Research   ↓checkpoint   ↓Analysis   ↓checkpoint   ↓Writer fails   ↓retry writer with fallback

يوفّر CrewAI حالياً تخزين نقاط تحقق يحفظ حالة التنفيذ ويسمح باستئناف التشغيل بعد الفشل. السلوك الموثّق لنقاط التحقق يتخطّى المهام المكتملة ويتابع العمل اللاحق من الحالة المحفوظة.

هذا هو الهيكل الأفضل لسير العمل المكلِف أو الذي ينتج آثاراً جانبية.

---

كيف تضيف نقاط تحقق CrewAI؟

لعمليات الإنتاج، فعّل نقاط التحقق على الطاقم:

crew = Crew(    agents=[researcher, analyst, writer],    tasks=[        research_task,        analysis_task,        writing_task,    ],    process=Process.sequential,    checkpoint=True,    verbose=True,)

يمكن لنظام نقاط التحقق في CrewAI أن يخزن حالة التنفيذ بعد اكتمال المهمة ويستعيد الطاقم من نقطة تحقق.

على سبيل المثال، يمكن لتشغيل مستعاد استخدام:

from crewai import CheckpointConfigresult = crew.kickoff(    from_checkpoint=CheckpointConfig(        restore_from="./.checkpoints/checkpoint.json",    ))

يجب أن يتبع ضبط نقاط التحقق نسخة CrewAI المستخدمة في مشروعك.

النقطة المعمارية المهمة هي:

احفظ نقطة التحقق أولاً، ثم التراجع.

هذا يمنع فشل النموذج المؤقت من إجبار العمل المكتمل المكلف على التشغيل مرة أخرى.

---

كيف تطبّق تراجعاً بسيطاً ومحدوداً؟

لأغراض الدليل، لا يزال بإمكانك إظهار تراجع بسيط على مستوى الطاقم.

def run_with_fallback(topic: str):    routes = [        PRIMARY_MODELS,        {            **PRIMARY_MODELS,            "writer": FALLBACK_MODELS["writer"],        },        {            **PRIMARY_MODELS,            "analyst": FALLBACK_MODELS["analyst"],            "writer": FALLBACK_MODELS["writer"],        },    ]    last_error = None    for attempt, model_map in enumerate(routes, start=1):        try:            crew = build_crew(model_map)            result = crew.kickoff(                inputs={"topic": topic}            )            return result, model_map        except Exception as error:            last_error = error            if not should_fallback(error):                raise            if attempt == len(routes):                raise            print(                f"Transient failure on attempt {attempt}. "                f"Trying bounded fallback route.",                flush=True,            )    raise RuntimeError(        "Crew execution failed after all fallback routes."    ) from last_error

لاحظ الفارق المهم:

هذا لا يدّعي أن الاستثناء حدّد الوكيل الذي فشل بالضبط.

إنه استراتيجية تراجع محدودة على مستوى الطاقم.

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

---

كيف تتتبّع استخدام الرموز في CrewAI؟

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

في نهاية التشغيل، افحص نتيجة CrewAI:

result, selected_models = run_with_fallback(topic)print("Selected models:")print(selected_models)print("Final result:")print(result.raw)print("Usage:")print(result.token_usage)

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

يُفضّل أن يحتوي سجل الاستخدام الإنتاجي على:

job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at

هذا يتيح لك الإجابة على أسئلة مثل:

أي وكيل يستهلك معظم الميزانية؟

كم مرّة يتراجع المحلل؟

أي نموذج لديه أعلى كمون؟

كم يكلّف كل سير عمل؟

---

كيف تتحكّم بالتكلفة على مستوى الوكيل؟

يكون توجيه النماذج المتعددة أكثر فائدة عندما يعكس اختلافات عبء العمل الفعلية.

على سبيل المثال:

Researcher→ high volume→ lower-cost modelAnalyst→ low volume→ stronger reasoning modelWriter→ medium volume→ general-purpose production model

يمكنك أيضاً تقييد التكلفة عبر ضبط الوكيل.

على سبيل المثال:

max_iter=3

يحدّ حلقة تكرار الوكيل. ولا ينبغي تفسيره على أنه حد صارم لثلاثة استدعاءات API أو ثلاثة ميزانيات رموز.

عناصر تحكم إضافية تشمل:

• الحد من سياق المهمة
• تلخيص المخرجات الوسيطة
• تخزين البحث القابل لإعادة الاستخدام مؤقتاً (caching)
• تحديد الحد الأقصى لحجم الإدخال
• تحديد الحد الأقصى لرموز الإخراج حيثما كان مدعوماً
• تقييد استدعاءات الأدوات
• تحديد ميزانيات لكل مستخدم
• تحديد ميزانيات لكل سير عمل
• تتبّع تكرار التراجع

---

كيف تتحقق من النماذج قبل النشر؟

لا تقم بترميز معرّفات النماذج بشكل دائم.

قد يصبح النموذج:

• غير متاح
• مُعاد التسمية
• مُهمل (deprecated)
• مقيّداً
• متغيّراً في القدرة
• متغيّراً في التسعير
• غير متوافق مع معلمة يستخدمها تطبيقك

يوفّر CometAPI نقطة نهاية لكاتالوج النماذج يمكن الاستعلام عنها برمجياً، بينما يمكن استخدام دليل النماذج العام للاكتشاف البشري.

يمكن أن يبدو فحص النشر كالتالي:

curl -s \  https://api.cometapi.com/api/models \  -H "Authorization: Bearer $COMETAPI_KEY"

ثم تحقق من أن معرّفات النماذج المضبوطة لديك موجودة قبل النشر.

على سبيل المثال، يمكن لعملية CI لديك التحقق:

gemini-3.7-flash → availableclaude-opus-5    → availablegpt-5.6          → available

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

---

كيف يبدو مثال CrewAI الكامل؟

فيما يلي تنفيذ موحّد:

import jsonimport osimport sysfrom collections.abc import Iteratorfrom dotenv import load_dotenvfrom openai import (    APIConnectionError,    APIStatusError,    APITimeoutError,)from crewai import Agent, Crew, LLM, Process, Taskload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv(    "COMETAPI_BASE_URL",    "https://api.cometapi.com/v1",)PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",}FALLBACK_MODELS = {    "researcher": "gpt-5.6",    "analyst": "gpt-5.6",    "writer": "gemini-3.7-flash",}def cometapi_llm(model_id: str) -> LLM:    return LLM(        model=model_id,        base_url=COMETAPI_BASE_URL,        api_key=COMETAPI_KEY,        timeout=60.0,        max_retries=0,    )def build_crew(model_map: dict[str, str]) -> Crew:    researcher = Agent(        role="Market Researcher",        goal="Collect the facts needed to answer the topic",        backstory=(            "You create concise, source-aware research briefs "            "and distinguish facts from assumptions."        ),        llm=cometapi_llm(model_map["researcher"]),        max_iter=3,        allow_delegation=False,    )    analyst = Agent(        role="Product Analyst",        goal="Turn research into a defensible recommendation",        backstory=(            "You evaluate evidence, assumptions, risks, "            "and trade-offs."        ),        llm=cometapi_llm(model_map["analyst"]),        max_iter=3,        allow_delegation=False,    )    writer = Agent(        role="Technical Writer",        goal="Produce a concise technical decision memo",        backstory=(            "You write clear technical explanations "            "without unnecessary hype."        ),        llm=cometapi_llm(model_map["writer"]),        max_iter=3,        allow_delegation=False,    )    research_task = Task(        description=(            "Research this topic: {topic}. "            "Return the key facts, uncertainties, "            "and relevant sources."        ),        expected_output=(            "A concise research brief with facts "            "and open questions."        ),        agent=researcher,    )    analysis_task = Task(        description=(            "Using the research brief, analyze {topic}. "            "Identify the strongest conclusion and "            "explain the major trade-offs."        ),        expected_output=(            "A decision outline with evidence, "            "assumptions, risks, and trade-offs."        ),        agent=analyst,        context=[research_task],    )    writing_task = Task(        description=(            "Write a concise technical decision memo "            "about {topic}. State the recommendation early "            "and preserve important caveats."        ),        expected_output=(            "A polished technical decision memo in Markdown."        ),        agent=writer,        context=[            research_task,            analysis_task,        ],    )    return Crew(        agents=[            researcher,            analyst,            writer,        ],        tasks=[            research_task,            analysis_task,            writing_task,        ],        process=Process.sequential,        verbose=True,    )def exception_chain(    error: BaseException,) -> Iterator[BaseException]:    current = error    seen: set[int] = set()    while current is not None and id(current) not in seen:        seen.add(id(current))        yield current        current = (            current.__cause__            or current.__context__        )def should_fallback(error: BaseException) -> bool:    for current in exception_chain(error):        if isinstance(            current,            (                APIConnectionError,                APITimeoutError,            ),        ):            return True        if isinstance(current, APIStatusError):            return (                current.status_code in {408, 429}                or current.status_code >= 500            )    return Falsedef run_with_fallback(topic: str):    routes = [        PRIMARY_MODELS,        {            **PRIMARY_MODELS,            "writer": FALLBACK_MODELS["writer"],        },        {            **PRIMARY_MODELS,            "analyst": FALLBACK_MODELS["analyst"],            "writer": FALLBACK_MODELS["writer"],        },    ]    last_error = None    for attempt, model_map in enumerate(        routes,        start=1,    ):        try:            crew = build_crew(model_map)            result = crew.kickoff(                inputs={                    "topic": topic,                }            )            return result, model_map        except Exception as error:            last_error = error            if not should_fallback(error):                raise            if attempt == len(routes):                raise            print(                f"Transient failure on attempt "                f"{attempt}; trying fallback.",                file=sys.stderr,            )    raise RuntimeError(        "No model route completed the crew."    ) from last_errordef main():    topic = (        sys.argv[1]        if len(sys.argv) > 1        else (            "Should a small SaaS add "            "AI-generated meeting summaries?"        )    )    result, selected_models = (        run_with_fallback(topic)    )    output = {        "selected_models": selected_models,        "raw": result.raw,        "tasks_output": [            task.raw            for task in result.tasks_output        ],        "token_usage": str(            result.token_usage        ),    }    print(        json.dumps(            output,            indent=2,            default=str,        )    )if __name__ == "__main__":    main()

التحسين المهم مقارنة بالإصدار الأصلي هو أن الشفرة لم تعد توحي بشكل خاطئ بأن الاستثناء يحدّد الوكيل الذي فشل بالضبط.

إنها عملية تراجع محدودة على مستوى الطاقم بشكل صريح.

للإنتاج، اجمع سياسة التوجيه نفسها مع نقاط تحقق CrewAI.

---

كيف تشغّل سير عمل CrewAI؟

احفظ الملف باسم:

crewai_multi_model.py

ثم شغّل:

python crewai_multi_model.py \  "Should a small SaaS add AI-generated meeting summaries?"

سيحتوي الرد الناجح على معلومات شبيهة بـ:

{  "selected_models": {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6"  },  "raw": "<final decision memo>",  "tasks_output": [    "<research output>",    "<analysis output>",    "<writing output>"  ],  "token_usage": "<usage information>"}

تعتمد الاستجابة وقيَم الاستخدام الدقيقة على الإدخال وسلوك النموذج ونسخة CrewAI ومسار التنفيذ.

إذا فعّل خطأ قابل لإعادة المحاولة مسار التراجع، فسيعرض كائن selected_models المسار المستخدم لتنفيد ذلك الطاقم.

---

كيف تصمّم توجيه النماذج للإنتاج؟

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

دالة قرار مفيدة هي:

Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency

يمكنك تطبيقها على عدة مستويات.

توجيه قائم على التكلفة

Simple task → economical modelComplex task → premium model

توجيه قائم على الكمون

Interactive request → fast modelBackground workflow → higher-quality model

توجيه قائم على الموثوقية

Primary model     ↓transient failure     ↓fallback model

توجيه قائم على المهمة

Research → Model AAnalysis → Model BWriting → Model CCode → Model D

النهج الأخير طبيعي خصوصاً مع CrewAI لأن الإطار يمنح كل وكيل دوراً مميزاً بالفعل.

---

كيف تجعل التراجع آمناً؟

يجب أن يفرض نظام تراجع قوي أربع قواعد.

لا تتراجع عند أخطاء المصادقة

إذا كان مفتاح API غير صالح:

401

فلن يصلح تغيير النماذج المشكلة.

لا تتراجع عند الطلبات غير الصالحة

إذا كان الطلب غير صالح:

400422

فأصلح الطلب بدلاً من ذلك.

لا تتراجع بلا حدود

ضع حدّاً صارماً:

MAX_FALLBACK_ATTEMPTS = 2

نظام التراجع بدون حدّ يمكن أن يتحول إلى حلقة إعادة محاولة مكلفة.

اجعل نماذج التراجع متوافقة مع الطلب

يجب أن يدعم نموذج التراجع الميزات التي يتطلبها وكيلك.

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

التوافق مع OpenAI لا يعني التوافق في الميزات.

---

ما أكثر أخطاء CrewAI + CometAPI شيوعاً؟

---

كيف تفصل أخطاء CrewAI عن أخطاء النماذج؟

هذا التفريق مهم عند استكشاف الأخطاء.

أخطاء الضبط

Missing API keyInvalid model IDInvalid base URLUnsupported parameter

يجب أن تفشل بسرعة.

أخطاء المزوّد/الـAPI

401403404429500503

تتطلب معالجة مختلفة وفقاً للحالة.

أخطاء التطبيق

Agent output invalidTool returned malformed dataTask context missingSide effect failed

قد لا تُحل بمجرد تغيير النماذج.

لذا يجب أن يحتوي نظام الوكلاء الناضج على معالجة منفصلة لـ:

configuration      ↓API transport      ↓model execution      ↓agent logic      ↓tool execution      ↓application side effects

هذا أكثر أماناً بكثير من:

except Exception:    use_fallback()

---

كيف تحمي الآثار الجانبية الخارجية؟

يصبح التراجع أكثر تعقيداً بكثير عندما يقوم الوكلاء بأكثر من مجرد توليد نصوص.

على سبيل المثال، تخيّل وكيلاً يقوم بـ:

1. إنشاء سجل في قاعدة بيانات
2. إرسال رسالة بريد إلكتروني
3. استدعاء API خارجي
4. تحديث CRM

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

استخدم:

• مفاتيح عدم التكرار (idempotency keys)
• نقاط تحقق للمهام
• حدود المعاملات
• معرفات تنفيذ
• حالة مهام مستدامة
• تأكيد صريح للآثار الجانبية

على سبيل المثال:

job_id = crew_run_123task_id = writer_456

احفظ هذه المعرفات مع العمليات الخارجية بحيث يمكن لإعادة المحاولة تحديد ما إذا كانت العملية قد حدثت بالفعل.

---

كيف تراقب سير عمل CrewAI متعدد النماذج؟

على الأقل، سجّل:

workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens

لا تسجّل:

API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output

لكل نموذج، راقب:

الموثوقية

success ratetimeout rate5xx ratefallback rate

الأداء

p50 latencyp95 latencyp99 latency

التكلفة

input tokensoutput tokenscost per taskcost per completed workflow

الجودة

task success ratehuman evaluationstructured-output validitytool-call success

هذا يحوّل توجيه النماذج من تفضيل مُرمّز إلى نظام هندسي قابل للملاحظة.

---

كيف تختار بين واجهات مزوّدين مباشرة وCometAPI؟

يعتمد الاختيار على هندستك.

إذا كان تطبيقك يحتاج مزوّداً واحداً فقط وقدراته الأصلية، فقد يكون التكامل المباشر مناسباً تماماً.

إذا كان تطبيق CrewAI لديك يحتاج نماذج من عدة مزوّدين وتريد طبقة وصول واحدة، يصبح CometAPI أكثر جاذبية.

النقطة المهمة أن CometAPI لا يحل محل CrewAI.

بدلاً من ذلك:

CrewAIAgent orchestration       ↓CometAPIModel access       ↓Multiple models

لكل طبقة مسؤولية مختلفة.

---

كيف تتوسع هذه الهندسة؟

بمجرد فصل سياسة التوجيه عن تعريفات الوكلاء، فإن إضافة نموذج آخر لا تتطلب إعادة بناء التطبيق بأكمله.

على سبيل المثال:

PRIMARY_MODELS = {    "researcher": "gemini-3.7-flash",    "analyst": "claude-opus-5",    "writer": "gpt-5.6",    "coder": "YOUR_CODE_MODEL",}

يمكن للهندسة نفسها أن تدعم:

وكيل بحثوكيل تحليلوكيل ترميزوكيل مراجعةوكيل كتابةوكيل تدقيق الحقائق

يمكن لكل وكيل أن يمتلك نموذجاً مختلفاً مع مشاركة طبقة وصول CometAPI نفسها.

الخطوة التالية هي جعل التوجيه ديناميكياً.

بدلاً من:

"analyst": "claude-opus-5"

يمكنك لاحقاً استخدام:

select_model(    task="analysis",    budget=budget,    latency_target=latency_target,)

عندها يمكن لنظام التوجيه أن يختار من النماذج المعتمدة بناءً على متطلبات التطبيق.

---

ما أفضل هندسة إنتاج لـ CrewAI + CometAPI؟

لسير عمل صغير:

User Input   ↓CrewAI   ↓CometAPI   ↓Models

للإنتاج:

                     ┌───────────────┐                     │ Model Catalog │                     └───────┬───────┘                             │                             ▼User → CrewAI → Routing Policy → CometAPI           │          │             │           │          │             ├── Gemini           │          │             ├── Claude           │          │             └── GPT           │          │           │          ▼           │     Cost / Quality /           │     Latency / Policy           │           ▼      Checkpoints           │           ▼      Usage Tracking

مكوّنات الإنتاج الأساسية هي:

1. قائمة سماح للنماذج (allowlist)
2. توجيه لكل وكيل
3. محاولات محدودة
4. نقاط تحقق للمهام
5. تتبّع الاستخدام
6. ضوابط التكلفة
7. اختبار توافق النماذج
8. قابلية الرصد
9. آثار جانبية عديمة التكرار (idempotent)

هذه الهندسة أكثر متانة بكثير من مجرد إضافة try/except حول crew.kickoff().

---

مفتاح CometAPI واحد، نماذج مختلفة، وأدوار وكلاء أوضح

أفضل طريقة للتفكير في CrewAI وCometAPI معاً هي أنهما طبقتان تكميليتان.

CrewAI يعرّف ما يفعله الوكلاء.

CometAPI يعرّف كيف يصل هؤلاء الوكلاء إلى النماذج.

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

أبسط تنفيذ يستخدم مفتاح CometAPI واحداً وعنوان base URL متوافقاً مع OpenAI واحداً:

https://api.cometapi.com/v1

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

هذا يمنحك نمطاً أكثر متانة بكثير من مجرد ربط CrewAI بنموذج LLM واحد:

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

---

أسئلة شائعة

هل يمكن لـ CrewAI استخدام عدة نماذج ذكاء اصطناعي في الطاقم نفسه؟

نعم. عيّن إعداد LLM مختلفاً لكل وكيل في CrewAI. يمكن لكل إعداد تحديد نموذجه الخاص بينما يستخدم مفتاح CometAPI وعنوان base URL نفسيهما.

هل يمكن لـ CrewAI الاتصال بواجهة API متوافقة مع OpenAI؟

نعم. يدعم ضبط LLM في CrewAI عنوان base_url مخصصاً ومفتاح API لنقاط النهاية المتوافقة مع OpenAI.

مع CometAPI، عنوان base URL هو:

https://api.cometapi.com/v1

هل أحتاج مفاتيح API منفصلة لـ GPT وClaude وGemini؟

عند الوصول إلى هذه النماذج عبر CometAPI، يمكن للتطبيق استخدام بيانات اعتماد CometAPI ونقطة النهاية بدلاً من تنفيذ بيانات اعتماد مزوّدين منفصلة في كل وكيل CrewAI.

هل يعني مفتاح API واحد أن النماذج تمتلك قدرات متطابقة؟

لا. يمكن توحيد واجهة API بينما تختلف قدرات النماذج. قد تختلف نوافذ السياق، ودعم الأدوات، والمعلمات، وسلوك الإخراج، والكمون، والتسعير حسب النموذج.

هل ينبغي إعادة محاولة سير عمل CrewAI بأكمله عند فشل نموذج واحد؟

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

وظيفة نقاط التحقق الحالية في CrewAI مصمّمة للحفاظ على حالة التنفيذ واستئناف التشغيل بعد الفشل.

هل كل استثناء في CrewAI يجب أن يفعّل تراجع النموذج؟

لا. المصادقة، والطلبات غير المهيكلة، ومعرّفات النماذج غير الصالحة، والمعلمات غير المدعومة تتطلب غالباً تغييرات ضبط بدلاً من نموذج مختلف.

التراجع أفضل عند حصره في الإخفاقات العابرة مثل انتهاء المهلة، وحدود المعدّل، واستجابات 5xx المؤقتة.

كيف أتتبّع تكلفة كل وكيل CrewAI؟

سجّل اسم الوكيل، معرّف النموذج، استخدام الرموز، الكمون، حالة التنفيذ، ومعلومات التراجع لكل مهمة. استخدم البيانات الناتجة لحساب تكلفة كل وكيل وكل سير عمل.

هل يمكنني تغيير النموذج المُسند إلى وكيل بشكل ديناميكي؟

نعم. احتفظ بمعرّفات النماذج في ضبط التوجيه بدلاً من تضمينها مباشرة في تعريفات الوكلاء. يمكن لتطبيقك حينئذٍ اختيار النماذج بناءً على التكلفة أو الكمون أو نوع المهمة أو التوافر.

هل CometAPI بديل عن CrewAI؟

لا. يعملاّن في طبقات مختلفة. CrewAI ينظّم الوكلاء والمهام، بينما يوفّر CometAPI طبقة وصول موحّدة للنماذج.

أين أجد نماذج CometAPI الحالية؟

استخدم CometAPI Model Directory لاكتشاف النماذج بشرياً وواجهة نماذج API للتحقق البرمجي. صفحة البدء السريع الحالية لـ CometAPI تسرد 500+ نموذج عبر فئات النص والصورة والفيديو والصوت.

---

المصادر

• CometAPI Quick Start
• CometAPI Model Directory
• CometAPI API Documentation
• CometAPI Python SDK
• CrewAI Documentation
• CrewAI LLM Configuration
• CrewAI Checkpointing