بناء نظام متعدد الوكلاء باستخدام Novita و CrewAI

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

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

لكن ظهرت مشكلة أخرى. إدارة وكلاء متعددين ليست سهلة مثل إدارة وكيل واحد. لحل هذه المشكلة، تم تقديم عدة أطر عمل لأنظمة متعددة الوكلاء، وأحد أكثرها وعدًا هو CrewAI.

يساعد CrewAI المطورين على بناء أنظمة متعددة الوكلاء. باستخدام CrewAI، يمكنك إدارة وكلائك ومهامهم وسير عملهم. في هذا المقال، سنستخدم CrewAI جنبًا إلى جنب مع مجموعة نماذج اللغات الكبيرة (LLMs) المتاحة على Novita لبناء أنظمة متعددة الوكلاء عملية وذكية.

ما هو النظام متعدد الوكلاء ولماذا نحتاج إليه؟

منذ إدخال نماذج اللغات الكبيرة (LLMs)، تم استكشاف عدة نهجات لجعلها أكثر استقلالية. كان أحد أكثر النهجات شيوعًا هو نمط ReAct الذي تم تقديمه في الورقة البحثية مزامنة الاستدلال والفعل في نماذج اللغات. قامت أطر عمل مثل LangChain بتطبيق ReAct بسرعة لمنح نماذج اللغات الكبيرة قدرات وكيلية.

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

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

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

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

لبناء نظام متعدد الوكلاء، تحتاج إلى مكونين أساسيين: موفر نماذج يقدم نماذج اللغات الكبيرة (LLMs)، وإطار عمل متعدد الوكلاء لإدارة سير العمل الوكيل بالكامل. في هذا المقال، سنستخدم Novita كموفر نماذج لنا، و CrewAI كإطار عمل من اختيارنا.

لماذا تحتاج إلى إطار عمل مثل CrewAI

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

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

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

فهم المفاهيم الأساسية لـ CrewAI

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

• الوكلاء
• المهام
• الطواقم
• العمليات
• التدفقات

الوكلاء

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

المهام

تحدد المهام العمل الذي needs to be completed. في CrewAI، يمكنك إنشاء مهام وتعيينها مباشرة لوكلاء محددين أو السماح للوكلاء بالتقاطها إذا كانت تتوافق مع متطلبات المهمة. تتضمن كل مهمة عادةً وصفًا واضحًا، ومخرجًا متوقعًا، ومعلمات أخرى ذات صلة مثل وكيل اختياري يمكنه التعامل مع المهمة.

الطاقم

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

العمليات

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

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

التدفقات

بينما العمليات هي سير عمل محددة مسبقًا، تمنح تدفقات CrewAI المطورين المرونة لتصميم سير عمل وكلاء مخصصة. يمكن أن تتضمن التدفقة وكلاء ومهام وحتى طواقم كاملة، مما يسمح بتفاعلات أكثر تعقيدًا.

بناء الطواقم باستخدام Novita

الآن بعد أن غطينا المفاهيم الأساسية لـ CrewAI، دعنا ننتقل إلى بناء طاقم. في هذا المثال، سنقوم بإنشاء طاقم مصمم لمساعدة المستخدمين على بناء منتج قابل للتطبيق الحد الأدنى (MVP) بسرعة. يتكون هذا الطاقم من ثلاثة وكلاء متخصصين:

• المهندس المعماري: يحدد بنية البرنامج ويوضح نطاق MVP. يعمل هذا الوكيل بنموذج moonshotai/kimi-k2-instruct المستضاف على Novita.
• المبرمج: ينفذ خطة المهندس المعماري ويكتب الكود الفعلي. يستخدم أداة FileWriter لإنشاء جميع ملفات المشروع ويعمل على نموذج Novita qwen/qwen3-coder-480b-a35b-instruct.
• المراجع: يحلل الكود الذي يولده المبرمج ويقترح تحسينات باستخدام ملاحظات بنمط الفروقات. مثل المهندس المعماري، فإنه يستخدم أيضًا نموذج moonshotai/kimi-k2-instruct.

