Guía práctica: frameworks de agentes de código abierto y ElevenAgents

En nuestra publicación anterior sobre Integración de agentes externos con la orquestación de voz de ElevenLabs, explicamos cómo los equipos pueden conectar su orquestación de agentes basada en texto a ElevenLabs mediante Custom LLM. A partir de esa base, esta guía muestra cómo adaptar e implementar los principales frameworks de agentes de código abierto tras la interfaz de Custom LLM. El resultado es una arquitectura flexible que añade una capa de voz a sistemas de agentes consolidados sin comprometer la gestión del estado, la orquestación de herramientas ni el control específico de cada aplicación. En todos los frameworks seguimos el mismo patrón de tres pasos: crear una solicitud de generación, extraer la respuesta de texto final y reformatearla en un formato Server-Sent Events (SSE) compatible con OpenAI. ElevenLabs admite los formatos Chat Completions y Responses. Aunque esta guía cubre cuatro frameworks muy utilizados, los patrones se pueden aplicar a cualquier entorno de ejecución capaz de producir salida en streaming compatible con OpenAI.

Configuración general

Los ejemplos de esta sección usan Python y FastAPI, aunque funcionará cualquier stack que gestione solicitudes HTTP POST y respuestas SSE en streaming. Cuando la orquestación de voz de ElevenLabs detecta un posible final de turno, envía una solicitud de generación a la ruta de API de Custom LLM configurada. Esta sección explica los componentes principales de esa capa de traducción: el puente o proxy que permite que la orquestación de voz y el framework de agentes hablen el mismo idioma.

Es comprensible que clientes elijan cada framework por su familiaridad general o por su capacidad para cumplir un propósito concreto. LlamaIndex, por ejemplo, se desarrolló originalmente para simplificar la configuración de la generación aumentada por recuperación (RAG), mientras que CrewAI se creó para automatizar tareas definidas en la era de los agentes. Objetivos de diseño distintos producen estructuras de respuesta diferentes, y cada una requiere un tratamiento específico. Transmitir fragmentos a medida que el LLM los genera, en lugar de esperar a que finalice el turno, es fundamental porque permite que el modelo de Texto a Voz (TTS) empiece a generar voz antes y reduzca así la latencia percibida. Nos centramos en cuatro frameworks populares: LangGraph, Google ADK, CrewAI y LlamaIndex.

Nota sobre el código compartido

Cada framework debe transmitir respuestas como fragmentos SSE compatibles con OpenAI. Presentamos una pequeña función auxiliar que se usa en todos los ejemplos para construir estos fragmentos.

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"

Con esta base, empecemos con LangGraph. 

LangGraph

LangGraph modela agentes como grafos, donde los nodos representan pasos individuales y las aristas definen el flujo de control entre ellos. La configuración mínima es sencilla: inicializa un modelo de chat, define las herramientas del agente y crea el entorno de ejecución del grafo de agentes.

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

Para cada solicitud de generación, el agente de LangGraph recibe todo el historial de la conversación, lo que le permite mantener internamente el estado necesario. LangGraph admite persistencia en el servidor mediante Checkpoints, aunque no los tratamos aquí para mantener una implementación mínima.

Una vez resuelta la gestión del estado, la siguiente decisión específica de LangGraph es el modo de streaming, para el que ofrece dos opciones, cada una adecuada para un caso de uso distinto:

• stream_mode="values" proporciona instantáneas del estado del grafo. Es más sencillo de implementar, pero incluye un estado de mensajes más completo en cada respuesta, lo que añade latencia a los flujos conversacionales en tiempo real.
• stream_mode="messages" transmite fragmentos incrementales de mensajes del modelo. En general, es preferible para interacciones de voz en tiempo real, ya que reduce el tiempo hasta el primer audio en la capa de orquestación de ElevenLabs.

Más concretamente, la implementación de mensajes del bucle del agente incluye pasos intermedios, como actualizaciones de llamadas a herramientas, que no deberían pronunciarse en voz alta. El proxy las filtra y solo pasa el texto de respuesta dirigido al usuario a la capa de TTS. Veamos un ejemplo de un turno que usa una herramienta.

[1] El modelo decide llamar a una herramienta (tool_calls=["get_price"])
[2] La herramienta se ejecuta y devuelve datos (result="$24.99") 
[3] El modelo genera una respuesta con el resultado (content="Cuesta $24.99") 

Naturalmente, solo los fragmentos del paso 3 deben reenviarse en el stream SSE. En la práctica, dos comprobaciones gestionan este filtrado en el bucle de streaming: una para conservar solo los eventos langgraph_node == "model" y otra para omitir el contenido vacío. Juntas, garantizan que solo el texto del asistente dirigido al usuario se reenvíe a ElevenLabs como SSE. Con estos conceptos reunidos, presentamos una implementación ligera del proxy de solicitudes.

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

Esto garantiza que solo se reenvíen a ElevenLabs los fragmentos del modelo dirigidos al usuario. Como LangGraph muestra la ejecución interna de sus herramientas a través del stream de estado, el filtrado es explícito y lo controla el proxy. 

A continuación, veremos los matices de trabajar con el Agent Development Kit (ADK) de Google.

Google ADK

El ADK de Google abstrae el bucle de ejecución tras unas pocas primitivas principales: Agent, Runner y SessionService. El Runner de ADK se sitúa entre la capa HTTP y la definición del agente. Gestiona el enrutamiento de mensajes, la orquestación de herramientas, el ciclo de vida de las sesiones y el 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
)

