Praktyczny przewodnik: open-source agent frameworks i ElevenAgents

W poprzednim artykule o Integrating External Agents with ElevenLabs Voice Orchestration pokazaliśmy, jak zespoły mogą połączyć istniejącą orkiestrację agentów tekstowych z ElevenLabs przez Custom LLM. Ten przewodnik pokazuje, jak dostosować i wdrożyć czołowe frameworki agentów open source za interfejsem Custom LLM. Powstaje elastyczna architektura, która dodaje głos do dojrzałych systemów agentowych bez utraty zarządzania stanem, orkiestracji narzędzi ani kontroli specyficznej dla aplikacji. Niezależnie od frameworka stosujemy ten sam schemat w trzech krokach: tworzymy żądanie generowania, wyciągamy końcową odpowiedź tekstową i formatujemy ją jako zgodne z OpenAI Server-Sent Events (SSE). ElevenLabs obsługuje formaty Chat Completions i Responses. Choć ten przewodnik omawia cztery popularne frameworki, te wzorce działają w każdym środowisku wykonawczym, które może generować strumieniowe dane zgodne z OpenAI.

Ogólna konfiguracja

Przykłady w tej sekcji używają Pythona i FastAPI, ale sprawdzi się każdy stack obsługujący żądania HTTP POST i strumieniowe odpowiedzi SSE. Gdy orkiestracja głosu ElevenLabs wykryje prawdopodobny koniec tury, wysyła żądanie generowania do skonfigurowanego endpointu Custom LLM. W tej sekcji omawiamy główne elementy tej warstwy tłumaczącej — mostu lub proxy, dzięki któremu orkiestracja głosu i framework agentowy mówią tym samym językiem.

Klienci mogą wybierać dany framework, bo go znają lub dlatego, że spełnia konkretny cel. LlamaIndex powstał na przykład, by ułatwić konfigurację Retrieval-Augmented Generation (RAG), a CrewAI — by automatyzować określone zadania w erze agentów. Różne cele projektowe dają różne struktury odpowiedzi, a każda wymaga innej obsługi. Strumieniowanie fragmentów podczas generowania przez LLM, zamiast czekania na pełną turę, jest kluczowe: model Text-to-Speech (TTS) może wcześniej zacząć generować mowę, co zmniejsza odczuwalne opóźnienie. Skupiamy się na czterech popularnych frameworkach: LangGraph, Google ADK, CrewAI i LlamaIndex.

Uwaga o wspólnym kodzie

Każdy framework musi przesyłać odpowiedzi jako fragmenty SSE zgodne z OpenAI. Wprowadzamy małą funkcję pomocniczą, używaną we wszystkich przykładach do tworzenia tych fragmentów.

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"

Mając tę bazę, zacznijmy od LangGraph. 

LangGraph

LangGraph przedstawia agentów jako grafy, w których węzły oznaczają poszczególne kroki, a krawędzie definiują przepływ sterowania między nimi. Minimalna konfiguracja jest prosta: inicjalizujesz model czatu, definiujesz narzędzia agenta i tworzysz środowisko wykonawcze grafu agenta.

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

Przy każdym żądaniu generowania agent LangGraph otrzymuje pełną historię rozmowy, dzięki czemu może wewnętrznie utrzymywać wymagany stan. LangGraph obsługuje trwałość po stronie serwera przez Checkpoints, ale dla zachowania minimalnej implementacji nie omawiamy ich tutaj.

Gdy zarządzanie stanem jest gotowe, kolejną decyzją specyficzną dla LangGraph jest tryb strumieniowania. LangGraph oferuje dwie opcje, każda do innego zastosowania:

• stream_mode="values" zwraca migawki stanu grafu. Jest prostszy we wdrożeniu, ale każda odpowiedź zawiera pełniejszy stan wiadomości, co zwiększa opóźnienie w rozmowach w czasie rzeczywistym.
• stream_mode="messages" przesyła z modelu przyrostowe fragmenty wiadomości. Zwykle lepiej sprawdza się w interakcjach głosowych w czasie rzeczywistym, ponieważ skraca czas do pierwszego audio w ElevenLabs warstwie orkiestracji.