سيتبع طاقمنا عملية تسلسلية.

التثبيت والإعداد

الآن بعد أن وضعنا الخطة، لنبدأ بإعداد بيئتنا. سنبدأ بتثبيت CrewAI جنبًا إلى جنب مع دعم الأدوات المدمج الخاص به:

pip install 'crewai[tools]'

بعد ذلك، ستحتاج إلى مفتاح API الخاص بـ Novita. بمجرد حصولك عليه، أضفه إلى متغيرات البيئة الخاصة بك باسم NOVITA_API_KEY:

export NOVITA_API_KEY=your_api_key_here

بمجرد الانتهاء من ذلك، نحن جاهزون لبدء بناء الطاقم.

إنشاء نماذج اللغات الكبيرة (LLMs)

لنبدأ باستيراد التبعيات الضرورية:

import os
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import FileWriterTool

بعد ذلك، سنقوم بإنشاء نسخة من أداة FileWriterTool، التي سيستخدمها وكيل المبرمج لكتابة الملفات على القرص:

file_writer_tool = FileWriterTool()

مع تجهيز الأدوات، يمكننا الآن تهيئة نماذج اللغات الكبيرة (LLMs) لكل من وكلائنا:

architect_llm = LLM(
   model="novita/moonshotai/kimi-k2-instruct",
   temperature=0.5,
   api_base="https://api.novita.ai/v3/openai",
   api_key=os.environ['NOVITA_API_KEY']
)

coder_llm = LLM(
   model="novita/qwen/qwen3-coder-480b-a35b-instruct",
   temperature=0.4,
   api_base="https://api.novita.ai/v3/openai",
   api_key=os.environ['NOVITA_API_KEY']
)

reviewer_llm = LLM(
   model="novita/moonshotai/kimi-k2-instruct",
   temperature=0.5,
   api_base="https://api.novita.ai/v3/openai",
   api_key=os.environ['NOVITA_API_KEY']
)

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

تعريف الوكلاء

الآن لننشئ الوكلاء الثلاثة:

architect = Agent(
   role="Software Architect for MVP Projects",
   goal="Define a basic system structure and simple feature set to guide MVP development",
   backstory="""You specialize in quickly outlining software architectures and project scopes for
minimum viable products. Your plans help guide coders with enough structure to get started while staying lean.""",
   llm=architect_llm,
   verbose=True
)

coder = Agent(
   role="Developer for MVP Projects",
   goal="Implement the MVP using simple, clear code, and write all necessary code files using the FileWriter tool",
   backstory="""You're a practical developer focused on speed. You prioritize working code over polish.
Simulate any runtime behavior if needed, and make sure to keep code clear and modular.""",
   llm=coder_llm,
   verbose=True,
   tools=[file_writer_tool]
)

reviewer = Agent(
   role="Code Reviewer for MVP Projects",
   goal="Read the code files, provide helpful feedback, and write improvements as diffs into a markdown file",
   backstory="""You're a fast but thoughtful reviewer. You check code for clarity, obvious bugs, and
provide improvements as diffs. Instead of modifying code directly, you 
save your suggestions in a markdown file
for easy inspection.""",
   llm=reviewer_llm,
   verbose=True,
   tools=[file_writer_tool]
)

يتم تعيين دور وهدف وقصة خلفية محددة لكل وكيل.

تعريف المهام

لننشئ مهام الوكلاء:

architect_task = Task(
   description="""
Create a basic plan for building a simple MVP version of a {project}.
Focus on the core features, basic file structure, and tech stack.
Make the structure minimal but enough to get a working demo.
""",
   expected_output="""
A simple architectural overview with:
- Project goals
- Key components or files
- Basic data flow or structure
""",
   agent=architect,
   output_file="architecture.md"
)

