We want to enable multiple voices in a single room of a agent. The voice should change to voice assigned to when (<<SOME_NAME>>) appears in the transcript, we extract the name and then we want to change the voice. How can we do this? Has someone already build this? Give me verified solution backed with sources

Thanks for reaching out, Piyush. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

below is my agent. What needs to be corrected to fix issues below as we are facing issue with live captions in the ui as when we process the transcript, it is not sending the actual transcript (with the name tag), to be used by ui to highlight which user is speaking.

import asyncio
import logging
import time
import re
from typing import AsyncIterable
from asyncio import Queue

import requests
from livekit import rtc
from livekit.agents import Agent, ModelSettings, tts, tokenize

from config import WEBHOOK_URL
from egress import stop_egress_if_running
from fallback_providers import create_tts_with_fallback
from report import create_parseable_transcript_object_from_chat_ctx

logger = logging.getLogger("ai-assistant")


class MultiPersonaAssistant(Agent):
    opener_line = None
    SPEAKER_RE = re.compile(
        r"<(?P<speaker>[^<>]+)>(?P<text>.*?)</(?P=speaker)>",
        re.IGNORECASE | re.DOTALL,
    )
    SPEAKER_TTS = {}
    parsed_to_original = []

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._raw_llm_queue = Queue()

    async def set_agents_description(self, prompts, selected_tts_backend, tts_model, vocabulary=None):
        combined_instructions = """When a message should be spoken by a particular person, use markup: "<CHARACTER>message</CHARACTER>" where X is the character. For any text outside of the xml tags, default character will be used.

For example:
Then out of sudden Jenny said, <Jenny>Hey I think I see it!</Jenny> and the picture fell on the ground.

Available voices are as follows:

default: any text outside of the CHARACTER tags, use when none of below applies"""

        primary_agent_instructions = ""
        for prompt in prompts:
            if prompt["is_primary"]:
                self.SPEAKER_TTS["default"] = create_tts_with_fallback(selected_tts_backend,
                                                                       {"id": prompt["voice_id"]},
                                                                       gender=prompt["gender"],
                                                                       language=prompt["language"],
                                                                       vocabulary=vocabulary,
                                                                       tts_model=tts_model)
                primary_agent_instructions = f'\n\nAgent character description: {prompt["bot_prompt"]}'
                continue
            self.SPEAKER_TTS[prompt['agent_name']] = create_tts_with_fallback(selected_tts_backend,
                                                                              {"id": prompt["voice_id"]},
                                                                              gender=prompt["gender"],
                                                                              language=prompt["language"],
                                                                              vocabulary=vocabulary,
                                                                              tts_model=tts_model)
            combined_instructions += f"\n<{prompt['agent_name']}> {prompt['bot_prompt']}</{prompt['agent_name']}>\n"

        combined_instructions += f"{primary_agent_instructions}"

        await self.update_instructions(combined_instructions)

    @staticmethod
    async def _trigger_post_processing(payload):
        response = await asyncio.to_thread(
            <http://requests.post|requests.post>,
            WEBHOOK_URL,
            json=payload,
            timeout=30
        )

        response.raise_for_status()
        return response

    async def on_enter(self):
        if self.opener_line:
            await self.session.say(self.opener_line, allow_interruptions=True)

    async def on_exit(self):
        <http://logger.info|logger.info>("transcript on exit called")
        chat_ctx = self.chat_ctx.to_dict()
        session_id = self.session.userdata["session_id"]
        recording_started_at = self.session.userdata["recording_started_at"]
        recording_stopped_at = time.time()
        try:
            egress_recording_stopped_at = await stop_egress_if_running(self.session.userdata.get("egress_id", None))
            if egress_recording_stopped_at:
                recording_stopped_at = egress_recording_stopped_at
        except Exception:
            pass
        parsed_transcript = create_parseable_transcript_object_from_chat_ctx(chat_ctx, recording_started_at,
                                                                             change_developer_to_gatekeeper=True)

        pending_multivoice_messages = self.session.userdata.get(
            "pending_multivoice_messages",
            [],
        )
        multivoice_index = 0

        for item in parsed_transcript:
            item["multivoice_message"] = None

            if item["speaker"] not in ["assistant", "developer", "gatekeeper"]:
                continue

            if multivoice_index >= len(pending_multivoice_messages):
                continue

            multivoice_message = pending_multivoice_messages[multivoice_index]
            multivoice_index += 1
            if not multivoice_message:
                item["multivoice_message"] = None
            else:
                item["multivoice_message"] = {
                    "parts": multivoice_message["parts"],
                }

                item["utterance"] = self._clean_multivoice_utterance(
                    multivoice_message["parts"]
                )

        usage_summary = None
        try:
            usage = self.session.userdata.get("usage")
            if usage:
                usage_summary = usage.get_summary().__dict__
                <http://logger.info|logger.info>(f"[USAGE SUMMARY] {usage_summary}")
        except Exception:
            logger.exception("Failed to read usage summary")
        webhook_payload = {"event": "transcript_report",
                           "session_id": session_id,
                           "backend": "livekit",
                           "data": {
                               "transcript": parsed_transcript,
                               "started_at": recording_started_at * 1000,
                               "duration": (recording_stopped_at - recording_started_at) * 1000,
                               "termination_reason": self.session.userdata.get("termination_reason",
                                                                               "Participant exited"),
                               "usage_summary": usage_summary
                           }}
        <http://logger.info|logger.info>("[Report] sending webhook payload...")
        <http://logger.info|logger.info>(f"[Report] webhook_payload: {webhook_payload}")
        webhook_response = await self._trigger_post_processing(webhook_payload)
        <http://logger.info|logger.info>(f"[Report] webhook_response: {webhook_response.text}")
        <http://logger.info|logger.info>("[Report] sent webhook payload...")

    def _parse_llm_response_to_speakers(self, chunk):
        pos = 0

        for match in self.SPEAKER_RE.finditer(chunk):
            before = chunk[pos:match.start()].strip()
            if before:
                yield "default", before

            speaker = match.group("speaker")
            spoken_text = match.group("text").strip()
            if spoken_text:
                yield speaker, spoken_text

            pos = match.end()

        after = chunk[pos:].strip()
        if after:
            yield "default", after

    @staticmethod
    def _clean_multivoice_utterance(parts):
        return " ".join(part["text"] for part in parts).strip()

    async def transcription_node(self, text: AsyncIterable[str], model_settings: ModelSettings,) -> AsyncIterable[str]:
        while True:
            while self.parsed_to_original:
                yield self.parsed_to_original.pop(0)

    async def tts_node(
            self,
            text: AsyncIterable[str],
            model_settings: ModelSettings,
    ) -> AsyncIterable[rtc.AudioFrame]:
        buffer = ""
        async for chunk in text:
            buffer += chunk
        self.parsed_to_original.append(buffer)
        parsed_segments = list(self._parse_llm_response_to_speakers(buffer))
        has_multivoice = any(
            speaker != "default"
            for speaker, _ in parsed_segments
        )
        multivoice_message = None

        try:
            user_session_data = self.session.userdata
        except ValueError:
            user_session_data = {}

        if has_multivoice:
            now_secs = time.time() - user_session_data["recording_started_at"]
            parts_message = []
            for speaker, spoken_text in parsed_segments:
                parts_message.append({
                    "text": spoken_text,
                    "voice_label": speaker,
                    "time_in_call_secs": int(now_secs),
                })

            multivoice_message = {
                "raw": buffer,
                "parts": parts_message
            }

        user_session_data.setdefault(
            "pending_multivoice_messages",
            []
        ).append(multivoice_message)
        for speaker, spoken_text in parsed_segments:
            wrapped_tts = self.SPEAKER_TTS.get(
                speaker,
                self.SPEAKER_TTS["default"],
            )

            if not wrapped_tts.capabilities.streaming:
                wrapped_tts = tts.StreamAdapter(
                    tts=wrapped_tts,
                    sentence_tokenizer=tokenize.blingfire.SentenceTokenizer(
                        retain_format=True
                    ),
                )

            conn_options = self.session.conn_options.tts_conn_options

            async with wrapped_tts.stream(
                    conn_options=conn_options
            ) as stream:
                <http://logger.info|logger.info>(f"[TTS] spoken_text: {spoken_text} {len(spoken_text)}")
                stream.push_text(spoken_text)
                stream.end_input()

                async for ev in stream:
                    yield ev.frame

