https://livekit.io logo
Join Slack
Powered by
if room have 3 standard paticipants, could just 1 ...
# ask-ai
t
if room have 3 standard paticipants, could just 1 agent instace can subscribe and do transcript for all member
t
Thanks for reaching out, Quinn. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Absolutely! A single agent instance can definitely subscribe to and transcribe audio for all participants in a room with 3 standard participants. To make this happen smoothly, you’ll want to create one 
AgentSession
per participant so their audio is processed independently. Here’s a friendly overview of how it works: • The agent worker process runs once per room and dynamically manages sessions as participants join or leave, keeping everything organized. • Each participant gets their own dedicated 
AgentSession
, which helps keep the transcription streams nicely separated. • With 
RoomIO
, you can connect an 
AgentSession
to a participant’s audio, enabling real-time transcription effortlessly. • The transcriptions come with metadata to identify the speaker, like the 
participant_identity
, so you always know who’s talking. If you’d like to dive into the details, check out the multi-user transcriber example. And please don’t hesitate to reach out if you have any more questions or need a hand—I’m here to help! Sources: protobufs/livekit_models.proto | src/room/participant/LocalParticipant.ts | Anatomy of an Agent | LiveKit Docs | examples/other/translation/multi-user-translator.py
t
Copy code
import {
  AutoSubscribe,
  cli,
  defineAgent,
  type JobContext,
  type JobProcess,
  metrics,
  voice,
  WorkerOptions,
} from '@livekit/agents';
import * as deepgram from '@livekit/agents-plugin-deepgram';
import * as silero from '@livekit/agents-plugin-silero';

// import * as cartesia from '@livekit/agents-plugin-cartesia';
// import * as livekit from '@livekit/agents-plugin-livekit';
// import * as openai from '@livekit/agents-plugin-openai';
// import { BackgroundVoiceCancellation } from '@livekit/noise-cancellation-node';
import dotenv from 'dotenv';
import { fileURLToPath } from 'node:url';
import { ParticipantKind } from '@livekit/rtc-node';

dotenv.config({ path: '.env.local' });

