Multi-Agent — وقتی چند Agent با هم کار می‌کنن

مقدمه: یه Agent کافی نیست

تا اینجای سری، یاد گرفتیم یه Agent بسازیم که فکر می‌کنه، ابزار استفاده می‌کنه، حافظه داره و برنامه‌ریزی می‌کنه. ولی یه مشکل هست: بعضی کارها برای یه Agent تنها خیلی سنگینه.

مثلاً فکر کن بخوای یه اپلیکیشن کامل بسازی. یه Agent باید هم طراحی UI کنه، هم بک‌اند بنویسه، هم تست بنویسه، هم دیپلوی کنه. خیلی سخته یه Agent توی همه اینا متخصص باشه.

راه‌حل؟ چند Agent که هر کدوم یه تخصص دارن و با هم همکاری می‌کنن. مثل یه تیم واقعی از آدم‌ها. این می‌شه Multi-Agent Systems.

چرا Multi-Agent؟

چند دلیل اصلی:

۱. تخصصی‌سازی

یه Agent که فقط کدنویسی بلده، کدش بهتر از Agent ای ه که هم کد می‌نویسه هم طراحی می‌کنه هم مدیریت می‌کنه. مثل دنیای واقعی: یه برنامه‌نویس فرانت‌اند با یه بک‌اند دولوپر، بهتر از یه آدم که هر دو رو نصفه‌نیمه بلده.

۲. موازی‌سازی

چند Agent می‌تونن همزمان روی بخش‌های مختلف کار کنن. سرعت کار بالا می‌ره.

۳. کنترل پیچیدگی

به جای یه prompt غول‌پیکر با ۱۰۰ تا instruction، هر Agent یه وظیفه مشخص داره. ساده‌تر، قابل فهم‌تر، قابل دیباگ‌تر.

۴. قابلیت اطمینان

اگه یه Agent fail کنه، بقیه می‌تونن کار رو ادامه بدن. مثل یه تیم که اگه یه نفر مریض بشه، بقیه جبران می‌کنن.

الگوی ۱: Orchestrator (هماهنگ‌کننده)

رایج‌ترین الگو. یه Agent اصلی (Orchestrator) وظایف رو بین Agent های تخصصی تقسیم می‌کنه و نتایج رو جمع‌آوری می‌کنه.

class OrchestratorAgent:
    def __init__(self):
        self.client = OpenAI()
        self.specialists = {
            "coder": CoderAgent(),
            "reviewer": ReviewerAgent(),
            "tester": TesterAgent(),
            "writer": DocumentationAgent(),
        }
    
    def solve(self, task: str) -> str:
        # ۱. تحلیل وظیفه و تعیین Agent ها
        plan = self._create_plan(task)
        
        # ۲. تخصیص و اجرا
        results = {}
        for step in plan:
            agent_name = step["assigned_to"]
            agent = self.specialists[agent_name]
            
            # اطلاعات مراحل قبلی رو بده
            context = self._get_step_context(step, results)
            result = agent.execute(step["task"], context)
            results[step["id"]] = result
            
            print(f"[{agent_name}] {step['task']}")
            print(f"  ← {result[:100]}...\n")
        
        # ۳. جمع‌بندی
        return self._compile_final(task, results)
    
    def _create_plan(self, task: str) -> list:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": """تو یه مدیر پروژه هستی.
وظیفه‌ات تقسیم کار بین اعضای تیمه.
اعضا: coder (کدنویسی), reviewer (بررسی کد), 
       tester (تست), writer (مستندات)
خروجی JSON: لیست مراحل با assigned_to"""},
                {"role": "user", "content": task}
            ],
            response_format={"type": "json_object"}
        )
        return json.loads(
            response.choices[0].message.content
        )["steps"]

یه Agent تخصصی هم اینطوری ساخته می‌شه:

class CoderAgent:
    def __init__(self):
        self.client = OpenAI()
        self.system_prompt = """تو یه برنامه‌نویس حرفه‌ای پایتون هستی.
کد تمیز، خوانا و با داکیومنت بنویس.
همیشه type hints استفاده کن.
اصول SOLID رو رعایت کن."""
    
    def execute(self, task: str, context: str = "") -> str:
        messages = [
            {"role": "system", "content": self.system_prompt}
        ]
        if context:
            messages.append({
                "role": "system", 
                "content": f"اطلاعات قبلی:\n{context}"
            })
        messages.append({"role": "user", "content": task})
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
        )
        return response.choices[0].message.content


class ReviewerAgent:
    def __init__(self):
        self.client = OpenAI()
        self.system_prompt = """تو یه Code Reviewer حرفه‌ای هستی.