coder_task = Task(
   description="""
Based on the architect's plan, implement the core parts of the {project} MVP.
Keep it lean and functional. Use the FileWriter tool to save all your code files.
If you need to simulate behavior (e.g. without a real interpreter), do so with clear comments or mocked logic.
""",
   expected_output="""
Working code files saved using the FileWriter tool that implement the key features defined in the plan.
Code should be readable and logically structured.
""",
   agent=coder,
   context=[architect_task]
)

review_task = Task(
   description="""
Review the code files created for the {project} MVP. Read each file using the FileRead tool.
Suggest improvements using diffs. Do not modify the original code files directly.
Instead, save all your suggested diffs into a markdown file using the FileWriter tool.
""",
   expected_output="""
A markdown file containing diff-style suggestions for each file reviewed.
""",
   agent=reviewer,
   context=[coder_task],
   output_file="code_review_diffs.md"
)

إنشاء الطاقم

الآن بعد أن عرفنا وكلائنا ونماذج اللغات الكبيرة (LLMs) الخاصة بهم ومهامهم، دعنا نجمع كل شيء معًا من خلال تجميع الطاقم:

code_crew = Crew(
   agents=[architect, coder, reviewer],
   tasks=[architect_task, coder_task, review_task],
   process=Process.sequential,
   verbose=True
)

مع إعداد كل شيء، يمكننا الآن تشغيل الطاقم:

if __name__ == "__main__":
   code_crew.kickoff(inputs={"project": "Todo App"})

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

إليك كيف يعمل:

1. يبدأ وكيل المهندس المعماري الأمور من خلال تصميم بنية البرنامج وحفظ الخطة في ملف يسمى architecture.md.
2. ثم يلتقط وكيل المبرمج مخرج المهندس المعماري وينشئ الأدلة وملفات الكود الضرورية لتنفيذ تطبيق المهام.
3. أخيرًا، يراجع وكيل المراجع الكود المُولد ويقدم اقتراحات باستخدام الفروقات، ويحفظ ملاحظاته في ملف يسمى code_review_diffs.md.

سير عمل التطوير القائم على الطواقم هذا قوي ووحدوي، لكنه لا يخلو من بعض القيود:

• موجهات مدمجة بشكل ثابت: من خلال دمج الموجهات مباشرة في الكود، يصبح من الأصعب صيانتها أو تحسينها بمرور الوقت.
• بنية متناثرة: يتم تعريف الوكلاء والمهام مباشرة في الكود، مما يمكن أن يؤدي إلى نص فوضوي وأصعب في التنقل مع نمو المشروع.

في القسم التالي، سننظر إلى كيفية تنظيف ذلك عن طريق نقل تعريفات الوكلاء وموجهات المهام إلى ملفات YAML خارجية.

بناء طواقم قائمة على الفئات

يتيح لنا CrewAI إنشاء طواقم قائمة على الفئات. دعنا نحدث الكود السابق ليتollow هذا الهيكل. أولاً، أنشئ مجلدًا يسمى config. سيحتوي هذا المجلد على ملفات YAML تحدد التكوين للوكلاء والمهام.

لننشئ ملفًا يسمى agents.yaml، الذي سيعمل كملف تكوين للوكلاء:

architect:
 role: >
   Software Architect for MVP Projects
 goal: >
   Define a basic system structure and simple feature set to guide MVP development
 backstory: >
   You specialize in quickly outlining software architectures and project scopes for minimum viable products.
  Your plans help guide coders with enough structure to get started while staying lean.

coder:
 role: >
   Developer for MVP Projects
 goal: >
   Implement the MVP using simple, clear code, and write all necessary code files using the FileWriter tool
 backstory: >
   You're a practical developer focused on speed. You prioritize working code over polish.
  Simulate any runtime behavior if needed, and make sure to keep code clear and modular.

reviewer:
 role: >
   Code Reviewer for MVP Projects
 goal: >
   Read the code files, provide helpful feedback, and write improvements as diffs into a markdown file
 backstory: >
   You're a fast but thoughtful reviewer. You check code for clarity, obvious bugs, and provide improvements
  as diffs. Instead of modifying code directly, you save your suggestions in a markdown file for easy inspection.

