Praktisk guide: open source-agentramverk och ElevenAgents

I vårt tidigare inlägg om Integrering av externa agenter med ElevenLabs Voice Orchestration beskrev vi hur team kan koppla sin befintliga textbaserade agentorkestrering till ElevenLabs via Custom LLM. Med den grunden visar den här guiden hur ledande ramverk för agenter med öppen källkod kan anpassas och distribueras bakom Custom LLM-gränssnittet. Resultatet är en flexibel arkitektur där röst läggs ovanpå etablerade agentsystem utan att kompromissa med tillståndshantering, verktygsorkestrering eller applikationsspecifik kontroll. Oavsett ramverk följer vi samma trestegsmönster: skapa en genereringsbegäran, extrahera det slutliga textsvar och formatera om det i ett OpenAI-kompatibelt Server-Sent Events-format (SSE). ElevenLabs har stöd för både Chat Completions och Responses-formaten. Även om den här guiden täcker fyra välanvända ramverk kan mönstren tillämpas på alla runtime-miljöer som kan producera OpenAI-kompatibel strömmande utdata.

Allmän konfiguration

Exemplen i det här avsnittet använder Python och FastAPI, men vilken stack som helst som hanterar HTTP POST-begäranden och strömmande SSE-svar fungerar. När ElevenLabs röstorkestrering upptäcker ett troligt turavslut skickar den en genereringsbegäran till den konfigurerade Custom LLM-slutpunkten. I det här avsnittet går vi igenom huvudkomponenterna i det översättningslager – bryggan eller proxyn som får röstorkestreringen och agentramverket att tala samma språk.

Det är förståeligt att kunder väljer ramverk utifrån allmän kännedom eller dess förmåga att fylla ett specifikt syfte. LlamaIndex utvecklades till exempel ursprungligen för att förenkla konfigurationen av Retrieval-Augmented Generation (RAG), medan CrewAI byggdes för att automatisera definierade uppgifter i agenternas era. Olika designmål ger olika svarsstrukturer, och varje struktur kräver särskild hantering. Att strömma delar medan LLM:en genererar dem, i stället för att vänta på en komplett tur, är avgörande eftersom Text-to-Speech- (TTS) modellen då kan börja generera tal tidigare och därmed minska den upplevda latensen. Vi fokuserar på de fyra populära ramverken LangGraph, Google ADK, CrewAI och LlamaIndex.

En kommentar om delad kod

Varje ramverk måste strömma svar som OpenAI-kompatibla SSE-delar. Vi introducerar en liten hjälpfunktion som används i alla exempel för att skapa dessa delar.

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"

Med den grunden på plats börjar vi med LangGraph. 

LangGraph

LangGraph modellerar agenter som grafer, där noder representerar enskilda steg och kanter definierar kontrollflödet mellan dem. Den minsta konfigurationen är enkel: initiera en chattmodell, definiera agentverktyg och skapa runtime-miljön för agentgrafen.

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

För varje genereringsbegäran tar LangGraph Agent emot hela konversationshistoriken, vilket gör att den kan behålla det nödvändiga tillståndet internt. LangGraph stöder beständig lagring på serversidan via Checkpoints, men vi tar inte upp dem här för att hålla implementationen så enkel som möjligt.

När tillståndshanteringen är löst är nästa LangGraph-specifika beslut strömningsläget. LangGraph erbjuder två alternativ, vart och ett lämpat för olika användningsfall:

• stream_mode="values" ger ögonblicksbilder av graftillståndet. Det är enklare att implementera, men innehåller ett mer komplett meddelandetillstånd i varje svar, vilket ökar latensen i konversationsflöden i realtid.
• stream_mode="messages" strömmar stegvisa meddelandedelar från modellen. Detta föredras generellt för röstinteraktioner i realtid, eftersom det minskar tiden till första ljud i ElevenLabs orkestreringslager.

Mer specifikt innehåller meddelandeimplementeringen av agentloopen mellansteg som verktygsanrop-uppdateringar, som inte ska läsas upp. Proxyn filtrerar bort dessa och skickar bara användarriktad svarstext till TTS-lagret. Här är ett exempel på en tur med verktygsstöd.

[1] Modellen väljer att anropa ett verktyg (tool_calls=["get_price"])
[2] Verktyget körs och returnerar data (result="$24.99") 
[3] Modellen producerar ett svar med resultatet (content="Det kostar $24.99") 

Naturligtvis ska endast delarna från steg 3 skickas vidare i SSE-strömmen. I praktiken hanterar två kontrollvillkor filtreringen i strömningsloopen: ett som endast behåller händelser där langgraph_node == "model", och ett som hoppar över tomt innehåll. Tillsammans säkerställer dessa kontroller att endast användarriktad assistenttext skickas vidare till ElevenLabs som SSE. Med dessa koncept på plats visar vi en lättviktig implementation av begärandeproxyn.

@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")

Detta säkerställer att endast användarriktade modelldelar skickas vidare till ElevenLabs. Eftersom LangGraph gör intern verktygskörning synlig genom tillståndsströmmen är filtreringen tydlig och styrs av proxyn. 

Härnäst går vi igenom nyanserna i att arbeta med Googles Agent Development Kit (ADK)

Google ADK

Googles ADK abstraherar runtime-loopen bakom några centrala primitiver: Agent, Runner och SessionService. ADK:s Runner ligger mellan HTTP-lagret och agentdefinitionen. Den hanterar meddelanderoutning, verktygsorkestrering, sessionslivscykeln och händelseströmning. 

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
)

