Praktyczny przewodnik: frameworki agentów open source i ElevenAgents

W poprzednim artykule o Integracji zewnętrznych agentów z orkiestracją głosu ElevenLabs, pokazaliśmy, jak zespoły mogą połączyć istniejącą orkiestrację agentów opartą na tekście z ElevenLabs przez Custom LLM. Ten przewodnik pokazuje, jak dostosować i wdrożyć popularne frameworki agentów open source za interfejsem Custom LLM. Powstaje elastyczna architektura, w której głos nakłada się na dojrzałe systemy agentowe bez uszczerbku dla zarządzania stanem, orkiestracji narzędzi czy kontroli specyficznej dla aplikacji. Niezależnie od frameworka stosujemy ten sam trzyetapowy schemat: utworzenie żądania generowania, wyodrębnienie końcowej odpowiedzi tekstowej i sformatowanie jej jako zgodne z OpenAI Server-Sent Events (SSE). ElevenLabs obsługuje formaty Chat Completions i Responses. Chociaż ten przewodnik obejmuje cztery popularne frameworki, opisane wzorce działają w każdym środowisku wykonawczym, które może tworzyć strumieniowe dane wyjściowe zgodne z OpenAI.

Konfiguracja podstawowa

Przykłady w tej sekcji używają Pythona i FastAPI, ale sprawdzi się każdy stos obsługujący żądania HTTP POST i strumieniowe odpowiedzi SSE. Gdy orkiestracja głosu ElevenLabs wykryje prawdopodobny koniec wypowiedzi, wysyła żądanie generowania do skonfigurowanego endpointu Custom LLM. Ta sekcja omawia 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 ze względu na znajomość narzędzia lub jego przydatność do konkretnego zadania. LlamaIndex powstał na przykład po to, by uprościć konfigurację Retrieval-Augmented Generation (RAG), a CrewAI — by automatyzować określone zadania w erze agentów. Różne cele projektowe prowadzą do różnych struktur odpowiedzi, które wymagają odpowiedniej obsługi. Strumieniowanie fragmentów w trakcie generowania przez LLM, zamiast czekania na całą wypowiedź, ma kluczowe znaczenie: model Text-to-Speech (TTS) może wtedy 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ę podstawę, zacznijmy od LangGraph.

LangGraph

LangGraph modeluje agentów jako grafy, w których węzły reprezentują pojedyncze kroki, a krawędzie określają przepływ sterowania między nimi. Minimalna konfiguracja jest prosta: zainicjuj model czatu, zdefiniuj narzędzia agenta i utwórz ś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 nie omawiamy ich tutaj, aby zachować minimalną implementację.

Po obsłużeniu zarządzania stanem kolejną decyzją specyficzną dla LangGraph jest tryb strumieniowania. LangGraph oferuje dwie opcje, każda do innego zastosowania:

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

Dokładniej mówiąc, implementacja pętli agenta z messages obejmuje kroki pośrednie, takie jak aktualizacje wywoływania narzędzi, których nie należy odczytywać na głos. Proxy je filtruje i przekazuje do warstwy TTS tylko tekst odpowiedzi widoczny dla użytkownika. Poniżej przykład wypowiedzi 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ź na podstawie wyniku (content="Kosztuje $24.99")

Oczywiście w strumieniu SSE należy przekazać tylko fragmenty z kroku 3. W praktyce filtrowanie w pętli strumieniowania obsługują dwa warunki: jeden zachowuje tylko zdarzenia langgraph_node == "model", a drugi pomija pustą treść. Razem zapewniają, że do ElevenLabs jako SSE trafia tylko tekst asystenta widoczny dla użytkownika. Łącząc te założenia, otrzymujemy lekką 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ą wyłącznie 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 przyjrzymy się niuansom pracy z Google Agent Development Kit (ADK)

Google ADK

