प्रैक्टिकल गाइड: ओपन-सोर्स एजेंट फ्रेमवर्क्स और ElevenAgents

हमारी पिछली पोस्ट में External Agents को ElevenLabs Voice Orchestration से जोड़ना, हमने बताया था कि टीमें अपने मौजूदा टेक्स्ट-बेस्ड एजेंट ऑर्केस्ट्रेशन को ElevenLabs से कैसे जोड़ सकती हैं कस्टम LLM के ज़रिए। उसी आधार पर, यह गाइड दिखाता है कि कैसे प्रमुख ओपन-सोर्स एजेंट फ्रेमवर्क्स को Custom LLM इंटरफेस के पीछे एडॉप्ट और डिप्लॉय किया जा सकता है। इसका नतीजा है एक फ्लेक्सिबल आर्किटेक्चर, जिसमें वॉइस को मैच्योर एजेंट सिस्टम्स पर लेयर किया जा सकता है, बिना स्टेट मैनेजमेंट, टूल ऑर्केस्ट्रेशन या एप्लिकेशन-स्पेसिफिक कंट्रोल से समझौता किए। हर फ्रेमवर्क में हम एक ही तीन-स्टेप पैटर्न फॉलो करते हैं: जनरेशन रिक्वेस्ट बनाएं, फाइनल टेक्स्ट रिस्पॉन्स निकालें, और उसे OpenAI-कम्पैटिबल Server-Sent Events (SSE) फॉर्मेट में रीफॉर्मेट करें। ElevenLabs दोनों चैट कम्प्लीशनऔर जवाब फॉर्मेट्स सपोर्ट करता है। यह गाइड चार पॉपुलर फ्रेमवर्क्स को कवर करता है, लेकिन ये पैटर्न किसी भी ऐसे रनटाइम पर लागू किए जा सकते हैं जो OpenAI-कम्पैटिबल स्ट्रीमिंग आउटपुट दे सकता है।

जनरल सेटअप

इस सेक्शन के उदाहरण Python और FastAPI का इस्तेमाल करते हैं, लेकिन कोई भी स्टैक जो HTTP POST रिक्वेस्ट और स्ट्रीमिंग SSE रिस्पॉन्स हैंडल कर सकता है, चलेगा। जब ElevenLabs की वॉइस ऑर्केस्ट्रेशन को टर्न एंड का अंदेशा होता है, तो वह कन्फ़िगर किए गए Custom LLM एंडपॉइंट पर जनरेशन रिक्वेस्ट भेजता है। यह सेक्शन उस ट्रांसलेशन लेयर के कोर कंपोनेंट्स को समझाता है, यानी वह ब्रिज या प्रॉक्सी जो वॉइस ऑर्केस्ट्रेशन और एजेंट फ्रेमवर्क को एक ही भाषा में बात करने देता है।

स्वाभाविक है कि हर फ्रेमवर्क को ग्राहक अपनी जरूरत या परिचिती के हिसाब से चुनते हैं। जैसे LlamaIndex को शुरू में Retrieval-Augmented Generation (RAG) को आसान बनाने के लिए बनाया गया था, वहीं CrewAI को एजेंट्स के दौर में डिफाइंड टास्क्स को ऑटोमेट करने के लिए। अलग-अलग डिज़ाइन गोल्स अलग रिस्पॉन्स स्ट्रक्चर बनाते हैं, और हर एक के लिए अलग हैंडलिंग चाहिए। जैसे-जैसे LLM जनरेट करता है, वैसे-वैसे चंक्स स्ट्रीम करना जरूरी है, ताकि टेक्स्ट टू स्पीच (TTS) मॉडल जल्दी बोलना शुरू कर सके और लेटेंसी कम हो। हम चार पॉपुलर फ्रेमवर्क्स पर फोकस करते हैं: LangGraph, Google ADK, CrewAI और LlamaIndex।

शेयर्ड कोड पर एक नोट

