Armá un Agente de Voz en 2026: Pipeline con LiveKit, Tool Calls y Presupuesto de Latencia

1. El Problema: la Voz es un Pipeline con Reloj

Un turno de voz cruza al menos cinco etapas: la red, la detección de actividad de voz, el speech-to-text, el LLM y el text-to-speech. El usuario escucha silencio durante la suma de todo lo que no superpusiste. La documentación de LiveKit dice que una conversación se siente natural cuando la latencia de punta a punta queda por debajo de un segundo, y la introducción de Pipecat dice que sus pipelines suelen caer entre 500 y 800 ms. Ambas son afirmaciones de proveedor, no garantías, y ninguna conoce tu STT, tu modelo ni tu región.

La solución es estructural, no un modelo más rápido: streaming en cada etapa, decidir los límites del turno con algo más listo que un temporizador de silencio, y tratar la interrupción como un evento de primera clase. Eso es lo que te dan los frameworks de abajo, y por eso no armaría este loop a mano.

Para la alternativa speech-to-speech y su precio por minuto, mirá mi post de Grok Voice 3.0. Este es el camino del pipeline, armado a mano.

2. Conceptos Mínimos Antes de Tocar Código

Seis ideas explican casi todas las perillas que vas a tocar.

3. LiveKit Agents vs Pipecat: Qué Verifiqué

Ambos son frameworks Python open source para exactamente este pipeline, y ambos son buenos. La tabla lista solo hechos que leí en la documentación oficial, en tags del código fuente o en índices de paquetes el 14 de octubre de 2026.

Uso LiveKit para el ejemplo por una sola razón: su quickstart es un archivo único y completo, así que pude chequear cada llamada contra el tag 1.8.7. El quickstart de los docs de Pipecat muestra fragmentos, y sus docs y su repositorio de quickstart hoy usan nombres de clase distintos (ver la nota debajo de la tabla).

4. Paso 1: Armá el Agente y Corré la Base

Todo en este paso sale del quickstart de Voice AI de LiveKit. No lo corrí contra un proyecto real: lo copié de la documentación y revisé imports y firmas contra el tag livekit-agents 1.8.7 en GitHub.

# 1) LiveKit CLI (pick your OS), then link your LiveKit Cloud project
winget install LiveKit.LiveKitCLI             # Windows
brew install livekit-cli                      # macOS
curl -sSL https://get.livekit.io/cli | bash   # Linux
lk cloud auth

# 2) Scaffold the official Python starter and run it
lk agent init my-agent --template agent-starter-python
cd my-agent
uv sync
lk agent console   # talk to it in your terminal (add --text for text only)
lk agent dev       # connect it to LiveKit and use the Agent Console

from dotenv import load_dotenv

from livekit import agents
from livekit.agents import AgentServer, AgentSession, Agent, inference, room_io, TurnHandlingOptions
from livekit.plugins import ai_coustics