Dokładniej: implementacja pętli agenta w trybie messages obejmuje kroki pośrednie, takie jak aktualizacje wywołań narzędzi, których nie należy odczytywać na głos. Proxy je odfiltrowuje, przekazując do warstwy TTS tylko tekst odpowiedzi widoczny dla użytkownika. Oto przykład tury z użyciem narzędzia.

[1] Model decyduje się wywołać narzędzie (tool_calls=["get_price"])
[2] Narzędzie wykonuje się i zwraca dane (result="$24.99") 
[3] Model tworzy odpowiedź z użyciem wyniku (content="Kosztuje $24.99") 

Oczywiście w strumieniu SSE należy przekazywać tylko fragmenty z kroku 3. W praktyce filtrowanie w pętli strumieniowania obsługują dwa sprawdzenia: jedno zachowuje wyłącznie zdarzenia langgraph_node == "model", a drugie pomija pustą zawartość. Razem zapewniają, że do ElevenLabs jako SSE trafia tylko tekst asystenta widoczny dla użytkownika. Łącząc te elementy, otrzymujemy prostą implementację proxy żądań.

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

Dzięki temu do ElevenLabs trafiają tylko fragmenty modelu widoczne dla użytkownika. Ponieważ LangGraph pokazuje wewnętrzne wykonanie narzędzi w strumieniu stanu, filtrowanie jest jawne i kontrolowane przez proxy. 

Teraz przejdźmy do niuansów pracy z Agent Development Kit (ADK) od Google

Google ADK

ADK od Google ukrywa pętlę środowiska wykonawczego za kilkoma podstawowymi elementami: Agent, Runner i SessionService. Runner ADK znajduje się między warstwą HTTP a definicją agenta. Obsługuje kierowanie wiadomości, orkiestrację narzędzi, cykl życia sesji i strumieniowanie zdarzeń. 

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
)

Po zainicjowaniu agenta, backendu sesji i runnera proxy wyszukuje lub tworzy sesję ADK dla każdego przychodzącego żądania. W ADK session_id kontroluje trwałość pamięci: ponowne użycie tego samego session_id w kolejnych turach automatycznie zachowuje historię, wywołania narzędzi i wcześniejsze odpowiedzi. Ponieważ tożsamość rozmowy jest ustalana wcześniej w ElevenLabs, proxy obsługuje to mapowanie jawnie. Przekazując właściwy identyfikator w żądaniu generowania, SDK może wewnętrznie obsłużyć wcześniejszy kontekst. Dowolny identyfikator przekazujemy podczas rozpoczęcia rozmowy przez dodatkowe parametry dodane do treści żądania.  

Gdy wiadomość i sesja są gotowe, można wywołać runnera. Wywołania narzędzi i ich wyniki nadal pojawiają się podczas wykonania jako wewnętrzne zdarzenia ADK, ale są traktowane jako pośrednie kroki orkiestracji, a nie dane widoczne dla użytkownika. Dzięki temu nie trzeba ręcznie filtrować danych, jak w frameworkach, gdzie wywołania narzędzi pojawiają się jako tekst widoczny dla użytkownika. 

Poniższy handler to uproszczona implementacja z logiką wyszukiwania sesji oraz jej pobierania lub tworzenia bezpośrednio w kodzie.

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

Teraz przyjrzymy się CrewAI, które z założenia jest bardziej skupione na zadaniach.

CrewAI

CrewAI stworzono do orkiestracji workflow wielu agentów wokół uporządkowanych zadań (badanie, pisanie, podsumowanie), a nie otwartych pętli dialogowych. Agenci są definiowani przez rolę, cel i historię. Wykonanie skupia się na obiektach Task, z których każdy ma jasny opis i oczekiwany wynik. 

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