हर फ्रेमवर्क को OpenAI-कम्पैटिबल SSE चंक्स के रूप में रिस्पॉन्स स्ट्रीम करना होता है। हम एक छोटा हेल्पर फंक्शन देते हैं, जो सभी उदाहरणों में इन चंक्स को बनाने के लिए इस्तेमाल होता है।

def sse_chunk(response_id: str, delta: dict, finish_reason=None) -> str:

    payload = {

अब जब बेस तैयार है, तो चलिए LangGraph से शुरू करते हैं।

LangGraph

LangGraph एजेंट्स को ग्राफ़ के रूप में मॉडल करता है, जहाँ नोड्स हर स्टेप को दिखाते हैं और एजेस उनके बीच कंट्रोल फ्लो तय करते हैं। बेसिक सेटअप आसान है: एक चैट मॉडल इनिशियलाइज़ करें, एजेंट टूल्स सेट करें, और एजेंट ग्राफ़ रनटाइम बनाएं।

    }

    return f"data: {json.dumps(payload)}\n\n"

हर जनरेशन रिक्वेस्ट पर, LangGraph एजेंट को पूरी बातचीत का इतिहास मिलता है, जिससे वह ज़रूरी स्टेट को अपने अंदर बनाए रख सकता है। LangGraph सर्वर-साइड परसिस्टेंस को सपोर्ट करता है

स्टेट मैनेजमेंट हो जाने के बाद, अगला LangGraph-स्पेसिफिक डिसीजन पॉइंट है स्ट्रीमिंग मोड, जिसमें LangGraph दो ऑप्शन देता है, जो अलग-अलग यूज़ केस के लिए बने हैं:

LangGraph एजेंट्स को ग्राफ के रूप में मॉडल करता है, जहां नोड्स इंडिविजुअल स्टेप्स होते हैं और एजेस उनके बीच कंट्रोल फ्लो तय करते हैं। इसका मिनिमल सेटअप सीधा है: एक चैट मॉडल इनिशियलाइज़ करें, एजेंट टूल्स डिफाइन करें, और एजेंट ग्राफ रनटाइम बनाएं।

from langchain.agents import create_agent

from langchain_openai import ChatOpenAI

from langchain_core.tools import tool

llm = ChatOpenAI(

    model=model_id,

    api_key=os.getenv("OPENAI_API_KEY"),

)

agent = create_agent(

    llm,
    tools=tool_list,
    system_prompt=system_prompt,
)

हर जनरेशन रिक्वेस्ट पर, LangGraph Agent को पूरी बातचीत का इतिहास मिलता है, जिससे वह जरूरी स्टेट को इंटरनली मेंटेन कर सकता है। LangGraph सर्वर-साइड परसिस्टेंस के लिए चेकपॉइंट्स भी सपोर्ट करता है, लेकिन यहां हम इसे सिंपल रखने के लिए कवर नहीं कर रहे।

स्टेट मैनेजमेंट हो जाने के बाद, अगला LangGraph-स्पेसिफिक डिसीजन पॉइंट है स्ट्रीमिंग मोड, जिसमें LangGraph दो ऑप्शन देता है, हर एक अलग यूज़ केस के लिए:

• stream_mode="values" ग्राफ स्टेट के स्नैपशॉट्स देता है। इसे इम्प्लीमेंट करना आसान है, लेकिन हर रिस्पॉन्स में पूरा मैसेज स्टेट आता है, जिससे रियल-टाइम कन्वर्सेशन में लेटेंसी बढ़ती है।
• एक्स्ट्रा पैरामीटर्स

मैसेज और सेशन तैयार होने के बाद, रनर को कॉल किया जा सकता है। टूल कॉल्स और टूल रिजल्ट्स अब भी एक्जीक्यूशन के दौरान इंटरनल ADK इवेंट्स के रूप में दिखते हैं, लेकिन इन्हें यूज़र के लिए आउटपुट की बजाय इंटरमीडिएट ऑर्केस्ट्रेशन स्टेप्स माना जाता है। इससे उन फ्रेमवर्क्स के मुकाबले मैन्युअल फ़िल्टर की ज़रूरत नहीं पड़ती, जहाँ टूल कॉल्स यूज़र-विज़िबल टेक्स्ट के रूप में आते हैं।

