Realtime and live models

• What it gives you - what goes wrong without it
• Quickstart - an agent on a realtime model
• What is supported - features by framework and model kind
• How a fixed line is said - lines with no TTS
• Instructions and values - what the model reads
• Handoffs - keep the whole conversation
• What is refused - and the words you will see
• Turn detection - set in your own code
• On Pipecat - extra wiring
• What is not here - gaps
• Where to go next

​

What it gives you

​

Quickstart

from livekit.agents import AgentSession, JobContext
from livekit.plugins import openai
from pydantic import BaseModel

from unmute_core.agent import AgentSpec
from unmute_core.call import CallHooks
from unmute_core.greeting import Greeting
from unmute_core.livekit import UnmuteAgent, open_call
from unmute_core.state import CallState
from unmute_core.tools import ToolRegistry, local_tool

PROMPT = "You answer calls for a corner shop. Keep replies short: they are spoken aloud."


class State(BaseModel):
    order_id: str | None = None


class OrderStatus(BaseModel):
    order_id: str
    status: str


def order_status(order_id: str) -> OrderStatus:
    """Look up an order by its id and return where it is."""
    return OrderStatus(order_id=order_id, status="out for delivery")


async def entrypoint(ctx: JobContext) -> None:
    session = AgentSession(llm=openai.realtime.RealtimeModel(voice="marin"))
    state = CallState(State)
    call = CallHooks(state)
    tools = ToolRegistry([local_tool(order_status)])
    spec = AgentSpec(PROMPT, tools=tools, save_values=True)
    agent = UnmuteAgent(spec, state=state, call=call)
    await session.start(agent=agent, room=ctx.room)
    await open_call(session, Greeting(text="Hi, how can I help?"), call)

import os

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService
from pydantic import BaseModel

from unmute_core.agent import AgentSpec
from unmute_core.call import CallHooks
from unmute_core.greeting import Greeting
from unmute_core.pipecat import UnmuteFlows, open_call
from unmute_core.state import CallState
from unmute_core.tools import ToolRegistry, local_tool

PROMPT = "You answer calls for a corner shop. Keep replies short: they are spoken aloud."


class State(BaseModel):
    order_id: str | None = None


class OrderStatus(BaseModel):
    order_id: str
    status: str


def order_status(order_id: str) -> OrderStatus:
    """Look up an order by its id and return where it is."""
    return OrderStatus(order_id=order_id, status="out for delivery")


async def run_call(transport) -> None:
    # transport comes from your own Pipecat runner
    llm = OpenAIRealtimeLLMService(api_key=os.environ["OPENAI_API_KEY"])
    context = LLMContext()
    aggregators = LLMContextAggregatorPair(context)
    state = CallState(State)
    call = CallHooks(state)
    tools = ToolRegistry([local_tool(order_status)])
    spec = AgentSpec(PROMPT, tools=tools, save_values=True)
    flows = UnmuteFlows(
        spec, llm=llm, context=context, aggregators=aggregators, state=state, call=call
    )
    flows.install()
    pipeline = Pipeline(
        [transport.input(), aggregators.user(), llm, transport.output(), aggregators.assistant()]
    )
    worker = PipelineWorker(pipeline)

    @transport.event_handler("on_client_connected")
    async def greet(transport, client) -> None:
        await open_call(
            worker, Greeting(text="Hi, how can I help?"), call, context=context, llm=llm
        )

    # run the worker with your own WorkerRunner, then: await call.end("ended")

​

What is supported

​

How a fixed line is said

1. The library has fixed lines: the greeting, an idle nudge, the failure line, the goodbye of end_call_tool, a handoff’s line, and the lines a transfer says once it is over.
2. With a TTS on the call, as in a half cascade, the TTS says them. Nothing changes from a cascade call.
3. A realtime model with no TTS beside it speaks itself. LiveKit’s session.say would raise, and Pipecat’s TTSSpeakFrame would play nothing.
4. So the model is told to say the line: “Say this to the caller now, in these words as far as you can: …”. This text is SAY_LINE in unmute_core.realtime.
5. The line is a reply of the model’s own. It joins the model’s conversation.

​

Instructions and values

• The prompt hooks run on the prompt before it is filled, as on a cascade call. See Hooks.
• After each save, the model gets the instructions again: LiveKit’s update_instructions, or Pipecat’s LLMUpdateSettingsFrame. A value a background prefetch saves arrives the same way.
• Each caller message runs the call’s tools on events. The model answers that message meanwhile and nothing waits for them. Their notes reach the model with the next instructions. See The call’s life.
• A live model gets the instructions once. On LiveKit that is as the agent starts. On Pipecat it is as UnmuteFlows or attach_realtime is set up. A value saved later never reaches it, prefetch included, and neither do the notes of tools on events.
• A plain Pipecat pipeline without UnmuteFlows gets the same with attach_realtime, before the pipeline runs.

​

attach_realtime

await attach_realtime(llm, worker, PROMPT, state, call=call)

​

Handoffs

return await self.hand_off(
    UnmuteAgent(billing, state=self.state), context=HandoffContext(history="full")
)

1. Pass HandoffContext(history="full"). Any other history is refused, because the library cannot reshape what the model holds. Tool calls must stay in, which is the default.
2. The model says the handoff’s line first.
3. The next agent’s instructions and tools replace the first agent’s, on the same connection.
4. The next agent speaks first.

​

What is refused and why

• Tasks. A step needs its own history and a finish check. Both are requests the library sends, and a realtime model has none.
• before_model, wrap_model and after_model hooks. No model request passes through the library, so they would never run. The prompt hooks do run.
• save_values on a live model. Its instructions are fixed.
• A handoff on a live model, or one without history="full". See above.

​

Turn detection

• endpointing(), vad_params() and smart_turn_params(). The model’s own VAD sets the pace.
• interruption(enabled=False) on LiveKit. It raises when the session starts, because LiveKit cannot stop the model from being cut off.
• Pipecat’s interruption strategies, like Backchannels. They act on local VAD frames, and the server cancels its reply itself.

​

On Pipecat

• Pass the service to open_call(llm=...), with UnmuteFlows too, and to IdleNudgeObserver(llm=...). They then say the greeting, the failure line and the nudges through the model. Left out, these lines become TTSSpeakFrames that nothing plays, and nothing reports it.
• Any other realtime service, like Gemini Live, is taken for a cascade LLM. Its fixed lines are not said.
• A greeting goes to the model as a developer message. Pipecat drops a first system message when the service has its own instructions.
• When the caller speaks first, open_call gives the service its context and tools at once. Without that, Pipecat sets the conversation up in the middle of the caller’s first turn. It sends the turn again as a “previously saved conversation”, asks for a second reply, and the model gets its tools only then.
• A live service opens its session on a first request. When the caller speaks first, open_call still sends one.
• A line said while the model replies waits for the reply to end. The server refuses a second reply while one runs.
• TelemetryObserver records one llm span per reply. Pipecat starts a realtime reply twice, and starts none for a reply that only calls a tool. The observer handles both.

​

What is not here

• Gemini Live, Nova Sonic, Ultravox, Grok and Inworld on Pipecat. Their services take instructions, tools and replies in other ways. On LiveKit any RealtimeModel works.
• The SLNG Context Router. It has no realtime endpoint.
• Building the model. Its voice, its turn detection and its transcription stay in your code.

​

Where to go next