W przeciwieństwie do modelu pętli agenta używanego w LangGraph i ADK, CrewAI zwykle tworzy Task i Crew dla każdego żądania, aby zdefiniować jednostkę pracy dla danej tury rozmowy. Kontekst rozmowy przenosimy, wstawiając poprzednie tury do kolejnego zadania przez placeholder. Zmienna {crew_chat_messages} jest przy każdym żądaniu wypełniana bieżącą historią rozmowy, a następnie wstawiana do opisu zadania podczas wykonania. Chcemy też uzyskać czysty tekst gotowy do odczytania, więc jawnie odfiltrowujemy pośrednie wzorce śledzenia (Thought, Action, Action Input, Observation) i emitujemy tylko tekst końcowej odpowiedzi. 

Poniższy handler łączy tworzenie zadań dla każdego żądania, wstawianie historii, strumieniowanie na poziomie Crew, filtrowanie śledzenia i formatowanie danych wyjściowych. 

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

Teraz przejdźmy do LlamaIndex, które wybiera inną drogę i skupia się na natywnym modelu strumieniowania opartym na zdarzeniach.

LlamaIndex

W przeciwieństwie do pozostałych frameworków omawianych w tym artykule LlamaIndex stworzono do łączenia LLM-ów z zewnętrznymi źródłami danych, takimi jak magazyny dokumentów, indeksy i pipeline’y wyszukiwania. Jego warstwa agentowa, FunctionAgent, bazuje na tym fundamencie, aby wyszukiwać i analizować uporządkowany kontekst, zamiast prowadzić otwarty dialog lub wykonywać zadania.

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

Aby zachować ciągłość rozmowy, proxy przekształca przychodzące wiadomości w wiadomości czatu LlamaIndex, a następnie dzieli je na najnowszą turę użytkownika (user_msg) i wcześniejsze tury (chat_history). Pole event.delta każdego zdarzenia AgentStream zawiera kolejny fragment tekstu, który bezpośrednio mapuje się na fragment delta.content w stylu OpenAI. Niepuste delty można przekazywać bez zmian, co czyni ten most strumieniowy najprostszym w tym przewodniku. Strumień zawiera zarówno zdarzenia orkiestracji (wywołania narzędzi, wyniki), jak i zdarzenia mowy (delty tekstu asystenta). Aby dane głosowe były czyste, proxy zachowuje tylko zdarzenia AgentStream i pomija puste delty.

[1] AgentStream (delta='')       ← pominięto
[2] ToolCall                     ← pominięto
[3] ToolCallResult               ← pominięto
[4] AgentStream (delta='To')     ← przekazano ✓
[5] AgentStream (delta=' kosztuje') ← przekazano ✓
[6] AgentStream (delta=' $49.99')← przekazano ✓

Ten podział nie dopuszcza pośrednich mechanizmów narzędzi do odczytywanego tekstu, a jednocześnie zachowuje przyrostowe generowanie mowy z niskim opóźnieniem. Poniższy gotowy do użycia handler łączy te kroki.

@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 mniej narzuca wzorce działania konwersacyjnego runtime’u end-to-end niż frameworki z bardziej rozbudowanymi wbudowanymi warstwami orkiestracji. W środowisku produkcyjnym klienci zwykle muszą więc wdrożyć obsługę sesji, zabezpieczenia odpowiedzi, orkiestrację narzędzi i śledzenie.

Podsumowanie

Każdy framework w tym przewodniku łączy się z ElevenLabs przez ten sam kontrakt: przyjmuje żądanie Completions lub Responses w stylu OpenAI i odsyła strumień fragmentów SSE. Dzięki temu zespoły mogą dodać orkiestrację głosu do istniejącej implementacji agenta przy minimalnych zmianach, zachowując to, co już zbudowały, i zyskując działające w czasie rzeczywistym konwersacyjne AI. Ta modułowość jest jedną z podstaw platformy ElevenAgents. Niezależnie od tego, czy organizacja rozwija istniejącego agenta, czy od początku tworzy rozwiązanie głosowe, orkiestracja głosu ElevenAgents dopasowuje się do jej potrzeb.

Jeśli masz już agenta opartego na frameworku open source i chcesz dodać głos, wypróbuj to podejście i daj nam znać, co o nim myślisz.