नीचे दिया गया हैंडलर एक सिंपल इम्प्लीमेंटेशन है जिसमें सेशन रिज़ॉल्यूशन और गेट-ऑर-क्रिएट लॉजिक इनलाइन है।

[2] टूल एक्सीक्यूट होता है और डेटा लौटाता है (result="$24.99")

[3] मॉडल रिजल्ट का इस्तेमाल कर रिस्पॉन्स बनाता है (content="It costs $24.99")

अब हम CrewAI देखते हैं, जो डिज़ाइन से ही ज़्यादा टास्क-केंद्रित है।

CrewAI

CrewAI को मल्टी-एजेंट वर्कफ़्लो को स्ट्रक्चर्ड टास्क्स (जैसे रिसर्च, लिखना, समरी बनाना) के इर्द-गिर्द ऑर्केस्ट्रेट करने के लिए बनाया गया है, न कि ओपन-एंडेड डायलॉग लूप्स के लिए। एजेंट्स को रोल, गोल और बैकस्टोरी के साथ डिफाइन किया जाता है। एक्जीक्यूशन Task ऑब्जेक्ट्स के इर्द-गिर्द होता है, जिनमें क्लियर डिस्क्रिप्शन और एक्सपेक्टेड आउटपुट होता है।

    input = {"messages": req.messages}

    async def stream():

LangGraph और ADK के एजेंट-लूप मॉडल के उलट, CrewAI आमतौर पर हर रिक्वेस्ट पर Task और Crew बनाता है, ताकि उस टर्न के लिए काम की यूनिट तय हो सके। हम बातचीत का कॉन्टेक्स्ट आगे बढ़ाते हैं, पिछले टर्न्स को अगले टास्क में एक प्लेसहोल्डर के ज़रिए डालकर। {crew_chat_messages} वेरिएबल हर रिक्वेस्ट पर चल रही बातचीत के हिस्ट्री से भरा जाता है, फिर एक्जीक्यूशन के समय टास्क डिस्क्रिप्शन में इंटरपोलट किया जाता है। हम इंटरमीडिएट ट्रेसिंग पैटर्न्स (Thought, Action, Action Input, Observation) को साफ़ तौर पर फ़िल्टर करके सिर्फ फाइनल-आंसर टेक्स्ट आउटपुट करते हैं, ताकि टेक्स्ट बोलने के लिए तैयार रहे।

नीचे दिया गया हैंडलर हर रिक्वेस्ट पर टास्क बनाना, हिस्ट्री इंटरपोलट करना, Crew-लेवल स्ट्रीमिंग, ट्रेस फ़िल्टरिंग और आउटपुट फॉर्मेटिंग—all एक साथ लाता है।

        async for message_chunk, metadata in agent.astream(input, stream_mode="messages"):

            # सिर्फ मॉडल टेक्स्ट चंक्स फॉरवर्ड करें; टूल अपडेट्स और नॉन-टेक्स्ट इवेंट्स स्किप करें।

अब हम LlamaIndex देखते हैं, जो एक अलग रास्ता अपनाता है और नेटिव इवेंट-ड्रिवन स्ट्रीमिंग मॉडल पर फोकस करता है।

LlamaIndex

इस पोस्ट में बताए गए बाकी फ्रेमवर्क्स से अलग, LlamaIndex को LLMs को बाहरी डेटा सोर्सेज (डॉक्युमेंट स्टोर्स, इंडेक्स, रिट्रीवल पाइपलाइन्स) से जोड़ने के लिए डिज़ाइन किया गया है। इसका एजेंट लेयर, FunctionAgent, इसी बेस पर बना है ताकि स्ट्रक्चर्ड कॉन्टेक्स्ट को रिट्रीव और रीजन किया जा सके, न कि ओपन-डायलॉग या टास्क एक्जीक्यूशन के लिए।

            if not content:

                continue