class Assistant extends voice.Agent {
  constructor() {
    super({
      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.
      `,
    });
  }
}

export default defineAgent({
  prewarm: async (proc: JobProcess) => {
    proc.userData.vad = await silero.VAD.load();
  },
  entry: async (ctx: JobContext) => {
    // Set up a voice AI pipeline using OpenAI, Cartesia, Deepgram, and the LiveKit turn detector
    const session = new voice.AgentSession({
      // A Large Language Model (LLM) is your agent's brain, processing user input and generating a response
      // See all providers at <https://docs.livekit.io/agents/integrations/llm/>
      // llm: new openai.LLM({ model: 'gpt-4o-mini' }),
      // Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand
      // See all providers at <https://docs.livekit.io/agents/integrations/stt/>
      stt: new deepgram.STT({ model: 'nova-3' }),
      // Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear
      // See all providers at <https://docs.livekit.io/agents/integrations/tts/>
      // tts: new cartesia.TTS({
      //   voice: '6f84f4b8-58a2-430c-8c79-688dad597532',
      // }),
      // VAD and turn detection are used to determine when the user is speaking and when the agent should respond
      // See more at <https://docs.livekit.io/agents/build/turns>
      // turnDetection: new livekit.turnDetector.MultilingualModel(),
      // vad: ctx.proc.userData.vad! as silero.VAD,
    });

    // To use a realtime model instead of a voice pipeline, use the following session setup instead:
    // const session = new voice.AgentSession({
    //   // See all providers at <https://docs.livekit.io/agents/integrations/realtime/>
    //   llm: new openai.realtime.RealtimeModel({ voice: 'marin' }),
    // });

    // Metrics collection, to measure pipeline performance
    // For more information, see <https://docs.livekit.io/agents/build/metrics/>
    const usageCollector = new metrics.UsageCollector();
    session.on(voice.AgentSessionEventTypes.MetricsCollected, (ev) => {
      metrics.logMetrics(ev.metrics);
      usageCollector.collect(ev.metrics);
    });

    session.on(voice.AgentSessionEventTypes.UserInputTranscribed, (ev) => {
      console.log(ev);
    });

    session.on(voice.AgentSessionEventTypes.UserStateChanged, (ev) => {
      console.log(ev);
    });

    session.on(voice.AgentSessionEventTypes.SpeechCreated, (ev) => {
      console.log('speech created:', ev);
    });

    session.on(voice.AgentSessionEventTypes.Error, (ev) => {
      console.log('error', ev);
    });

    const logUsage = async () => {
      const summary = usageCollector.getSummary();
      console.log(`Usage: ${JSON.stringify(summary)}`);
    };

    ctx.addShutdownCallback(logUsage);

    // Start the session, which initializes the voice pipeline and warms up the models
    await session.start({
      agent: new Assistant(),
      room: ctx.room,
      inputOptions: {
      },
      outputOptions: {
      }
    });


    // eslint-disable-next-line @typescript-eslint/no-unused-vars
    const roomName = ctx?.job?.room?.name || '';


    await ctx.connect(undefined, AutoSubscribe.SUBSCRIBE_ALL);
  },
});

cli.runApp(new WorkerOptions({ agent: fileURLToPath(import.meta.url) }));
But i see this only transcibe for the first participant
t
Thanks for sharing your code! It looks like it’s setting up just one 
AgentSession
, which could be why only the first participant is being transcribed. Are you hoping to transcribe everyone in the room at the same time? If that’s the case, I’d be happy to help you create multiple sessions dynamically for each participant. Just let me know!
t
so just 1 session could not generate for all participant?
t
You're absolutely right! A single 
AgentSession
can only transcribe audio for one participant at a time. Each 
AgentSession
is built to handle one participant's audio stream individually. So, to capture the audio from everyone in the room, you’ll want to create a separate 
AgentSession
for each participant and manage these sessions as people join or leave. If you'd like, I’d be happy to walk you through how to set up multiple sessions for all participants or even share an example to make things clearer. Just let me know how you’d like to move forward—I’m here to help!
t
yes guide me
t
To help you transcribe audio for all participants in a room, you can create one 
AgentSession
dynamically for each participant. I’ve put together a friendly step-by-step guide to walk you through this:
t
show me the code
t
I'm happy to share a complete example that shows how to dynamically create and manage 
AgentSession
instances for all participants in a room. This way, every participant can be supported seamlessly:
Copy code
import { AutoSubscribe, defineAgent, type JobContext, voice } from '@livekit/agents';
import * as deepgram from '@livekit/agents-plugin-deepgram';

class Assistant extends voice.Agent {
  constructor() {
    super({
      instructions: `You are a helpful voice AI assistant.`,
    });
  }
}

export default defineAgent({
  entry: async (ctx: JobContext) => {
    const sessions = {}; // Store sessions by participant identity

    // Function to start a session for a participant
    async function startSession(participant) {
      const session = new voice.AgentSession({
        stt: new deepgram.STT({ model: 'nova-3' }),
      });

      const roomIO = new voice.RoomIO({
        agentSession: session,
        room: ctx.room,
        participant: participant,
        outputOptions: { transcriptionEnabled: true },
      });

      await roomIO.start();
      await session.start({ agent: new Assistant() });

      return session;
    }

    // Handle participant connections
    ctx.room.on('participant_connected', async (participant) => {
      sessions[participant.identity] = await startSession(participant);
    });

    // Handle participant disconnections
    ctx.room.on('participant_disconnected', async (participant) => {
      const session = sessions[participant.identity];
      if (session) {
        await session.drain();
        await session.aclose();
        delete sessions[participant.identity];
      }
    });

    // Start sessions for already-connected participants
    for (const participant of Object.values(ctx.room.remoteParticipants)) {
      sessions[participant.identity] = await startSession(participant);
    }

    // Clean up on shutdown
    ctx.addShutdownCallback(async () => {
      for (const session of Object.values(sessions)) {
        await session.drain();
        await session.aclose();
      }
    });

    // Connect to the room
    await ctx.connect(undefined, AutoSubscribe.SUBSCRIBE_ALL);
  },
});
This approach ensures that each participant gets their own 
AgentSession
, so everyone in the room can be transcribed at the same time without any hassle. If you have any questions or need help customizing this further, please feel free to reach out—I’m here to help!
t
Copy code
async connect(
  e2ee?: E2EEOptions,
  autoSubscribe: AutoSubscribe = AutoSubscribe.SUBSCRIBE_ALL,
  rtcConfig?: RtcConfiguration,
) {
  if (this.connected) {
    return;
  }

  const opts = {
    e2ee,
    autoSubscribe: autoSubscribe == AutoSubscribe.SUBSCRIBE_ALL,
    rtcConfig,
    dynacast: false,
  };

  await this.#room.connect(this.#info.url, this.#info.token, opts);
  this.#onConnect();

  this.#room.remoteParticipants.forEach(this.onParticipantConnected);

  if ([AutoSubscribe.AUDIO_ONLY, AutoSubscribe.VIDEO_ONLY].includes(autoSubscribe)) {
    this.#room.remoteParticipants.forEach((p) => {
      p.trackPublications.forEach((pub) => {
        if (
          (autoSubscribe === AutoSubscribe.AUDIO_ONLY && pub.kind === TrackKind.KIND_AUDIO) ||
          (autoSubscribe === AutoSubscribe.VIDEO_ONLY && pub.kind === TrackKind.KIND_VIDEO)
        ) {
          pub.setSubscribed(true);
        }
      });
    });
  }
  this.connected = true;
}
But i thought it would
Copy code
pub.setSubscribed(true);
for each participant
t
You're absolutely right that 
AutoSubscribe.SUBSCRIBE_ALL
automatically subscribes to all participants' tracks, including both audio and video. This is a great way to make sure the agent stays connected to all existing and future tracks published by participants in the room. That said, while this setting takes care of subscribing to the tracks, it doesn't automatically create separate 
AgentSession
instances for each participant. To transcribe audio for everyone, you'll still want to dynamically create and manage individual 
AgentSession
instances for each participant, just like in the example we discussed earlier. If you have any questions or need a hand with this, please don’t hesitate to reach out—I’m here to help! Sources: Rooms, participants, and tracks | LiveKit Docs | livekit-agents/livekit/agents/job.py | Receiving and publishing tracks | LiveKit Docs
t
is there other way to implement transcript for multiple participant, i still see the above solution is complex
t
If you're looking for an easier way to set up transcription for multiple participants, I’d love to share two friendly approaches with you: 1. Using 
RoomIO
for Automatic Transcription: This is a great option that automatically manages transcription for everyone by creating a 
RoomIO
object for each participant. It makes things simple by sending audio straight to the Speech-to-Text (STT) engine. All you need to do is set 
transcription_enabled=True
when configuring 
RoomIO
, and the transcriptions will be shared back in the room as a data stream. You can see this in action in the multi-user transcriber example. 2. Using 
STTSegmentsForwarder
for Direct Control: If you prefer more hands-on control, this approach lets you manage audio tracks yourself. You’d listen for the 
track_subscribed
event, create an 
AudioStream
for each track, and then use 
STTSegmentsForwarder
to handle and forward the transcription events. It’s perfect if you want to customize the transcription process more closely. Both of these methods make things simpler compared to juggling multiple 
AgentSession
instances. If you’d like, I’m here to help with detailed guidance or code snippets for either option—just let me know! Sources: README.md | agents/src/voice/room_io/room_io.ts | main.py | lib/src/context/transcription_context.dart | VoiceAgent/Chat/Receive/TranscriptionDelegateReceiver.swift
t
why do we need RoomID while sessions.start already have rooomInput, outpout with the same options Is them the same/?
t
Thank you for your question! The 
RoomID
isn’t required in 
session.start
because the 
room
object already holds all the important details about the room, including its identifier. You can easily configure media and transcription settings using the 
roomInput
and 
roomOutput
options within the 
room
object, which takes care of managing the communication context behind the scenes. That said, 
RoomID
does play a key role in backend operations like egress management or certain service-level functions where only the identifier is available, rather than the full 
room
object. For instance, it’s used in recording or media routing tasks where targeting specific resources precisely is important. For most agent session setups, the 
room
object helps simplify things by removing the need to handle 
RoomID
directly. If you’d like, I’m here to help clarify anything else or walk you through it step-by-step! Sources: Avoiding Room ID Reuse When Re-Creating Rooms With the Same Name | livekit-agents/livekit/agents/voice/agent_session.py | agents/src/voice/agent_session.ts | pkg/rtc/types/typesfakes/fake_room.go | src/room/Room.ts
t
it could work with multiple agent participants. Which options is better. Dispatch to create new agent participants every time a new participant join Or create new agent session
t
I totally understand where you’re coming from! It’s usually best to create new 
AgentSession
instances for each participant instead of adding new agent participants. Let me explain why this approach works well: • Isolation and Scalability: Each 
AgentSession
is built to manage one participant’s audio and interactions separately, which helps keep things neat and organized in terms of state and media processing. • Standard Approach: The LiveKit framework is designed to handle multiple 
AgentSession
instances within a single agent worker—this is a common pattern you’ll see in scenarios like multi-user transcription and IVR workflows. • Resource Efficiency: Using 
AgentSession
means you avoid the extra overhead of creating new agent participants, which isn’t typically recommended in LiveKit. If you find that you need to manage resource-heavy tasks for each participant, you can also set up the worker to use 
WorkerType.PUBLISHER
, which will automatically create a new agent instance for each participant. Please feel free to reach out if you’d like me to walk you through this in more detail or help with the implementation—I’m here to help! Sources: Docs::Agents | Docs::Agents | Agent-assisted warm transfer | LiveKit Docs | Worker options | LiveKit Docs | Building an Automated IVR Menu Caller | LiveKit Docs
t
convert multiple user transciber from python to node version
Copy code
import asyncio
import logging

from dotenv import load_dotenv

from livekit import rtc
from livekit.agents import (
    Agent,
    AgentSession,
    AutoSubscribe,
    JobContext,
    JobProcess,
    RoomInputOptions,
    RoomIO,
    RoomOutputOptions,
    StopResponse,
    WorkerOptions,
    cli,
    llm,
    utils,
)
from livekit.plugins import deepgram, silero

load_dotenv()

logger = logging.getLogger("transcriber")


# This example demonstrates how to transcribe audio from multiple remote participants.
# It creates agent sessions for each participant and transcribes their audio.


class Transcriber(Agent):
    def __init__(self, *, participant_identity: str):
        super().__init__(
            instructions="not-needed",
            stt=deepgram.STT(),
        )
        self.participant_identity = participant_identity

    async def on_user_turn_completed(self, chat_ctx: llm.ChatContext, new_message: llm.ChatMessage):
        user_transcript = new_message.text_content
        <http://logger.info|logger.info>(f"{self.participant_identity} -> {user_transcript}")

        raise StopResponse()


class MultiUserTranscriber:
    def __init__(self, ctx: JobContext):
        self.ctx = ctx
        self._sessions: dict[str, AgentSession] = {}
        self._tasks: set[asyncio.Task] = set()

    def start(self):
        self.ctx.room.on("participant_connected", self.on_participant_connected)
        self.ctx.room.on("participant_disconnected", self.on_participant_disconnected)

    async def aclose(self):
        await utils.aio.cancel_and_wait(*self._tasks)

        await asyncio.gather(*[self._close_session(session) for session in self._sessions.values()])

        self.ctx.room.off("participant_connected", self.on_participant_connected)
        self.ctx.room.off("participant_disconnected", self.on_participant_disconnected)

    def on_participant_connected(self, participant: rtc.RemoteParticipant):
        if participant.identity in self._sessions:
            return

        <http://logger.info|logger.info>(f"starting session for {participant.identity}")
        task = asyncio.create_task(self._start_session(participant))
        self._tasks.add(task)

        def on_task_done(task: asyncio.Task):
            try:
                self._sessions[participant.identity] = task.result()
            finally:
                self._tasks.discard(task)

        task.add_done_callback(on_task_done)

    def on_participant_disconnected(self, participant: rtc.RemoteParticipant):
        if (session := self._sessions.pop(participant.identity)) is None:
            return

        <http://logger.info|logger.info>(f"closing session for {participant.identity}")
        task = asyncio.create_task(self._close_session(session))
        self._tasks.add(task)
        task.add_done_callback(lambda _: self._tasks.discard(task))

    async def _start_session(self, participant: rtc.RemoteParticipant) -> AgentSession:
        if participant.identity in self._sessions:
            return self._sessions[participant.identity]

        session = AgentSession(
            vad=self.ctx.proc.userdata["vad"],
        )
        room_io = RoomIO(
            agent_session=session,
            room=self.ctx.room,
            participant=participant,
            input_options=RoomInputOptions(
                # text input is not supported for multiple room participants
                # if needed, register the text stream handler by yourself
                # and route the text to different sessions based on the participant identity
                text_enabled=False,
            ),
            output_options=RoomOutputOptions(
                transcription_enabled=True,
                audio_enabled=False,
            ),
        )
        await room_io.start()
        await session.start(
            agent=Transcriber(
                participant_identity=participant.identity,
            )
        )
        return session

    async def _close_session(self, sess: AgentSession) -> None:
        await sess.drain()
        await sess.aclose()


async def entrypoint(ctx: JobContext):
    transcriber = MultiUserTranscriber(ctx)
    transcriber.start()

    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    for participant in ctx.room.remote_participants.values():
        # handle all existing participants
        transcriber.on_participant_connected(participant)

    async def cleanup():
        await transcriber.aclose()

    ctx.add_shutdown_callback(cleanup)


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))
t
Hi there! Converting the multi-user transcriber example from Python to Node.js is definitely doable, and I’m here to guide you through it. The key is to replicate the same logic using the LiveKit Node.js SDK. Below, you’ll find a Node.js implementation inspired by the Python example:
Copy code
import { AutoSubscribe, defineAgent, type JobContext, voice } from '@livekit/agents';
import * as deepgram from '@livekit/agents-plugin-deepgram';
import * as silero from '@livekit/agents-plugin-silero';

class Transcriber extends voice.Agent {
  constructor(participantIdentity) {
    super({
      instructions: "not-needed",
      stt: new deepgram.STT(),
    });
    this.participantIdentity = participantIdentity;
  }

  async onUserTurnCompleted(chatCtx, newMessage) {
    const userTranscript = newMessage.textContent;
    console.log(`${this.participantIdentity} -> ${userTranscript}`);
  }
}

class MultiUserTranscriber {
  constructor(ctx) {
    this.ctx = ctx;
    this.sessions = new Map();
  }

  start() {
    this.ctx.room.on('participant_connected', this.onParticipantConnected.bind(this));
    this.ctx.room.on('participant_disconnected', this.onParticipantDisconnected.bind(this));
  }

  async onParticipantConnected(participant) {
    if (this.sessions.has(participant.identity)) return;

    console.log(`Starting session for ${participant.identity}`);
    const session = new voice.AgentSession({ vad: this.ctx.proc.userData.vad });

    const roomIO = new voice.RoomIO({
      agentSession: session,
      room: this.ctx.room,
      participant,
      inputOptions: { textEnabled: false },
      outputOptions: { transcriptionEnabled: true, audioEnabled: false },
    });

    await roomIO.start();
    await session.start(new Transcriber(participant.identity));

    this.sessions.set(participant.identity, session);
  }

  async onParticipantDisconnected(participant) {
    const session = this.sessions.get(participant.identity);
    if (!session) return;

    console.log(`Closing session for ${participant.identity}`);
    await session.drain();
    await session.aclose();
    this.sessions.delete(participant.identity);
  }

  async close() {
    for (const session of this.sessions.values()) {
      await session.drain();
      await session.aclose();
    }
    this.sessions.clear();
  }
}

export default defineAgent({
  prewarm: async (proc) => {
    proc.userData.vad = await silero.VAD.load();
  },
  entry: async (ctx) => {
    const transcriber = new MultiUserTranscriber(ctx);
    transcriber.start();

    await ctx.connect(undefined, AutoSubscribe.AUDIO_ONLY);

    for (const participant of Object.values(ctx.room.remoteParticipants)) {
      await transcriber.onParticipantConnected(participant);
    }

    ctx.addShutdownCallback(async () => {
      await transcriber.close();
    });
  },
});
This Node.js version closely follows the Python example by dynamically creating and managing 
AgentSession
instances for each participant. If anything feels unclear or if you’d like me to walk you through any part of this, please don’t hesitate to ask—I’m here to help!
t
add typescript for it
t
I'm happy to share the Node.js implementation of the multi-user transcriber example, now updated with TypeScript to give you that extra layer of type safety:
t
i do not see the code
t
Here is the TypeScript version of the multi-user transcriber example:
t
Copy code
If you find that you need to manage resource-heavy tasks for each participant, you can also set up the worker to use WorkerType.PUBLISHER, which will automatically create a new agent instance for each participant.
how to set up a new agent instance for each participant.
t
To set up a new agent instance for each participant, you’ll want to configure the worker to use 
WorkerType.PUBLISHER
. This way, a separate agent instance is created for every participant who joins the room. Here’s a simple example to guide you:
Copy code
import { defineAgent, WorkerOptions, WorkerType } from '@livekit/agents';

export default defineAgent({
  entry: async (ctx) => {
    // ctx.publisher represents the specific participant for this agent instance
    console.log(`Agent started for publisher: ${ctx.publisher.identity}`);

    // Feel free to add your media processing or interaction logic here
  },
  workerOptions: new WorkerOptions({
    workerType: WorkerType.PUBLISHER, // This creates one agent instance per participant
  }),
});
With this setup, the 
entry
function runs once for each participant, and 
ctx.publisher
gives you access to that participant’s details. It’s a great approach if you’re handling resource-intensive tasks and want to keep things nicely isolated for each participant. If anything feels unclear or you’d like a hand with the next steps, just let me know—I’m here to help! Sources: Worker options | LiveKit Docs | Anatomy of an Agent | LiveKit Docs | Docs::Agents | livekit-agents/livekit/agents/worker.py
2 Views
Open in Slack
PreviousNext
Powered by