Con el agente, el backend de sesiones y el runner inicializados, el proxy resuelve o crea una sesión de ADK para cada solicitud entrante. En ADK, session_id controla la persistencia de la memoria: reutilizar el mismo session_id entre turnos conserva automáticamente el historial, las llamadas a herramientas y las respuestas anteriores. Como la identidad de la conversación se encuentra aguas arriba en ElevenLabs, el proxy gestiona este mapeo de forma explícita. Al pasar el identificador correcto para la solicitud de generación, el SDK puede gestionar internamente el contexto anterior. Pasamos el identificador arbitrario durante el inicio de la conversación mediante parámetros adicionales enviados en el cuerpo de la solicitud.  

Con el mensaje y la sesión preparados, se puede invocar el runner. Las llamadas a herramientas y sus resultados siguen apareciendo como eventos internos de ADK durante la ejecución, pero se tratan como pasos intermedios de orquestación, no como salida dirigida al usuario. Esto elimina la necesidad de un filtro manual en comparación con los frameworks en los que las llamadas a herramientas aparecen como texto visible para el usuario. 

El siguiente controlador es una implementación simplificada que incluye en línea la resolución de sesiones y la lógica de obtener o crear.

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

A continuación, veremos CrewAI, que por diseño está más centrado en tareas.

CrewAI

CrewAI se diseñó para orquestar flujos de trabajo multiagente en torno a tareas estructuradas (investigar, escribir, resumir), en lugar de bucles de diálogo abiertos. Los agentes se definen con un rol, un objetivo y un trasfondo. La ejecución se centra en objetos Task, cada uno con una descripción clara y una salida 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,
)

A diferencia del modelo de bucle de agentes utilizado en LangGraph y ADK, CrewAI suele construir Task y Crew por solicitud para definir la unidad de trabajo de ese turno de la conversación. Mantenemos el contexto conversacional inyectando los turnos anteriores en la siguiente tarea mediante un marcador de posición. La variable {crew_chat_messages} se completa en cada solicitud con el historial de conversación acumulado y, después, se interpola en la descripción de la tarea durante la ejecución. También buscamos producir texto limpio y listo para locución filtrando explícitamente patrones de trazado intermedios (Thought, Action, Action Input, Observation) y emitiendo solo el texto de la respuesta final. 

El siguiente controlador reúne la construcción de tareas por solicitud, la interpolación del historial, el streaming a nivel de Crew, el filtrado de trazas y el formato de salida. 

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

A continuación, veremos LlamaIndex, que sigue una vía distinta centrada en un modelo de streaming nativo basado en eventos.

LlamaIndex

A diferencia de los demás frameworks tratados en esta publicación, LlamaIndex se diseñó para conectar LLM a fuentes de datos externas (almacenes de documentos, índices y pipelines de recuperación). Su capa de agentes, FunctionAgent, se apoya en esa base para recuperar información y razonar sobre contexto estructurado, en lugar de gestionar diálogos abiertos o ejecutar tareas.

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 la continuidad conversacional, el proxy transforma los mensajes entrantes en mensajes de chat de LlamaIndex y luego los divide entre el último turno del usuario (user_msg) y los turnos anteriores (chat_history). El campo event.delta de cada evento AgentStream contiene el siguiente fragmento de texto, que se asigna directamente a un fragmento delta.content al estilo de OpenAI. Los deltas no vacíos se pueden reenviar tal cual, lo que convierte este en el puente de streaming más directo de la guía. El stream contiene tanto eventos de orquestación (llamadas a herramientas y resultados) como eventos de voz (deltas de texto del asistente). Para mantener limpia la salida de voz, el proxy conserva solo los eventos AgentStream y omite los deltas vacíos.

[1] AgentStream (delta='')       ← omitido
[2] ToolCall                     ← omitido
[3] ToolCallResult               ← omitido
[4] AgentStream (delta='Esto')   ← reenviado ✓
[5] AgentStream (delta=' cuesta')← reenviado ✓
[6] AgentStream (delta=' $49.99')← reenviado ✓

Esta separación mantiene la mecánica intermedia de las herramientas fuera de la salida hablada y preserva la voz incremental de baja latencia. El controlador listo para usar que aparece a continuación reúne estos pasos.

@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 es menos prescriptivo sobre los patrones de ejecución conversacional de extremo a extremo que los frameworks con capas de orquestación integradas más completas. Para implementaciones de producción, normalmente requiere que clientes implementen la gestión de sesiones, medidas de seguridad para las respuestas, orquestación de herramientas y trazado.

Conclusión

Cada framework de esta guía se conecta a ElevenLabs mediante el mismo contrato: acepta una solicitud de Completions o Responses al estilo de OpenAI y devuelve fragmentos SSE en streaming. Esto permite a los equipos añadir orquestación de voz a una implementación de agentes existente con cambios mínimos, conservando lo que ya han creado y habilitando la IA conversacional en tiempo real. Esta modularidad es un principio fundamental de la plataforma ElevenAgents. Tanto si las organizaciones amplían un agente existente como si crean una solución nativa de voz desde el principio, la orquestación de voz de ElevenAgents está diseñada para adaptarse a sus necesidades.

Si ya utilizas un agente con un framework de código abierto y quieres habilitar la voz, prueba este enfoque y cuéntanos qué te parece.