बातचीत की निरंतरता बनाए रखने के लिए, प्रॉक्सी इनकमिंग मैसेजेस को LlamaIndex चैट मैसेजेस में बदलता है, फिर उन्हें लेटेस्ट यूज़र टर्न (user_msg) और पुराने टर्न्स (chat_history) में बांटता है। हर AgentStream इवेंट का event.delta फील्ड अगला टेक्स्ट फ्रैगमेंट रखता है, जो सीधे OpenAI-स्टाइल delta.content चंक में मैप होता है। नॉन-एम्प्टी डेल्टास को वैसे ही फॉरवर्ड किया जा सकता है, जिससे ये गाइड का सबसे सिंपल स्ट्रीमिंग ब्रिज बन जाता है। स्ट्रीम में ऑर्केस्ट्रेशन इवेंट्स (टूल कॉल्स, रिजल्ट्स) और स्पीच इवेंट्स (असिस्टेंट टेक्स्ट डेल्टास) दोनों होते हैं। वॉइस आउटपुट क्लीन रखने के लिए, प्रॉक्सी सिर्फ AgentStream इवेंट्स रखता है और खाली डेल्टास को स्किप करता है।

कन्वर्सेशनल कंटिन्युटी बनाए रखने के लिए, प्रॉक्सी इनकमिंग मैसेज को LlamaIndex चैट मैसेज में बदलता है, फिर उन्हें लेटेस्ट यूज़र टर्न (user_msg) और पिछले टर्न्स (chat_history) में बांटता है। हर AgentStream इवेंट के event.delta फील्ड में अगला टेक्स्ट फ्रैगमेंट होता है, जो सीधे OpenAI-स्टाइल delta.content चंक में मैप होता है। नॉन-एम्प्टी डेल्टा को वैसे ही फॉरवर्ड किया जा सकता है, जिससे यह गाइड का सबसे सीधा स्ट्रीमिंग ब्रिज बन जाता है। स्ट्रीम में ऑर्केस्ट्रेशन इवेंट्स (टूल कॉल्स, रिजल्ट्स) और स्पीच इवेंट्स (असिस्टेंट टेक्स्ट डेल्टा) दोनों होते हैं। वॉइस आउटपुट साफ रखने के लिए, प्रॉक्सी सिर्फ AgentStream इवेंट्स रखता है और खाली डेल्टा को स्किप करता है।

[1] AgentStream (delta='')       ← नजरअंदाज

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

            yield sse_chunk(response_id, {"content": content})

LlamaIndex बाकी फ्रेमवर्क्स की तरह एंड-टू-एंड कन्वर्सेशनल रनटाइम पैटर्न्स पर ज़्यादा ज़ोर नहीं देता। प्रोडक्शन डिप्लॉयमेंट्स में आमतौर पर कस्टमर्स को सेशन हैंडलिंग, रिस्पॉन्स गार्डरेल्स, टूल ऑर्केस्ट्रेशन और ट्रेसिंग खुद इम्प्लीमेंट करनी पड़ती है।

LlamaIndex में एंड-टू-एंड कन्वर्सेशनल रनटाइम पैटर्न्स को लेकर उतनी सख्ती नहीं है, जितनी उन फ्रेमवर्क्स में जिनमें भारी ऑर्केस्ट्रेशन लेयर होती है। प्रोडक्शन में आमतौर पर कस्टमर्स को खुद सेशन हैंडलिंग, रिस्पॉन्स गार्डरेल्स, टूल ऑर्केस्ट्रेशन और ट्रेसिंग इम्प्लीमेंट करनी पड़ती है।

निष्कर्ष

इस गाइड में हर फ्रेमवर्क ElevenLabs से एक ही कॉन्ट्रैक्ट के ज़रिए जुड़ता है: OpenAI-स्टाइल Completions या Responses रिक्वेस्ट स्वीकार करें और SSE चंक्स स्ट्रीम करें। इससे टीमें मौजूदा एजेंट इम्प्लीमेंटेशन पर वॉइस ऑर्केस्ट्रेशन आसानी से जोड़ सकती हैं, जिससे पहले से बना सिस्टम भी बना रहता है और रियल-टाइम

