You're asking whether you can send the agent's speech audio alongside the user's audio so your conversational STT gets full context (you'll ignore the agent segments using diarization). Yes — you can do this by mixing/publishing both audio streams into the STT input. A concise plan and quick questions to tailor guidance follow. What to do (high-level) 1. Ensure the agent's synthesized speech is published as an audio track in the room (a LocalAudioTrack). 2. In the agent-side process, subscribe/access both the user's audio track and the agent's audio track. 3. Mix or multiplex those two audio streams into a single, time-aligned PCM stream (matching sample rate, channels, and timestamps). 4. Stream the combined audio to your STT streaming endpoint. 5. Use the STT's speaker-diarization output to filter out agent segments and forward user segments to downstream logic. Implementation notes and caveats • Keep the streams synchronized (timestamps) and use the same sample rate and channel layout before mixing. • Be mindful of added latency from mixing; design buffers to minimize delay. • Avoid feedback loops (agent audio being re-captured as user audio) by routing synthesized agent audio directly to the mixer rather than re-recording from playback. • If real-time mixing is too heavy, recording/egress captures both sides but is not real-time. Quick questions to tailor exact steps • Which LiveKit client SDK or server language are you using (JavaScript, Python, Go, etc.)? • Are you using a specific STT plugin/provider (Speechmatics, other)? • Do you need sample code for subscribing, mixing PCM streams, or for streaming into the STT API? Tell me the SDK and STT provider and I’ll give you a short, specific code pattern you can use. Sources: Camera & microphone | LiveKit docs | Speech-to-text (STT) models | LiveKit Docs | Speechmatics STT plugin guide | LiveKit Docs | Agent insights in LiveKit Cloud | LiveKit docs | OpenAI and LiveKit partner to turn Advanced Voice into an API

Thanks — using Python + Soniox is clear. Below is a concise, actionable approach and a focused Python pattern you can adapt to mix the agent's synthesized audio with user audio and stream the combined feed to Soniox for diarized transcription. What I'll cover: subscribing to user audio, capturing agent TTS audio, low-latency mixing, and pushing mixed frames to Soniox. Solution outline 1. Capture user audio via LiveKit

AudioStream

from the remote track. 2. Capture agent TTS audio by intercepting your agent's TTS node (emit frames to a local buffer/sink). 3. Real-time mix frames (ensure same sample rate/channels), scale to avoid clipping, and align by timestamp. 4. Push mixed frames into Soniox's streaming API and use speaker diarization to ignore agent segments. Minimal Python pattern (adapt to your app structure) • Use numpy for PCM mixing. • Use a small ring buffer for agent frames to align with incoming user frames. • Prefer 16-bit PCM with same sample rate (e.g., 48kHz mono) for both sources. Example pattern (concise):

import asyncio
import numpy as np
from livekit import rtc
from livekit.plugins import soniox

# Shared buffer for agent frames (thread-safe)
agent_frames = asyncio.Queue()

# In your agent TTS node, push frames to agent_frames
async def tts_node_capture(base_stream):
    async for frame in base_stream:
        await agent_frames.put(frame)   # capture for mixing
        yield frame                    # still publish to room