يحدد هذا الملف الأدوار والأهداف والقصص الخلفية لكل وكيل.

بعد ذلك، سنقوم بإنشاء ملف tasks.yaml لحفظ تكوينات المهام:

بدلاً من ذلك، احفظ كل الفروقات المقترحة الخاصة بك في ملف markdown باستخدام أداة FileWriter.

architect_task:
 description: >
   Create a basic plan for building a simple MVP version of a {project}.
  Focus on the core features, basic file structure, and tech stack.
  Make the structure minimal but enough to get a working demo.
 expected_output: >
   A simple architectural overview with:
  - Project goals
  - Key components or files
  - Basic data flow or structure
 agent: architect
 output_file: architecture.md

coder_task:
 description: >
   Based on the architect's plan, implement the core parts of the {project} MVP.
  Keep it lean and functional. Use the FileWriter tool to save all your code files.
  If you need to simulate behavior (e.g. without a real interpreter), do so with clear comments or mocked logic.
 expected_output: >
   Working code files saved using the FileWriter tool that implement the key features defined in the plan.
  Code should be readable and logically structured.
 agent: coder
 context:
   - architect_task

review_task:
 description: >
   Review the code files created for the {project} MVP. Read each file using the FileRead tool.
  Suggest improvements using diffs. Do not modify the original code files directly.
  Instead, save all your suggested diffs into a markdown file using the FileWriter tool.
 expected_output: >
   A markdown file containing diff-style suggestions for each file reviewed.
 agent: reviewer
 context:
   - coder_task
 output_file: code_review_diffs.md

بعد ذلك، لننشئ طاقمنا القائم على الفئات. للقيام بذلك، نحدد فئة ونزينها بزخرفة @CrewBase التي يوفرها CrewAI.

@CrewBase
class CodeCrew():
   """Three-agent code crew: architect, coder, reviewer"""

   @agent
   def architect(self) -> Agent:
       pass

   @agent
   def coder(self) -> Agent:
       pass

   @agent
   def reviewer(self) -> Agent:
       pass

   @task
   def architect_task(self) -> Task:
       pass

   @task
   def code_task(self) -> Task:
       pass

   @task
   def review_task(self) -> Task:
       pass

   @crew
   def crew(self) -> Crew:
       pass

في هذا الهيكل:

• تُستخدم زخرفة @agent لتحديد طرق إنتاج الوكلاء.
• تحدد زخرفة @task طرق إنتاج المهام.
• تحدد زخرفة @crew الطريقة التي تجمع الطاقم بالكامل.

الآن، لننظر إلى التنفيذ الكامل:

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

from crewai_tools import FileWriterTool

# Instantiate tool
file_writer_tool = FileWriterTool()

@CrewBase
class CodeCrew():

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

   agents_config = 'config/agents.yaml'
   tasks_config = 'config/tasks.yaml'
  
   @agent
   def architect(self) -> Agent:
       llm = LLM(
           model="novita/moonshotai/kimi-k2-instruct",
           temperature=0.5,
           api_base="https://api.novita.ai/v3/openai",
           api_key=os.environ['NOVITA_API_KEY']
       )
       return Agent(
           config=self.agents_config['architect'],
           llm=llm,
           verbose=True
       )

   @agent
   def coder(self) -> Agent:
       llm = LLM(
           model="novita/qwen/qwen3-coder-480b-a35b-instruct",
           temperature=0.4,
           api_base="https://api.novita.ai/v3/openai",
           api_key=os.environ['NOVITA_API_KEY']
       )
       return Agent(
           config=self.agents_config['coder'],
           llm=llm,
           verbose=True,
           tools=[file_writer_tool]
       )

   @agent
   def reviewer(self) -> Agent:
       llm = LLM(
           model="novita/moonshotai/kimi-k2-instruct",
           temperature=0.5,
           api_base="https://api.novita.ai/v3/openai",
           api_key=os.environ['NOVITA_API_KEY']
       )
       return Agent(
           config=self.agents_config['reviewer'],
           llm=llm,
           verbose=True,
       )

   @task
   def architect_task(self) -> Task:
       return Task(config=self.tasks_config['architect_task'])

   @task
   def coder_task(self) -> Task:
       return Task(config=self.tasks_config['coder_task'])

   @task
   def review_task(self) -> Task:
       return Task(config=self.tasks_config['review_task'])

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

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