अगर आप पहले से किसी ओपन-सोर्स फ्रेमवर्क के साथ एजेंट चला रहे हैं और उसमें वॉइस जोड़ना चाहते हैं, तो यह तरीका आज़माएं और हमें बताएं कि आपको कैसा लगा।

अब हम Google के Agent Development Kit (ADK) के साथ काम करने की बारीकियों में उतरते हैं

Google ADK

Google का ADK रनटाइम लूप को कुछ कोर प्रिमिटिव्स के पीछे एब्स्ट्रैक्ट करता है: Agent, Runner, और SessionService। ADK का Runner HTTP लेयर और एजेंट डिफिनिशन के बीच बैठता है। यह मैसेज रूटिंग, टूल ऑर्केस्ट्रेशन, सेशन लाइफसाइकल और इवेंट स्ट्रीमिंग हैंडल करता है।

from google.adk.agents import Agent

from google.adk.runners import Runner

from google.adk.agents.run_config import RunConfig, StreamingMode

from google.adk.sessions import InMemorySessionService

from google.genai import types as genai_types

agent = Agent(

    name=name,

    model=model,

    instruction=instruction,

    tools=[tool_list],

)

session_service = InMemorySessionService()

runner = Runner(

agent=agent,

app_name=app_name,

session_service=session_service

)

एजेंट, सेशन बैकएंड और रनर इनिशियलाइज़ होने के बाद, प्रॉक्सी हर इनकमिंग रिक्वेस्ट के लिए ADK सेशन को रिजॉल्व या क्रिएट करता है। ADK में session_id मेमोरी परसिस्टेंस कंट्रोल करता है: एक ही session_id को बार-बार यूज़ करने से हिस्ट्री, टूल कॉल्स और पुराने रिस्पॉन्स अपने आप आगे बढ़ते हैं। क्योंकि कन्वर्सेशन आइडेंटिटी ElevenLabs में रहती है, प्रॉक्सी यह मैपिंग एक्सप्लिसिटली करता है। सही आइडेंटिफायर जनरेशन रिक्वेस्ट में पास करने से SDK पुराने कॉन्टेक्स्ट को इंटरनली हैंडल कर लेता है। हम कन्वर्सेशन शुरू करते समय अतिरिक्त पैरामीटर को रिक्वेस्ट बॉडी में पास करते हैं।

मैसेज और सेशन तैयार होने के बाद, रनर को इनवोक किया जा सकता है। टूल कॉल्स और टूल रिजल्ट्स अब भी ADK के इंटरनल इवेंट्स के रूप में दिखते हैं, लेकिन इन्हें यूज़र के लिए आउटपुट की बजाय इंटरमीडिएट ऑर्केस्ट्रेशन स्टेप्स माना जाता है। इससे उन फ्रेमवर्क्स के मुकाबले मैन्युअल फिल्टर की जरूरत नहीं रहती, जहां टूल कॉल्स यूज़र-विजिबल टेक्स्ट के रूप में आते हैं।

