मल्टी-एजेंट पैटर्न: ऑर्केस्ट्रेटर, Workers और पाइपलाइन

एकल एजेंट सीमित दायरे वाले कार्यों में अच्छे होते हैं। जिस क्षण कोई कार्य विभिन्न डोमेन में विशेष ज्ञान, बड़े इनपुट पर समानांतर काम, या निष्पादन से पहले सत्यापित किए जाने वाले निर्णयों की मांग करता है — आपको कई एजेंटों की आवश्यकता होती है।

मल्टी-एजेंट सिस्टम स्वाभाविक रूप से अधिक जटिल नहीं हैं — वे बस अलग तरह से संरचित हैं। मुख्य बात यह है कि समस्या के लिए सही पैटर्न चुनें। तीन पैटर्न अधिकांश उपयोग मामलों को कवर करते हैं: ऑर्केस्ट्रेटर-Worker, पाइपलाइन और पैरेलल फैन-आउट।

कई एजेंट क्यों

मल्टी-एजेंट सिस्टम का मामला तीन व्यावहारिक कारणों पर आधारित है।

विशेषज्ञता। संदर्भ में 50 टूल वाला एक एजेंट भ्रमित हो जाता है। 5 फोकस्ड टूल वाला विशेष एजेंट बेहतर प्रदर्शन करता है। डोमेन के अनुसार विभाजित करें — रिसर्च, लेखन, कोड, सत्यापन — और हर एजेंट एक काम अच्छे से करे।

समानांतरता। कुछ कार्य स्वतंत्र उप-कार्यों में विभाजित होते हैं। 20 दस्तावेजों का क्रमिक विश्लेषण धीमा है; समानांतर एजेंटों के साथ एक साथ विश्लेषण तेज है।

सत्यापन। एक एजेंट द्वारा आउटपुट तैयार करना और एक दूसरे एजेंट द्वारा स्वतंत्र रूप से आलोचना करना उन त्रुटियों को पकड़ता है जो आत्म-समीक्षा चूक जाती है। समीक्षक का मूल उत्तर का बचाव करने में कोई हित नहीं है।

पैटर्न 1: ऑर्केस्ट्रेटर-Worker

एक ऑर्केस्ट्रेटर एजेंट योजना बनाता है और कार्य सौंपता है। Worker एजेंट विशिष्ट कार्य निष्पादित करते हैं और परिणाम लौटाते हैं। ऑर्केस्ट्रेटर अंतिम आउटपुट को असेंबल करता है।

यह सबसे लचीला पैटर्न है। ऑर्केस्ट्रेटर मध्यवर्ती परिणामों के आधार पर अपनी योजना को अनुकूलित कर सकता है, विफल कार्यों को पुनः प्रयास कर सकता है, या किसी भिन्न Worker तक एस्केलेट कर सकता है।

import anthropic
import json

client = anthropic.Anthropic()

