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

हमारी पिछली पोस्ट ElevenLabs वॉइस ऑर्केस्ट्रेशन के साथ एक्सटर्नल एजेंट्स को इंटीग्रेट करना में, हमने बताया था कि टीमें अपने मौजूदा टेक्स्ट-आधारित एजेंट ऑर्केस्ट्रेशन को Custom LLM के ज़रिए ElevenLabs से कैसे कनेक्ट कर सकती हैं। उसी आधार पर, यह गाइड दिखाती है कि प्रमुख ओपन-सोर्स एजेंट फ्रेमवर्क्स को Custom LLM इंटरफ़ेस के पीछे कैसे अनुकूलित और डिप्लॉय किया जा सकता है। नतीजा एक लचीला आर्किटेक्चर है, जिसमें state management, टूल ऑर्केस्ट्रेशन या ऐप्लिकेशन-विशिष्ट नियंत्रण से समझौता किए बिना, परिपक्व एजेंट सिस्टम्स में वॉइस जोड़ी जाती है। फ्रेमवर्क कोई भी हो, हम एक ही तीन-स्टेप पैटर्न अपनाते हैं: generation request बनाना, अंतिम टेक्स्ट रिस्पॉन्स निकालना, और उसे OpenAI-संगत Server-Sent Events (SSE) फ़ॉर्मैट में बदलना। ElevenLabs Chat Completions और Responses दोनों फ़ॉर्मैट सपोर्ट करता है। यह गाइड चार व्यापक रूप से अपनाए गए फ्रेमवर्क्स को कवर करती है, लेकिन ये पैटर्न किसी भी ऐसे runtime पर लागू होते हैं जो OpenAI-संगत streaming output दे सकता है।

सामान्य सेटअप

इस सेक्शन के उदाहरण Python और FastAPI का उपयोग करते हैं, हालांकि HTTP POST requests और streaming SSE responses संभालने वाला कोई भी stack काम करेगा। जब ElevenLabs का वॉइस ऑर्केस्ट्रेशन संभावित turn end पहचानता है, तो यह कॉन्फ़िगर किए गए Custom LLM endpoint पर generation request भेजता है। यह सेक्शन उस translation layer के मुख्य हिस्सों की जानकारी देता है—वह bridge या proxy जो वॉइस ऑर्केस्ट्रेशन और एजेंट फ्रेमवर्क को एक ही भाषा में बात करने देता है।

स्वाभाविक रूप से, ग्राहक किसी फ्रेमवर्क को अपनी सामान्य परिचितता या किसी खास उद्देश्य को पूरा करने की क्षमता के आधार पर चुन सकते हैं। उदाहरण के लिए, LlamaIndex को मूल रूप से Retrieval-Augmented Generation (RAG) सेट अप करना आसान बनाने के लिए विकसित किया गया था, जबकि CrewAI को एजेंट्स के दौर में तय कार्यों को ऑटोमेट करने के लिए बनाया गया था। अलग-अलग डिज़ाइन लक्ष्य अलग-अलग response structures बनाते हैं, और हर एक को खास तरीके से संभालना पड़ता है। पूरा turn पूरा होने का इंतज़ार करने के बजाय LLM द्वारा जनरेट किए जाते ही chunks को stream करना ज़रूरी है, क्योंकि इससे Text-to-Speech (TTS) मॉडल पहले ही स्पीच जनरेट करना शुरू कर सकता है और महसूस होने वाली latency कम हो जाती है। हम चार लोकप्रिय फ्रेमवर्क्स—मुख्य रूप से LangGraph, Google ADK, CrewAI और LlamaIndex—पर ध्यान देते हैं।

शेयर किए गए कोड पर एक नोट

हर फ्रेमवर्क को OpenAI-संगत SSE chunks के रूप में responses stream करने होंगे। इन chunks को बनाने के लिए हम उदाहरणों में इस्तेमाल होने वाला एक छोटा helper function पेश करते हैं।

