دليل CrewAI 2026: بناء أنظمة وكلاء ذكاء اصطناعي متعددين بلغة بايثون

mkdir market-crew && cd market-crew
uv venv --python 3.11
source .venv/bin/activate
uv pip install "crewai[tools]" python-dotenv fastapi uvicorn

python -m venv .venv
source .venv/bin/activate
pip install "crewai[tools]" python-dotenv fastapi uvicorn

crewai version
# CrewAI version: 0.86.x

cd .. && crewai create crew market-crew

OPENAI_API_KEY=sk-...
SERPER_API_KEY=...      # من serper.dev، اختياري
MODEL=gpt-4o-mini

researcher:
  role: >
    باحث سوقي أول متخصص في {industry}
  goal: >
    جمع أحدث الحقائق والإحصاءات والاقتباسات الموثّقة حول {topic}
    من مصادر ويب عامة نُشرت خلال آخر اثني عشر شهراً.
  backstory: >
    قضيت عقداً كاملاً في McKinsey تدير البحوث الأولية والثانوية.
    تتحقّق من كل ادعاء عبر مصدرين مستقلّين على الأقل، وتُرفق كل
    معلومة بتاريخ نشرها ورابطها.
  verbose: true
  allow_delegation: false
 
analyst:
  role: >
    محلل استراتيجي في قطاع {industry}
  goal: >
    تحويل ملاحظات الباحث الخام إلى SWOT ومصفوفة تنافسية وثلاث
    توصيات قابلة للتنفيذ موجّهة إلى {target_audience}.
  backstory: >
    تكتب نوع المذكرات ذات الصفحتين التي يقرؤها الرؤساء التنفيذيون،
    وترفض إدراج أي ادعاء غير مدعوم بمصدر موثّق من ملاحظات الباحث.
  verbose: true
 
writer:
  role: >
    كاتب تحريري أول
  goal: >
    إنتاج موجز تنفيذي من 1200 كلمة حول {topic} يستطيع مستثمر
    قراءته في أقل من ست دقائق وإعادة توجيهه إلى شركائه.
  backstory: >
    سبق أن كتبت في Stratechery أسبوعياً. تفضّل الجمل القصيرة
    والأمثلة الملموسة وأطروحة واحدة قوية لكل مقال.
  verbose: true
 
editor:
  role: >
    رئيس التحرير
  goal: >
    التحقّق من الموجز مقابل قائمة مصادر الباحث، وإصلاح الصياغة
    الضعيفة، والتأكّد من أن كل ادعاء يستشهد بمصدره.
  backstory: >
    أدرت قسم المعايير في The Economist لمدة ثماني سنوات. لا تسمح
    لأي رقم غير موثّق بالمرور.
  verbose: true

research_task:
  description: >
    اجمع ما لا يقل عن 12 معلومة عالية الجودة حول {topic} في قطاع
    {industry}. لكل معلومة سجّل: الادعاء، رابط المصدر، تاريخ
    النشر، وملخّص في جملة واحدة.
  expected_output: >
    قائمة markdown تحتوي على 12 نقطة على الأقل. كل نقطة يجب أن
    تتضمّن رابط المصدر وتاريخه. لا إعادة صياغة بدون استشهاد.
  agent: researcher
 
analysis_task:
  description: >
    باستخدام نتائج الباحث، أنتج تحليل SWOT لـ {target_audience}
    يدخلون سوق {topic}، ومصفوفة تنافسية من 5 صفوف، وبالضبط ثلاث
    توصيات مرتّبة حسب الأثر.
  expected_output: >
    مستند markdown بثلاثة أقسام: SWOT، المصفوفة التنافسية،
    التوصيات. كل ادعاء رقمي يجب أن يستشهد بمعلومة من الباحث.
  agent: analyst
  context:
    - research_task
 
writing_task:
  description: >
    اكتب موجزاً تنفيذياً من 1200 كلمة حول {topic} موجّهاً إلى
    {target_audience}. ابدأ بأطروحة من جملة واحدة. استخدم SWOT
    وتوصيات المحلل. اختم بقسم "ما يجب مراقبته" يحتوي على ثلاثة
    مؤشرات استشرافية.
  expected_output: >
    موجز markdown مصقول بين 1100 و 1300 كلمة، بعناوين H2
    واستشهادات مدمجة كروابط markdown.
  agent: writer
  context:
    - research_task
    - analysis_task
 
editing_task:
  description: >
    حرّر الموجز للوضوح والدقّة. كل ادعاء رقمي يجب أن يستشهد بمصدر
    من قائمة الباحث. أعلم عن أي ادعاء غير مدعوم بإستبداله بتعليق
    TODO.
  expected_output: >
    الموجز النهائي بصيغة markdown محفوظاً في `output/brief.md`.
  agent: editor
  context:
    - research_task
    - analysis_task
    - writing_task
  output_file: output/brief.md

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool, FileWriterTool
 