नीचे दिया हैंडलर सेशन रिजॉल्यूशन और गेट-ऑर-क्रिएट लॉजिक को सिंपल तरीके से दिखाता है।

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest, request: Request):

    # प्रोडक्शन में, अपने अपस्ट्रीम सिस्टम से स्टेबल आइडेंटिफायर यूज़ करें।

    session_id = req.elevenlabs_extra_body.arbitrary_identifier

    session = await session_service.get_session(

        app_name="elevenlabs", user_id="user", session_id=session_id

    )

    if not session:

        session = await session_service.create_session(

            app_name="elevenlabs", user_id="user", session_id=session_id

        )

    user_text = next((m["content"] for m in reversed(req.messages) if m["role"] == "user"), "")

    content = genai_types.Content(role="user", parts=[genai_types.Part(text=user_text)])

   async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        sent_role = False

        async for event in runner.run_async(

            user_id="user",

            session_id=session.id,

            new_message=content,

            run_config=RunConfig(streaming_mode=StreamingMode.SSE),

        ):

            if not event.content or not event.content.parts:

                continue

            # SSE मोड में, ADK पार्टियल (इनक्रिमेंटल) और फाइनल (कम्प्लीट) इवेंट्स एमिट करता है।

            # सिर्फ पार्टियल इवेंट्स फॉरवर्ड करने से फुल टेक्स्ट डुप्लिकेट नहीं होता।
            # नोट: SSE स्ट्रीमिंग ADK में एक्सपेरिमेंटल है। प्रोडक्शन में,

            # दोनों इवेंट टाइप्स को हैंडल करें, अगर मॉडल बैकएंड पार्टियल्स नहीं भेजता।

            if not getattr(event, "partial", False):

                continue

            text = "".join((getattr(p, "text", "") or "") for p in event.content.parts)

            if not text:

                continue

            if not sent_role:

                yield sse_chunk(response_id, {"role": "assistant"})

                sent_role = True

            yield sse_chunk(response_id, {"content": text})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

अब हम CrewAI देखते हैं, जो डिज़ाइन में ज्यादा टास्क-केंद्रित है।

CrewAI

CrewAI को स्ट्रक्चर्ड टास्क्स (जैसे रिसर्च, लिखना, समरी बनाना) के इर्द-गिर्द मल्टी-एजेंट वर्कफ़्लो ऑर्केस्ट्रेट करने के लिए डिज़ाइन किया गया था, न कि ओपन-एंडेड डायलॉग लूप्स के लिए। एजेंट्स को एक रोल, गोल और बैकस्टोरी के साथ डिफाइन किया जाता है। एक्सीक्यूशन Task ऑब्जेक्ट्स के इर्द-गिर्द होता है, जिनमें क्लियर डिस्क्रिप्शन और अपेक्षित आउटपुट होता है।

from crewai import Agent, Task, Crew, Process, LLM

from crewai.tools import tool

from crewai.types.streaming import StreamChunkType

llm = LLM(

    model=model_id,

    api_key=os.getenv("OPENAI_API_KEY")

)

store_agent = Agent(

    role=role,

    goal=goal,

    backstory=backstory,

    tools=tools,

    llm=llm,

    verbose=False,

)

LangGraph और ADK के एजेंट-लूप मॉडल के उलट, CrewAI आमतौर पर हर रिक्वेस्ट पर Task और Crew बनाता है, ताकि उस टर्न के लिए वर्क यूनिट डिफाइन हो सके। हम कन्वर्सेशनल कॉन्टेक्स्ट को अगले टास्क में पिछले टर्न्स को एक प्लेसहोल्डर के ज़रिए डालकर आगे बढ़ाते हैं। {crew_chat_messages} वेरिएबल हर रिक्वेस्ट पर चलती बातचीत के हिस्ट्री से भरा जाता है, फिर टास्क डिस्क्रिप्शन में इंटरपोलेट किया जाता है। हम आउटपुट को क्लीन और स्पीच-रेडी बनाने के लिए इंटरमीडिएट ट्रेसिंग पैटर्न्स (Thought, Action, Action Input, Observation) को फिल्टर करते हैं और सिर्फ फाइनल-आंसर टेक्स्ट ही भेजते हैं।