کد رو از نظر این موارد بررسی کن:
- باگ‌های احتمالی
- مشکلات امنیتی
- کیفیت کد و خوانایی
- بهینه‌سازی عملکرد
نقد سازنده و مشخص بده."""
    
    def execute(self, task: str, context: str = "") -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", 
             "content": f"این کد رو بررسی کن:\n{context}\n\n{task}"}
        ]
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
        )
        return response.choices[0].message.content

الگوی ۲: Debate (مناظره)

توی این الگو، دو یا چند Agent درباره یه موضوع بحث و مناظره می‌کنن. هر کدوم دیدگاه خودش رو داره و سعی می‌کنه بقیه رو قانع کنه. نتیجه نهایی از ترکیب دیدگاه‌ها به دست میاد.

این الگو خصوصاً برای تصمیم‌گیری‌های پیچیده خوبه:

class DebateSystem:
    def __init__(self):
        self.client = OpenAI()
    
    def debate(self, topic: str, rounds: int = 3) -> str:
        # دو Agent با دیدگاه‌های مختلف
        agent_a_prompt = f"""تو طرفدار استفاده از {topic} هستی. 
استدلال‌های قوی بیار. منطقی باش."""
        
        agent_b_prompt = f"""تو منتقد {topic} هستی.
مشکلات و ریسک‌ها رو بگو. منطقی باش."""
        
        debate_history = []
        
        for round_num in range(rounds):
            # Agent A حرف می‌زنه
            a_response = self._agent_speak(
                agent_a_prompt, topic, debate_history, "A"
            )
            debate_history.append(f"Agent A: {a_response}")
            
            # Agent B جواب می‌ده
            b_response = self._agent_speak(
                agent_b_prompt, topic, debate_history, "B"
            )
            debate_history.append(f"Agent B: {b_response}")
            
            print(f"--- راند {round_num + 1} ---")
            print(f"موافق: {a_response[:150]}...")
            print(f"مخالف: {b_response[:150]}...\n")
        
        # داور نهایی
        return self._judge(topic, debate_history)
    
    def _judge(self, topic: str, debate: list) -> str:
        """قضاوت نهایی"""
        judge_prompt = f"""بعد از شنیدن بحث زیر درباره "{topic}"، 
یه نتیجه‌گیری متوازن و عملی بنویس.

بحث:
{chr(10).join(debate)}