@CrewBase
class MarketCrew:
    """طاقم بحث سوقي من أربعة متخصصين."""
 
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"
 
    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"],
            tools=[SerperDevTool()],
        )
 
    @agent
    def analyst(self) -> Agent:
        return Agent(config=self.agents_config["analyst"])
 
    @agent
    def writer(self) -> Agent:
        return Agent(config=self.agents_config["writer"])
 
    @agent
    def editor(self) -> Agent:
        return Agent(
            config=self.agents_config["editor"],
            tools=[FileWriterTool()],
        )
 
    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config["research_task"])
 
    @task
    def analysis_task(self) -> Task:
        return Task(config=self.tasks_config["analysis_task"])
 
    @task
    def writing_task(self) -> Task:
        return Task(config=self.tasks_config["writing_task"])
 
    @task
    def editing_task(self) -> Task:
        return Task(config=self.tasks_config["editing_task"])
 
    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

from market_crew.crew import MarketCrew
 
def run():
    inputs = {
        "topic": "وكلاء الذكاء الاصطناعي للبرمجة في منصات المطوّرين المؤسسية",
        "industry": "أدوات المطوّرين",
        "target_audience": "قادة الهندسة في شركات Series B",
    }
    MarketCrew().crew().kickoff(inputs=inputs)
 
if __name__ == "__main__":
    run()

crewai run

from crewai import Agent
 
@crew
def crew(self) -> Crew:
    manager = Agent(
        role="مدير التحرير",
        goal="تحديد ترتيب المهام وتفويضها إلى المتخصص المناسب",
        backstory="مدير مكتب سابق في Bloomberg يُدير قاعات تحرير منضبطة.",
        allow_delegation=True,
        verbose=True,
    )
    return Crew(
        agents=self.agents,
        tasks=self.tasks,
        process=Process.hierarchical,
        manager_agent=manager,
        verbose=True,
    )

from crewai import Crew, Process
 
@crew
def crew(self) -> Crew:
    return Crew(
        agents=self.agents,
        tasks=self.tasks,
        process=Process.sequential,
        memory=True,
        embedder={
            "provider": "openai",
            "config": {"model": "text-embedding-3-small"},
        },
        verbose=True,
    )

from crewai.tasks.task_output import TaskOutput
 
def on_task_complete(output: TaskOutput) -> None:
    print(f"[task:{output.task.description[:60]}] -> {len(output.raw)} chars")
 
def on_step(step: dict) -> None:
    if step.get("tool"):
        print(f"  tool call: {step['tool']} args={step['tool_input']}")
 
@task
def research_task(self) -> Task:
    return Task(
        config=self.tasks_config["research_task"],
        callback=on_task_complete,
    )

from fastapi import FastAPI
from pydantic import BaseModel
from market_crew.crew import MarketCrew
 
app = FastAPI()
 
class BriefRequest(BaseModel):
    topic: str
    industry: str
    target_audience: str
 
@app.post("/brief")
def generate_brief(req: BriefRequest):
    result = MarketCrew().crew().kickoff(inputs=req.model_dump())
    return {"output": result.raw, "token_usage": result.token_usage}

uv run uvicorn market_crew.api:app --reload --port 8000

curl -X POST http://localhost:8000/brief \
  -H "content-type: application/json" \
  -d '{"topic":"AI coding agents","industry":"developer tools","target_audience":"CTOs"}'

import logging, json, time
 
class CrewLogFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts": int(time.time() * 1000),
            "level": record.levelname,
            "msg": record.getMessage(),
        }
        return json.dumps(payload)
 
handler = logging.StreamHandler()
handler.setFormatter(CrewLogFormatter())
logging.getLogger("crewai").addHandler(handler)
logging.getLogger("crewai").setLevel(logging.INFO)

from crewai.llm import LLM
from market_crew.crew import MarketCrew
 
class FakeLLM(LLM):
    def __init__(self):
        super().__init__(model="fake")
        self.replies = iter([
            "- finding 1 (https://example.com 2026-01-01)",
            "## SWOT\n- strength: x",
            "Brief draft...",
            "Edited brief.",
        ])
 
    def call(self, messages, **kwargs):
        return next(self.replies)
 
def test_crew_smoke(monkeypatch):
    monkeypatch.setattr("crewai.agent.LLM", FakeLLM)
    out = MarketCrew().crew().kickoff(inputs={
        "topic": "x", "industry": "y", "target_audience": "z",
    })
    assert "Edited brief." in out.raw