def sse_chunk(response_id: str, delta: dict, finish_reason=None) -> str:
    payload = {
        "id": response_id,
        "object": "chat.completion.chunk",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
    }
    return f"data: {json.dumps(payload)}\n\n"

आधार तैयार है, तो आइए LangGraph से शुरू करें। 

LangGraph

LangGraph एजेंट्स को graphs के रूप में मॉडल करता है, जहाँ nodes अलग-अलग steps को दर्शाते हैं और edges उनके बीच control flow तय करते हैं। न्यूनतम सेटअप सरल है: chat model initialize करें, agent tools तय करें, और agent graph runtime बनाएं।

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,
)

हर generation request के लिए LangGraph Agent को पूरी conversation history मिलती है, जिससे वह ज़रूरी state को अंदर ही बनाए रख सकता है। LangGraph server-side persistence को Checkpoints के ज़रिए सपोर्ट करता है, हालाँकि implementation को न्यूनतम रखने के लिए हमने उन्हें यहाँ कवर नहीं किया है।

state management संभालने के बाद, LangGraph से जुड़ा अगला निर्णय streaming mode है। LangGraph दो विकल्प देता है, जिनमें से हर एक अलग use case के लिए उपयुक्त है:

• stream_mode="values" graph state snapshots देता है। इसे लागू करना आसान है, लेकिन हर response में अधिक पूर्ण message state शामिल होती है, जिससे real-time conversational flows में latency बढ़ती है।
• stream_mode="messages" मॉडल से incremental message chunks stream करता है। यह आमतौर पर realtime voice interactions के लिए बेहतर होता है, क्योंकि इससे ElevenLabs ऑर्केस्ट्रेशन लेयर में time-to-first-audio कम होता है।

और खास तौर पर, agent loop के messages implementation में टूल कॉलिंग updates जैसे intermediate steps शामिल होते हैं, जिन्हें ज़ोर से नहीं बोलना चाहिए। proxy इन्हें फ़िल्टर कर देता है और TTS layer को सिर्फ़ यूज़र के लिए response text भेजता है। यहाँ tool-enabled turn का एक उदाहरण है।

[1] मॉडल टूल कॉल करने का निर्णय लेता है (tool_calls=["get_price"])
[2] टूल चलकर डेटा लौटाता है (result="$24.99") 
[3] मॉडल result का इस्तेमाल करके response बनाता है (content="इसकी कीमत $24.99 है") 

स्वाभाविक रूप से, SSE stream में सिर्फ़ step 3 के chunks आगे भेजे जाने चाहिए। व्यवहार में, streaming loop में दो guard checks यह filtering करते हैं: एक केवल langgraph_node == "model" events रखने के लिए, और दूसरा empty content को छोड़ने के लिए। साथ मिलकर, ये checks सुनिश्चित करते हैं कि SSE के रूप में केवल यूज़र के लिए assistant text ही ElevenLabs को भेजा जाए। इन अवधारणाओं को मिलाकर, हम request proxy का एक हल्का implementation देते हैं।

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    input = {"messages": req.messages}
    async def stream():
        response_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
        sent_role = False
        async for message_chunk, metadata in agent.astream(input, stream_mode="messages"):
            # Only forward model text chunks; skip tool updates and non-text events.
            if metadata.get("langgraph_node") != "model":
                continue
            content = getattr(message_chunk, "content", None)
            if not content:
                continue
            if not sent_role:
                yield sse_chunk(response_id, {"role": "assistant"})
                sent_role = True
            # Send incremental token-like chunks to ElevenLabs in OpenAI format.
            yield sse_chunk(response_id, {"content": content})
         # Signal natural completion before using the finish_reason: "stop" [DONE]
        yield sse_chunk(response_id, {}, finish_reason="stop")
        yield "data: [DONE]\n\n"
    return StreamingResponse(stream(), media_type="text/event-stream")

