Speech-to-Speech Orchestration

Multi-agent orchestration using Gemini Live (or OpenAI Realtime) speech-to-speech, where the AI provider handles audio recognition, reasoning, and speech generation natively — no STT/LLM/TTS pipeline.

Overview

Traditional orchestration routes text events between Agent channels backed by an AIProvider. Each agent is an intelligence channel in the room, and VoiceChannel handles the audio conversion (STT on the way in, TTS on the way out).

Speech-to-speech orchestration uses the same Agent definitions — same role, description, scope, voice, system_prompt — but wires them into a RealtimeVoiceChannel instead. The AI reasoning happens inside the provider (Gemini Live), not in a separate channel. On handoff, the framework reconnects the provider session with the new agent's configuration, preserving conversation history via Gemini's session resumption.

Traditional (STT → LLM → TTS):
  User speaks → VoiceChannel (STT) → text → Agent (AIChannel) → text → VoiceChannel (TTS)

Speech-to-speech:
  User speaks → RealtimeVoiceChannel → Gemini Live (audio in → AI → audio out) → User hears
                                         ↑
                                    Agent config (system_prompt, voice, tools)
                                    swapped on handoff via session resumption

The developer writes nearly identical code for both modes. The only difference is whether Agent has a provider or not:

# Traditional (STT → LLM → TTS)
triage = Agent(
    "agent-triage",
    provider=GeminiAIProvider(config),    # <-- has AI provider
    role="Triage receptionist",
    system_prompt="Greet callers warmly. Under 30 words.",
    voice="en-US-Journey-F",             # TTS voice ID
)
voice = VoiceChannel("voice", stt=deepgram, tts=elevenlabs)

# Speech-to-speech
triage = Agent(
    "agent-triage",                       # <-- no provider (config-only)
    role="Triage receptionist",
    system_prompt="Greet callers warmly. Under 30 words.",
    voice="Aoede",                        # Gemini native voice
)
voice = RealtimeVoiceChannel("voice", provider=GeminiLiveProvider(...), transport=...)

Both use the same pipeline.install() call. The framework detects the channel type and wires things differently under the hood.

How it works

Handoff via session resumption

Gemini Live does not support mid-session configuration changes. However, it supports session resumption with different configuration — you can disconnect and reconnect with a new system_instruction, voice, and tools while preserving the conversation history. The model parameter is the only thing that cannot change.

On handoff, the framework:

1. Builds the new agent's config (system_prompt + identity block, voice, handoff tools)
2. Disconnects the current Gemini session
3. Reconnects with the new config + the session resumption handle
4. Gemini remembers the full conversation but responds with the new agent's personality/voice

The reconnect takes ~200-500ms — a natural pause, similar to a phone transfer.

Wiring

When pipeline.install() detects that voice_channel_id points to a RealtimeVoiceChannel:

1. Agents are NOT registered as channels — they have no provider, they're config-only
2. A session starts with its room's active agent — system_prompt, voice, and tools are passed to provider.connect(), read from the room's conversation state when the session opens. The tools are the channel's own, as they are when the session opens or the handoff lands (a channel reconfigured after the install keeps them under every agent), the agent's own (Agent(tools=...)), and the handoff tool. A name the channel's tools carry (its host tools, its human-input tools, and the ones it serves itself: Tool Search's, the skills') is the channel's: an agent tool of the same name is neither declared nor served, and a warning names it at the install; it stays undeclared if the channel later drops its own, since nothing of the agent's serves it. The channel's own system_prompt, voice and tools are left as they are
3. The handoff tool is served by the RealtimeVoiceChannel — Gemini calls it natively via its built-in tool calling. A call to one of the active agent's own tools is served by the tool_handler the agent was given; a call to a channel tool and to an agent tool the agent has no handler for, by the channel's. The active agent's tool_policy holds on its session beside the channel's: a tool either denies is not declared, nor listed by Tool Search, and a call to it is refused with the policy's words. A handoff applies the next agent's policy to every session of the room, for its participant's role, and each call is judged by the agent the room talks to, the one that serves it. An allow-list policy hides the handoff tool too, as on text turns: allow handoff_conversation for an agent that hands off. An agent that carries what a realtime session never serves for an agent is refused at the install (ValueError, before anything is installed), each cause named: its skills, a registry even empty (pass them to RealtimeVoiceChannel(skills=...)), its human_input_handler (pass it to RealtimeVoiceChannel(human_input_handler=...)), enable_planning, a sandbox or an external_tool_handler; a realtime session serves the channel's skills and human-input tools only. The agent's own tools stay served
4. On handoff, reconfigure_session() is called for the room's sessions — disconnects and reconnects with the new agent's config. Another room's sessions, and the next session of another room, keep their own room's agent