Thanks — I’m responding to your request about why live captions don’t receive the transcript with the NAME tags and what to change in your agent. Root cause (brief) • Your

transcription_node

never yields the original tagged transcript: it waits on

self.parsed_to_original

, which is only populated by

tts_node

after TTS processing. That flow strips/changes the text before the UI gets it. • TTS processing should speak parsed segments, but the UI needs the original tagged string (for highlighting). Those responsibilities must be separated. What to change (concise, actionable) 1) Fix transcription_node to forward original text (preserve tags) • Replace the blocking loop with a simple pass-through that yields each incoming chunk unchanged so the UI gets the tagged transcript immediately. Example replacement:

async def transcription_node(self, text: AsyncIterable[str], model_settings: ModelSettings,) -> AsyncIterable[str]:
    async for chunk in text:
        # forward original chunk (with <Name>...</Name> tags) to the UI
        yield chunk

2) Change tts_node to consume text for speaking only (don’t rely on parsed_to_original for UI) • Read and buffer incoming text, parse tags for TTS selection, and stream audio. Do not use

parsed_to_original

as the source for UI transcriptions. • Remove appending stripped text into

self.parsed_to_original

as the primary way to surface captions. Sketch (integrate into your existing logic):

async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings,) -> AsyncIterable[rtc.AudioFrame]:
    buffer = ""
    async for chunk in text:
        buffer += chunk
        # parse only the new content (or full buffer) into segments for speaking
        segments = list(self._parse_llm_response_to_speakers(buffer))
        # build multivoice_message and append to userdata.pending_multivoice_messages if needed
        # stream each segment via the appropriate TTS (same as your current streaming loop)
    # yield audio frames as you already do