इससे केवल यूज़र के लिए model chunks ही ElevenLabs को भेजे जाते हैं। LangGraph अपने internal tool execution को state stream के ज़रिए दिखाता है, इसलिए filtering स्पष्ट रूप से होती है और proxy द्वारा नियंत्रित होती है। 

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

Google ADK

Google का ADK runtime loop को कुछ core primitives—Agent, Runner और SessionService—के पीछे abstract करता है। ADK का Runner HTTP layer और agent definition के बीच होता है। यह message routing, टूल ऑर्केस्ट्रेशन, session lifecycle और event streaming संभालता है। 

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
)

agent, session backend और runner initialize होने के बाद, proxy हर incoming request के लिए ADK session ढूँढता या बनाता है। ADK में session_id memory persistence को नियंत्रित करता है: turns के बीच एक ही session_id का दोबारा इस्तेमाल करने पर history, tool calls और पिछले responses अपने-आप आगे बने रहते हैं। चूँकि conversation identity ElevenLabs में upstream पर रहती है, proxy इस mapping को स्पष्ट रूप से संभालता है। generation request के लिए सही identifier पास करने पर SDK पिछले context को अंदर ही संभाल सकता है। हम conversation शुरू करते समय extra parameters के ज़रिए arbitrary identifier को request body में पास करते हैं।  

message और session तैयार होने पर runner को invoke किया जा सकता है। execution के दौरान tool calls और tool results अब भी internal ADK events के रूप में आते हैं, लेकिन उन्हें यूज़र के लिए output के बजाय intermediary orchestration steps माना जाता है। इससे उन फ्रेमवर्क्स की तुलना में manual filter की ज़रूरत नहीं रहती, जिनमें tool calls यूज़र को दिखने वाले text के रूप में आते हैं। 

नीचे दिया गया handler एक सरल implementation है, जिसमें session resolution और get-or-create logic inline शामिल है।

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest, request: Request):
    # In production, prefer a stable identifier from your upstream system.
    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
            # In SSE mode, ADK emits partial (incremental) and final (complete) events.
            # Forwarding only partial events avoids duplicating the full text.
            # Note: SSE streaming is experimental in ADK. For production, reconcile
            # both event types in case the model backend doesn't emit partials.
            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 देखते हैं, जो डिज़ाइन के हिसाब से ज़्यादा task-centric है।

CrewAI

CrewAI को open-ended dialogue loops के बजाय structured tasks (research, write, summarize) के इर्द-गिर्द multi-agent workflows orchestrate करने के लिए डिज़ाइन किया गया था। Agents को role, goal और backstory के साथ परिभाषित किया जाता है। execution, स्पष्ट description और expected output वाले Task objects पर केंद्रित होता है। 

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 में इस्तेमाल होने वाले agent-loop model के विपरीत, CrewAI आमतौर पर conversation के उस turn के work unit को तय करने के लिए हर request पर Task और Crew बनाता है। हम placeholder के ज़रिए पिछले turns को अगले task में डालकर conversational context आगे ले जाते हैं। {crew_chat_messages} variable को हर request पर चल रही conversation history से भरा जाता है, फिर execution के समय उसे task description में interpolate किया जाता है। हम intermediary tracing patterns (Thought, Action, Action Input, Observation) को स्पष्ट रूप से फ़िल्टर करके और सिर्फ़ final-answer text भेजकर साफ़, speech-ready text बनाने का भी लक्ष्य रखते हैं। 

नीचे दिया गया handler per-request task construction, history interpolation, Crew-level streaming, trace filtering और output formatting को एक साथ लाता है। 

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    # Task and Crew are assembled per request (not at startup).	
    task = Task(
        description=(
            "Conversation history:\n{crew_chat_messages}\n\n"
            "Respond to the user's latest message."
        ),
        expected_output=expected_output,
        agent=store_agent,
    )
    # stream=True returns CrewStreamingOutput instead of a single 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:
            # Skip non-text events (e.g. tool calls).
            if chunk.chunk_type != StreamChunkType.TEXT or not chunk.content:
                continue
            # Only forward text after the "Final Answer:" marker
            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
            # Clean up any trailing markdown artifacts from CrewAI output.
            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})
        # Fallback to handle short responses without the "Final Answer:" marker
        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 देखते हैं, जो native event-driven streaming model पर केंद्रित एक अलग रास्ता अपनाता है।