نتیجه‌گیری:"""
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": judge_prompt}]
        )
        return response.choices[0].message.content
    
    def _agent_speak(self, system: str, topic: str,
                     history: list, agent_id: str) -> str:
        messages = [
            {"role": "system", "content": system},
            {"role": "user", 
             "content": f"موضوع: {topic}\n\n"
                       f"بحث تا الان:\n"
                       f"{chr(10).join(history) if history else 'شروع بحث'}\n\n"
                       f"نوبت توئه:"}
        ]
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
        )
        return response.choices[0].message.content

الگوی ۳: Pipeline (خط لوله)

مثل خط تولید کارخانه. هر Agent یه مرحله از کار رو انجام می‌ده و خروجیش رو به Agent بعدی می‌ده:

class AgentPipeline:
    def __init__(self):
        self.stages = [
            ("researcher", ResearchAgent()),
            ("writer", WriterAgent()),
            ("editor", EditorAgent()),
            ("fact_checker", FactCheckAgent()),
        ]
    
    def run(self, initial_input: str) -> str:
        current_output = initial_input
        
        for stage_name, agent in self.stages:
            print(f"مرحله: {stage_name}")
            current_output = agent.process(current_output)
            print(f"  خروجی: {current_output[:100]}...\n")
        
        return current_output

مثال: تولید محتوا

1. Researcher: درباره موضوع تحقیق می‌کنه
2. Writer: از تحقیق، مقاله می‌نویسه
3. Editor: مقاله رو ویرایش می‌کنه
4. Fact Checker: اطلاعات رو چک می‌کنه

الگوی ۴: Supervisor (سرپرست)

مشابه Orchestrator ولی با یه تفاوت مهم: Supervisor فقط هدایت می‌کنه و کنترل کیفیت داره. Agent ها خودشون تصمیم‌های جزئی رو می‌گیرن.

class SupervisorAgent:
    def __init__(self):
        self.client = OpenAI()
        self.workers = {
            "analyst": AnalystAgent(),
            "developer": DeveloperAgent(),
            "qa": QAAgent(),
        }
    
    def manage(self, project: str) -> str:
        status = {"completed": [], "in_progress": None, "todo": []}
        
        # تعیین وظایف اولیه
        tasks = self._plan_tasks(project)
        status["todo"] = tasks
        
        while status["todo"]:
            # انتخاب وظیفه بعدی
            task = status["todo"].pop(0)
            
            # تخصیص به Agent مناسب
            worker_name = self._assign_worker(task)
            worker = self.workers[worker_name]
            
            # اجرا
            result = worker.execute(task)
            
            # بررسی کیفیت
            quality_check = self._review_quality(task, result)
            
            if quality_check["approved"]:
                status["completed"].append({
                    "task": task, 
                    "result": result
                })
                print(f"[تایید] {task[:50]}...")
            else:
                # برگردون برای اصلاح
                revised_task = f"{task}\n\nایرادات: " \
                              f"{quality_check['feedback']}"
                status["todo"].insert(0, revised_task)
                print(f"[برگشت] {task[:50]}...")
        
        return self._compile_report(status["completed"])

آشنایی با CrewAI

CrewAI یه فریمورک پایتون برای ساخت سیستم‌های Multi-Agent ه. ایده‌ش ساده‌ست: یه “خدمه” (Crew) از Agent ها تعریف می‌کنی، هر کدوم نقش و وظیفه مشخص دارن.

from crewai import Agent, Task, Crew

# تعریف Agent ها
researcher = Agent(
    role="محقق ارشد",
    goal="تحقیق جامع و دقیق درباره موضوعات فنی",
    backstory="تو یه محقق با ۱۰ سال تجربه هستی "
              "که مهارت بالایی در پیدا کردن اطلاعات دقیق داری.",
    verbose=True,
    llm="gpt-4o",
)

writer = Agent(
    role="نویسنده فنی",
    goal="نوشتن محتوای فنی واضح و جذاب",
    backstory="تو یه نویسنده فنی هستی که می‌تونی "
              "مفاهیم پیچیده رو ساده توضیح بدی.",
    verbose=True,
    llm="gpt-4o",
)

editor = Agent(
    role="ویراستار",
    goal="ویرایش و بهبود کیفیت محتوا",
    backstory="تو یه ویراستار حرفه‌ای هستی "
              "با چشم تیز برای جزئیات.",
    verbose=True,
    llm="gpt-4o",
)

# تعریف وظایف
research_task = Task(
    description="درباره {topic} تحقیق کن. "
                "منابع معتبر پیدا کن و نکات کلیدی رو لیست کن.",
    expected_output="گزارش تحقیق با نکات کلیدی و منابع",
    agent=researcher,
)

writing_task = Task(
    description="با استفاده از تحقیق انجام‌شده، "
                "یه مقاله فنی ۱۵۰۰ کلمه‌ای بنویس.",
    expected_output="مقاله فنی کامل",
    agent=writer,
    context=[research_task],  # وابسته به تحقیق
)

editing_task = Task(
    description="مقاله رو ویرایش کن. "
                "اشتباهات رو اصلاح کن و کیفیت رو بالا ببر.",
    expected_output="مقاله ویرایش‌شده نهایی",
    agent=editor,
    context=[writing_task],  # وابسته به نوشتن
)

# ساخت Crew و اجرا
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    verbose=True,
)

result = crew.kickoff(inputs={"topic": "RAG در هوش مصنوعی"})
print(result)

معماری عملی Multi-Agent

بیا یه معماری واقعی رو ببینیم — یه سیستم پشتیبانی مشتری:

class CustomerSupportSystem:
    """سیستم پشتیبانی مشتری با چند Agent"""
    
    def __init__(self):
        self.router = RouterAgent()      # مسیریابی پیام
        self.tech = TechSupportAgent()   # پشتیبانی فنی
        self.billing = BillingAgent()    # مالی
        self.sales = SalesAgent()        # فروش
        self.escalation = HumanAgent()   # ارجاع به انسان
    
    def handle_message(self, customer_msg: str, 
                       customer_id: str) -> str:
        # ۱. مسیریابی: کدوم Agent باید جواب بده؟
        route = self.router.classify(customer_msg)
        
        print(f"مسیریابی: {route['category']} "
              f"(اطمینان: {route['confidence']})")
        
        # ۲. اگه اطمینان کمه، از مشتری بپرس
        if route["confidence"] < 0.7:
            return self._ask_clarification(customer_msg)
        
        # ۳. ارجاع به Agent مناسب
        agent_map = {
            "technical": self.tech,
            "billing": self.billing,
            "sales": self.sales,
            "complex": self.escalation,
        }
        
        agent = agent_map.get(route["category"], self.tech)
        
        # ۴. گرفتن تاریخچه مشتری
        history = self._get_customer_history(customer_id)
        
        # ۵. پاسخ
        response = agent.respond(customer_msg, history)
        
        # ۶. ذخیره تعامل
        self._save_interaction(customer_id, customer_msg, 
                               response, route["category"])
        
        return response


class RouterAgent:
    """Agent مسیریاب — تشخیص نوع درخواست"""
    
    def classify(self, message: str) -> dict:
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",  # مدل سبک‌تر برای سرعت
            messages=[
                {"role": "system", "content": """نوع درخواست رو تشخیص بده:
