Guia prático: frameworks de agentes open-source e ElevenAgents

No nosso post anterior sobre Integração de agentes externos à orquestração de voz da ElevenLabs, explicamos como as equipes podem conectar a orquestração de agentes baseada em texto que já utilizam à ElevenLabs por meio do LLM personalizado. Com base nisso, este guia mostra como os principais frameworks de agentes de código aberto podem ser adaptados e implantados por trás da interface de LLM personalizado. O resultado é uma arquitetura flexível em que a voz é adicionada a sistemas de agentes maduros sem comprometer o gerenciamento de estado, a orquestração de ferramentas ou o controle específico da aplicação. Em todos os frameworks, seguimos o mesmo padrão de três etapas: criar uma solicitação de geração, extrair a resposta final em texto e reformatá-la em um formato Server-Sent Events (SSE) compatível com OpenAI. A ElevenLabs oferece suporte aos formatos Chat Completions e Responses. Embora este guia aborde quatro frameworks amplamente adotados, os padrões se aplicam a qualquer ambiente de execução capaz de gerar saída em streaming compatível com OpenAI.

Configuração geral

Os exemplos desta seção usam Python e FastAPI, mas qualquer stack que lide com solicitações HTTP POST e respostas SSE em streaming funciona. Quando a orquestração de voz da ElevenLabs detecta um provável fim de turno, ela envia uma solicitação de geração ao endpoint de LLM personalizado configurado. Esta seção aborda os componentes principais dessa camada de tradução, a ponte ou proxy que faz a orquestração de voz e o framework de agentes falarem a mesma língua.

Naturalmente, os clientes podem escolher cada framework por já estarem familiarizados com ele ou por sua capacidade de atender a uma finalidade específica. O LlamaIndex, por exemplo, foi desenvolvido originalmente para simplificar a configuração da Geração Aumentada por Recuperação (RAG), enquanto o CrewAI foi criado para automatizar tarefas definidas na era dos agentes. Objetivos de design diferentes geram estruturas de resposta diferentes, e cada uma exige um tratamento específico. Transmitir blocos à medida que o LLM os gera, em vez de esperar um turno completo, é essencial porque permite que o modelo de Text to Speech (TTS) comece a gerar fala mais cedo, reduzindo assim a latência percebida. Vamos nos concentrar em quatro frameworks populares: LangGraph, Google ADK, CrewAI e LlamaIndex.

Uma observação sobre o código compartilhado

Cada framework precisa transmitir respostas como blocos SSE compatíveis com OpenAI. Apresentamos uma pequena função auxiliar, usada em todos os exemplos, para construir esses blocos.

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"

Com essa base estabelecida, vamos começar pelo LangGraph. 

LangGraph

O LangGraph modela agentes como grafos, em que os nós representam etapas individuais e as arestas definem o fluxo de controle entre elas. A configuração mínima é simples: inicialize um modelo de chat, defina as ferramentas do agente e crie o ambiente de execução do grafo do agente.

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

A cada solicitação de geração, o agente do LangGraph recebe todo o histórico da conversa, o que permite manter internamente o estado necessário. O LangGraph oferece suporte à persistência no servidor por meio de Checkpoints, embora não os abordemos aqui para manter a implementação mínima.

Com o gerenciamento de estado resolvido, a próxima decisão específica do LangGraph é o modo de streaming. O LangGraph oferece duas opções, cada uma adequada a um caso de uso diferente:

• stream_mode="values" fornece snapshots do estado do grafo. É mais simples de implementar, mas inclui um estado mais completo das mensagens em cada resposta, o que aumenta a latência em fluxos conversacionais em tempo real.
• stream_mode="messages" transmite blocos incrementais de mensagens do modelo. Em geral, essa é a opção preferida para interações de voz em tempo real, pois reduz o tempo até o primeiro áudio na camada de orquestração da ElevenLabs.

Mais especificamente, a implementação de mensagens do loop do agente inclui etapas intermediárias, como atualizações de chamadas de ferramentas, que não devem ser faladas em voz alta. O proxy filtra essas atualizações e envia apenas o texto de resposta destinado ao usuário para a camada de TTS. Veja um exemplo de turno com uso de ferramenta.

[1] O modelo decide chamar uma ferramenta (tool_calls=["get_price"])
[2] A ferramenta é executada e retorna dados (result="$24.99") 
[3] O modelo produz uma resposta usando o resultado (content="Custa $24.99") 

Naturalmente, apenas os blocos da etapa 3 devem ser encaminhados no fluxo SSE. Na prática, duas verificações controlam essa filtragem no loop de streaming: uma para manter apenas eventos com langgraph_node == "model" e outra para ignorar conteúdo vazio. Juntas, essas verificações garantem que apenas o texto do assistente destinado ao usuário seja encaminhado à ElevenLabs como SSE. Reunindo esses conceitos, apresentamos uma implementação leve do proxy de solicitações.

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

Isso garante que apenas os blocos do modelo destinados ao usuário sejam encaminhados à ElevenLabs. Como o LangGraph torna visível a execução interna das ferramentas pelo fluxo de estado, a filtragem é explícita e controlada pelo proxy. 

Agora, vamos explorar as particularidades de trabalhar com o Agent Development Kit (ADK) do Google

Google ADK

O ADK do Google abstrai o loop do ambiente de execução por trás de algumas primitivas principais: Agent, Runner e SessionService. O Runner do ADK fica entre a camada HTTP e a definição do agente. Ele gerencia o roteamento de mensagens, a orquestração de ferramentas, o ciclo de vida da sessão e o streaming de eventos. 

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
)