नीचे दिया हैंडलर हर रिक्वेस्ट पर टास्क कंस्ट्रक्शन, हिस्ट्री इंटरपोलेशन, Crew-लेवल स्ट्रीमिंग, ट्रेस फिल्टरिंग और आउटपुट फॉर्मेटिंग को जोड़ता है।

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest):

    # Task और Crew हर रिक्वेस्ट पर बनाए जाते हैं (स्टार्टअप पर नहीं)।

    task = Task(

        description=(

            "कन्वर्सेशन हिस्ट्री:\n{crew_chat_messages}\n\n"

            "यूज़र के लेटेस्ट मैसेज का जवाब दें।"

        ),

        expected_output=expected_output,

        agent=store_agent,

    )

    # stream=True से CrewStreamingOutput मिलता है, न कि एक CrewOutput।

    crew = Crew(

        agents=[store_agent],

        tasks=[task],

        process=Process.sequential,

        verbose=False,

        stream=True,

    )

    async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        sent_role = False

        final_marker = "final answer:"

        marker_buffer = ""

        marker_found = False

        emitted_any_content = False

        streaming = await crew.kickoff_async(

            inputs={"crew_chat_messages": json.dumps(req.messages)}

        )

       async for chunk in streaming:

            # नॉन-टेक्स्ट इवेंट्स (जैसे टूल कॉल्स) स्किप करें।

            if chunk.chunk_type != StreamChunkType.TEXT or not chunk.content:

                continue

            # सिर्फ "Final Answer:" मार्कर के बाद का टेक्स्ट फॉरवर्ड करें

            if not marker_found:

                marker_buffer += chunk.content

                idx = marker_buffer.lower().find("final answer:")

                if idx == -1:

                    continue

                marker_found = True

                content = marker_buffer[idx + 13:].lstrip()

                marker_buffer = ""

            else:

                content = chunk.content

            # CrewAI आउटपुट से कोई भी ट्रेलिंग मार्कडाउन आर्टिफैक्ट्स हटा दें।

            content = content.rstrip("`").rstrip()

            if not content:

                continue

            if not sent_role:

                yield sse_chunk(response_id, {"role": "assistant"})

                sent_role = True

            yield sse_chunk(response_id, {"content": content})

        # शॉर्ट रिस्पॉन्स के लिए फॉलबैक, जिसमें "Final Answer:" मार्कर नहीं है

        if not sent_role:

            raw = getattr(streaming, "result", None)

            fallback = (raw.raw if raw else marker_buffer).strip().rstrip("`").rstrip()

            if fallback:

                yield sse_chunk(response_id, {"role": "assistant"})

                yield sse_chunk(response_id, {"content": fallback})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

अब हम LlamaIndex देखते हैं, जो एक अलग रास्ता अपनाता है और नेटिव इवेंट-ड्रिवन स्ट्रीमिंग मॉडल पर फोकस करता है।

LlamaIndex

इस पोस्ट में कवर किए गए बाकी फ्रेमवर्क्स के उलट, LlamaIndex को LLMs को एक्सटर्नल डेटा सोर्सेज (डॉक्युमेंट स्टोर्स, इंडेक्स, रिट्रीवल पाइपलाइन्स) से जोड़ने के लिए डिज़ाइन किया गया था। इसका एजेंट लेयर, FunctionAgent, इसी आधार पर स्ट्रक्चर्ड कॉन्टेक्स्ट को रिट्रीव और रीजन करने के लिए बैठता है, न कि ओपन-डायलॉग या टास्क एक्सीक्यूशन के लिए।

from llama_index.llms.openai import OpenAI

from llama_index.core.agent.workflow import FunctionAgent, AgentStream

from llama_index.core.base.llms.types import ChatMessage, MessageRole

llm = OpenAI(

    model=model,

    api_key=os.getenv("OPENAI_API_KEY")

)

agent = FunctionAgent(

    tools=[list_inventory, get_item_price],

    llm=llm,

    system_prompt=system_prompt,

)

कन्वर्सेशनल कंटीन्यूटी बनाए रखने के लिए, प्रॉक्सी इनकमिंग मैसेजेस को LlamaIndex चैट मैसेजेस में बदलता है, फिर उन्हें लेटेस्ट यूज़र टर्न (user_msg) और पुराने टर्न्स (chat_history) में बांटता है। हर AgentStream इवेंट का event.delta फील्ड अगला टेक्स्ट फ्रैगमेंट देता है, जिसे सीधे OpenAI-स्टाइल delta.content चंक में मैप किया जा सकता है। नॉन-एम्प्टी डेल्टाज को ऐसे ही फॉरवर्ड किया जा सकता है, जिससे यह गाइड का सबसे सीधा स्ट्रीमिंग ब्रिज बन जाता है। स्ट्रीम में ऑर्केस्ट्रेशन इवेंट्स (टूल कॉल्स, रिजल्ट्स) और स्पीच इवेंट्स (असिस्टेंट टेक्स्ट डेल्टाज) दोनों होते हैं। वॉइस आउटपुट क्लीन रखने के लिए, प्रॉक्सी सिर्फ AgentStream इवेंट्स रखता है और खाली डेल्टाज को स्किप करता है।

