Guía práctica: frameworks de agentes open-source y ElevenAgents

En nuestro artículo anterior sobre Cómo integrar 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 el LLM personalizado. A partir de esa base, esta guía muestra cómo adaptar e implementar los principales frameworks de agentes de código abierto detrás de la interfaz de LLM personalizado. El resultado es una arquitectura flexible que añade 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 de eventos enviados por el servidor (SSE) compatible con OpenAI. ElevenLabs admite los formatos Chat Completions y Responses. Aunque esta guía abarca cuatro frameworks muy utilizados, los patrones se pueden aplicar a cualquier entorno de ejecución que genere 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 LLM personalizado configurada. Esta sección explica los componentes básicos 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 fin 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. Cada objetivo de diseño genera estructuras de respuesta distintas y requiere un tratamiento específico. Transmitir fragmentos a medida que el LLM los genera, en lugar de esperar a que termine un turno completo, 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 utiliza 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: inicializar un modelo de chat, definir las herramientas del agente y crear 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,
)

En cada solicitud de generación, el agente de LangGraph recibe todo el historial de conversación, lo que le permite mantener internamente el estado necesario. LangGraph admite persistencia en el servidor mediante puntos de control, aunque no los abordamos aquí para mantener la implementación al mínimo.

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 desde el 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 deben pronunciarse. 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 usando el resultado (content="Cuesta $24.99")

Naturalmente, solo deben reenviarse en el flujo SSE los fragmentos del paso 3. En la práctica, dos comprobaciones gestionan este filtrado en el bucle de streaming: una para conservar únicamente los eventos langgraph_node == "model" y otra para omitir contenido vacío. Juntas, estas comprobaciones garantizan que solo el texto del asistente dirigido al usuario se reenvíe a ElevenLabs como SSE. A continuación, reunimos estos conceptos en 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 flujo de estado, el filtrado es explícito y está controlado por el proxy.

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

Google ADK

El ADK de Google abstrae el bucle de ejecución mediante varias primitivas básicas: 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
)

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

Con el mensaje y la sesión preparados, se puede invocar el runner. Durante la ejecución, las llamadas a herramientas y sus resultados siguen apareciendo como eventos internos de ADK, 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 frameworks donde las llamadas a herramientas aparecen como texto visible para el usuario.

El controlador siguiente 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, redactar, resumir), en lugar de bucles de diálogo abiertos. Los agentes se definen con un rol, un objetivo y una historia de fondo. La ejecución se centra en objetos Task, cada uno con una descripción clara y un resultado esperado.

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 agente utilizado en LangGraph y ADK, CrewAI suele crear una Task y un Crew por solicitud para definir la unidad de trabajo de ese turno de la conversación. Conservamos el contexto conversacional inyectando los turnos anteriores en la siguiente tarea mediante un marcador de posición. La variable {crew_chat_messages} se rellena en cada solicitud con el historial acumulado de la conversación y después se interpola en la descripción de la tarea durante la ejecución. Además, buscamos generar texto limpio y listo para voz filtrando explícitamente patrones de trazado intermedios (Thought, Action, Action Input, Observation) y emitiendo solo el texto de la respuesta final.

El controlador siguiente reúne la creació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 un enfoque distinto centrado en un modelo de streaming nativo basado en eventos.

LlamaIndex

A diferencia de los demás frameworks tratados en este artículo, LlamaIndex se diseñó para conectar LLM con fuentes de datos externas (repositorios de documentos, índices y canalizaciones de recuperación). Su capa de agentes, FunctionAgent, se apoya en esa base para recuperar información y razonar sobre contexto estructurado, en vez de centrarse en diálogos abiertos o en la ejecución de 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 después los divide entre el último turno de 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 de estilo 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 flujo 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='') ← ignorado
[2] ToolCall ← ignorado
[3] ToolCallResult ← ignorado
[4] AgentStream (delta='Esto') ← reenviado ✓
[5] AgentStream (delta=' cuesta')← reenviado ✓
[6] AgentStream (delta=' $49.99')← reenviado ✓

Esta separación mantiene los mecanismos intermedios de las herramientas fuera de la salida de voz, al tiempo que conserva una generación de 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. En implementaciones de producción, esto suele requerir que clientes implementen la gestión de sesiones, medidas de protección para las respuestas, la orquestación de herramientas y el 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 agente existente con cambios mínimos, preservando así lo que ya han creado y habilitando 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 desarrollan soluciones nativas de voz desde el principio, la orquestación de voz de ElevenAgents está diseñada para adaptarse a su punto de partida.

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.