def run_worker(system_prompt: str, task: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    return response.content[0].text

def orchestrator(user_request: str) -> str:
    # चरण 1: काम की योजना बनाएं
    plan_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""आप एक योजना एजेंट हैं। उपयोगकर्ता अनुरोध दिए जाने पर, इसे
2-4 विशिष्ट उप-कार्यों में विभाजित करें। केवल कार्य विवरणों की JSON array लौटाएं।
उदाहरण: ["X खोजें", "Y का विश्लेषण करें", "निष्कर्षों को संश्लेषित करें"]""",
        messages=[{"role": "user", "content": user_request}]
    )

    tasks = json.loads(plan_response.content[0].text)

    # चरण 2: विशेष Worker के साथ प्रत्येक कार्य चलाएं
    results = []
    for task in tasks:
        result = run_worker(
            system_prompt="आप एक फोकस्ड निष्पादन एजेंट हैं। असाइन किए गए कार्य को पूरी तरह से पूरा करें।",
            task=task
        )
        results.append({"task": task, "result": result})

    # चरण 3: संश्लेषण
    synthesis_prompt = f"""मूल अनुरोध: {user_request}

Worker परिणाम:
{json.dumps(results, indent=2, ensure_ascii=False)}

इन परिणामों को एक सुसंगत अंतिम प्रतिक्रिया में संश्लेषित करें।"""

    final = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": synthesis_prompt}]
    )
    return final.content[0].text

ऑर्केस्ट्रेटर पैटर्न तब सबसे अच्छा काम करता है जब कार्य की संरचना पहले से ज्ञात नहीं होती। यदि आपको पता नहीं है कि समस्या का विश्लेषण करने तक आपको कितने उप-कार्यों की आवश्यकता होगी, तो ऑर्केस्ट्रेटर का उपयोग करें।

एक आम समस्या: ऑर्केस्ट्रेटर बेमतलब उप-कार्यों को मतिभ्रमित कर सकते हैं। Workers चलाने से पहले आउटपुट फॉर्मेट (JSON array, नंबर्ड लिस्ट) को बाधित करें और उसे सत्यापित करें। JSON पार्सिंग के आसपास try/except और फॉलबैक के रूप में पुनः-योजना चरण इसे सुचारू रूप से संभालता है।

पैटर्न 2: पाइपलाइन

एजेंट एक अनुक्रमिक श्रृंखला बनाते हैं। प्रत्येक एजेंट इनपुट को रूपांतरित करता है और अपना आउटपुट अगले को पास करता है। कोई भी एजेंट दूसरों के बारे में नहीं जानता — वे इनपुट प्राप्त करते हैं और आउटपुट उत्पन्न करते हैं।

यह लागू करने और तर्क करने में सबसे सरल पैटर्न है। यह अच्छी तरह परिभाषित चरणों वाले रूपांतरण कार्यों के लिए अच्छा काम करता है।

def run_pipeline(input_text: str) -> str:
    stages = [
        {
            "name": "शोधकर्ता",
            "system": "इनपुट से सभी प्रमुख तथ्यों को निकालें और व्यवस्थित करें। "
                      "उपलब्ध होने पर स्रोतों के साथ संरचित सूची के रूप में फॉर्मेट करें।",
        },
        {
            "name": "लेखक",
            "system": "शोध नोट्स को स्पष्ट, पठनीय गद्य में रूपांतरित करें। "
                      "सभी तथ्यात्मक सामग्री बनाए रखें। तकनीकी दर्शकों को लक्षित करें।",
        },
        {
            "name": "संपादक",
            "system": "स्पष्टता और संक्षिप्तता में सुधार करें। अनावश्यक दोहराव हटाएं। "
                      "तथ्य मत बदलें। केवल बेहतर टेक्स्ट लौटाएं।",
        },
        {
            "name": "तथ्य जांचकर्ता",
            "system": "आंतरिक स्थिरता की समीक्षा करें। किसी भी दावे को चिह्नित करें जो "
                      "एक-दूसरे का विरोध करते हों या असमर्थित लगते हों। "
                      "यदि कोई समस्या नहीं है, तो 'सत्यापित: ' के बाद मूल टेक्स्ट लौटाएं।",
        },
    ]

    current = input_text
    for stage in stages:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=stage["system"],
            messages=[{"role": "user", "content": current}]
        )
        current = response.content[0].text
        print(f"[{stage['name']}] पूर्ण ({len(current)} chars)")

    return current

पाइपलाइन त्रुटियाँ जमा करती हैं। यदि शोधकर्ता कुछ चूक जाता है, तो लेखक इसे वापस नहीं जोड़ सकता। अपने चरणों को हानिरहित के बजाय योगात्मक बनाएं — ऐसे चरणों से बचें जो जानकारी हटाते हैं जो अगले चरण को चाहिए हो।

एक व्यावहारिक समायोजन: प्रत्येक चरण के आउटपुट के साथ मूल इनपुट पास करें जब डाउनस्ट्रीम एजेंटों को उस संदर्भ की आवश्यकता हो जो पहले के चरणों ने संकुचित किया हो।

पैटर्न 3: पैरेलल फैन-आउट

एक बड़े इनपुट को स्वतंत्र खंडों में विभाजित करें, प्रत्येक को अलग एजेंटों के साथ एक साथ प्रोसेस करें, फिर परिणाम एकत्रित करें।

यह सही पैटर्न है जब आप एक संदर्भ विंडो में सहजता से फिट होने से अधिक डेटा प्रोसेस कर रहे हों, या जब प्रोसेसिंग समय महत्वपूर्ण हो।

import asyncio

async def analyze_document(doc: str, index: int) -> dict:
    """एक दस्तावेज़ को असिंक्रोनस रूप से विश्लेषण करें।"""
    system = """इस दस्तावेज़ का विश्लेषण करें और निम्नलिखित के साथ JSON ऑब्जेक्ट लौटाएं:
- "sentiment": positive/negative/neutral
- "key_topics": 3-5 मुख्य विषयों की सूची
- "summary": 2-3 वाक्य का सारांश
- "flags": चिंताओं की सूची (यदि कोई नहीं तो खाली सूची)"""

    result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            system=system,
            messages=[{"role": "user", "content": doc}]
        ).content[0].text
    )
    return {"index": index, **json.loads(result)}

async def parallel_analysis(documents: list[str]) -> dict:
    # फैन आउट: सभी दस्तावेजों का एक साथ विश्लेषण करें
    tasks = [analyze_document(doc, i) for i, doc in enumerate(documents)]
    analyses = await asyncio.gather(*tasks)

    # एक समर्पित सिंथेसिस एजेंट के साथ एकत्रित करें
    synthesis_input = json.dumps({
        "document_count": len(documents),
        "analyses": analyses
    }, ensure_ascii=False)

    aggregate_result = await asyncio.to_thread(
        lambda: client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="दस्तावेज़ विश्लेषणों को संश्लेषित करें: समग्र भावना वितरण, "
                   "सभी दस्तावेजों में शीर्ष विषय और संकेतों में उल्लेखनीय पैटर्न। "
                   "JSON के रूप में लौटाएं।",
            messages=[{"role": "user", "content": synthesis_input}]
        ).content[0].text
    )

    return {
        "individual": analyses,
        "aggregate": json.loads(aggregate_result)
    }

एकत्रीकरण चरण वह जगह है जहाँ अधिकांश कार्यान्वयन शॉर्टकट लेते हैं। परिणामों को जोड़ें नहीं — उन्हें एक ऐसे एजेंट को पास करें जो एकत्रीकरण कार्य को समझता है। 20 विश्लेषणों की स्ट्रिंग जॉइन उपयोगी नहीं है; एक संश्लेषित सारांश है।

सही पैटर्न चुनना

ये पैटर्न मिलकर काम करते हैं। एक वास्तविक सिस्टम एक ऐसे ऑर्केस्ट्रेटर का उपयोग कर सकता है जो कुछ कार्यों को समानांतर में वितरित करता है जबकि अन्य को पाइपलाइन के माध्यम से चलाता है। सबसे सरल पैटर्न से शुरू करें जो फिट हो और जटिलता तभी जोड़ें जब सरल दृष्टिकोण विफल हो।

व्यावहारिक विचार

लागत। मल्टी-एजेंट सिस्टम API कॉल को कई गुना बढ़ाते हैं। एक 4-चरण पाइपलाइन एकल कॉल की तुलना में 4 गुना अधिक लागत और सिंथेसिस ओवरहेड पड़ सकती है। मॉडलों को रणनीतिक रूप से मिलाएं: जहाँ निर्णय महत्वपूर्ण है वहाँ ऑर्केस्ट्रेशन और योजना के लिए Opus का उपयोग करें, उच्च-मात्रा निष्पादन कार्यों के लिए Haiku का।

त्रुटि प्रसार। पहले से तय करें कि प्रत्येक एजेंट विफलताओं को कैसे संभालता है। विकल्प: त्रुटि प्रचारित करें (रोकें), एक त्रुटि ऑब्जेक्ट लौटाएं (ऑर्केस्ट्रेटर को निर्णय लेने दें), या संशोधित प्रॉम्प्ट के साथ पुनः प्रयास करें (सुचारू रूप से ठीक होता है, विलंबता बढ़ाता है)। अधिकांश उत्पादन प्रणालियों के लिए, संरचित त्रुटि ऑब्जेक्ट लौटाना और ऑर्केस्ट्रेटर को तय करने देना सही डिफ़ॉल्ट है।

ट्रेसिंग। एक मल्टी-एजेंट सिस्टम जहाँ आप नहीं देख सकते कि प्रत्येक एजेंट ने क्या किया, डीबग करना एक दुःस्वप्न है। प्रत्येक एजेंट कॉल को लॉग करें: इनपुट, आउटपुट, मॉडल, विलंबता और टोकन काउंट के साथ। पूर्ण निष्पादन पथ को पुनर्निर्माण करने में सक्षम होने के लिए प्रत्येक कॉल को एक ट्रेस ID के साथ टैग करें।

संदर्भ पासिंग। प्रत्येक एजेंट को कौन सा संदर्भ मिलता है, इसके बारे में सचेत रहें। प्रत्येक एजेंट को पूर्ण वार्तालाप इतिहास पास करना महंगा और अक्सर भ्रामक है — एजेंट असंबंधित पूर्व संदर्भ से विचलित हो जाते हैं। केवल वही पास करें जो प्रत्येक एजेंट को अपना विशिष्ट कार्य करने के लिए चाहिए।

आगे क्या बनाएं

यहाँ के पैटर्न नींव हैं। आप उन पर क्या बनाते हैं यह आपकी समस्या पर निर्भर करता है:

• Workers में उपकरण उपयोग जोड़ें — विशेष एजेंटों को APIs कॉल करने, डेटाबेस क्वेरी करने, या कोड चलाने दें
• ह्यूमन-इन-द-लूप चेकपॉइंट जोड़ें जहाँ ऑर्केस्ट्रेटर उच्च-जोखिम वाले कार्यों से पहले रुकता है
• एजेंट आउटपुट को एक वेक्टर स्टोर में पर्सिस्ट करके मेमोरी जोड़ें जिसे भविष्य के एजेंट क्वेरी कर सकते हैं
• उपयोगकर्ता को वापस करने से पहले आउटपुट को एक जज एजेंट के माध्यम से रूट करके मूल्यांकन जोड़ें

मल्टी-एजेंट सिस्टम वह जगह है जहाँ AI इंजीनियरिंग में अभी सबसे दिलचस्प काम हो रहा है। पैटर्न सरल हैं; निर्णय उन्हें आपकी विशिष्ट समस्या पर सही ढंग से लागू करने में है।

संबंधित लेख

• टूल उपयोग पैटर्न: विश्वसनीय एजेंट-टूल इंटरफेस बनाना
• एजेंट एरर रिकवरी: प्रोडक्शन विश्वसनीयता के लिए 5 पैटर्न
• एजेंट मेमोरी सिस्टम: अपनी AI को स्थायी संदर्भ देना
• स्वायत्त एजेंट सिस्टम में डिबगिंग और ऑब्जर्वेबिलिटी