- technical: مشکل فنی، باگ، خطا
- billing: فاکتور، پرداخت، اشتراک
- sales: خرید، قیمت، مقایسه
- complex: نیاز به انسان
خروجی JSON: {"category": "...", "confidence": 0.0-1.0}"""},
                {"role": "user", "content": message}
            ],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)

چالش‌های Multi-Agent

۱. هماهنگی (Coordination)

وقتی چند Agent همزمان کار می‌کنن، باید هماهنگ باشن. اگه Agent A یه فایل رو تغییر بده و Agent B هم همون فایل رو تغییر بده، conflict داریم.

۲. ارتباطات (Communication)

Agent ها باید بتونن با هم حرف بزنن. چه اطلاعاتی رد و بدل بشه؟ چه فرمتی؟ هر چقدر ارتباطات بهتر باشه، نتیجه بهتره.

۳. هزینه (Cost)

هر Agent یعنی API call اضافی. یه سیستم Multi-Agent ممکنه ۵ تا ۱۰ برابر یه Agent تنها هزینه داشته باشه. باید حساب کتاب کنی.

۴. دیباگ کردن

وقتی نتیجه غلطه، کدوم Agent اشتباه کرده؟ دیباگ سیستم‌های Multi-Agent سخت‌تره. logging خوب اینجا حیاتیه.

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("multi_agent")

class LoggedAgent:
    def __init__(self, name: str):
        self.name = name
    
    def execute(self, task: str, context: str = "") -> str:
        logger.info(f"[{self.name}] شروع: {task[:80]}...")
        logger.debug(f"[{self.name}] Context: {context[:200]}")
        
        try:
            result = self._do_work(task, context)
            logger.info(f"[{self.name}] موفق: {result[:80]}...")
            return result
        except Exception as e:
            logger.error(f"[{self.name}] خطا: {e}")
            raise

کی از Multi-Agent استفاده کنیم؟

قانون ساده:

• کار ساده (سوال-جواب، محاسبه) → یه Agent کافیه
• کار متوسط (تحقیق + نوشتن) → Pipeline دو-سه Agent ای
• کار پیچیده (پروژه نرم‌افزاری، تحلیل جامع) → Orchestrator + Agent های تخصصی
• تصمیم‌گیری (معماری، استراتژی) → الگوی Debate

آینده Multi-Agent

Multi-Agent Systems هنوز توی مراحل اولیه‌شونه ولی خیلی سریع داره پیشرفت می‌کنه:

• Agent Marketplaces: جایی که Agent های آماده رو می‌تونی پیدا کنی و ترکیب کنی
• استانداردسازی ارتباطات: پروتکل‌های استاندارد برای حرف زدن Agent ها با هم (مثل MCP)
• Agent های خودسازمان‌ده: Agent هایی که خودشون تیم تشکیل می‌دن و وظایف رو تقسیم می‌کنن
• Agent Economy: اقتصادی که Agent ها با هم معامله و همکاری می‌کنن

جمع‌بندی سری

بیا یه نگاه کلی به کل سری بندازیم:

• اپیزود ۱: Agent چیه — فرق Agent با Chatbot و ۴ قابلیت اصلی
• اپیزود ۲: Tool Use — وقتی Agent می‌تونه از ابزارها استفاده کنه
• اپیزود ۳: حافظه — کوتاه‌مدت، بلندمدت، و Vector DB
• اپیزود ۴: Planning — ReAct, CoT, تجزیه وظایف، Self-Reflection
• اپیزود ۵: Multi-Agent — وقتی چند Agent با هم کار می‌کنن

این ۵ مفهوم پایه‌های ساخت Agent هستن. با ترکیب اینا می‌تونی Agent هایی بسازی که واقعاً کارهای پیچیده انجام بدن.

دنیای Agent ها خیلی سریع داره تغییر می‌کنه. هر هفته ابزار جدید، فریمورک جدید، و ایده جدید میاد. ولی اگه این پایه‌ها رو خوب بلد باشی، هر ابزار جدیدی رو راحت یاد می‌گیری. چون همه‌شون از همین مفاهیم ساخته شدن.

موفق باشی!