pipeline.install(kit, [triage, advisor], voice_channel_id="voice")

Internally:
  1. Does NOT call kit.register_channel() for agents
  2. Builds per-agent handoff tools (enum-constrained targets)
  3. Serves the handoff and the agents' tools on RealtimeVoiceChannel
  4. Starts each new session with its room's active agent
  5. HandoffHandler.handle() → rtv.reconfigure_session(session, new_config)
     for the sessions of the room that handed off

Agent identity in speech-to-speech

Agent identity (role, description, scope) is appended to the system_prompt passed to provider.connect(), just as it would be appended to the AIChannel's prompt in traditional mode. The AI sees the same identity block regardless of mode:

Greet callers warmly. Under 30 words.

--- Agent Identity ---
Role: Triage receptionist
Description: Routes callers to the right financial specialist
Scope: Financial advisory services only

Quick start

from roomkit import Agent, RealtimeVoiceChannel, RoomKit
from roomkit.orchestration import ConversationPipeline, PipelineStage
from roomkit.providers.gemini.realtime import GeminiLiveProvider
from roomkit.voice.backends.local import LocalAudioBackend

# 1. Define agents (no provider — config-only for speech-to-speech)
triage = Agent(
    "agent-triage",
    role="Triage receptionist",
    description="Routes callers to the right specialist",
    system_prompt="Greet callers. Ask what they need. Under 30 words.",
    voice="Aoede",
)

advisor = Agent(
    "agent-advisor",
    role="Portfolio advisor",
    description="Provides financial and investment advice",
    system_prompt="Give concise financial advice. Under 50 words.",
    voice="Charon",
)

# 2. Define pipeline (same as traditional orchestration)
pipeline = ConversationPipeline(
    stages=[
        PipelineStage(phase="triage", agent_id="agent-triage", next="advising"),
        PipelineStage(phase="advising", agent_id="agent-advisor", next=None),
    ],
)

# 3. Create RealtimeVoiceChannel with Gemini Live provider
kit = RoomKit()

provider = GeminiLiveProvider(api_key="...")
transport = LocalAudioBackend(
    input_sample_rate=24000,
    output_sample_rate=24000,
)

voice = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
)
kit.register_channel(voice)

# 4. One-liner: wires everything (detects RealtimeVoiceChannel automatically)
router, handler = pipeline.install(kit, [triage, advisor], voice_channel_id="voice")

# 5. Create room, start session — same as any realtime voice app
await kit.create_room(room_id="call-1")
await kit.attach_channel("call-1", "voice")
session = await voice.start_session("call-1", "caller-1", connection=None)

When the caller asks for investment advice, Gemini (as the triage agent) calls the handoff_conversation tool. The framework reconnects the Gemini session with the advisor's config — the caller hears the advisor's voice and personality without any gap in context.

Gemini Live constraints

