Praxisleitfaden: Open-Source-Agenten-Frameworks und ElevenAgents

In unserem vorherigen Beitrag über Externe Agenten in die ElevenLabs-Sprachorchestrierung integrieren, haben wir erläutert, wie Teams ihre bestehende textbasierte Agentenorchestrierung über das Custom LLM mit ElevenLabs verbinden können. Darauf aufbauend zeigt dieser Leitfaden, wie sich führende Open-Source-Frameworks für Agenten anpassen und hinter der Custom-LLM-Schnittstelle bereitstellen lassen. Das Ergebnis ist eine flexible Architektur, die Sprache auf ausgereifte Agentensysteme aufsetzt, ohne Zustandsverwaltung, Tool-Orchestrierung oder anwendungsspezifische Kontrolle zu beeinträchtigen. Frameworkübergreifend folgen wir demselben dreistufigen Muster: eine Generierungsanfrage erstellen, die finale Textantwort extrahieren und sie in ein OpenAI-kompatibles Server-Sent-Events-Format (SSE) überführen. ElevenLabs unterstützt sowohl die Formate Chat Completions als auch Responses. Dieser Leitfaden behandelt vier weit verbreitete Frameworks, doch die Muster lassen sich auf jede Runtime übertragen, die OpenAI-kompatiblen Streaming-Output erzeugen kann.

Allgemeines Setup

Die Beispiele in diesem Abschnitt verwenden Python und FastAPI. Jeder Stack, der HTTP-POST-Anfragen und gestreamte SSE-Antworten verarbeitet, funktioniert jedoch ebenfalls. Erkennt die Sprachorchestrierung von ElevenLabs ein wahrscheinliches Turn-Ende, sendet sie eine Generierungsanfrage an den konfigurierten Custom-LLM-Endpunkt. Dieser Abschnitt erläutert die Kernkomponenten dieser Übersetzungsschicht: die Bridge bzw. den Proxy, durch den Sprachorchestrierung und Agenten-Framework dieselbe Sprache sprechen.

Kunden wählen ein Framework häufig, weil sie damit vertraut sind oder es einen bestimmten Zweck erfüllt. LlamaIndex wurde beispielsweise ursprünglich entwickelt, um die Einrichtung von Retrieval-Augmented Generation (RAG) zu vereinfachen. CrewAI wurde geschaffen, um klar definierte Aufgaben im Zeitalter der Agenten zu automatisieren. Unterschiedliche Entwicklungsziele führen zu unterschiedlichen Antwortstrukturen, die jeweils spezifisch verarbeitet werden müssen. Entscheidend ist, Chunks zu streamen, während das LLM sie generiert, statt auf einen vollständigen Turn zu warten. So kann das Text-to-Speech-(TTS)-Modell früher mit der Spracherzeugung beginnen, was die wahrgenommene Latenz reduziert. Wir konzentrieren uns auf vier verbreitete Frameworks: LangGraph, Google ADK, CrewAI und LlamaIndex.

Hinweis zum gemeinsamen Code

Jedes Framework muss Antworten als OpenAI-kompatible SSE-Chunks streamen. Wir führen eine kleine Hilfsfunktion ein, die in allen Beispielen zum Erstellen dieser Chunks verwendet wird.

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"

Mit dieser Grundlage beginnen wir mit LangGraph. 

LangGraph

LangGraph modelliert Agenten als Graphen: Knoten stellen einzelne Schritte dar, Kanten definieren den Kontrollfluss zwischen ihnen. Das minimale Setup ist einfach: ein Chatmodell initialisieren, Agenten-Tools definieren und die Runtime des Agentengraphen erstellen.

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

Bei jeder Generierungsanfrage erhält der LangGraph-Agent den vollständigen Gesprächsverlauf und kann dadurch den erforderlichen Zustand intern verwalten. LangGraph unterstützt serverseitige Persistenz über Checkpoints, die wir hier jedoch nicht behandeln, um die Implementierung schlank zu halten.

Nach der Zustandsverwaltung folgt die nächste LangGraph-spezifische Entscheidung: der Streaming-Modus. LangGraph bietet zwei Optionen, die jeweils für einen anderen Anwendungsfall geeignet sind:

• stream_mode="values" liefert Momentaufnahmen des Graphzustands. Er ist einfacher zu implementieren, enthält aber bei jeder Antwort einen umfassenderen Nachrichtenzustand, was die Latenz in Echtzeit-Gesprächsabläufen erhöht.
• stream_mode="messages" streamt inkrementelle Nachrichten-Chunks des Modells. Dies wird für Echtzeit-Sprachinteraktionen im Allgemeinen bevorzugt, da es die Zeit bis zum ersten Audio in der ElevenLabs-Orchestrierung reduziert.

Genauer gesagt enthält die Messages-Implementierung der Agentenschleife Zwischenschritte wie Updates zu Tool-Aufrufen, die nicht vorgelesen werden sollten. Der Proxy filtert diese heraus und leitet nur nutzerseitigen Antworttext an die TTS-Schicht weiter. Hier ein Beispiel für einen Turn mit Tool-Nutzung.

[1] Modell entscheidet sich für einen Tool-Aufruf (tool_calls=["get_price"])
[2] Tool wird ausgeführt und gibt Daten zurück (result="$24.99") 
[3] Modell erzeugt anhand des Ergebnisses eine Antwort (content="Es kostet $24.99") 

Natürlich sollten nur die Chunks aus Schritt 3 im SSE-Stream weitergeleitet werden. In der Praxis übernehmen zwei Prüfungen diese Filterung in der Streaming-Schleife: Eine behält nur Ereignisse mit langgraph_node == "model", die andere überspringt leeren Inhalt. Zusammen stellen diese Prüfungen sicher, dass nur nutzerseitiger Assistententext als SSE an ElevenLabs weitergeleitet wird. Diese Konzepte kombiniert die folgende schlanke Implementierung des Anfrage-Proxys.

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

So werden nur nutzerseitige Modell-Chunks an ElevenLabs weitergeleitet. Da LangGraph seine interne Tool-Ausführung über den Zustandsstream sichtbar macht, ist die Filterung explizit und wird durch den Proxy gesteuert. 

Als Nächstes betrachten wir die Besonderheiten bei der Arbeit mit Googles Agent Development Kit (ADK).

Google ADK

Googles ADK abstrahiert die Runtime-Schleife hinter einigen zentralen Grundbausteinen: Agent, Runner und SessionService. Der Runner von ADK sitzt zwischen der HTTP-Schicht und der Agentendefinition. Er übernimmt Nachrichtenrouting, Tool-Orchestrierung, Sitzungslebenszyklus und 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
)

Nachdem Agent, Sitzungs-Backend und Runner initialisiert sind, löst der Proxy für jede eingehende Anfrage eine ADK-Sitzung auf oder erstellt sie. In ADK steuert session_id die Persistenz des Speichers: Wird dieselbe session_id über mehrere Turns hinweg wiederverwendet, bleiben Verlauf, Tool-Aufrufe und frühere Antworten automatisch erhalten. Da die Gesprächsidentität in ElevenLabs vorgelagert ist, verarbeitet der Proxy diese Zuordnung explizit. Durch die Übergabe des richtigen Bezeichners für die Generierungsanfrage kann das SDK den vorherigen Kontext intern verarbeiten. Den beliebigen Bezeichner übergeben wir bei der Gesprächsinitialisierung über zusätzliche Parameter im Body der Anfrage.  

Sobald Nachricht und Sitzung vorbereitet sind, kann der Runner aufgerufen werden. Tool-Aufrufe und Tool-Ergebnisse erscheinen während der Ausführung weiterhin als interne ADK-Events, werden jedoch als Zwischenschritte der Orchestrierung und nicht als nutzerseitige Ausgabe behandelt. Anders als bei Frameworks, in denen Tool-Aufrufe als sichtbarer Text erscheinen, ist daher kein manueller Filter erforderlich. 

Der folgende Handler ist eine vereinfachte Implementierung, die Sitzungsauflösung und Get-or-Create-Logik direkt enthält.

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

Als Nächstes betrachten wir CrewAI, das von Grund auf stärker auf Aufgaben ausgerichtet ist.

CrewAI

CrewAI wurde entwickelt, um Multi-Agent-Workflows rund um strukturierte Aufgaben zu orchestrieren – etwa Recherche, Schreiben und Zusammenfassen – statt um offene Dialogschleifen. Agenten werden mit einer Rolle, einem Ziel und einer Vorgeschichte definiert. Die Ausführung konzentriert sich auf Task-Objekte, jeweils mit einer klaren Beschreibung und erwarteten Ausgabe. 

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