Com o agente, o backend de sessões e o runner inicializados, o proxy resolve ou cria uma sessão do ADK para cada solicitação recebida. No ADK, session_id controla a persistência da memória: reutilizar o mesmo session_id entre turnos mantém automaticamente o histórico, as chamadas de ferramentas e as respostas anteriores. Como a identidade da conversa fica na camada anterior da ElevenLabs, o proxy lida explicitamente com esse mapeamento. Ao enviar o identificador correto na solicitação de geração, o SDK consegue processar internamente o contexto anterior. Enviamos o identificador arbitrário durante o início da conversa por meio de parâmetros extras incluídos no corpo da solicitação.  

Com a mensagem e a sessão preparadas, o runner pode ser invocado. As chamadas de ferramentas e seus resultados ainda aparecem como eventos internos do ADK durante a execução, mas são tratados como etapas intermediárias de orquestração, e não como saída destinada ao usuário. Isso elimina a necessidade de um filtro manual em comparação com frameworks nos quais as chamadas de ferramentas aparecem como texto visível ao usuário. 

O manipulador abaixo é uma implementação simplificada que inclui em linha a resolução de sessão e a lógica de obter ou criar.

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

Agora, vamos analisar o CrewAI, que por design é mais centrado em tarefas.

CrewAI

O CrewAI foi projetado para orquestrar workflows multiagente em torno de tarefas estruturadas (pesquisar, escrever, resumir), em vez de loops de diálogo abertos. Os agentes são definidos com uma função, um objetivo e um histórico. A execução é centrada em objetos Task, cada um com uma descrição clara e uma saída esperada. 

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

Diferentemente do modelo de loop de agentes usado no LangGraph e no ADK, o CrewAI geralmente constrói uma Task e uma Crew por solicitação para definir a unidade de trabalho daquele turno da conversa. Mantemos o contexto da conversa inserindo os turnos anteriores na próxima tarefa por meio de um placeholder. A variável {crew_chat_messages} é preenchida a cada solicitação com o histórico acumulado da conversa e, depois, interpolada na descrição da tarefa no momento da execução. Também buscamos gerar um texto limpo e pronto para ser falado, filtrando explicitamente padrões de rastreamento intermediários (Thought, Action, Action Input, Observation) e emitindo apenas o texto da resposta final. 

O manipulador abaixo reúne a construção de tarefas por solicitação, a interpolação do histórico, o streaming no nível da Crew, a filtragem de rastreamentos e a formatação da saída. 

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

Agora, vamos ver o LlamaIndex, que segue um caminho diferente, com foco em um modelo nativo de streaming orientado a eventos.

LlamaIndex

Ao contrário dos outros frameworks abordados neste post, o LlamaIndex foi projetado para conectar LLMs a fontes de dados externas, como repositórios de documentos, índices e pipelines de recuperação. Sua camada de agentes, FunctionAgent, se baseia nisso para recuperar e interpretar contexto estruturado, em vez de executar diálogos abertos ou tarefas.

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

Para preservar a continuidade da conversa, o proxy transforma as mensagens recebidas em mensagens de chat do LlamaIndex e as separa no turno mais recente do usuário (user_msg) e nos turnos anteriores (chat_history). O campo event.delta de cada evento AgentStream contém o próximo fragmento de texto, que é mapeado diretamente para um bloco delta.content no estilo OpenAI. Deltas não vazios podem ser encaminhados como estão, tornando esta a ponte de streaming mais direta do guia. O fluxo contém tanto eventos de orquestração, como chamadas de ferramentas e resultados, quanto eventos de fala, como deltas de texto do assistente. Para manter a saída de voz limpa, o proxy conserva apenas os eventos AgentStream e ignora deltas vazios.

[1] AgentStream (delta='')       ← ignorado
[2] ToolCall                     ← ignorado
[3] ToolCallResult               ← ignorado
[4] AgentStream (delta='Isso')   ← encaminhado ✓
[5] AgentStream (delta=' custa') ← encaminhado ✓
[6] AgentStream (delta=' $49.99')← encaminhado ✓

Essa separação mantém os mecanismos intermediários das ferramentas fora da saída falada, preservando ao mesmo tempo a fala incremental de baixa latência. O manipulador pronto para uso abaixo reúne essas etapas.

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

O LlamaIndex é menos prescritivo quanto aos padrões de execução conversacional de ponta a ponta do que frameworks com camadas de orquestração integradas mais robustas. Em implantações de produção, isso normalmente exige que os clientes implementem gerenciamento de sessões, proteções para respostas, orquestração de ferramentas e rastreamento.

Conclusão

Cada framework deste guia se conecta à ElevenLabs pelo mesmo contrato: aceita uma solicitação de Completions ou Responses no estilo OpenAI e transmite blocos SSE como resposta. Isso permite que as equipes adicionem orquestração de voz a uma implementação de agente existente com poucas mudanças, preservando o que já construíram e habilitando IA conversacional em tempo real. Essa modularidade é um princípio central da plataforma ElevenAgents. Seja para ampliar um agente existente ou criar uma solução nativa de voz desde o início, a orquestração de voz do ElevenAgents foi criada para atender às organizações onde elas estão.

Se você já executa um agente com um framework de código aberto e quer habilitar a voz, experimente essa abordagem e conte para nós o que achou.