Aspect	Supported	Notes
Change system_instruction on reconnect	Yes	Via session resumption
Change voice on reconnect	Yes	Via session resumption
Change tools on reconnect	Yes	Via session resumption
Change model on reconnect	No	Must be same model
Mid-session config update (no reconnect)	No	No session.update API
Conversation history preserved	Yes	Session resumption handle
Reconnect latency	~200-500ms	Natural handoff pause

Comparison with traditional orchestration

Aspect	Traditional (STT→LLM→TTS)	Speech-to-Speech
Agent definition	Agent(..., provider=AIProvider)	Agent(...) (no provider)
Voice channel	VoiceChannel with STT + TTS	RealtimeVoiceChannel
AI reasoning	In Agent (AIChannel)	In provider (Gemini Live)
Latency	Higher (STT + LLM + TTS)	Lower (native audio)
Voice quality	Depends on TTS provider	Native (Gemini voices)
Provider flexibility	Mix providers per agent	Single provider, all agents
Tool calling	Via AIChannel tool loop	Via provider native tools
Handoff mechanism	Route events to different Agent	Reconnect provider session
install() call	Identical	Identical

Implementation details

reconfigure_session() on RealtimeVoiceChannel

async def reconfigure_session(
    self,
    session: VoiceSession,
    *,
    system_prompt: str | None = None,
    voice: str | None = None,
    tools: list[dict[str, Any]] | None = None,
) -> None:
    """Switch agent config by reconnecting with session resumption.

    Disconnects the current provider session and reconnects with
    new configuration. Gemini preserves conversation history via
    the session resumption handle.
    """

It changes the session it is given, and nothing else (RFC §12.4): the channel's configuration for future sessions changes through configure(), never as a side effect of one session's reconfiguration.

reconfigure() on RealtimeVoiceProvider

async def reconfigure(
    self,
    session: VoiceSession,
    *,
    system_prompt: str | None = None,
    voice: str | None = None,
    tools: list[dict[str, Any]] | None = None,
    temperature: float | None = None,
    provider_config: dict[str, Any] | None = None,
) -> None:
    """Rebuild config and reconnect with session resumption."""

GeminiLiveProvider implements this by:

1. Building a new LiveConnectConfig from the provided parameters
2. Storing it as _live_configs[session.id]
3. Calling _reconnect(session) which uses the stored resumption handle

Config-only Agent

When Agent is created without a provider, it uses an internal _NullAIProvider placeholder. The agent is still an AIChannel instance (isinstance checks work), but calling generate() raises a clear error. This is intentional — config-only agents are never registered as channels.

# Config-only (speech-to-speech)
triage = Agent("agent-triage", role="Triage", system_prompt="...")
assert isinstance(triage, AIChannel)  # True
# triage._provider.generate() would raise RuntimeError

# Full agent (traditional)
triage = Agent("agent-triage", provider=GeminiAIProvider(config), role="Triage", ...)
# triage._provider.generate() works normally

install() detection logic

def install(self, kit, agents, *, voice_channel_id=None, ...):
    vc = kit._channels.get(voice_channel_id) if voice_channel_id else None

    if isinstance(vc, RealtimeVoiceChannel):
        # Speech-to-speech mode
        self._wire_realtime(kit, agents, vc)
    else:
        # Traditional mode (existing behavior)
        self._wire_handoff(agents, handler)
        if voice_channel_id:
            self._wire_voice_map(kit, agents, voice_channel_id)

In speech-to-speech mode, _wire_realtime():

1. Builds each agent's config with identity block, and its handoff tool; its tools (the channel's as they are, the agent's, the handoff tool) are read when a session opens or a handoff lands
2. Builds per-agent handoff tools (enum-constrained targets with descriptions)
3. Serves handoff_conversation and the agents' own tools on the RealtimeVoiceChannel
4. Gives the channel the config a new session of a room starts with: the room's active agent's
5. A handoff calls HandoffHandler.handle(), which triggers reconfigure_session() for that room's sessions

Examples

• examples/orchestration_realtime_triage.py — Voice triage with Gemini Live speech-to-speech