الأطقم

فهم واستخدام الأطقم في إطار عمل CrewAI مع خصائص ووظائف شاملة.

​

نظرة عامة

يمثل الطاقم في CrewAI مجموعة تعاونية من الوكلاء يعملون معًا لتحقيق مجموعة من المهام. يحدد كل طاقم استراتيجية تنفيذ المهام وتعاون الوكلاء وسير العمل العام.

​

خصائص الطاقم

الخاصية	المعامل	الوصف
المهام	tasks	قائمة المهام المعيّنة للطاقم.
الوكلاء	agents	قائمة الوكلاء الذين يشكلون جزءًا من الطاقم.
العملية (اختياري)	process	تدفق العملية (مثل تسلسلي، هرمي) الذي يتبعه الطاقم. الافتراضي sequential.
الوضع المفصل (اختياري)	verbose	مستوى التفصيل في التسجيل أثناء التنفيذ. الافتراضي False.
LLM المدير (اختياري)	manager_llm	نموذج اللغة المستخدم بواسطة وكيل المدير في العملية الهرمية. مطلوب عند استخدام العملية الهرمية.
LLM استدعاء الدوال (اختياري)	function_calling_llm	إذا مُرر، سيستخدم الطاقم هذا LLM لاستدعاء دوال الأدوات لجميع الوكلاء. يمكن لكل وكيل أن يكون له LLM خاص يتجاوز LLM الطاقم.
التهيئة (اختياري)	config	إعدادات تهيئة اختيارية للطاقم، بتنسيق Json أو Dict[str, Any].
الحد الأقصى لـ RPM (اختياري)	max_rpm	الحد الأقصى للطلبات في الدقيقة. الافتراضي None.
الذاكرة (اختياري)	memory	تُستخدم لتخزين ذاكرات التنفيذ (قصيرة المدى، طويلة المدى، ذاكرة الكيانات).
التخزين المؤقت (اختياري)	cache	يحدد ما إذا كان يُستخدم تخزين مؤقت لنتائج تنفيذ الأدوات. الافتراضي True.
المُضمّن (اختياري)	embedder	تهيئة المُضمّن المستخدم من قبل الطاقم. الافتراضي {"provider": "openai"}.
دالة الخطوة (اختياري)	step_callback	دالة تُستدعى بعد كل خطوة لكل وكيل.
دالة المهمة (اختياري)	task_callback	دالة تُستدعى بعد اكتمال كل مهمة.
مشاركة الطاقم (اختياري)	share_crew	ما إذا كنت تريد مشاركة معلومات الطاقم الكاملة وتنفيذه مع فريق CrewAI.
ملف سجل المخرجات (اختياري)	output_log_file	عيّن True لحفظ السجلات كـ logs.txt أو وفّر مسار ملف. الافتراضي None.
وكيل المدير (اختياري)	manager_agent	يعيّن وكيلًا مخصصًا سيُستخدم كمدير.
التخطيط (اختياري)	planning	يضيف قدرة التخطيط للطاقم.
LLM التخطيط (اختياري)	planning_llm	نموذج اللغة المستخدم بواسطة AgentPlanner في عملية التخطيط.
مصادر المعرفة (اختياري)	knowledge_sources	مصادر المعرفة المتاحة على مستوى الطاقم، يمكن لجميع الوكلاء الوصول إليها.
البث (اختياري)	stream	تفعيل مخرجات البث لتلقي تحديثات في الوقت الفعلي. الافتراضي False.

الحد الأقصى لـ RPM للطاقم: تعيّن خاصية max_rpm الحد الأقصى للطلبات في الدقيقة التي يمكن للطاقم تنفيذها لتجنب حدود المعدل وستتجاوز إعدادات max_rpm الفردية للوكلاء إذا عيّنتها.

​

إنشاء الأطقم

هناك طريقتان لإنشاء الأطقم في CrewAI: باستخدام تهيئة YAML (موصى بها) أو تعريفها مباشرة في الكود.

​

تهيئة YAML (موصى بها)

توفر تهيئة YAML طريقة أنظف وأكثر قابلية للصيانة لتعريف الأطقم وتتسق مع كيفية تعريف الوكلاء والمهام في مشاريع CrewAI.

code

from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff
from crewai.agents.agent_builder.base_agent import BaseAgent
from typing import List

