بناء فريق وكلاء AI (Multi-Agent System) بكلود — الدليل العربي الأول

pip install crewai crewai-tools anthropic

import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_anthropic import ChatAnthropic

# إعداد نموذج كلود
claude_opus = ChatAnthropic(
    model="claude-opus-4-5",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"]
)

claude_haiku = ChatAnthropic(
    model="claude-haiku-4-5",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"]
)

# أداة البحث في الإنترنت
search_tool = SerperDevTool()

# الوكيل الباحث
researcher = Agent(
    role="باحث محتوى متخصص",
    goal="جمع معلومات دقيقة وموثوقة عن الموضوع المطلوب من مصادر متعددة",
    backstory="""أنت باحث خبير في جمع المعلومات وتحليلها. تتقن البحث العميق
    في الإنترنت وتمييز المصادر الموثوقة من غيرها. تنتج ملخصات شاملة ومنظمة.""",
    tools=[search_tool],
    llm=claude_opus,
    verbose=True,
    max_iter=5  # الحد الأقصى لمحاولات إكمال المهمة
)

# الوكيل الكاتب
writer = Agent(
    role="كاتب محتوى احترافي",
    goal="كتابة محتوى جذاب ومقنع يخدم القارئ ويحقق أهداف المنشأة",
    backstory="""أنت كاتب محترف يمتلك أسلوباً سلساً وقادراً على تبسيط المعلومات
    المعقدة. تُراعي القارئ العربي وتكتب بلغة واضحة ومؤثرة.""",
    llm=claude_opus,
    verbose=True
)

# وكيل SEO
seo_optimizer = Agent(
    role="خبير SEO وتحسين محركات البحث",
    goal="تحسين المحتوى لمحركات البحث مع الحفاظ على جودة القراءة",
    backstory="""أنت خبير SEO متخصص في المحتوى العربي. تُتقن تحسين العناوين
    والمقدمات والكلمات المفتاحية وبنية المحتوى لأعلى ترتيب في جوجل.""",
    llm=claude_haiku,  # Haiku يكفي لمهمة التحسين
    verbose=True
)

# وكيل التحرير والجودة
editor = Agent(
    role="محرر أول ومراقب جودة",
    goal="التحقق من دقة المعلومات وجودة الأسلوب واتساق المحتوى",
    backstory="""أنت محرر خبير بعين نقدية حادة. تُراجع المحتوى للتأكد من دقته
    وسلاسته وخلوه من الأخطاء اللغوية والمعلوماتية.""",
    llm=claude_opus,
    verbose=True
)

# مهمة البحث
research_task = Task(
    description="""ابحث عن موضوع: {topic}
    اجمع:
    1. أهم 5 حقائق وإحصائيات حديثة
    2. 3 مصادر موثوقة
    3. أبرز التطورات الأخيرة خلال 6 أشهر الماضية
    4. المخاوف والتحديات المرتبطة بالموضوع
    """,
    expected_output="ملخص منظم بنقاط واضحة مع روابط المصادر",
    agent=researcher
)

# مهمة الكتابة
writing_task = Task(
    description="""استناداً على نتائج البحث، اكتب مقالاً شاملاً عن {topic}:
    - عنوان جذاب
    - مقدمة تُشدّ الانتباه (150-200 كلمة)
    - 4-5 أقسام رئيسية مع عناوين فرعية
    - خاتمة تلخّص النقاط وتحث على الفعل
    - الطول الإجمالي: 800-1200 كلمة
    """,
    expected_output="مقال كامل بتنسيق Markdown",
    agent=writer,
    context=[research_task]  # يستخدم نتائج البحث كسياق
)

# مهمة تحسين SEO
seo_task = Task(
    description="""حسّن المقال لمحركات البحث:
    1. تحسين العنوان الرئيسي ليحتوي الكلمة المفتاحية الأساسية
    2. إضافة كلمات مفتاحية ثانوية بشكل طبيعي
    3. تحسين Meta Description (150-160 حرف)
    4. تحسين العناوين الفرعية H2 وH3
    5. إضافة روابط داخلية وخارجية مقترحة
    """,
    expected_output="المقال المحسّن مع تقرير SEO مختصر",
    agent=seo_optimizer,
    context=[writing_task]
)

# مهمة التحرير النهائي
editing_task = Task(
    description="""راجع المقال النهائي وتأكد من:
    - صحة المعلومات وتطابقها مع البحث الأصلي
    - سلامة اللغة العربية والأسلوب
    - الاتساق في التنسيق والعرض
    - جاهزية المقال للنشر الفوري
    أضف [تمت المراجعة] أو [يحتاج تعديلات] في البداية مع ملاحظاتك
    """,
    expected_output="المقال النهائي جاهز للنشر مع تقرير المراجعة",
    agent=editor,
    context=[research_task, writing_task, seo_task]
)

# تجميع الفريق
content_crew = Crew(
    agents=[researcher, writer, seo_optimizer, editor],
    tasks=[research_task, writing_task, seo_task, editing_task],
    process=Process.sequential,  # تنفيذ متسلسل بالترتيب
    verbose=2,
    memory=True  # تفعيل الذاكرة المشتركة
)

# تشغيل خط الإنتاج
result = content_crew.kickoff(
    inputs={"topic": "تأثير الذكاء الاصطناعي على قطاع العقارات في السعودية"}
)

print("="*50)
print("المقال النهائي:")
print(result)
print("="*50)

# حفظ النتيجة
with open("article_output.md", "w", encoding="utf-8") as f:
    f.write(result)

from crewai import Process

# تنفيذ متوازٍ — الباحث والمحلل التنافسي يعملان معاً
parallel_crew = Crew(
    agents=[researcher, competitive_analyst, writer, editor],
    tasks=[research_task, competitive_task, writing_task, editing_task],
    process=Process.hierarchical,  # مدير يُوزّع المهام
    manager_llm=claude_opus,
    verbose=True
)

# إنتاج 10 مقالات بالتوازي
import asyncio
from concurrent.futures import ThreadPoolExecutor

topics = [
    "الذكاء الاصطناعي في البنوك",
    "الأتمتة في التجزئة",
    "AI في الرعاية الصحية",
    # ... باقي المواضيع
]

def produce_article(topic):
    crew_instance = content_crew.kickoff(inputs={"topic": topic})
    return crew_instance

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(produce_article, topics))

from crewai import Task

# مهمة تحتاج موافقة بشرية قبل النشر
publishing_task = Task(
    description="نشر المقال على منصة WordPress عبر API",
    expected_output="رابط المقال المنشور",
    agent=publisher_agent,
    human_input=True  # توقف وانتظار موافقة بشرية
)