generate_reply()

generate_reply()

generate_reply()

generate_reply()

update_*

generate_reply()

// 1) Create session and seed initial history before starting
const session = await createLiveAgentSession({ model: "gemini-3-flash-live" });

// Seed initial messages as history (do this before starting/connecting)
await session.send_client_content({
  type: "history_seed",
  content: [
    { role: "system", content: "You are a helpful assistant." },
    { role: "user", content: "Hello — please greet me briefly." }
  ]
});

// 2) Start or connect the session
await session.start();

// 3) Listen for text output events (model response arrives here)
session.on("output", (evt) => {
  if (evt.type === "text") {
    console.log("Model text reply:", evt.text);
  }
});

// 1) Create session configured for audio with the Flash Live model
const session = await createLiveAgentSession({
  model: "gemini-3-flash-live",
  audio: { input: true, output: true }
});

// Seed initial history (before starting)
await session.send_client_content({
  type: "history_seed",
  content: [
    { role: "system", content: "You are a helpful voice assistant." },
    { role: "user", content: "Hello — please greet me briefly." }
  ]
});

// 2) Start session / connect audio streams
await session.start();

// 3) Listen for audio (TTS) output events
session.on("output", (evt) => {
  if (evt.type === "audio") {
    // evt.audio may be an audio buffer or stream chunk depending on SDK
    playAudioChunk(evt.audio);
  }
  if (evt.type === "text") {
    console.log("Model text (transcript):", evt.text);
  }
});

generate_reply()

output

generate_reply()

# Minimal example (pseudo-API names; adapt to your SDK)
from livekit_agent import LiveAgentSession

# 1) create session configured for gemini-3-flash-live
session = LiveAgentSession(model="gemini-3-flash-live", audio=False)

# 2) seed initial history BEFORE starting/connecting
session.send_client_content({
    "type": "history_seed",
    "content": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello — please greet me briefly."}
    ]
})

# 3) start/connect the session
session.start()

# 4) subscribe to output events to receive the model's reply
def on_output(evt):
    if evt.get("type") == "text":
        print("Model text reply:", evt.get("text"))

session.on("output", on_output)

# Minimal example (pseudo-API names; adapt to your SDK)
from livekit_agent import LiveAgentSession
import sounddevice as sd
import numpy as np

# 1) create session with audio input/output enabled
session = LiveAgentSession(model="gemini-3-flash-live", audio={"input": True, "output": True})

# 2) seed initial history BEFORE starting
session.send_client_content({
    "type": "history_seed",
    "content": [
        {"role": "system", "content": "You are a helpful voice assistant."},
        {"role": "user", "content": "Hello — please greet me briefly."}
    ]
})

# 3) start/connect the session and audio streams
session.start()

# 4) play audio chunks from output events
def play_audio_chunk(chunk_bytes):
    # chunk_bytes -> PCM float32 or int16 depending on SDK; adapt as needed
    audio = np.frombuffer(chunk_bytes, dtype=np.float32)
    sd.play(audio, samplerate=24000)

def on_output(evt):
    if evt.get("type") == "audio":
        play_audio_chunk(evt.get("audio"))
    if evt.get("type") == "text":
        print("Model transcript:", evt.get("text"))

session.on("output", on_output)

generate_reply()

evt["audio"]

AgentSession

Agent

livekit.agents

google.realtime.RealtimeModel

"gemini-3.1-flash-live-preview"

"gemini-3-flash-live"

tts

google.realtime.RealtimeModel

google.realtime.RealtimeModel

session.on(...)

session.subscribe(...)

session.stream()

google.realtime.RealtimeModel

session.on()

generate_reply()

import asyncio
from livekit.agents import AgentSession
from livekit.plugins import google

async def main():
    session = AgentSession(
        llm=google.realtime.RealtimeModel(
            model="gemini-3.1-flash-live-preview",
            instructions="You are a helpful assistant.",
            modalities=["text"],  # use text modality when using separate TTS
        ),
        tts="cartesia/sonic-3",
    )

    # seed initial history BEFORE start
    await session.send_client_content({
        "type": "history_seed",
        "content": [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Hello — please greet me briefly."}
        ],
    })

    # subscribe to output events
    def on_output(evt):
        if evt.get("type") == "text":
            print("Model text reply:", evt.get("text"))

    session.on("output", on_output)

    # start session (connects and begins model realtime processing)
    await session.start()

    # keep running to receive events (replace with your app lifecycle)
    await asyncio.sleep(10)
    await session.stop()

asyncio.run(main())

import asyncio
import numpy as np
import sounddevice as sd
from livekit.agents import AgentSession
from livekit.plugins import google

async def main():
    session = AgentSession(
        llm=google.realtime.RealtimeModel(
            model="gemini-3.1-flash-live-preview",
            instructions="You are a helpful voice assistant.",
            modalities=["text"],  # model handles input; TTS handles output
        ),
        tts="cartesia/sonic-3",
        audio={"input": True, "output": True},  # enable audio streams if needed
    )

    # seed initial history BEFORE start
    await session.send_client_content({
        "type": "history_seed",
        "content": [
            {"role": "system", "content": "You are a helpful voice assistant."},
            {"role": "user", "content": "Hello — please greet me briefly."}
        ],
    })

    # audio playback helper (adapt dtype/samplerate to your SDK's audio format)
    def play_audio_chunk(chunk_bytes):
        audio = np.frombuffer(chunk_bytes, dtype=np.float32)
        sd.play(audio, samplerate=24000)

    # subscribe to output events
    def on_output(evt):
        t = evt.get("type")
        if t == "text":
            print("Model transcript:", evt.get("text"))
        elif t == "audio":
            play_audio_chunk(evt.get("audio"))

    session.on("output", on_output)

    # start session
    await session.start()

    # keep running to receive events
    await asyncio.sleep(15)
    await session.stop()

asyncio.run(main())

generate_reply()

update_instructions()

update_chat_ctx()

session.start()

output