load_dotenv(".env.local")


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor.""",
        )

server = AgentServer()

@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: agents.JobContext):
    session = AgentSession(
        stt=inference.STT(model="assemblyai/universal-3-6-pro", language="en"),
        llm=inference.LLM(model="google/gemma-4-31b-it"),
        tts=inference.TTS(
            model="fishaudio/s2.1-pro",
            voice="fa4c9eb3dccc4806b382b40d61c6b10a",
        ),
        turn_handling=TurnHandlingOptions(
            turn_detection=inference.TurnDetector(),
        ),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=ai_coustics.audio_enhancement(model=ai_coustics.EnhancerModel.QUAIL_VF_S),
            ),
        ),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )


if __name__ == "__main__":
    agents.cli.run_app(server)

Leé el archivo como un pipeline. inference.STT, inference.LLM e inference.TTS son las tres etapas en streaming, servidas por LiveKit Inference: no necesitás keys de proveedores pero sí un proyecto de LiveKit Cloud. inference.TurnDetector() es el modelo de fin de turno, y la sesión aporta el VAD que necesita. Las instrucciones le piden al modelo respuestas breves sin emojis ni asteriscos, porque la salida se habla; además la sesión quita markdown y emojis por defecto en 1.8.7.

5. Paso 2: Turnos, VAD y Barge-In

El manejo de turnos es donde un agente de voz se siente humano o robótico, y en 1.8.x vive en un solo objeto: TurnHandlingOptions. Esta es la forma de la configuración inicial recomendada por LiveKit, recortada a lo que importa.

from livekit.agents import AgentSession, TurnHandlingOptions, inference

session = AgentSession(
    stt=inference.STT(model="assemblyai/universal-3-6-pro", language="en"),
    llm=inference.LLM(model="google/gemma-4-31b-it"),
    tts=inference.TTS(model="fishaudio/s2.1-pro", voice="fa4c9eb3dccc4806b382b40d61c6b10a"),
    turn_handling=TurnHandlingOptions(
        # audio model that predicts end of turn on top of VAD
        turn_detection=inference.TurnDetector(),
        # wait after detected silence before closing the turn
        # (0.3 / 2.5 are the documented defaults with the audio turn detector)
        endpointing={"mode": "fixed", "min_delay": 0.3, "max_delay": 2.5},
        interruption={
            "mode": "adaptive",   # default on LiveKit Cloud; use "vad" when self-hosting
            "min_duration": 0.5,  # seconds of speech that count as a barge-in
            "min_words": 0,       # raise it (needs STT) to ignore "okay", "uh-huh"
            "resume_false_interruption": True,
            "false_interruption_timeout": 2.0,
        },
        # on by default; preemptive TTS trades wasted compute for lower latency
        preemptive_generation={"preemptive_tts": False},
    ),
)

# uv add "livekit-agents[silero,deepgram,openai,cartesia]~=1.8"
# env: DEEPGRAM_API_KEY, OPENAI_API_KEY, CARTESIA_API_KEY
from livekit.agents import AgentSession, TurnHandlingOptions, inference
from livekit.plugins import cartesia, deepgram, openai, silero

session = AgentSession(
    vad=silero.VAD.load(),
    stt=deepgram.STT(model="nova-3", language="en"),
    llm=openai.responses.LLM(model="gpt-4.1"),
    tts=cartesia.TTS(model="sonic-3", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"),
    turn_handling=TurnHandlingOptions(
        turn_detection=inference.TurnDetector(version="v1-mini"),  # local CPU
    ),
)

Tres defaults que conviene conocer, todos de la documentación de LiveKit: con el audio turn detector la sesión cierra un turno tras 0,3 s mínimo y 2,5 s máximo, contra 0,5 s y 3,0 s sin él; la generación preventiva está activa y la de TTS desactivada; y el manejo adaptativo de interrupciones, que filtra backchannel con un modelo de audio, es una función de LiveKit Cloud.

6. Paso 3: Function Calling en Plena Llamada sin Silencios Muertos

Un tool call en un agente de voz es un silencio que el usuario escucha. LiveKit te da tres comportamientos, y elegir el correcto por tool es casi todo el trabajo: una consulta corta de solo lectura que cancelás si el usuario interrumpe, una escritura que nunca debe quedar a medias y un trabajo lento que mantiene al agente hablando.

import asyncio

from livekit.agents import Agent, RunContext, function_tool
from livekit.agents.llm import ToolError


class SupportAgent(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a phone support agent. Short sentences, no formatting.")

    # 1) Short read-only lookup: stop the work if the user talks over the agent
    @function_tool
    async def get_order_status(self, order_id: str, run_ctx: RunContext) -> str | None:
        """Look up the status of an order when the user asks about it.

        Args:
            order_id: The ID of the order to look up.
        """
        lookup = asyncio.ensure_future(self._lookup_order(order_id))  # your backend call
        await run_ctx.speech_handle.wait_if_not_interrupted([lookup])
        if run_ctx.speech_handle.interrupted:
            lookup.cancel()  # Python does not cancel it for you
            return None      # empty result, no reply
        return lookup.result()

    # 2) Write: a barge-in must not leave it half done
    @function_tool
    async def cancel_order(self, order_id: str, run_ctx: RunContext) -> str:
        """Cancel an order. Call only after the user confirmed out loud.

        Args:
            order_id: The ID of the order to cancel.
        """
        run_ctx.disallow_interruptions()
        ok = await self._cancel_in_backend(order_id)  # your backend call, with a timeout
        if not ok:
            raise ToolError("The order could not be cancelled. Offer to transfer the caller to a person.")
        return f"Order {order_id} is cancelled."

    # 3) Slow job: acknowledge now, fill the silence, answer when done
    @function_tool()
    async def export_statement(self, ctx: RunContext, month: str) -> str:
        """Generate a statement and email it. Takes up to a minute.

        Args:
            month: Month to export, YYYY-MM.
        """
        await ctx.update(f"Generating the statement for {month}. This can take a minute.")
        async with ctx.with_filler("Still working on it, hang on a sec.", delay=5):
            await self._build_statement(month)  # your slow backend call
        return f"The statement for {month} was sent to the email on file."

7. Paso 4: Medí Cada Turno Antes de Afinar

LiveKit adjunta un reporte de métricas a cada mensaje del chat. Los campos mapean uno a uno con la tabla de presupuesto, así ves qué etapa se comió el turno.

from livekit.agents import ConversationItemAddedEvent
from livekit.agents.llm import ChatMessage


@session.on("conversation_item_added")
def on_item(ev: ConversationItemAddedEvent):
    if not isinstance(ev.item, ChatMessage):
        return
    m = ev.item.metrics  # per-turn MetricsReport
    if ev.item.role == "user":
        print("stt:", m.get("transcription_delay"), "eot:", m.get("end_of_turn_delay"))
    elif ev.item.role == "assistant":
        print(
            "llm ttft:", m.get("llm_node_ttft"),
            "tts ttfb:", m.get("tts_node_ttfb"),
            "e2e:", m.get("e2e_latency"),
        )

Registrá esto por turno y mirá p50 y p95 por etapa en unas pocas decenas de llamadas reales desde tu región objetivo. llm_node_ttft y tts_node_ttfb solo se completan en modo pipeline. Las notas de la versión 1.8.7 sumaron la misma latencia de punta a punta para agentes realtime cuando hay VAD local configurado. Cambiá una perilla por vez: la propia página de tuning de LiveKit avisa que la generación preventiva no siempre reduce la latencia.

8. Llamadas Telefónicas: el Traspaso a Nivel Conceptual

El código del agente no cambia cuando quien llama está en un teléfono: un llamador SIP es un participante más en una sala. Lo que cambia es la puerta de entrada y qué hacés cuando el agente debe dejar de atender.

9. Modos de Falla: Eco, Salas Ruidosas, Tool Calls Largos

Las demos esconden estos tres. Las llamadas reales los encuentran en la primera semana.

10. Errores Comunes (y los Arreglos Aburridos)

Casi todos los bugs de agentes de voz sobre los que leí caen en estos baldes. La columna izquierda es lo que aguanta; la derecha es lo que te despierta.

Conclusión

Ya tenés el loop completo: un pipeline STT → LLM → TTS en streaming sobre LiveKit Agents, manejo de turnos sobre el que podés razonar, tools que se portan bien bajo interrupción, un camino telefónico en borrador y una métrica por etapa. Pipecat te daría la misma forma con otros nombres y otra historia de transporte, y la tabla comparativa está para que elijas sin guerra de benchmarks.

Esta semana: generá el quickstart, sumá un tool de solo lectura con manejo de interrupción, registrá las cuatro métricas por turno en 20 llamadas y completá tu propia tabla de presupuesto. Después decidí si el pipeline o el modelo speech-to-speech de mi post anterior gana con tu tráfico.

Fuentes