3) (Recommended) Emit a structured transcription event for the UI • Instead of relying on ad-hoc lists, explicitly send the original tagged transcript or a structured object to the front end using the agent text/transcription stream (so the UI can parse and highlight). • If available in your session API, call the output/transcription helper (or just yield the original tagged text via

transcription_node

) so the frontend subscribes to

lk.transcription

. Example (after you parse/generate the original_with_tags):

# ensure transcription_node yields original_with_tags; or
# call session.output.add_transcription(...) if available to attach attributes (track id, etc.)

4) Remove/replace parsed_to_original usage •

self.parsed_to_original

is currently used as a fragile queue. Stop using it to transfer data between TTS and transcription nodes. Let the transcription node forward the original text and let tts_node parse independently for audio. Why this fixes the UI highlighting • The UI listens to the transcription text stream. If that stream receives the original Name…/Name markup, the frontend can parse the tags and highlight speakership. Separating the TTS parsing from the transcription stream ensures the UI sees the unmodified tagged text immediately. Quick checklist to apply • [ ] Replace

transcription_node

with pass-through yielding incoming chunks. • [ ] Remove filling/consuming

self.parsed_to_original

as the primary UI transcript transport. • [ ] Keep TTS parsing for audio only; build and store any multivoice metadata in userdata (as you already do), but not as the only source for UI captions. • [ ] Optionally push structured transcription events (original text + attributes) so the frontend can easily highlight speakers. If you want, tell me which SDK (Python) and the UI client you use and I’ll give a minimal patch diff for your file and a short frontend parse example that extracts NAME tags for highlighting. Sources: Text and transcriptions | LiveKit Documentation, with interactive elements

using python

livekit==1.0.23
livekit-agents==1.3.12
livekit-api==1.1.0
livekit-blingfire==1.1.0
livekit-plugins-anthropic==1.3.12
livekit-plugins-assemblyai==1.3.12
livekit-plugins-aws==1.3.12
livekit-plugins-cartesia==1.3.12
livekit-plugins-deepgram==1.3.12
livekit-plugins-elevenlabs==1.3.12
livekit-plugins-liveavatar==1.3.12
livekit-plugins-noise-cancellation==0.2.5
livekit-plugins-openai==1.3.12
livekit-plugins-silero==1.3.12
livekit-plugins-tavus==1.3.12
livekit-plugins-turn-detector==1.3.12
livekit-protocol==1.1.2

Give me the complete updated code that i shared to fix the issues. Make sure to keep everything same only change / remove the functions which needs to be added to fix this. Remove any function which wont be used any where/

Thanks — I updated only the functions required to fix the caption issue: I removed the unused shared queue, made

transcription_node

pass through the original tagged text immediately, and updated

tts_node

to parse for speaking and stream audio without relying on

parsed_to_original

. Everything else is unchanged. Updated agent (full file):