När agenten, sessionsbackend och runnern har initierats hämtar eller skapar proxyn en ADK-session för varje inkommande begäran. I ADK styr session_id minnespersistensen: genom att återanvända samma session_id mellan turer följer historik, verktygsanrop och tidigare svar automatiskt med. Eftersom konversationsidentiteten finns uppströms i ElevenLabs hanterar proxyn denna mappning uttryckligen. Genom att skicka rätt identifierare för genereringsbegäran kan SDK:t hantera tidigare kontext internt. Vi skickar den godtyckliga identifieraren när konversationen initieras via extra parametrar som skickas i begärans brödtext.  

När meddelandet och sessionen är förberedda kan runnern anropas. Verktygsanrop och verktygsresultat visas fortfarande som interna ADK-händelser under körningen, men behandlas som mellanliggande orkestreringssteg snarare än användarriktad utdata. Därmed behövs inget manuellt filter, till skillnad från ramverk där verktygsanrop visas som användarsynlig text. 

Hanteraren nedan är en förenklad implementation med logik för sessionsupplösning och hämta-eller-skapa direkt i koden.

@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")

Härnäst tittar vi på CrewAI, som är mer uppgiftscentrerat till sin utformning.

CrewAI

CrewAI skapades för att orkestrera arbetsflöden med flera agenter kring strukturerade uppgifter (undersöka, skriva, sammanfatta), snarare än öppna dialogloopar. Agenter definieras med en roll, ett mål och en bakgrundshistoria. Körningen kretsar kring Task-objekt, vart och ett med en tydlig beskrivning och förväntad utdata. 

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

Till skillnad från agentloopmodellen i LangGraph och ADK skapar CrewAI vanligtvis Task och Crew per begäran för att definiera arbetsenheten för den turen i konversationen. Vi för vidare konversationskontext genom att injicera tidigare turer i nästa uppgift via en platshållare. Variabeln {crew_chat_messages} fylls vid varje begäran med den löpande konversationshistoriken och infogas sedan i uppgiftsbeskrivningen vid körning. Vi strävar också efter att skapa ren, talfärdig text genom att uttryckligen filtrera bort mellanliggande spårningsmönster (Thought, Action, Action Input, Observation) och endast skicka text från slutsvaret. 

Hanteraren nedan kombinerar uppgiftskonstruktion per begäran, interpolering av historik, strömning på Crew-nivå, filtrering av spårning och formatering av utdata. 

@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"

Härnäst tittar vi på LlamaIndex, som väljer en annan väg med fokus på en inbyggd händelsestyrd strömningsmodell.

LlamaIndex

Till skillnad från de andra ramverken i det här inlägget skapades LlamaIndex för att koppla LLM:er till externa datakällor (dokumentarkiv, index, hämtningspipelines). Dess agentlager, FunctionAgent, bygger ovanpå den grunden för att hämta och resonera utifrån strukturerad kontext, snarare än öppen dialog eller uppgiftskörning.

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

För att bevara kontinuiteten i konversationen omvandlar proxyn inkommande meddelanden till LlamaIndex-chattmeddelanden och delar sedan upp dem i den senaste användarturen (user_msg) och tidigare turer (chat_history). Fältet event.delta i varje AgentStream-händelse innehåller nästa textfragment, som direkt mappas till en delta.content-del i OpenAI-stil. Icke-tomma delta kan skickas vidare som de är, vilket gör detta till den mest direkta strömningsbryggan i guiden. Strömmen innehåller både orkestreringshändelser (verktygsanrop, resultat) och talhändelser (assistentens textdelta). För att hålla röstutdata ren behåller proxyn endast AgentStream-händelser och hoppar över tomma delta.

[1] AgentStream (delta='')       ← ignoreras
[2] ToolCall                     ← ignoreras
[3] ToolCallResult               ← ignoreras
[4] AgentStream (delta='Det')   ← vidarebefordras ✓
[5] AgentStream (delta=' kostar') ← vidarebefordras ✓
[6] AgentStream (delta=' $49.99')← vidarebefordras ✓

Denna uppdelning håller mellanliggande verktygsmekanik borta från talad utdata, samtidigt som stegvis tal med låg latens bevaras. Den färdiga hanteraren nedan för samman dessa steg.

@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 är mindre styrande kring mönster för konversationsruntime från början till slut än ramverken med mer omfattande inbyggda orkestreringslager. I produktionsdistributioner behöver kunder därför vanligtvis implementera sessionshantering, skyddsräcken för svar, verktygsorkestrering och spårning.

Slutsats

Varje ramverk i den här guiden ansluter till ElevenLabs genom samma kontrakt: ta emot en Completions- eller Responses-begäran i OpenAI-stil och strömma tillbaka SSE-delar. Det gör att team kan lägga röstorkestrering ovanpå en befintlig agentimplementation med minimala ändringar. De bevarar därmed det de redan har byggt och kan samtidigt använda konversations-AI i realtid. Denna modularitet är en kärnprincip för ElevenAgents-plattformen. Oavsett om organisationer bygger ut en befintlig agent eller bygger röstbaserat från början är ElevenAgents röstorkestrering utformad för att möta dem där de är.

Om du redan kör en agent med ett ramverk med öppen källkod och vill aktivera röst kan du prova det här tillvägagångssättet och berätta vad du tycker.