Google ADK ukrywa pętlę środowiska wykonawczego za kilkoma podstawowymi elementami: Agent, Runner i SessionService. Runner w 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: użycie tego samego session_id w kolejnych wypowiedziach automatycznie przenosi historię, wywołania narzędzi i wcześniejsze odpowiedzi. Tożsamość rozmowy istnieje po stronie ElevenLabs, więc proxy jawnie obsługuje to mapowanie. Po przekazaniu właściwego identyfikatora dla żądania generowania SDK może wewnętrznie obsłużyć wcześniejszy kontekst. Dowolny identyfikator przekazujemy podczas rozpoczęcia rozmowy przez dodatkowe parametry przekazane w treści żądania.

Po przygotowaniu wiadomości i sesji można wywołać runner. 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 wyjściowe dla użytkownika. Eliminuje to potrzebę ręcznego filtrowania, wymaganego w frameworkach, w których wywołania narzędzi pojawiają się jako tekst widoczny dla użytkownika.

Poniższy handler to uproszczona implementacja z obsługą sesji oraz logiką pobierania lub tworzenia sesji 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")

Następnie przyjrzymy się CrewAI, które z założenia bardziej skupia się na zadaniach.

CrewAI

CrewAI zaprojektowano do orkiestracji workflowów wieloagentowych wokół ustrukturyzowanych zadań (badanie, pisanie, podsumowanie), a nie otwartych pętli dialogowych. Agenci są definiowani przez rolę, cel i historię. Wykonanie koncentruje 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 wypowiedzi w rozmowie. Przenosimy kontekst rozmowy, wstawiając wcześniejsze wypowiedzi do kolejnego zadania przez placeholder. Zmienna {crew_chat_messages} jest przy każdym żądaniu wypełniana bieżącą historią rozmowy, a następnie interpolowana w opisie zadania podczas wykonania. Chcemy też uzyskać czysty tekst gotowy do odczytu, więc jawnie odfiltrowujemy pośrednie wzorce śledzenia (Thought, Action, Action Input, Observation) i generujemy wyłącznie tekst końcowej odpowiedzi.

Poniższy handler łączy tworzenie zadań dla każdego żądania, interpolację historii, strumieniowanie na poziomie Crew, filtrowanie śladów 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 przyjrzymy się LlamaIndex, które podąża inną ścieżką i skupia się na natywnym, sterowanym zdarzeniami modelu strumieniowania.

LlamaIndex

W przeciwieństwie do innych frameworków omówionych w tym artykule LlamaIndex zaprojektowano do łączenia LLM-ów z zewnętrznymi źródłami danych (repozytoriami dokumentów, indeksami, potokami wyszukiwania). Warstwa agenta, FunctionAgent, bazuje na tym fundamencie, aby wyszukiwać i analizować ustrukturyzowany 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ą wypowiedź użytkownika (user_msg) i wcześniejsze wypowiedzi (chat_history). Pole event.delta każdego zdarzenia AgentStream zawiera kolejny fragment tekstu, który bezpośrednio odpowiada fragmentowi delta.content w stylu OpenAI. Niepuste delty można przekazywać bez zmian, co czyni ten most strumieniowania najprostszym w 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średniej mechaniki narzędzi do odczytywanego tekstu, a jednocześnie zachowuje przyrostową mowę o niskim opóźnieniu. 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 jest mniej nakazowy w kwestii wzorców działania konwersacyjnego runtime end-to-end niż frameworki z bardziej rozbudowanymi wbudowanymi warstwami orkiestracji. W środowisku produkcyjnym zwykle wymaga to od klientów wdrożenia obsługi sesji, zabezpieczeń odpowiedzi, orkiestracji narzędzi i śledzenia.

Podsumowanie

Każdy framework z tego przewodnika łączy się z ElevenLabs przez tę samą umowę: przyjmuje żądanie Completions lub Responses w stylu OpenAI i przesyła z powrotem fragmenty SSE. Dzięki temu zespoły mogą z niewielkimi zmianami dodać orkiestrację głosu do istniejącej implementacji agenta, zachowując to, co już zbudowały, i odblokowując działające w czasie rzeczywistym Conversational AI. Ta modułowość jest jedną z głównych zasad platformy ElevenAgents. Niezależnie od tego, czy organizacje rozwijają istniejącego agenta, czy od początku budują rozwiązanie głosowe, orkiestracja głosu ElevenAgents jest stworzona tak, by działać z tym, co już mają.

Jeśli już korzystasz z 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.