how to process the text before TTS in all agents and Agenttasks in python

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

Can I manually add this node in all my agents (Multi agent system). how to add in Agenttask

Thanks — I’m responding to your question about whether you can manually add a custom text-processing node (for example an overridden

tts_node

) across all agents in a multi-agent system, and how to do this for

AgentTask

. Short answer • Yes — you can add a custom

tts_node

(or other pipeline nodes) for every agent by defining it on a shared base

Agent

class that all agents inherit from. •

AgentTask

instances use the agent/session pipeline, so they automatically get those overrides. You can also override nodes directly in an

AgentTask

subclass when you need task-specific behavior. How to apply this across all agents (recommended) 1. Create a base agent that overrides

tts_node

with your preprocessing logic. 2. Have all agent classes inherit from that base class so the custom node runs for every agent and during handoffs. Example (base agent approach)

from typing import AsyncIterable
from livekit.agents import Agent, ModelSettings
import rtc  # replace with actual AudioFrame import

class BaseTTSAgent(Agent):
    async def tts_node(
        self, text: AsyncIterable[str], model_settings: ModelSettings
    ) -> AsyncIterable[rtc.AudioFrame]:
        async def preprocess(stream: AsyncIterable[str]) -> AsyncIterable[str]:
            async for chunk in stream:
                # Example processing: normalize and expand abbreviations
                modified = chunk.strip().replace("API", "A P I").replace("LLM", "L L M")
                yield modified

        async for frame in Agent.default.tts_node(self, preprocess(text), model_settings):
            yield frame

# All your agents inherit:
class AgentA(BaseTTSAgent):
    ...

class AgentB(BaseTTSAgent):
    ...

How

AgentTask

gets the node •

AgentTask

runs inside the agent/session pipeline; if the agent starting the task has a custom

tts_node

, the task will use it automatically. • If you need task-specific processing, you can override the node in the

AgentTask

subclass directly: Example (task-specific override)

from livekit.agents import AgentTask

class MyTask(AgentTask[bool]):
    async def tts_node(self, text, model_settings):
        async def preprocess(stream):
            async for chunk in stream:
                # Task-specific transform
                yield chunk.replace("foo", "bar")
        async for frame in Agent.default.tts_node(self, preprocess(text), model_settings):
            yield frame

Notes and recommendations • Keep preprocessing streaming-friendly (work on each text chunk) unless you need whole-response context — then buffer and reassemble chunks. • Factor shared processing into a helper function or module and call it from the base agent (DRY). • Use

Agent.default.tts_node()

to reuse default TTS behavior after processing. • If you need different behavior for specific handoffs, implement branching logic inside the base

tts_node

using session/task context. If you’d like, tell me: • Do you want one uniform preprocessing for every agent, or different rules per agent/task? • Do you need SSML insertion, profanity filtering, or pronunciation overrides? With that I can provide a tailored snippet (streaming vs. whole-response) and show how to wire it into your AgentSession or TaskGroup. Sources: Pipeline nodes and hooks | LiveKit Documentation | Agent speech and audio | LiveKit Documentation | Agents and handoffs | LiveKit Documentation | Tasks and task groups | LiveKit Documentation