Anders als beim Agentenschleifen-Modell von LangGraph und ADK erstellt CrewAI typischerweise pro Anfrage einen Task und eine Crew, um die Arbeitseinheit für diesen Gesprächs-Turn festzulegen. Wir führen den Gesprächskontext fort, indem wir vorherige Turns über einen Platzhalter in die nächste Aufgabe einfügen. Die Variable {crew_chat_messages} wird bei jeder Anfrage mit dem bisherigen Gesprächsverlauf befüllt und bei der Ausführung in die Task-Beschreibung eingefügt. Außerdem wollen wir sauberen, direkt sprechbaren Text erzeugen. Deshalb filtern wir Zwischenmuster für das Tracing (Thought, Action, Action Input, Observation) explizit heraus und geben nur finalen Antworttext aus. 

Der folgende Handler kombiniert die auf Anfrage basierende Task-Erstellung, die Einfügung des Verlaufs, Streaming auf Crew-Ebene, Trace-Filterung und Ausgabeformatierung. 

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

Als Nächstes betrachten wir LlamaIndex, das sich auf ein natives ereignisgesteuertes Streaming-Modell konzentriert und einen anderen Weg einschlägt.

LlamaIndex

Anders als die anderen in diesem Beitrag behandelten Frameworks wurde LlamaIndex entwickelt, um LLMs mit externen Datenquellen zu verbinden, etwa Dokumentenspeichern, Indizes und Retrieval-Pipelines. Seine Agentenschicht FunctionAgent baut darauf auf, um strukturierten Kontext abzurufen und daraus Schlussfolgerungen zu ziehen, statt offene Dialoge oder Aufgaben auszuführen.

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

Um den Gesprächszusammenhang zu erhalten, wandelt der Proxy eingehende Nachrichten in LlamaIndex-Chatnachrichten um und teilt sie dann in den neuesten Nutzer-Turn (user_msg) und vorherige Turns (chat_history) auf. Das Feld event.delta jedes AgentStream-Events enthält das nächste Textfragment, das sich direkt einem delta.content-Chunk im OpenAI-Stil zuordnen lässt. Nicht leere Deltas können unverändert weitergeleitet werden. Damit ist dies die direkteste Streaming-Bridge in diesem Leitfaden. Der Stream enthält sowohl Orchestrierungs-Events (Tool-Aufrufe, Ergebnisse) als auch Sprach-Events (Text-Deltas des Assistenten). Um die Sprachausgabe sauber zu halten, behält der Proxy nur AgentStream-Events und überspringt leere Deltas.

[1] AgentStream (delta='')       ← ignoriert
[2] ToolCall                     ← ignoriert
[3] ToolCallResult               ← ignoriert
[4] AgentStream (delta='Es')     ← weitergeleitet ✓
[5] AgentStream (delta=' kostet') ← weitergeleitet ✓
[6] AgentStream (delta=' $49.99')← weitergeleitet ✓

Diese Trennung hält die Mechanik der Zwischentools aus der gesprochenen Ausgabe heraus und erhält gleichzeitig inkrementelle Sprachausgabe mit geringer Latenz. Der folgende direkt einsetzbare Handler führt diese Schritte zusammen.

@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 ist bei End-to-End-Mustern für Gesprächsruntimes weniger präskriptiv als Frameworks mit umfangreicheren integrierten Orchestrierungsschichten. Für Produktivbereitstellungen müssen Kunden daher typischerweise Sitzungsverwaltung, Schutzmechanismen für Antworten, Tool-Orchestrierung und Tracing selbst implementieren.

Fazit

Jedes Framework in diesem Leitfaden verbindet sich über denselben Vertrag mit ElevenLabs: Es akzeptiert eine Completions- oder Responses-Anfrage im OpenAI-Stil und streamt SSE-Chunks zurück. So können Teams Sprachorchestrierung mit minimalen Änderungen auf eine bestehende Agentenimplementierung aufsetzen. Sie erhalten, was sie bereits entwickelt haben, und erschließen gleichzeitig Echtzeit-Conversational AI. Diese Modularität ist ein zentraler Grundsatz der ElevenAgents-Plattform. Ob Unternehmen einen bestehenden Agenten erweitern oder von Anfang an sprachbasiert entwickeln: Die Sprachorchestrierung von ElevenAgents ist darauf ausgelegt, sie dort abzuholen, wo sie stehen.

Wenn Sie bereits einen Agenten mit einem Open-Source-Framework betreiben und Sprache aktivieren möchten, probieren Sie diesen Ansatz aus und teilen Sie uns Ihre Meinung mit.