[1] AgentStream (delta='')       ← इग्नोर किया गया

[2] ToolCall                     ← इग्नोर किया गया

[3] ToolCallResult               ← इग्नोर किया गया

[4] AgentStream (delta='It')     ← फॉरवर्ड ✓

[5] AgentStream (delta=' costs') ← फॉरवर्ड ✓

[6] AgentStream (delta=' $49.99')← फॉरवर्ड ✓

यह सेपरेशन इंटरमीडिएट टूल मैकेनिक्स को बोले गए आउटपुट से बाहर रखता है और लो-लेटेंसी इनक्रिमेंटल स्पीच बनाए रखता है। नीचे दिया ड्रॉप-इन हैंडलर इन स्टेप्स को एक साथ लाता है।

@app.post("/chat/completions")

async def chat_completions(req: ChatCompletionRequest):

    # यह मानता है कि आखिरी मैसेज हमेशा यूज़र टर्न और स्ट्रिंग कंटेंट वाला है।

    # प्रोडक्शन में, नॉन-टेक्स्ट पेलोड्स के लिए रोल/कंटेंट हैंडलिंग जोड़ें।

    chat_history = [

        ChatMessage(role=MessageRole(m["role"]), content=m.get("content") or "")

        for m in req.messages

    ]

    user_text = chat_history.pop().content

    async def stream():

        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"

        handler = agent.run(user_msg=user_text, chat_history=chat_history)

        async for event in handler.stream_events():

            if not isinstance(event, AgentStream):

                continue

            if not event.delta:

                continue

            yield sse_chunk(response_id, {"content": event.delta})

        yield sse_chunk(response_id, {}, finish_reason="stop")

        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

LlamaIndex बाकी फ्रेमवर्क्स की तरह एंड-टू-एंड कन्वर्सेशनल रनटाइम पैटर्न्स पर जोर नहीं देता। प्रोडक्शन डिप्लॉयमेंट्स में आमतौर पर ग्राहकों को सेशन हैंडलिंग, रिस्पॉन्स गार्डरेल्स, टूल ऑर्केस्ट्रेशन और ट्रेसिंग खुद इम्प्लीमेंट करनी पड़ती है।

निष्कर्ष

इस गाइड के हर फ्रेमवर्क को ElevenLabs से एक ही कॉन्ट्रैक्ट के ज़रिए जोड़ा जाता है: OpenAI-स्टाइल Completions या Responses रिक्वेस्ट एक्सेप्ट करें और SSE चंक्स स्ट्रीम करें। इससे टीमें अपनी मौजूदा एजेंट इम्प्लीमेंटेशन पर वॉइस ऑर्केस्ट्रेशन आसानी से जोड़ सकती हैं, जिससे पहले से बना सिस्टम भी बना रहता है और रियल-टाइम कन्वर्सेशनल AI की ताकत भी मिलती है। यही मॉड्युलैरिटी ElevenAgents प्लेटफॉर्म की खासियत है। चाहे आप मौजूदा एजेंट को बढ़ा रहे हों या शुरुआत से वॉइस-नेटिव बना रहे हों, ElevenAgent की वॉइस ऑर्केस्ट्रेशन आपकी जरूरत के हिसाब से बनी है।

अगर आप पहले से किसी ओपन-सोर्स फ्रेमवर्क के साथ एजेंट चला रहे हैं और उसमें वॉइस जोड़ना चाहते हैं, तो यह तरीका आज़माएं और हमें बताएं कि आपको कैसा लगा।