async def handle_user_track(track: rtc.RemoteAudioTrack):
    audio_stream = rtc.AudioStream(track)
    stt = soniox.STT()
    stt_stream = stt.stream()

    async def stt_consume():
        async for evt in stt_stream:
            if evt.is_final:
                print("FINAL:", evt.alternatives[0].text)
            else:
                print("INTERIM:", evt.alternatives[0].text)

    asyncio.create_task(stt_consume())

    async for event in audio_stream:   # event.frame is rtc.AudioFrame
        user_frame = event.frame
        try:
            agent_frame = agent_frames.get_nowait()
        except asyncio.QueueEmpty:
            agent_frame = None

        if agent_frame and frames_compatible(user_frame, agent_frame):
            user_data = np.frombuffer(user_frame.data, dtype=np.int16).astype(np.int32)
            agent_data = np.frombuffer(agent_frame.data, dtype=np.int16).astype(np.int32)

            # Ensure same length: trim or pad
            n = min(len(user_data), len(agent_data))
            mixed = (user_data[:n] // 2) + (agent_data[:n] // 2)
            mixed = np.clip(mixed, -32768, 32767).astype(np.int16)
            mixed_frame = rtc.AudioFrame(
                data=mixed.tobytes(),
                sample_rate=user_frame.sample_rate,
                num_channels=user_frame.num_channels,
                samples_per_channel=n // user_frame.num_channels,
            )
            stt_stream.push_frame(mixed_frame)
        else:
            # No agent audio currently, send user frame
            stt_stream.push_frame(user_frame)

def frames_compatible(a: rtc.AudioFrame, b: rtc.AudioFrame) -> bool:
    return a.sample_rate == b.sample_rate and a.num_channels == b.num_channels

Key practical points • Synchronization: For best results align frames by timestamp; simple queueing works if TTS frames are emitted close to real-time. • Resampling: If sample rates differ, resample before mixing. • Avoid feedback: Route synthesized agent audio directly into the mixer (don’t re-capture from speaker output). • Level control: Scale sources (e.g., 0.5 each) to avoid clipping and favor user intelligibility. • Latency: Keep buffers small (one or two frames) to minimize added latency. • Soniox: Use the Soniox streaming interface (

stt.stream().push_frame(...)

) and rely on diarization to filter agent segments. If you want, I can: • Provide a complete runnable example that wires into your agent entrypoint and TTS node. • Show a timestamp-aligned mixing variant or a resampling helper. Which of those would help most? Sources: Soniox STT plugin guide | LiveKit Docs | Processing raw media tracks | LiveKit docs | Pipeline nodes and hooks | LiveKit docs

Below is a concise, practical example that follows the typical Python voice-agent quickstart structure and shows how to capture the agent's TTS audio, mix it with incoming user audio, and stream the mixed feed to Soniox for diarized STT. It uses LiveKit agent patterns (tts_node override) and the Soniox plugin. Adapt paths/imports to your project. Key points: • Intercept TTS output by overriding

tts_node

and push frames into an asyncio Queue. • Subscribe to user audio via

track_subscribed

and mix user frames with queued agent frames. • Push mixed frames to Soniox streaming via

stt.stream().push_frame(...)

. • Keep mixing simple (equal gain, clip) and keep buffers small to reduce latency. Example (quickstart-style):

import asyncio
import numpy as np
from livekit import rtc, cli
from livekit.agents import Agent, WorkerOptions
from livekit.plugins import soniox

# Small queue to hold recent agent TTS frames for mixing
AGENT_FRAMES_Q = asyncio.Queue(maxsize=100)

# Helper: check compatibility and simple mixer
def frames_compatible(a: rtc.AudioFrame, b: rtc.AudioFrame) -> bool:
    return a.sample_rate == b.sample_rate and a.num_channels == b.num_channels

def mix_pcm_frames(user_frame: rtc.AudioFrame, agent_frame: rtc.AudioFrame) -> rtc.AudioFrame:
    user = np.frombuffer(user_frame.data, dtype=np.int16).astype(np.int32)
    agent = np.frombuffer(agent_frame.data, dtype=np.int16).astype(np.int32)

    # Align lengths
    n = min(len(user), len(agent))
    mixed = (user[:n] // 2) + (agent[:n] // 2)
    mixed = np.clip(mixed, -32768, 32767).astype(np.int16)

    return rtc.AudioFrame(
        data=mixed.tobytes(),
        sample_rate=user_frame.sample_rate,
        num_channels=user_frame.num_channels,
        samples_per_channel=n // user_frame.num_channels,
    )

# Agent class override to capture TTS output
class MyAgent(Agent):
    async def tts_node(self, text_async, model_settings):
        # Call default TTS node to get audio frames
        base_stream = await super().tts_node(text_async, model_settings)
        async for frame in base_stream:
            # push a copy into buffer for mixing
            try:
                AGENT_FRAMES_Q.put_nowait(frame)
            except asyncio.QueueFull:
                # drop oldest if full
                try:
                    AGENT_FRAMES_Q.get_nowait()
                    AGENT_FRAMES_Q.put_nowait(frame)
                except asyncio.QueueEmpty:
                    pass
            yield frame  # still let frame continue to room publish

async def run_agent_entrypoint():
    stt = soniox.STT()
    stt_stream = stt.stream()

    async def stt_consumer():
        async for evt in stt_stream:
            if evt.is_final:
                print("SONIOX FINAL:", evt.alternatives[0].text)
            else:
                print("SONIOX INTERIM:", evt.alternatives[0].text)

    asyncio.create_task(stt_consumer())

    # Connect and subscribe to tracks
    async def on_track_subscribed(track: rtc.Track, publication, participant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return

        audio_stream = rtc.AudioStream(track)

        async for evt in audio_stream:
            user_frame = evt.frame

            # Try to get a recent agent frame to mix
            try:
                agent_frame = AGENT_FRAMES_Q.get_nowait()
            except asyncio.QueueEmpty:
                agent_frame = None

            if agent_frame and frames_compatible(user_frame, agent_frame):
                mixed = mix_pcm_frames(user_frame, agent_frame)
                stt_stream.push_frame(mixed)
            else:
                # No agent context available, push user frame alone
                stt_stream.push_frame(user_frame)

    # Register room-level handler (Room object available in quickstart env)
    rtc.Room.on("track_subscribed", on_track_subscribed)

    # Keep running
    await asyncio.Event().wait()

if __name__ == "__main__":
    # Run agent worker: adapt WorkerOptions to your quickstart setup
    opts = WorkerOptions(agent_class=MyAgent, entrypoint=run_agent_entrypoint)
    cli.run_app(opts)

Notes and tweaks to consider • Timestamp alignment: This example uses a simple queue; for overlapping or precise alignment, align by frame timestamps. • Resampling: If agent TTS uses different sample rate or channels, resample before mixing. • Gain control: Adjust weights (e.g., 0.7 user / 0.3 agent) to prioritize user audio. • Feedback: Ensure agent audio is routed directly into the queue (not captured from speakers) to avoid echo. • Latency: Keep AGENT_FRAMES_Q small and process frames immediately to minimize added delay. If you'd like, I can: • Produce a runnable example wired to your exact quickstart repo structure. • Replace the queue with timestamp-based alignment code or add resampling helpers. Which would you prefer?

You're asking for an AgentSession-based example that captures the agent's TTS audio, mixes it with incoming user audio, and streams the combined feed to Soniox for diarized STT. Below is a compact, practical AgentSession pattern you can adapt. What this does • Overrides the agent TTS path to capture outgoing frames into a session-local queue. • Subscribes to user audio in the AgentSession and mixes queued agent frames with each user frame. • Streams mixed frames to Soniox via

stt.stream().push_frame(...)

. • Keeps buffers small for low latency and performs simple PCM mixing with clipping protection. Code (AgentSession pattern)

import asyncio
import numpy as np
from livekit import rtc, cli
from livekit.agents import AgentSession, WorkerOptions, Agent
from livekit.plugins import soniox

class SessionAgent(Agent):
    # Agent class unchanged; session logic handled in AgentSession below
    pass

class MyAgentSession(AgentSession):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # session-local FIFO for agent TTS frames (small to keep latency low)
        self.agent_frames_q = asyncio.Queue(maxsize=8)
        self.stt = soniox.STT()
        self.stt_stream = None

    async def start(self):
        # Start STT streaming consumer
        self.stt_stream = self.stt.stream()
        asyncio.create_task(self._consume_stt_events())

        # Register track subscription handler
        self.room.on("track_subscribed", self._on_track_subscribed)

        # continue normal session startup
        await super().start()

    async def _consume_stt_events(self):
        async for evt in self.stt_stream:
            if evt.is_final:
                print("SONIOX FINAL:", evt.alternatives[0].text)
            else:
                print("SONIOX INTERIM:", evt.alternatives[0].text)

    # Override/extend TTS output at session level to capture agent audio
    async def tts_node(self, text_async, model_settings):
        # Get base TTS stream from default node
        base_stream = await super().tts_node(text_async, model_settings)

        async for frame in base_stream:
            # push frame into session queue for mixing (drop oldest if full)
            try:
                self.agent_frames_q.put_nowait(frame)
            except asyncio.QueueFull:
                try:
                    _ = self.agent_frames_q.get_nowait()
                    self.agent_frames_q.put_nowait(frame)
                except asyncio.QueueEmpty:
                    pass
            # still yield frame to be published by the agent into the room
            yield frame

    async def _on_track_subscribed(self, track, publication, participant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return

        audio_stream = rtc.AudioStream(track)

        async for evt in audio_stream:
            user_frame = evt.frame

            # try to get the most recent agent frame (non-blocking)
            try:
                agent_frame = self.agent_frames_q.get_nowait()
            except asyncio.QueueEmpty:
                agent_frame = None

            # If available and compatible, mix; else forward user frame alone
            if agent_frame and self._frames_compatible(user_frame, agent_frame):
                mixed = self._mix_frames(user_frame, agent_frame)
                self.stt_stream.push_frame(mixed)
            else:
                self.stt_stream.push_frame(user_frame)

    # Helpers
    def _frames_compatible(self, a: rtc.AudioFrame, b: rtc.AudioFrame) -> bool:
        return a.sample_rate == b.sample_rate and a.num_channels == b.num_channels

    def _mix_frames(self, user_frame: rtc.AudioFrame, agent_frame: rtc.AudioFrame) -> rtc.AudioFrame:
        u = np.frombuffer(user_frame.data, dtype=np.int16).astype(np.int32)
        v = np.frombuffer(agent_frame.data, dtype=np.int16).astype(np.int32)

        n = min(len(u), len(v))
        mixed = (u[:n] // 2) + (v[:n] // 2)
        mixed = np.clip(mixed, -32768, 32767).astype(np.int16)

        return rtc.AudioFrame(
            data=mixed.tobytes(),
            sample_rate=user_frame.sample_rate,
            num_channels=user_frame.num_channels,
            samples_per_channel=n // user_frame.num_channels,
        )

async def agent_entrypoint():
    # Called when worker starts; create session lifecycle managed by Fast Agent worker
    # AgentSession creation is managed by the framework; this function can wait
    await asyncio.Event().wait()

if __name__ == "__main__":
    opts = WorkerOptions(agent_class=SessionAgent, session_class=MyAgentSession, entrypoint=agent_entrypoint)
    cli.run_app(opts)

Practical notes • Alignment: This uses a simple queue to match agent frames to the next user frame. For precise sync, align by timestamps. • Sample rates: Ensure TTS and incoming audio share sample rate/channels; resample if needed. • Gains: Adjust mixing weights to favor user audio for clarity (e.g., 0.7 user / 0.3 agent). • Feedback: Ensure TTS frames are injected directly (not re-captured from speaker output) to avoid echo loops. • Latency: Keep queue small (1–4 frames) to minimize added delay. Would you like a timestamp-aligned mixing variant or a runnable repo-style example wired to your quickstart files?

You're asking how to avoid feedback by injecting TTS frames directly into the mixer (instead of re-capturing from speaker output). Below are concise, practical steps and a focused AgentSession code pattern showing safe injection, plus key best practices. Core idea • Never record the agent audio from playback devices. Instead, capture the synthesized TTS frames at the source (the TTS node) and feed those frames directly into the mixer/STT pipeline. This prevents the agent audio from being heard by the microphone and re-captured as user audio. How to do it (step-by-step) 1. Intercept the agent TTS output inside the AgentSession (override

tts_node

) and push frames into a session-local buffer/queue. 2. When mixing user frames, pull agent frames from that buffer and mix them with user frames before pushing to Soniox. 3. Ensure the agent frames you inject are the same format (sample rate, channels, bitdepth) as user frames; resample/convert if necessary. 4. Do not route agent audio to system speakers or any capture loop that the microphone can hear. If you must play audio locally for debugging, mute playback or use virtual audio devices isolated from microphone capture. 5. Use short buffers and timestamp alignment to avoid duplicating agent audio or mixing stale frames. AgentSession pattern (focused snippet)

# inside MyAgentSession (simplified)

# session-local queue for TTS frames (injected at source)
self.agent_frames_q = asyncio.Queue(maxsize=8)

# override tts_node to capture frames
async def tts_node(self, text_async, model_settings):
    base = await super().tts_node(text_async, model_settings)
    async for frame in base:
        # Inject frame into the session queue (direct source capture)
        try:
            self.agent_frames_q.put_nowait(frame)
        except asyncio.QueueFull:
            # drop oldest to keep latency low
            try:
                _ = self.agent_frames_q.get_nowait()
                self.agent_frames_q.put_nowait(frame)
            except asyncio.QueueEmpty:
                pass
        yield frame  # still publish to the room

# mixing loop for incoming user frames
async for evt in rtc.AudioStream(user_track):
    user_frame = evt.frame
    agent_frame = None
    try:
        agent_frame = self.agent_frames_q.get_nowait()
    except asyncio.QueueEmpty:
        pass

    if agent_frame and frames_compatible(user_frame, agent_frame):
        mixed = mix_frames(user_frame, agent_frame)   # mix PCM directly
        stt_stream.push_frame(mixed)
    else:
        stt_stream.push_frame(user_frame)

Practical safeguards to prevent feedback • Do not play TTS to the same machine microphone path. If you need local monitoring: ◦ Play TTS on a separate device/community test setup that is not routed into the capture pipeline. ◦ Use an isolated virtual audio output that is not captured by the mic. ◦ Use Push-to-Monitor or muted playback in production. • If your deployment mixes agent audio server-side (preferred), keep TTS generation and mixing entirely on the server or agent process so client microphones never capture it. • Add a short-integer session ID or metadata to agent frames if you need to recognize and drop re-captured agent audio (defensive measure). • Use voice activity detection (VAD) and a small grace window to avoid sending short re-captured snippets if they do appear. Troubleshooting tips • If you still see echoes in STT, check whether the client publishes system audio or stereo-mixes. Disable “Stereo Mix” or similar on client machines. • Log timestamps and sample counts for both sources to confirm alignment and detect stale frames being mixed. • Temporarily mute agent playback locally; if echoes disappear, the feedback path is via local playback. If you want, I can: • Provide a timestamp-aligned mixing example (matching and trimming frames by timestamp). • Show a small resampling helper if TTS uses a different sample rate. Which would you prefer?

show me an example with TTS node, AgentSession, track_published, local_track_published