يمكنك الآن تشغيل الطاقم عن طريق إضافة الكود التالي إلى النص البرمجي:

def main():
   code_crew = CodeCrew()
   code_crew.crew().kickoff(inputs={"project": "Todo App"})
if __name__ == '__main__':
   main()

بناء التدفقات باستخدام Novita

قمنا سابقًا ببناء طاقم يتبع عملية تسلسلية. الآن، دعنا نستكشف كيفية بناء تدفقة تتبع سير عمل مصمم خصيصًا.

سير العمل الخاص بنا

نريد تصميم نظام متعدد الوكلاء لدعم العملاء له وكيل رئيسي مسؤول عن تصنيف مشاكل العملاء. بناءً على التصنيف، سيحدد النظام أي وكيل متخصص هو الأكثر ملاءمة للتعامل مع المشكلة. يتكون الإعداد من أربعة وكلاء:

• وكيل العملاء: هذا هو الوكيل الرئيسي. يتلقى مشكلة العميل ويصنفها إلى واحدة من الفئات التالية: الفواتير، التقنية، أو عام.
• وكيل الفواتير: يتعامل مع جميع المشاكل المتعلقة بالفواتير.
• الوكيل التقني: يتعامل مع المشاكل التقنية.
• الوكيل العام: يتعامل مع أي مشاكل لا تقع في الفئات المذكورة أعلاه.

تعريف حالة التدفقة

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

لنستورد الوحدات الضرورية ونحدد حالتنا:

import json
import os
from typing import Literal, Dict, Any
from pydantic import BaseModel, ValidationError
from crewai import Agent, LLM
from crewai.flow.flow import Flow, start, router, listen

# Define the flow state
class SupportState(BaseModel):
   issue: Literal["billing", "technical", "general"] = "general"
   message: str = ""

إنشاء التدفقة

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

class CustomerSupportFlow(Flow[SupportState]):

   @start()
   def classify_issue(self):
       pass

   @router(classify_issue)
   def route_based_on_issue(self) -> str:
       print(f"📨 Routing to agent for issue type: {self.state.issue}")
       if self.state.issue == "billing":
           return "billing"
       if self.state.issue == "technical":
           return "technical"
       else:
           return "general"       

   @listen("billing")
   def handle_billing(self):
       pass

   @listen("technical")
   def handle_technical(self):
       pass

   @listen("general")
   def handle_general(self):
       pass

الطريقة المزخرفة بـ @start() هي نقطة دخول التدفقة. في هذه الحالة، تتعامل مع تصنيف المشكلة. يتم تنفيذ الطريقة المزخرفة بـ @router() بعد classify_issue وتحدد أي وكيل توجيه إليه. تستخدم الطرق الأخرى النظام المدفوع بالأحداث للرد عند توجيهها إليها.

تنفيذ طرق التدفقة

لننفذ الآن كل طريقة ونعين الوكيل المناسب لكل منها:

class CustomerSupportFlow(Flow[SupportState]):

   @start()
   def classify_issue(self) -> Dict[str, Any]:
       intake_agent = Agent(
           role="User Intake Agent",
           goal="Classify user support issue and summarize the message",
           backstory="You classify the user query into billing, technical, or general categories.",
           llm=LLM(
               model="novita/qwen/qwen2.5-vl-72b-instruct",
               temperature=0.3,
               api_base="https://api.novita.ai/v3/openai",
               api_key=os.environ["NOVITA_API_KEY"]
           ),
           verbose=True
       )
      
       prompt = f"""
           A user submitted this message: "{self.state.message}"

           Your task:
           1. Identify whether the issue is "billing", "technical", or "general".
           2. Rephrase or extract the message clearly.

           Respond in valid JSON format like the example below:

           {{
               "issue": "billing",
               "message": "The user has been charged twice for their subscription and is requesting a refund."
           }}
       """

       output = intake_agent.kickoff(prompt, response_format=SupportState)
      
       try:
           # Parse the JSON string
           print(output.raw)
           parsed_json = json.loads(output.raw)

           # Validate with Pydantic
           validated = SupportState(**parsed_json)

           # Save to state
           self.state.issue = validated.issue
           self.state.message = validated.message

       except (json.JSONDecodeError, ValidationError) as e:
           print("❌ Failed to parse or validate response:", e)
           self.state.issue = "general"

   @router(classify_issue)
   def route_based_on_issue(self) -> str:
       print(f"📨 Routing to agent for issue type: {self.state.issue}")
       if self.state.issue == "billing":
           return "billing"
       if self.state.issue == "technical":
           return "technical"
       else:
           return "general"       

   @listen("billing")
   def handle_billing(self):
       agent = Agent(
           role="Billing Agent",
           goal="Handle billing questions, refunds, and invoice issues",
           backstory="You resolve billing-related queries effectively.",
           llm=LLM(
               model="novita/meta-llama/llama-3.3-70b-instruct",
               temperature=0.5,
               api_base="https://api.novita.ai/v3/openai",
               api_key=os.environ["NOVITA_API_KEY"]
           ),
           verbose=True
       )
       result = agent.kickoff(self.state.message)
       print("💰 Billing Agent Response:", result)

   @listen("technical")
   def handle_technical(self):
       agent = Agent(
           role="Technical Support Agent",
           goal="Help users resolve technical issues",
           backstory="You're a technical expert providing troubleshooting support.",
           llm=LLM(
               model="novita/meta-llama/llama-3.1-8b-instruct",
               temperature=0.5,
               api_base="https://api.novita.ai/v3/openai",
               api_key=os.environ["NOVITA_API_KEY"]
           ),
           verbose=True
       )
       result = agent.kickoff(self.state.message)
       print("🛠 Technical Agent Response:", result)

   @listen("general")
   def handle_general(self):
       agent = Agent(
           role="General Support Agent",
           goal="Provide helpful answers to non-technical and non-billing questions",
           backstory="You're friendly and well-informed about our services.",
           llm=LLM(
               model="novita/minimaxai/minimax-m1-80k",
               temperature=0.5,
               api_base="https://api.novita.ai/v3/openai",
               api_key=os.environ["NOVITA_API_KEY"]
           ),
           verbose=True
       )
       result = agent.kickoff(self.state.message)
       print("📋 General Agent Response:", result)

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

def run():
   flow = CustomerSupportFlow()
   flow.plot("CustomerSupportFlowPlot")

   # Example input message
   example_input = {
       "message": "Hello, I was charged twice for my subscription and need a refund."
   }

   flow.kickoff(inputs=example_input)

if __name__ == "__main__":
   run()

الخلاصة

لبناء نظام متعدد الوكلاء، تحتاج إلى مكونين رئيسيين: موفر نماذج يقدم مجموعة واسعة من النماذج، وإطار عمل متعدد الوكلاء لتنسيق تفاعلاتهم. في هذا المقال، أوضحنا كيف يمكن لـ Novita و CrewAI أن يقدما هذه الأدوار بشكل فعال. لمعرفة المزيد عن CrewAI، تأكد من استكشاف التوثيق الخاص بهم. وإذا كنت تريد تجربة المزيد من النماذج لطواقمك، فراجع ملعب نماذج اللغات الكبيرة (LLM) من Novita.

مستودع الملفات - https://github.com/novitalabs/Novita-CollabHub/tree/main/examples/novita-crewai

حول Novita AI

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