LlamaIndex

इस पोस्ट में शामिल अन्य फ्रेमवर्क्स के विपरीत, LlamaIndex को LLMs को external data sources (document stores, indexes, retrieval pipelines) से जोड़ने के लिए डिज़ाइन किया गया था। इसकी agent layer, FunctionAgent, open dialogue या task execution के बजाय structured context को retrieve करने और उस पर reasoning करने के लिए इसी आधार पर काम करती है।

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,
)

conversational continuity बनाए रखने के लिए proxy incoming messages को LlamaIndex chat messages में बदलता है, फिर उन्हें सबसे नए user turn (user_msg) और पिछले turns (chat_history) में बाँटता है। हर AgentStream event के event.delta field में अगला text fragment होता है, जो सीधे OpenAI-style delta.content chunk में मैप हो जाता है। non-empty deltas को जैसे हैं वैसे आगे भेजा जा सकता है, जिससे यह इस गाइड का सबसे सीधा streaming bridge बनता है। stream में orchestration events (tool calls, results) और speech events (assistant text deltas), दोनों होते हैं। voice output को साफ़ रखने के लिए proxy सिर्फ़ AgentStream events रखता है और empty deltas छोड़ देता है।

[1] AgentStream (delta='')       ← अनदेखा किया गया
[2] ToolCall                     ← अनदेखा किया गया
[3] ToolCallResult               ← अनदेखा किया गया
[4] AgentStream (delta='यह')     ← आगे भेजा गया ✓
[5] AgentStream (delta='कीमत है') ← आगे भेजा गया ✓
[6] AgentStream (delta=' $49.99')← आगे भेजा गया ✓

यह अलगाव intermediate tool mechanics को बोले जाने वाले output से बाहर रखता है और साथ ही कम-लेटेंसी वाली incremental speech बनाए रखता है। नीचे दिया गया drop-in handler इन steps को एक साथ लाता है।

@app.post("/chat/completions")
async def chat_completions(req: ChatCompletionRequest):
    # This assumes the last message is always a user turn with string content.
    # For production, add defensive role/content handling for non-text payloads.
    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, भारी built-in orchestration layers वाले फ्रेमवर्क्स की तुलना में end-to-end conversational runtime patterns के बारे में कम निर्देश देता है। production deployments के लिए, ग्राहकों को आमतौर पर session handling, response guardrails, टूल ऑर्केस्ट्रेशन और tracing लागू करने की ज़रूरत पड़ती है।

निष्कर्ष

इस गाइड का हर फ्रेमवर्क एक ही contract के ज़रिए ElevenLabs से जुड़ता है: OpenAI-style Completions या Responses request स्वीकार करें और SSE chunks वापस stream करें। इससे टीमें मौजूदा agent implementation में बहुत कम बदलाव करके उसके ऊपर voice orchestration जोड़ सकती हैं। इस तरह वे अपने पहले से बनाए हुए काम को बचाते हुए real-time कन्वर्सेशनल AI का लाभ उठा सकती हैं। यह modularity ElevenAgents platform का एक मुख्य सिद्धांत है। संगठन किसी मौजूदा agent को बढ़ा रहे हों या शुरुआत से voice-native बना रहे हों, ElevenAgents का voice orchestration उन्हें उनके मौजूदा स्तर पर ही सहयोग देने के लिए बनाया गया है।

अगर आप पहले से किसी open-source framework के साथ agent चला रहे हैं और voice enable करना चाहते हैं, तो इस तरीके को आज़माएं और हमें बताएं कि आपको कैसा लगा।