@CrewBase
class YourCrewName:
    """Description of your crew"""

    agents: List[BaseAgent]
    tasks: List[Task]

    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @before_kickoff
    def prepare_inputs(self, inputs):
        inputs['additional_data'] = "Some extra information"
        return inputs

    @after_kickoff
    def process_output(self, output):
        output.raw += "\nProcessed after kickoff."
        return output

    @agent
    def agent_one(self) -> Agent:
        return Agent(
            config=self.agents_config['agent_one'], # type: ignore[index]
            verbose=True
        )

    @task
    def task_one(self) -> Task:
        return Task(
            config=self.tasks_config['task_one'] # type: ignore[index]
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

سيتم تنفيذ المهام بالترتيب الذي عُرّفت به.

فئة CrewBase، مع هذه المزيّنات، تؤتمت جمع الوكلاء والمهام، مما يقلل الحاجة للإدارة اليدوية.

​

تعريف مباشر في الكود (بديل)

بدلاً من ذلك، يمكنك تعريف الطاقم مباشرة في الكود بدون ملفات تهيئة YAML.

​

مخرجات الطاقم

تُغلّف مخرجات الطاقم في فئة CrewOutput. توفر هذه الفئة طريقة منظمة للوصول إلى نتائج تنفيذ الطاقم، بما في ذلك تنسيقات متنوعة مثل السلاسل النصية الخام وJSON ونماذج Pydantic.

​

خصائص مخرجات الطاقم

الخاصية	المعامل	النوع	الوصف
Raw	raw	str	المخرجات الخام للطاقم. هذا هو التنسيق الافتراضي.
Pydantic	pydantic	Optional[BaseModel]	كائن نموذج Pydantic يمثل المخرجات المنظمة.
JSON Dict	json_dict	Optional[Dict[str, Any]]	قاموس يمثل مخرجات JSON.
Tasks Output	tasks_output	List[TaskOutput]	قائمة كائنات TaskOutput، كل منها يمثل مخرجات مهمة.
Token Usage	token_usage	Dict[str, Any]	ملخص استخدام الرموز.

​

استخدام الذاكرة

يمكن للأطقم استخدام الذاكرة (قصيرة المدى، طويلة المدى، وذاكرة الكيانات) لتحسين تنفيذها وتعلمها بمرور الوقت.

​

استخدام التخزين المؤقت

يمكن استخدام التخزين المؤقت لتخزين نتائج تنفيذ الأدوات، مما يجعل العملية أكثر كفاءة.

​

مقاييس استخدام الطاقم

بعد تنفيذ الطاقم، يمكنك الوصول إلى خاصية usage_metrics لعرض مقاييس استخدام نموذج اللغة (LLM) لجميع المهام المنفذة.

Code

crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

​

عملية تنفيذ الطاقم

• العملية التسلسلية: تُنفذ المهام واحدة تلو الأخرى، مما يسمح بتدفق عمل خطي.
• العملية الهرمية: ينسق وكيل مدير الطاقم، ويفوّض المهام ويتحقق من النتائج.

​

تشغيل الطاقم

بمجرد تجميع طاقمك، ابدأ سير العمل بطريقة kickoff().

Code

result = my_crew.kickoff()
print(result)

​

طرق مختلفة لتشغيل الطاقم

​

الطرق المتزامنة

• kickoff(): يبدأ عملية التنفيذ وفقًا لتدفق العملية المحدد.
• kickoff_for_each(): ينفذ المهام بالتتابع لكل مدخل.

​

الطرق غير المتزامنة

الطريقة	النوع	الوصف
akickoff()	غير متزامن أصلي	async/await أصلي عبر سلسلة التنفيذ بأكملها
akickoff_for_each()	غير متزامن أصلي	تنفيذ غير متزامن أصلي لكل مدخل في قائمة
kickoff_async()	مبني على الخيوط	يغلّف التنفيذ المتزامن في asyncio.to_thread
kickoff_for_each_async()	مبني على الخيوط	غير متزامن مبني على الخيوط لكل مدخل في قائمة

لأحمال العمل عالية التزامن، يُوصى بـ akickoff() و akickoff_for_each() لأنها تستخدم async أصلي.

​

بث تنفيذ الطاقم

للرؤية في الوقت الفعلي لتنفيذ الطاقم، يمكنك تفعيل البث:

Code

crew = Crew(
    agents=[researcher],
    tasks=[task],
    stream=True
)

streaming = crew.kickoff(inputs={"topic": "AI"})
for chunk in streaming:
    print(chunk.content, end="", flush=True)

result = streaming.result

​

الإعادة من مهمة محددة

يمكنك الآن الإعادة من مهمة محددة باستخدام أمر CLI replay.

crewai log-tasks-outputs

ثم للإعادة من مهمة محددة:

crewai replay -t <task_id>