import asyncio
import logging
import time
import re
from typing import AsyncIterable

import requests
from livekit import rtc
from livekit.agents import Agent, ModelSettings, tts, tokenize

from config import WEBHOOK_URL
from egress import stop_egress_if_running
from fallback_providers import create_tts_with_fallback
from report import create_parseable_transcript_object_from_chat_ctx

logger = logging.getLogger("ai-assistant")


class MultiPersonaAssistant(Agent):
    opener_line = None
    SPEAKER_RE = re.compile(
        r"<(?P<speaker>[^<text>]+)>(?P</(?P=speaker)>.*?)<CHARACTER>",
        re.IGNORECASE | re.DOTALL,
    )
    SPEAKER_TTS = {}

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

    async def set_agents_description(self, prompts, selected_tts_backend, tts_model, vocabulary=None):
        combined_instructions = """When a message should be spoken by a particular person, use markup: "</CHARACTER>message<Jenny>" where X is the character. For any text outside of the xml tags, default character will be used.

For example:
Then out of sudden Jenny said, </Jenny>Hey I think I see it!<{prompt['agent_name']}> and the picture fell on the ground.

Available voices are as follows:

default: any text outside of the CHARACTER tags, use when none of below applies"""

        primary_agent_instructions = ""
        for prompt in prompts:
            if prompt["is_primary"]:
                self.SPEAKER_TTS["default"] = create_tts_with_fallback(selected_tts_backend,
                                                                       {"id": prompt["voice_id"]},
                                                                       gender=prompt["gender"],
                                                                       language=prompt["language"],
                                                                       vocabulary=vocabulary,
                                                                       tts_model=tts_model)
                primary_agent_instructions = f'\n\nAgent character description: {prompt["bot_prompt"]}'
                continue
            self.SPEAKER_TTS[prompt['agent_name']] = create_tts_with_fallback(selected_tts_backend,
                                                                              {"id": prompt["voice_id"]},
                                                                              gender=prompt["gender"],
                                                                              language=prompt["language"],
                                                                              vocabulary=vocabulary,
                                                                              tts_model=tts_model)
            combined_instructions += f"\n</{prompt['agent_name']}> {prompt['bot_prompt']}<Name>\n"

        combined_instructions += f"{primary_agent_instructions}"

        await self.update_instructions(combined_instructions)

    @staticmethod
    async def _trigger_post_processing(payload):
        response = await asyncio.to_thread(
            requests.post,
            WEBHOOK_URL,
            json=payload,
            timeout=30
        )

        response.raise_for_status()
        return response

    async def on_enter(self):
        if self.opener_line:
            await self.session.say(self.opener_line, allow_interruptions=True)

    async def on_exit(self):
        logger.info("transcript on exit called")
        chat_ctx = self.chat_ctx.to_dict()
        session_id = self.session.userdata["session_id"]
        recording_started_at = self.session.userdata["recording_started_at"]
        recording_stopped_at = time.time()
        try:
            egress_recording_stopped_at = await stop_egress_if_running(self.session.userdata.get("egress_id", None))
            if egress_recording_stopped_at:
                recording_stopped_at = egress_recording_stopped_at
        except Exception:
            pass
        parsed_transcript = create_parseable_transcript_object_from_chat_ctx(chat_ctx, recording_started_at,
                                                                             change_developer_to_gatekeeper=True)

        pending_multivoice_messages = self.session.userdata.get(
            "pending_multivoice_messages",
            [],
        )
        multivoice_index = 0

        for item in parsed_transcript:
            item["multivoice_message"] = None

            if item["speaker"] not in ["assistant", "developer", "gatekeeper"]:
                continue

            if multivoice_index >= len(pending_multivoice_messages):
                continue

            multivoice_message = pending_multivoice_messages[multivoice_index]
            multivoice_index += 1
            if not multivoice_message:
                item["multivoice_message"] = None
            else:
                item["multivoice_message"] = {
                    "parts": multivoice_message["parts"],
                }

                item["utterance"] = self._clean_multivoice_utterance(
                    multivoice_message["parts"]
                )

        usage_summary = None
        try:
            usage = self.session.userdata.get("usage")
            if usage:
                usage_summary = usage.get_summary().__dict__
                logger.info(f"[USAGE SUMMARY] {usage_summary}")
        except Exception:
            logger.exception("Failed to read usage summary")
        webhook_payload = {"event": "transcript_report",
                           "session_id": session_id,
                           "backend": "livekit",
                           "data": {
                               "transcript": parsed_transcript,
                               "started_at": recording_started_at * 1000,
                               "duration": (recording_stopped_at - recording_started_at) * 1000,
                               "termination_reason": self.session.userdata.get("termination_reason",
                                                                               "Participant exited"),
                               "usage_summary": usage_summary
                           }}
        logger.info("[Report] sending webhook payload...")
        logger.info(f"[Report] webhook_payload: {webhook_payload}")
        webhook_response = await self._trigger_post_processing(webhook_payload)
        logger.info(f"[Report] webhook_response: {webhook_response.text}")
        logger.info("[Report] sent webhook payload...")

    def _parse_llm_response_to_speakers(self, chunk):
        pos = 0

        for match in self.SPEAKER_RE.finditer(chunk):
            before = chunk[pos:match.start()].strip()
            if before:
                yield "default", before

            speaker = match.group("speaker")
            spoken_text = match.group("text").strip()
            if spoken_text:
                yield speaker, spoken_text

            pos = match.end()

        after = chunk[pos:].strip()
        if after:
            yield "default", after

    @staticmethod
    def _clean_multivoice_utterance(parts):
        return " ".join(part["text"] for part in parts).strip()

    async def transcription_node(self, text: AsyncIterable[str], model_settings: ModelSettings,) -> AsyncIterable[str]:
        # Pass original text (including </Name>... tags) through to the UI immediately.
        # The frontend should subscribe to the agent text/transcription stream to receive these chunks
        async for chunk in text:
            # chunk contains the original LLM response with tags; yield it unchanged
            yield chunk

    async def tts_node(
            self,
            text: AsyncIterable[str],
            model_settings: ModelSettings,
    ) -> AsyncIterable[rtc.AudioFrame]:
        """
        TTS node: consume incoming tagged text, parse into speaker segments for audio, stream audio frames.
        This function no longer relies on a shared list to pass transcription to the UI.
        """
        buffer = ""
        async for chunk in text:
            # accumulate buffer so multi-segment messages are parsed completely
            buffer += chunk

            # parse the current buffer into segments to be spoken
            parsed_segments = list(self._parse_llm_response_to_speakers(buffer))
            has_multivoice = any(speaker != "default" for speaker, _ in parsed_segments)

            try:
                user_session_data = self.session.userdata
            except ValueError:
                user_session_data = {}

            multivoice_message = None
            if has_multivoice:
                now_secs = time.time() - user_session_data.get("recording_started_at", time.time())
                parts_message = []
                for speaker, spoken_text in parsed_segments:
                    parts_message.append({
                        "text": spoken_text,
                        "voice_label": speaker,
                        "time_in_call_secs": int(now_secs),
                    })

                multivoice_message = {
                    "raw": buffer,
                    "parts": parts_message
                }

            # store multivoice metadata for on_exit reporting (if needed)
            user_session_data.setdefault(
                "pending_multivoice_messages",
                []
            ).append(multivoice_message)

            # stream each segment with appropriate TTS
            for speaker, spoken_text in parsed_segments:
                if not spoken_text:
                    continue

                wrapped_tts = self.SPEAKER_TTS.get(
                    speaker,
                    self.SPEAKER_TTS["default"],
                )

                if not wrapped_tts.capabilities.streaming:
                    wrapped_tts = tts.StreamAdapter(
                        tts=wrapped_tts,
                        sentence_tokenizer=tokenize.blingfire.SentenceTokenizer(
                            retain_format=True
                        ),
                    )

                conn_options = self.session.conn_options.tts_conn_options

                async with wrapped_tts.stream(
                        conn_options=conn_options
                ) as stream:
                    logger.info(f"[TTS] spoken_text: {spoken_text} {len(spoken_text)}")
                    stream.push_text(spoken_text)
                    stream.end_input()

                    async for ev in stream:
                        yield ev.frame

        # end of incoming text: nothing additional to yield here for audio

Notes • transcription_node now yields the incoming chunks unchanged;, with interactive elements

the issue in the first code I shared is we have llm response, we parse it in tts node, now using that we change the voice, when is it coming to transcript node, then the XML tags are removed. Hence the UI is not able to differente which agent is speakeking as the ra transcript is not being vbroadcasted to the client.

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!