Is there a way to stop routing LLM output to the T...
# ask-ais
sparse-napkin-34305
12/16/2025, 5:45 PMIs there a way to stop routing LLM output to the TTS node entirely (not just mute playback) so that TTS is never invoked when sound is off?
t
tall-airline-16756
12/16/2025, 5:45 PMThanks for reaching out, Anmy. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
12/16/2025, 5:45 PMAcknowledging your question about preventing LLM output from ever invoking TTS when sound is off.
Yes — disable audio output at the session level so TTS is never called. When you start the AgentSession set
audioEnabled: false Copy code
await session.start({
agent,
room: ctx.room,
outputOptions: {
audioEnabled: false,
},
});audioEnabled: falselk.transcriptions
sparse-napkin-34305
12/16/2025, 5:49 PMwhere should i import name outputoptions
t
tall-airline-16756
12/16/2025, 5:49 PMYou're not required to import a separate
OutputOptions Copy code
import { voice } from '@livekit/agents';
await session.start({
room: ctx.room,
agent: myAgent,
outputOptions: {
audioEnabled: false,
},
});room_io.RoomOptions Copy code
from livekit.agents.voice import room_io
await session.start(
room=ctx.room,
agent=my_agent,
room_options=room_io.RoomOptions(
audio_output=False
)
)outputOptionss
sparse-napkin-34305
12/16/2025, 5:52 PMI want like only when user clicks sound button off the llm to stop sending output to tts
t
tall-airline-16756
12/16/2025, 5:52 PMResponding to your request about only stopping TTS when the user clicks the sound-off button.
Use the session's runtime toggle for audio output. Call the session output API when the button is clicked so the agent stops invoking TTS without ending the session.
Short examples:
• TypeScript (frontend RPC -> backend):
Copy code
// frontend: call RPC when button toggled
await room.localParticipant.sendRpc('toggle_sound', { enabled: false });
// backend: registered RPC handler
ctx.room.localParticipant.registerRpcMethod('toggle_sound', async (data) => {
session.output.setAudioEnabled(data.enabled);
}); Copy code
@ctx.room.local_participant.register_rpc_method('toggle_sound')
async def toggle_sound(data):
session.output.set_audio_enabled(data['enabled'])s
sparse-napkin-34305
12/16/2025, 5:54 PMbut here when we use session.output.set_audio_enabled(False) TTS is still producing output audio right? I want to reduce the cost soo my aim is to stop tts
t
tall-airline-16756
12/16/2025, 5:54 PMYou're right to confirm — calling
session.output.set_audio_enabled(False)audioEnabled: falselk.transcriptions
sparse-napkin-34305
12/16/2025, 5:59 PMBut audio is generating right from TTS even if we session.output.set_audio_enabled(False)
t
tall-airline-16756
12/16/2025, 6:00 PMYou're right to be concerned — if you're still hearing TTS after calling
session.output.set_audio_enabled(False)session.output.set_audio_enabled(False)session.output.setAudioEnabled(false)session.interrupt()say()lk.transcriptionsession.interrupt()session.interrupt()session.output.audio_enabledtts_nodes
sparse-napkin-34305
12/16/2025, 6:02 PMNo like when we do session.output.set_audio_enabled(False) this TTS is still generating the audio but not giving to the client is that right?....My aim is to stop TTS from working when user turns off the sound
t
tall-airline-16756
12/16/2025, 6:02 PMThanks — I hear you: you want to ensure TTS stops being synthesized at all when the user turns sound off (not just blocked from the client). Let’s confirm exactly what’s happening and fix it.
Short answer: calling
session.output.set_audio_enabled(False)set_audio_enabled(False)setAudioEnabled(false)say()session.output.set_audio_enabled(False)session.interrupt()say()s
sparse-napkin-34305
12/16/2025, 6:03 PMits working like the audio is getting muted but i want to know if TTS is working
t
tall-airline-16756
12/16/2025, 6:04 PMYou're asking whether TTS actually runs when you disable audio (it appears muted) and how to verify that—good question.
Facts:
• Disabling audio with
session.output.set_audio_enabled(False)tts_node()session.interrupt()say()s
sparse-napkin-34305
12/16/2025, 6:26 PMfrom future import annotations
import json
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import uvicorn
import logging
import os
import certifi
import time
import multiprocessing
from livekit.plugins import deepgram, groq, elevenlabs
from dotenv import load_dotenv
from custom_deepgram_tts import CustomDeepgramTTS
from livekit.agents.voice import UserStateChangedEvent, AgentStateChangedEvent
from livekit.agents.voice.io import PlaybackFinishedEvent
from livekit import api, rtc
from livekit.agents import (
tts,
Agent,
AgentSession,
JobContext,
llm,
voice,
WorkerOptions,
cli,
room_io,
RunContext,
RoomInputOptions,
RoomOutputOptions,
)
from livekit.plugins import silero, noise_cancellation
from livekit.agents import metrics, MetricsCollectedEvent, ConversationItemAddedEvent
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Set SSL_CERT_FILE to use certifi's certificate bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
load_dotenv(".env.local")
# =============================================================================
# FASTAPI APP
# =============================================================================
app = FastAPI(title="BMMI HR Voice AI", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/check-online")
def check_online():
print("Success")
return "Success-12"
# =============================================================================
# CUSTOM AGENT
# =============================================================================
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conversation_history: list[tuple[float, str, str]] = []
self.sound_enabled = True # Track sound state
def log_transcription(self, timestamp: float, role: str, text: str):
"""Keep local conversation record."""
self.conversation_history.append((timestamp, role, text))
self.conversation_history.sort(key=lambda x: x[0])
logging.info(self.conversation_history)
class ConditionalTTS(tts.TTS):
"""
A TTS wrapper that only synthesizes audio if the agent's sound_enabled
flag is True.
"""
def __init__(self, agent: MyAgent, tts_service: tts.TTS):
super().__init__(
capabilities=tts_service.capabilities,
sample_rate=tts_service.sample_rate,
num_channels=tts_service.num_channels,
)
self._agent = agent
self._tts = tts_service
async def synthesize(self, text: str) -> tts.SynthesizeStream:
if not self._agent.sound_enabled:
# Return an empty stream if sound is disabled
return tts.SynthesizeStream.from_text_stream(asyncio.sleep(10000)) # sleep forever
return await self._tts.synthesize(text)
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[tts.SynthesizeStream]:
if not self._agent.sound_enabled:
# sleep forever to not process anything
yield tts.SynthesizeStream.from_text_stream(asyncio.sleep(10000))
return
async with self._tts.stream(**kwargs) as stream:
yield stream
class LoggingTTS(tts.TTS):
"""
A TTS wrapper that logs the text being synthesized.
"""
def __init__(self, tts_service: tts.TTS):
super().__init__(
capabilities=tts_service.capabilities,
sample_rate=tts_service.sample_rate,
num_channels=tts_service.num_channels,
)
self._tts = tts_service
async def synthesize(self, text: str) -> tts.SynthesizeStream:
logging.info(f"🎤 TTS synthesizing text: '{text}'")
return await self._tts.synthesize(text)
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[tts.SynthesizeStream]:
"""
Log text chunks as they are pushed to the TTS stream.
"""
async with self._tts.stream(**kwargs) as stream:
original_push_text = stream.push_text
def log_and_push(text: str):
logging.info(f"🎤 TTS streaming text chunk: '{text}'")
original_push_text(text)
stream.push_text = log_and_push
yield stream
# =============================================================================
# ENTRYPOINT
# =============================================================================
async def entrypoint(ctx: JobContext):
await ctx.connect()
participant = await ctx.wait_for_participant()
logging.info(f"👤 Participant connected: {participant.identity}")
# -- Setup Agent + LLM + TTS + STT --
agent = MyAgent(
instructions="You are a helpful assistant that answers company-related questions clearly."
)
last_spoken_info = {"text": "", "timestamp": 0.0}
async def on_full_sentence(text: str):
"""Called when a full sentence is ready for TTS."""
nonlocal last_spoken_info
timestamp = time.time()
last_spoken_info = {"text": text, "timestamp": timestamp}
agent.log_transcription(timestamp, "assistant", text)
session = AgentSession(
stt = elevenlabs.STT(),
llm=groq.LLM(model="llama-3.1-8b-instant"),
tts=LoggingTTS(
tts_service=elevenlabs.TTS(
voice_id="pNInz6obpgDQGcFmaJgB", model="eleven_turbo_v2"
)
),
vad=silero.VAD.load(),
)
# Disable mic until ready
session.input.set_audio_enabled(False)
@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
nonlocal last_spoken_info
if ev.new_state == "speaking":
logging.info("🗣️ Agent started speaking")
# 🧠 NEW: Send final sentence payload 1 second after the agent starts speaking.
if last_spoken_info["text"]:
async def send_with_delay():
await asyncio.sleep(1) # Wait for 1 second
try:
payload = {
"type": "final_sentence",
"role": "assistant",
"text": last_spoken_info["text"],
"timestamp": last_spoken_info["timestamp"],
}
await ctx.room.local_participant.publish_data(
json.dumps(payload).encode("utf-8"), reliable=True
)
logging.info("✅ Sent final_sentence payload after 1s delay.")
except Exception as e:
logging.error(f"❌ Failed to send final sentence from on_agent_state_changed: {e}")
asyncio.create_task(send_with_delay())
# -------------------------------------------------------------------------
# 🗣️ Log conversation items
# -------------------------------------------------------------------------
@session.on("conversation_item_added")
def on_conversation_item(event: ConversationItemAddedEvent):
if not event.item.text_content:
return
if event.item.role=="user":
agent.log_transcription(event.item.created_at, event.item.role, event.item.text_content)
# -------------------------------------------------------------------------
# 📩 Handle text-based interruptions
# -------------------------------------------------------------------------
@ctx.room.on("data_received")
def on_data_received(data: rtc.DataPacket):
try:
payload = getattr(data, "payload", None) or getattr(data, "data", None)
if not payload:
return
raw_text = payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else str(payload)
msg = None
try:
msg = json.loads(raw_text)
# Only process messages that contain sound toggle
if msg.get("hidden") is True and "User turned sound" in msg.get("message", ""):
handle_sound_toggle(session, msg["message"])
except json.JSONDecodeError:
pass
msg_text = msg.get("message") if isinstance(msg, dict) else raw_text
logging.info(f"💬 User Message: {msg_text}")
except Exception as e:
logging.error(f"Error handling data_received: {e}")
# -------------------------------------------------------------------------
# 🎬 Start the agent session
# -------------------------------------------------------------------------
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(
noise_cancellation=lambda params: noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC(),
),
),)
# Re-enable audio input after connection
async def fallback_reenable():
await asyncio.sleep(2)
if not session.input.audio_enabled:
session.input.set_audio_enabled(True)
logging.warning("Audio re-enabled via fallback")
asyncio.create_task(fallback_reenable())
@session.output.audio.on("playback_finished")
def _on_playback_finished(ev: PlaybackFinishedEvent):
if ev.interrupted:
logging.info("Agent was interrupted during playback")
interruption_payload = {
"type": "interrupted",
"message": "User interrupted the assistant.",
}
asyncio.create_task(
ctx.room.local_participant.publish_data(
json.dumps(interruption_payload).encode("utf-8"), reliable=True
))
await session.say("Hello, how can I help you today?")
# =============================================================================
# SERVER STARTUP
# =============================================================================
async def start_api_server(port: int):
logging.info(f"🚀 Starting FastAPI server on port {port}")
config = uvicorn.Config(app=app, host="0.0.0.0", port=port, log_level="info")
server = uvicorn.Server(config)
await server.serve()
def run_api_server(base_port):
asyncio.run(start_api_server(base_port))
def handle_sound_toggle(session: AgentSession, message: str):
if "on" in message.lower():
logging.info("🔊 User turned SOUND ON")
session.output.set_audio_enabled(True)
elif "off" in message.lower():
logging.info("🔇 User turned SOUND OFF")
session.output.set_audio_enabled(False)
def start_livekit_agent(port: int, agent_name: str):
logging.info(f"🤖 Starting LiveKit agent on port {port}")
options = WorkerOptions(entrypoint_fnc=entrypoint, port=port,agent_name="BMMI HR VoiceAI")
cli.run_app(options)
def main():
logging.info("Starting BMMI HR Voice AI system")
base_port = 5029
api_process = multiprocessing.Process(target=run_api_server, args=(base_port,))
api_process.start()
try:
start_livekit_agent(base_port + 1, "agent_name")
except KeyboardInterrupt:
logging.info("🛑 Shutdown signal received")
finally:
api_process.terminate()
api_process.join()
logging.info("✅ System shutdown complete")
if name == "__main__":
load_dotenv()
main()..........can you tell me if when output audio enable is made to false is the tts node processing?
t
tall-airline-16756
12/16/2025, 6:26 PMThanks — from your code and setup, calling
session.output.set_audio_enabled(False)session.interrupt()LoggingTTSsession.output.set_audio_enabled(False)synthesizeasyncio.sleep(10000)SynthesizeStreamself._tts.synthesize(...)sound_enabledsession.output.audio_enabled🎤 TTS synthesizing text:LoggingTTSsession.interrupt()playback_finishedev.interruptedConditionalTTSs
sparse-napkin-34305
12/16/2025, 6:30 PMis there other way to check which all are used on livekit dashboard or something like the cost
t
tall-airline-16756
12/16/2025, 6:30 PMYou can verify usage/costs and TTS activity using these options:
1) LiveKit metrics / dashboard
• Check the LiveKit dashboard’s Metrics/Usage sections to see published audio tracks, TTS-related events, and bandwidth/compute usage. Look for TTSMetrics or audio-publish counts to confirm synthesis activity.
• Check the Usage/Billing view for cost summaries tied to media processing.
2) Runtime logging & events (most reliable)
• Use your existing LoggingTTS wrapper: if you stop seeing the “🎤 TTS synthesizing text” log after disabling audio, TTS isn’t being invoked.
• Emit or collect TTSMetrics events server-side and confirm they stop after toggling audio off.
• Log
session.output.audio_enabledsession.interrupt()playback_finisheds
sparse-napkin-34305
12/18/2025, 4:44 AMfrom future import annotations
import json
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import uvicorn
import logging
import os
import certifi
import time
import multiprocessing
from livekit.plugins import deepgram, groq, elevenlabs
from dotenv import load_dotenv
from custom_deepgram_tts import CustomDeepgramTTS
from livekit.agents.voice import UserStateChangedEvent, AgentStateChangedEvent
from livekit.agents.voice.io import PlaybackFinishedEvent
from livekit import api, rtc
from livekit.agents import (
tts,
Agent,
AgentSession,
JobContext,
llm,
voice,
WorkerOptions,
cli,
room_io,
RunContext,
RoomInputOptions,
RoomOutputOptions,
)
from livekit.plugins import silero, noise_cancellation
from livekit.agents import metrics, MetricsCollectedEvent, ConversationItemAddedEvent
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Set SSL_CERT_FILE to use certifi's certificate bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
load_dotenv(".env.local")
# =============================================================================
# FASTAPI APP
# =============================================================================
app = FastAPI(title="BMMI HR Voice AI", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/check-online")
def check_online():
print("Success")
return "Success-12"
# =============================================================================
# CUSTOM AGENT
# =============================================================================
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conversation_history: list[tuple[float, str, str]] = []
self.sound_enabled = True # Track sound state
def log_transcription(self, timestamp: float, role: str, text: str):
"""Keep local conversation record."""
self.conversation_history.append((timestamp, role, text))
self.conversation_history.sort(key=lambda x: x[0])
logging.info(self.conversation_history)
class LoggingTTS(tts.TTS):
"""
A TTS wrapper that logs the text being synthesized.
"""
def __init__(self, tts_service: tts.TTS):
super().__init__(
capabilities=tts_service.capabilities,
sample_rate=tts_service.sample_rate,
num_channels=tts_service.num_channels,
)
self._tts = tts_service
async def synthesize(self, text: str) -> tts.SynthesizeStream:
logging.info(f"🎤 TTS synthesizing text: '{text}'")
return await self._tts.synthesize(text)
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[tts.SynthesizeStream]:
"""
Log text chunks as they are pushed to the TTS stream.
"""
async with self._tts.stream(**kwargs) as stream:
original_push_text = stream.push_text
def log_and_push(text: str):
logging.info(f"🎤 TTS streaming text chunk: '{text}'")
original_push_text(text)
stream.push_text = log_and_push
yield stream
# =============================================================================
# ENTRYPOINT
# =============================================================================
async def entrypoint(ctx: JobContext):
await ctx.connect()
participant = await ctx.wait_for_participant()
logging.info(f"👤 Participant connected: {participant.identity}")
# -- Setup Agent + LLM + TTS + STT --
agent = MyAgent(
instructions="You are a helpful assistant that answers company-related questions clearly."
)
last_spoken_info = {"text": "", "timestamp": 0.0}
async def on_full_sentence(text: str):
"""Called when a full sentence is ready for TTS."""
nonlocal last_spoken_info
timestamp = time.time()
last_spoken_info = {"text": text, "timestamp": timestamp}
agent.log_transcription(timestamp, "assistant", text)
session = AgentSession(
stt = elevenlabs.STT(),
llm=groq.LLM(model="llama-3.1-8b-instant"),
tts=LoggingTTS(
tts_service=elevenlabs.TTS(
voice_id="pNInz6obpgDQGcFmaJgB", model="eleven_turbo_v2"
)
),
vad=silero.VAD.load(),
)
# Disable mic until ready
session.input.set_audio_enabled(False)
@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
nonlocal last_spoken_info
if ev.new_state == "speaking":
logging.info("🗣️ Agent started speaking")
# 🧠 NEW: Send final sentence payload 1 second after the agent starts speaking.
if last_spoken_info["text"]:
async def send_with_delay():
await asyncio.sleep(1) # Wait for 1 second
try:
payload = {
"type": "final_sentence",
"role": "assistant",
"text": last_spoken_info["text"],
"timestamp": last_spoken_info["timestamp"],
}
await ctx.room.local_participant.publish_data(
json.dumps(payload).encode("utf-8"), reliable=True
)
logging.info("✅ Sent final_sentence payload after 1s delay.")
except Exception as e:
logging.error(f"❌ Failed to send final sentence from on_agent_state_changed: {e}")
asyncio.create_task(send_with_delay())
# -------------------------------------------------------------------------
# 🗣️ Log conversation items
# -------------------------------------------------------------------------
@session.on("conversation_item_added")
def on_conversation_item(event: ConversationItemAddedEvent):
if not event.item.text_content:
return
if event.item.role=="user":
agent.log_transcription(event.item.created_at, event.item.role, event.item.text_content)
# -------------------------------------------------------------------------
# 📩 Handle text-based interruptions
# -------------------------------------------------------------------------
@ctx.room.on("data_received")
def on_data_received(data: rtc.DataPacket):
try:
payload = getattr(data, "payload", None) or getattr(data, "data", None)
if not payload:
return
raw_text = payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else str(payload)
msg = None
try:
msg = json.loads(raw_text)
# Only process messages that contain sound toggle
if msg.get("hidden") is True and "User turned sound" in msg.get("message", ""):
handle_sound_toggle(session, msg["message"])
except json.JSONDecodeError:
pass
msg_text = msg.get("message") if isinstance(msg, dict) else raw_text
logging.info(f"💬 User Message: {msg_text}")
except Exception as e:
logging.error(f"Error handling data_received: {e}")
# -------------------------------------------------------------------------
# 🎬 Start the agent session
# -------------------------------------------------------------------------
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(
noise_cancellation=noise_cancellation.BVC(),
),
),
)
# Re-enable audio input after connection
async def fallback_reenable():
await asyncio.sleep(2)
if not session.input.audio_enabled:
session.input.set_audio_enabled(True)
logging.warning("Audio re-enabled via fallback")
asyncio.create_task(fallback_reenable())
@session.output.audio.on("playback_finished")
def _on_playback_finished(ev: PlaybackFinishedEvent):
if ev.interrupted:
logging.info("Agent was interrupted during playback")
interruption_payload = {
"type": "interrupted",
"message": "User interrupted the assistant.",
}
asyncio.create_task(
ctx.room.local_participant.publish_data(
json.dumps(interruption_payload).encode("utf-8"), reliable=True
))
await session.say("Hello, how can I help you today?")
# =============================================================================
# SERVER STARTUP
# =============================================================================
async def start_api_server(port: int):
logging.info(f"🚀 Starting FastAPI server on port {port}")
config = uvicorn.Config(app=app, host="0.0.0.0", port=port, log_level="info")
server = uvicorn.Server(config)
await server.serve()
def run_api_server(base_port):
asyncio.run(start_api_server(base_port))
def handle_sound_toggle(session: AgentSession, message: str):
if "on" in message.lower():
logging.info("🔊 User turned SOUND ON")
session.output.set_audio_enabled(True)
elif "off" in message.lower():
logging.info("🔇 User turned SOUND OFF")
session.output.set_audio_enabled(False)
def start_livekit_agent(port: int, agent_name: str):
logging.info(f"🤖 Starting LiveKit agent on port {port}")
options = WorkerOptions(entrypoint_fnc=entrypoint, port=port,agent_name="BMMI HR VoiceAI")
cli.run_app(options)
def main():
logging.info("Starting BMMI HR Voice AI system")
base_port = 5029
api_process = multiprocessing.Process(target=run_api_server, args=(base_port,))
api_process.start()
try:
start_livekit_agent(base_port + 1, "agent_name")
except KeyboardInterrupt:
logging.info("🛑 Shutdown signal received")
finally:
api_process.terminate()
api_process.join()
logging.info("✅ System shutdown complete")
if name == "__main__":
load_dotenv()
main()........... 101326.364 ERROR livekit task exception: <Task finished name='Task-84' coro=<AudioStream._run() done,
defined at
C:\Users\AnmyJosy\AppData\Local\Programs\Python\Python310\lib\site-packages\…
exception=AttributeError("'AudioStream' object has no attribute
'_processor'")>
Traceback (most recent call last):
File "C:\Users\AnmyJosy\AppData\Local\Programs\Python\Python310\lib\site-packages\livekit\rtc\audio_stream.py", line
276, in _run
if self._processor is not None and self._processor.enabled:
AttributeError: 'AudioStream' object has no attribute '_processor'
t
tall-airline-16756
12/18/2025, 4:44 AMThanks — I reviewed your code and the error. Short answers first, then targeted action items.
Short answers
• Calling
session.output.set_audio_enabled(False)session.output.set_audio_enabled(False)session.interrupt()session.output.audio_enabledSynthesizeStreamsay()elevenlabs.TTSsound_enableds
sparse-napkin-34305
12/18/2025, 4:45 AM101326.364 ERROR livekit task exception: <Task finished name='Task-84' coro=<AudioStream._run() done,
defined at
C:\Users\AnmyJosy\AppData\Local\Programs\Python\Python310\lib\site-packages\…
exception=AttributeError("'AudioStream' object has no attribute
'_processor'")>
Traceback (most recent call last):
File "C:\Users\AnmyJosy\AppData\Local\Programs\Python\Python310\lib\site-packages\livekit\rtc\audio_stream.py", line
276, in _run
if self._processor is not None and self._processor.enabled:
AttributeError: 'AudioStream' object has no attribute '_processor'....why i am i getting this error?
t
tall-airline-16756
12/18/2025, 4:45 AMThanks — that traceback points to an internal attribute access on the LiveKit SDK's AudioStream object. Short answer: it's caused by a mismatch between the SDK code you have and the runtime expectations (an internal attribute
_processorself._processor_run()_processor_run()_run()_run()s
sparse-napkin-34305
12/18/2025, 4:46 AM(venv) C:\Users\AnmyJosy\Desktop\full_transcript>py -3.10 -m pip show livekit livekit-agents
Name: livekit
Version: 1.0.21
Summary: Python Real-time SDK for LiveKit
Home-page: https://github.com/livekit/python-sdks
Author:
Author-email:
License: Apache-2.0
Location: c:\users\anmyjosy\appdata\local\programs\python\python310\lib\site-packages
Requires: aiofiles, numpy, protobuf, types-protobuf
Required-by: livekit-agents, livekit-plugins-groq, livekit-plugins-noise-cancellation
---
Name: livekit-agents
Version: 1.3.6
Summary: A powerful framework for building realtime voice AI agents
Home-page:
Author:
Author-email: LiveKit <hello@livekit.io>
License:
Location: c:\users\anmyjosy\appdata\local\programs\python\python310\lib\site-packages
Requires: aiofiles, aiohttp, av, certifi, click, colorama, docstring-parser, eval-type-backport, livekit, livekit-api, livekit-blingfire, livekit-protocol, nest-asyncio, numpy, openai, opentelemetry-api, opentelemetry-exporter-otlp, opentelemetry-sdk, prometheus-client, protobuf, psutil, pydantic, pyjwt, sounddevice, typer, types-protobuf, typing-extensions, watchfiles
Required-by: livekit-plugins-bey, livekit-plugins-deepgram, livekit-plugins-elevenlabs, livekit-plugins-groq, livekit-plugins-openai, livekit-plugins-silero, livekit-plugins-turn-detector
t
tall-airline-16756
12/18/2025, 4:47 AMThanks — from the versions you shared, the most likely cause is a version mismatch between
livekitlivekit-agentslivekits
sparse-napkin-34305
12/18/2025, 6:58 AMi was trying to do the same thing with input audio so to reduce the cost .....so i dont want deepgram to work on starting is that possible?
t
tall-airline-16756
12/18/2025, 6:58 AMYou can prevent Deepgram (or any STT) from running until you explicitly enable it. Two safe patterns:
1) Start session with input audio disabled, then enable when user turns mic on
• At session start set input disabled:
Copy code
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(...),
),
)
# ensure STT won't run until explicitly enabled
session.input.set_audio_enabled(False) Copy code
session.input.set_audio_enabled(True)session.input.audio_enabled Copy code
# create session without real STT
session = AgentSession(stt=NoopSTT(), llm=..., tts=..., ...)
# later, when enabling mic:
session.stt = deepgram.Deepgram(...) # attach real STTsession.input.audio_enabledsession.interrupt()s
sparse-napkin-34305
12/18/2025, 7:04 AMfrom future import annotations
import json
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import uvicorn
import logging
import os
import certifi
import time
import multiprocessing
from livekit.plugins import deepgram, groq, elevenlabs
from dotenv import load_dotenv
from custom_deepgram_tts import CustomDeepgramTTS
from livekit.agents.voice import UserStateChangedEvent, AgentStateChangedEvent
from livekit.agents.voice.io import PlaybackFinishedEvent
from livekit import api, rtc
from livekit.agents import (
tts,
Agent,
AgentSession,
JobContext,
llm,
voice,
WorkerOptions,
cli,
room_io,
RunContext,
RoomInputOptions,
RoomOutputOptions,
)
from livekit.plugins import silero, noise_cancellation
from livekit.agents import metrics, MetricsCollectedEvent, ConversationItemAddedEvent
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Set SSL_CERT_FILE to use certifi's certificate bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
load_dotenv(".env.local")
# =============================================================================
# FASTAPI APP
# =============================================================================
app = FastAPI(title="BMMI HR Voice AI", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/check-online")
def check_online():
print("Success")
return "Success-12"
# =============================================================================
# CUSTOM AGENT
# =============================================================================
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conversation_history: list[tuple[float, str, str]] = []
self.sound_enabled = True # Track sound state
def log_transcription(self, timestamp: float, role: str, text: str):
"""Keep local conversation record."""
self.conversation_history.append((timestamp, role, text))
self.conversation_history.sort(key=lambda x: x[0])
logging.info(self.conversation_history)
class LoggingTTS(tts.TTS):
"""
A TTS wrapper that logs the text being synthesized.
"""
def __init__(self, tts_service: tts.TTS):
super().__init__(
capabilities=tts_service.capabilities,
sample_rate=tts_service.sample_rate,
num_channels=tts_service.num_channels,
)
self._tts = tts_service
async def synthesize(self, text: str) -> tts.SynthesizeStream:
logging.info(f"🎤 TTS synthesizing text: '{text}'")
return await self._tts.synthesize(text)
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[tts.SynthesizeStream]:
"""
Log text chunks as they are pushed to the TTS stream.
"""
async with self._tts.stream(**kwargs) as stream:
original_push_text = stream.push_text
def log_and_push(text: str):
logging.info(f"🎤 TTS streaming text chunk: '{text}'")
original_push_text(text)
stream.push_text = log_and_push
yield stream
# =============================================================================
# ENTRYPOINT
# =============================================================================
async def entrypoint(ctx: JobContext):
await ctx.connect()
participant = await ctx.wait_for_participant()
logging.info(f"👤 Participant connected: {participant.identity}")
# -- Setup Agent + LLM + TTS + STT --
agent = MyAgent(
instructions="You are a helpful assistant that answers company-related questions clearly."
)
last_spoken_info = {"text": "", "timestamp": 0.0}
async def on_full_sentence(text: str):
"""Called when a full sentence is ready for TTS."""
nonlocal last_spoken_info
timestamp = time.time()
last_spoken_info = {"text": text, "timestamp": timestamp}
agent.log_transcription(timestamp, "assistant", text)
session = AgentSession(
stt = deepgram.STT(
model="nova-2-general",
),
llm=groq.LLM(model="llama-3.1-8b-instant"),
tts=LoggingTTS(
tts_service=elevenlabs.TTS(
voice_id="pNInz6obpgDQGcFmaJgB", model="eleven_turbo_v2"
)
),
vad=silero.VAD.load(),
)
@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
nonlocal last_spoken_info
if ev.new_state == "speaking":
logging.info("🗣️ Agent started speaking")
# 🧠 NEW: Send final sentence payload 1 second after the agent starts speaking.
if last_spoken_info["text"]:
async def send_with_delay():
await asyncio.sleep(1) # Wait for 1 second
try:
payload = {
"type": "final_sentence",
"role": "assistant",
"text": last_spoken_info["text"],
"timestamp": last_spoken_info["timestamp"],
}
await ctx.room.local_participant.publish_data(
json.dumps(payload).encode("utf-8"), reliable=True
)
logging.info("✅ Sent final_sentence payload after 1s delay.")
except Exception as e:
logging.error(f"❌ Failed to send final sentence from on_agent_state_changed: {e}")
asyncio.create_task(send_with_delay())
# -------------------------------------------------------------------------
# 🗣️ Log conversation items
# -------------------------------------------------------------------------
@session.on("conversation_item_added")
def on_conversation_item(event: ConversationItemAddedEvent):
if not event.item.text_content:
return
if event.item.role=="user":
agent.log_transcription(event.item.created_at, event.item.role, event.item.text_content)
# -------------------------------------------------------------------------
# 📩 Handle text-based interruptions
# -------------------------------------------------------------------------
@ctx.room.local_participant.register_rpc_method("sound_control")
async def handle_sound_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
sound_enabled = payload.get("soundEnabled")
if sound_enabled is True:
logging.info("🔊 User turned SOUND ON via RPC")
session.output.set_audio_enabled(True)
elif sound_enabled is False:
logging.info("🔇 User turned SOUND OFF via RPC")
session.output.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling sound_control RPC: {e}")
@ctx.room.local_participant.register_rpc_method("mic_control")
async def handle_mic_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
mic_enabled = payload.get("micEnabled")
if mic_enabled is True:
logging.info("🎤 User turned MIC ON via RPC")
session.input.set_audio_enabled(True)
elif mic_enabled is False:
logging.info("🔇 User turned MIC OFF via RPC")
session.input.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling mic_control RPC: {e}")
@ctx.room.on("data_received")
def on_data_received(data: rtc.DataPacket):
try:
payload = getattr(data, "payload", None) or getattr(data, "data", None)
if not payload:
return
raw_text = payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else str(payload)
logging.info(f"💬 User Message: {raw_text}")
except Exception as e:
logging.error(f"Error handling data_received: {e}")
# -------------------------------------------------------------------------
# 🎬 Start the agent session
# -------------------------------------------------------------------------
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(...),
),
)
session.input.set_audio_enabled(False)
@session.output.audio.on("playback_finished")
def _on_playback_finished(ev: PlaybackFinishedEvent):
if ev.interrupted:
logging.info("Agent was interrupted during playback")
interruption_payload = {
"type": "interrupted",
"message": "User interrupted the assistant.",
}
asyncio.create_task(
ctx.room.local_participant.publish_data(
json.dumps(interruption_payload).encode("utf-8"), reliable=True
))
# =============================================================================
# SERVER STARTUP
# =============================================================================
async def start_api_server(port: int):
logging.info(f"🚀 Starting FastAPI server on port {port}")
config = uvicorn.Config(app=app, host="0.0.0.0", port=port, log_level="info")
server = uvicorn.Server(config)
await server.serve()
def run_api_server(base_port):
asyncio.run(start_api_server(base_port))
def start_livekit_agent(port: int, agent_name: str):
logging.info(f"🤖 Starting LiveKit agent on port {port}")
options = WorkerOptions(entrypoint_fnc=entrypoint, port=port,agent_name="BMMI HR VoiceAI")
cli.run_app(options)
def main():
logging.info("Starting BMMI HR Voice AI system")
base_port = 5029
api_process = multiprocessing.Process(target=run_api_server, args=(base_port,))
api_process.start()
try:
start_livekit_agent(base_port + 1, "agent_name")
except KeyboardInterrupt:
logging.info("🛑 Shutdown signal received")
finally:
api_process.terminate()
api_process.join()
logging.info("✅ System shutdown complete")
if name == "__main__":
load_dotenv()
main()......is this correct
t
tall-airline-16756
12/18/2025, 7:04 AMYou're very close — overall approach is correct. A few concrete fixes and confirmations so your toggles truly stop STT/TTS and avoid errors.
1) STT (Deepgram) off until enabled
• Your pattern of creating session with
stt=deepgram.STT(...)session.input.set_audio_enabled(False) Copy code
session = AgentSession(
stt=NoopSTT(), # no network calls
llm=...,
tts=LoggingTTS(...),
vad=...
)
# when user enables mic
session.stt = deepgram.STT(model="nova-2-general")
session.input.set_audio_enabled(True)handle_mic_controlset_audio_enabled(True)session.stt = deepgram.STT(...)session.output.set_audio_enabled(False)LoggingTTS.synthesizeself._tts.synthesize(...) Copy code
async def synthesize(self, text: str) -> tts.SynthesizeStream:
if not agent.sound_enabled or not session.output.audio_enabled:
<http://logging.info|logging.info>("Skipping TTS synth because sound is disabled")
return tts.SynthesizeStream.empty() # or other immediate no-op
<http://logging.info|logging.info>(f"🎤 TTS synthesizing text: '{text}'")
return await self._tts.synthesize(text)session.interrupt()livekitlivekit-agents_processorsession.input.set_audio_enabled(False)await session.start(...)room_optionsLoggingTTS.streamoriginal_push_textctx.room.local_participantNoopSTTSynthesizeStream.empty()s
sparse-napkin-34305
12/18/2025, 7:07 AMNameError: name 'NoopSTT' is not defined
t
tall-airline-16756
12/18/2025, 7:07 AMThanks — that NameError means you referenced
NoopSTTNoopSTT Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import tts, voice
class NoopSTT(voice.STT):
"""A no-op STT that accepts audio but never sends it to an external STT."""
def __init__(self):
super().__init__(capabilities=None)
async def transcribe(self, audio_stream) -> AsyncIterator[voice.TranscriptionChunk]:
# Yield nothing — acts as if there is no transcription
if False:
yield # type: ignore
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[voice.TranscriptionStream]:
# Provide a stream object that has a no-op push_frame / close behavior.
class _NoopStream:
def push_frame(self, frame):
pass
async def close(self):
pass
yield _NoopStream()session.stt = deepgram.STT(...)s
sparse-napkin-34305
12/18/2025, 7:52 AM(venv) C:\Users\AnmyJosy\Desktop\full_transcript>py -3.10 agent.py dev
DEBUGasyncioUsing proactor: IocpProactor
132138.035 DEBUG asyncio Using proactor: IocpProactor
DEVlivekit.agentsWatching C:\Users\AnmyJosy\Desktop\full_transcript
132138.040 DEV livekit.agents Watching C:\Users\AnmyJosy\Desktop\full_transcript
132140.755 DEBUG asyncio Using proactor: IocpProactor
132140.761 INFO livekit.agents starting worker {"version": "1.3.6", "rtc-version": "1.0.22"}
INFO: Started server process [11688]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:5029 (Press CTRL+C to quit)
132140.762 INFO livekit.agents HTTP server listening on :5030
132140.959 INFO livekit.agents registered worker
{"agent_name": "BMMI HR VoiceAI", "id": "AW_iaJnHXmGYeFt", "url":
"wss://part12-cy0qmp6y.livekit.cloud", "region": "India South", "protocol": 16}
132141.011 INFO livekit.agents received job request
{"job_id": "AJ_VX28gK4D8348", "dispatch_id": "AD_6s5UBx8snLyu", "room":
"room-peQg-B9Op", "room_id": "RM_pByhcuuHh8y6", "agent_name": "BMMI HR VoiceAI", "resuming": false, "enable_recording":
false}
132141.042 DEBUG livekit.agents received assignment
{"agent_name": "BMMI HR VoiceAI", "room_id": "RM_pByhcuuHh8y6", "room":
"room-peQg-B9Op", "job_id": "AJ_VX28gK4D8348", "dispatch_id": "AD_6s5UBx8snLyu", "enable_recording": false}
132141.054 INFO livekit.agents initializing job runner {"tid": 25844}
132141.061 DEBUG asyncio Using proactor: IocpProactor
132141.064 INFO livekit.agents job runner initialized {"tid": 25844, "elapsed_time": 0.0}
132141.567 INFO root 👤 Participant connected: identity-ogZq
132141.724 DEBUG livekit.agents input stream detached
{"participant": null, "source": "SOURCE_UNKNOWN", "accepted_sources":
["SOURCE_MICROPHONE"]}
132141.730 DEBUG livekit.agents http_session(): creating a new httpclient ctx
132141.732 DEBUG livekit.agents using audio io:
RoomIOAgentSessionTranscriptSynchronizerRoomIOAgentSessionTranscriptSynchronizerRoomIOt
tall-airline-16756
12/18/2025, 7:53 AMYou're seeing Deepgram still attempt to connect because you constructed the session with
stt=deepgram.STT(...)session.input.set_audio_enabled(False) Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import voice
class NoopSTT(voice.STT):
"""No-op STT: never opens a network connection or yields transcription."""
def __init__(self):
super().__init__(capabilities=None)
async def transcribe(self, audio_stream) -> AsyncIterator[voice.TranscriptionChunk]:
if False:
yield # keep as an async generator but yield nothing
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[voice.TranscriptionStream]:
class _NoopStream:
def push_frame(self, frame):
pass
async def close(self):
pass
yield _NoopStream() Copy code
# create session without instantiating Deepgram
session = AgentSession(
stt=NoopSTT(),
llm=groq.LLM(...),
tts=LoggingTTS(...),
vad=silero.VAD.load(),
)
# when user enables mic (RPC handler)
session.stt = deepgram.STT(model="nova-2-general") # instantiate Deepgram here
session.input.set_audio_enabled(True)
# when user disables mic
session.input.set_audio_enabled(False)
session.stt = NoopSTT() # optionally detach Deepgram to stop background connectionsession.stt = NoopSTT()session.input.set_audio_enabled(False)s
sparse-napkin-34305
12/18/2025, 7:53 AMfrom future import annotations
import json
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import uvicorn
import logging
import os
import certifi
import time
import multiprocessing
from livekit.plugins import deepgram, groq, elevenlabs
from dotenv import load_dotenv
from custom_deepgram_tts import CustomDeepgramTTS
from livekit.agents.voice import UserStateChangedEvent, AgentStateChangedEvent
from livekit.agents.voice.io import PlaybackFinishedEvent
from livekit import api, rtc
from livekit.agents import (
tts,
stt,
Agent,
AgentSession,
JobContext,
llm,
voice,
WorkerOptions,
cli,
room_io,
RunContext,
RoomInputOptions,
RoomOutputOptions,
)
from livekit.plugins import silero, noise_cancellation
from livekit.agents import metrics, MetricsCollectedEvent, ConversationItemAddedEvent
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Set SSL_CERT_FILE to use certifi's certificate bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
load_dotenv(".env.local")
# =============================================================================
# FASTAPI APP
# =============================================================================
app = FastAPI(title="BMMI HR Voice AI", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/check-online")
def check_online():
print("Success")
return "Success-12"
# =============================================================================
# CUSTOM AGENT
# =============================================================================
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conversation_history: list[tuple[float, str, str]] = []
self.sound_enabled = True # Track sound state
def log_transcription(self, timestamp: float, role: str, text: str):
"""Keep local conversation record."""
self.conversation_history.append((timestamp, role, text))
self.conversation_history.sort(key=lambda x: x[0])
logging.info(self.conversation_history)
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import tts, voice
from livekit.agents import stt
class NoopSTT(stt.STT):
"""A no-op STT that accepts audio but never sends it to an external STT."""
def __init__(self):
super().__init__(capabilities=None)
async def _recognize_impl(self, buffer: rtc.AudioFrame) -> str:
"""A no-op recognize implementation."""
return ""
async def transcribe(self, audio_stream) -> AsyncIterator[stt.TranscriptionChunk]:
# Yield nothing — acts as if there is no transcription
if False:
yield # type: ignore
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[stt.TranscriptionStream]:
# Provide a stream object that has a no-op push_frame / close behavior.
class _NoopStream:
def push_frame(self, frame):
pass
async def close(self):
pass
yield _NoopStream()
# =============================================================================
# ENTRYPOINT
# =============================================================================
async def entrypoint(ctx: JobContext):
await ctx.connect()
participant = await ctx.wait_for_participant()
logging.info(f"👤 Participant connected: {participant.identity}")
# -- Setup Agent + LLM + TTS + STT --
agent = MyAgent(
instructions="You are a helpful assistant that answers company-related questions clearly."
)
last_spoken_info = {"text": "", "timestamp": 0.0}
async def on_full_sentence(text: str):
"""Called when a full sentence is ready for TTS."""
nonlocal last_spoken_info
timestamp = time.time()
last_spoken_info = {"text": text, "timestamp": timestamp}
agent.log_transcription(timestamp, "assistant", text)
session = AgentSession(
stt=deepgram.STT(model="nova-2-general"),
llm=groq.LLM(model="llama-3.1-8b-instant"),
tts=elevenlabs.TTS(
voice_id="pNInz6obpgDQGcFmaJgB", model="eleven_turbo_v2"
),
vad=silero.VAD.load(),
)
# Disable mic until ready
session.input.set_audio_enabled(False)
@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
nonlocal last_spoken_info
if ev.new_state == "speaking":
logging.info("🗣️ Agent started speaking")
# 🧠 NEW: Send final sentence payload 1 second after the agent starts speaking.
if last_spoken_info["text"]:
async def send_with_delay():
await asyncio.sleep(1) # Wait for 1 second
try:
payload = {
"type": "final_sentence",
"role": "assistant",
"text": last_spoken_info["text"],
"timestamp": last_spoken_info["timestamp"],
}
await ctx.room.local_participant.publish_data(
json.dumps(payload).encode("utf-8"), reliable=True
)
logging.info("✅ Sent final_sentence payload after 1s delay.")
except Exception as e:
logging.error(f"❌ Failed to send final sentence from on_agent_state_changed: {e}")
asyncio.create_task(send_with_delay())
# -------------------------------------------------------------------------
# 🗣️ Log conversation items
# -------------------------------------------------------------------------
@session.on("conversation_item_added")
def on_conversation_item(event: ConversationItemAddedEvent):
if not event.item.text_content:
return
if event.item.role=="user":
agent.log_transcription(event.item.created_at, event.item.role, event.item.text_content)
# -------------------------------------------------------------------------
# 📩 Handle text-based interruptions
# -------------------------------------------------------------------------
@ctx.room.local_participant.register_rpc_method("sound_control")
async def handle_sound_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
sound_enabled = payload.get("soundEnabled")
if sound_enabled is True:
logging.info("🔊 User turned SOUND ON via RPC")
session.output.set_audio_enabled(True)
elif sound_enabled is False:
logging.info("🔇 User turned SOUND OFF via RPC")
session.output.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling sound_control RPC: {e}")
@ctx.room.local_participant.register_rpc_method("mic_control")
async def handle_mic_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
mic_enabled = payload.get("micEnabled")
if mic_enabled is True:
logging.info("🎤 User turned MIC ON via RPC")
session.input.set_audio_enabled(True)
elif mic_enabled is False:
logging.info("🔇 User turned MIC OFF via RPC")
session.input.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling mic_control RPC: {e}")
@ctx.room.on("data_received")
def on_data_received(data: rtc.DataPacket):
try:
payload = getattr(data, "payload", None) or getattr(data, "data", None)
if not payload:
return
raw_text = payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else str(payload)
logging.info(f"💬 User Message: {raw_text}")
except Exception as e:
logging.error(f"Error handling data_received: {e}")
# -------------------------------------------------------------------------
# 🎬 Start the agent session
# -------------------------------------------------------------------------
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(...),
),
)
@session.output.audio.on("playback_finished")
def _on_playback_finished(ev: PlaybackFinishedEvent):
if ev.interrupted:
logging.info("Agent was interrupted during playback")
interruption_payload = {
"type": "interrupted",
"message": "User interrupted the assistant.",
}
asyncio.create_task(
ctx.room.local_participant.publish_data(
json.dumps(interruption_payload).encode("utf-8"), reliable=True
))
await session.say("Hello, how can I help you today?")
# =============================================================================
# SERVER STARTUP
# =============================================================================
async def start_api_server(port: int):
logging.info(f"🚀 Starting FastAPI server on port {port}")
config = uvicorn.Config(app=app, host="0.0.0.0", port=port, log_level="info")
server = uvicorn.Server(config)
await server.serve()
def run_api_server(base_port):
asyncio.run(start_api_server(base_port))
def start_livekit_agent(port: int, agent_name: str):
logging.info(f"🤖 Starting LiveKit agent on port {port}")
options = WorkerOptions(entrypoint_fnc=entrypoint, port=port,agent_name="BMMI HR VoiceAI")
cli.run_app(options)
def main():
logging.info("Starting BMMI HR Voice AI system")
base_port = 5029
api_process = multiprocessing.Process(target=run_api_server, args=(base_port,))
api_process.start()
try:
start_livekit_agent(base_port + 1, "agent_name")
except KeyboardInterrupt:
logging.info("🛑 Shutdown signal received")
finally:
api_process.terminate()
api_process.join()
logging.info("✅ System shutdown complete")
if name == "__main__":
load_dotenv()
main()
t
tall-airline-16756
12/18/2025, 7:53 AMYou're very close — the remaining issue is that you still construct the session with
deepgram.STT(...) Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import stt, voice, rtc
class NoopSTT(stt.STT):
"""No-op STT: never opens a network connection or yields transcription."""
def __init__(self):
super().__init__(capabilities=None)
async def transcribe(self, audio_stream) -> AsyncIterator[stt.TranscriptionChunk]:
if False:
yield # keep as async generator but yield nothing
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[stt.TranscriptionStream]:
class _NoopStream:
def push_frame(self, frame: rtc.AudioFrame):
pass
async def close(self):
pass
yield _NoopStream() Copy code
session = AgentSession(
stt = NoopSTT(),
llm = groq.LLM(model="llama-3.1-8b-instant"),
tts = elevenlabs.TTS(...),
vad = silero.VAD.load(),
) Copy code
@ctx.room.local_participant.register_rpc_method("mic_control")
async def handle_mic_control(data: rtc.RpcInvocationData):
payload = json.loads(data.payload)
mic_enabled = payload.get("micEnabled")
if mic_enabled is True:
<http://logging.info|logging.info>("🎤 User turned MIC ON via RPC")
# instantiate Deepgram only when needed
session.stt = deepgram.STT(model="nova-2-general")
session.input.set_audio_enabled(True)
elif mic_enabled is False:
<http://logging.info|logging.info>("🔇 User turned MIC OFF via RPC")
session.input.set_audio_enabled(False)
# optionally detach Deepgram to stop any background connection
session.stt = NoopSTT()deepgram.STT()s
sparse-napkin-34305
12/18/2025, 7:56 AMsession.stt = deepgram.STT(model="nova-2-general")
AttributeError: can't set attribute 'stt'
t
tall-airline-16756
12/18/2025, 7:57 AMYou're seeing that error because
AgentSession.stt Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import stt, voice, rtc
import logging
class ConditionalSTT(stt.STT):
def __init__(self, delegate_factory):
# delegate_factory: callable that returns a stt.STT instance when needed
super().__init__(capabilities=None)
self._delegate_factory = delegate_factory
self._delegate = None
self._enabled = False
def enable(self):
self._enabled = True
if self._delegate is None:
self._delegate = self._delegate_factory()
def disable(self):
self._enabled = False
# optional: keep delegate for reuse, or close if desired
# if self._delegate:
# await self._delegate.close()
async def transcribe(self, audio_stream) -> AsyncIterator[stt.TranscriptionChunk]:
if not self._enabled or self._delegate is None:
if False:
yield
return
async for chunk in self._delegate.transcribe(audio_stream):
yield chunk
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[stt.TranscriptionStream]:
if not self._enabled:
class _NoopStream:
def push_frame(self, frame: rtc.AudioFrame): pass
async def close(self): pass
yield _NoopStream()
return
# ensure delegate exists
if self._delegate is None:
self._delegate = self._delegate_factory()
async with self._delegate.stream(**kwargs) as s:
yield ssession.input.set_audio_enabled(True)session.stt.enable()session.input.set_audio_enabled(False)session.stt.disable().enable()session.interrupt()s
sparse-napkin-34305
12/18/2025, 7:59 AMso if we are not doing this at the starting the agent will connect to deepgram right automatically? even if we disable input audio?
t
tall-airline-16756
12/18/2025, 7:59 AMShort answer: Yes — if you pass
deepgram.STT(...)AgentSessionsession.input.set_audio_enabled(False) Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import stt, rtc
import logging
class ConditionalSTT(stt.STT):
def __init__(self, factory):
super().__init__(capabilities=None)
self._factory = factory
self._delegate = None
self._enabled = False
def enable(self):
self._enabled = True
if self._delegate is None:
self._delegate = self._factory()
def disable(self):
self._enabled = False
# optional: close delegate if it exposes a close method
async def transcribe(self, audio_stream) -> AsyncIterator[stt.TranscriptionChunk]:
if not self._enabled or self._delegate is None:
if False:
yield
return
async for chunk in self._delegate.transcribe(audio_stream):
yield chunk
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[stt.TranscriptionStream]:
if not self._enabled:
class _NoopStream:
def push_frame(self, frame: rtc.AudioFrame): pass
async def close(self): pass
yield _NoopStream()
return
if self._delegate is None:
self._delegate = self._factory()
async with self._delegate.stream(**kwargs) as s:
yield s.enable().enable()s
sparse-napkin-34305
12/18/2025, 8:38 AMfrom future import annotations
import json
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncIterator
import uvicorn
import logging
import os
import certifi
import time
import multiprocessing
from livekit.plugins import deepgram, groq, elevenlabs
from dotenv import load_dotenv
from custom_deepgram_tts import CustomDeepgramTTS
from livekit.agents.voice import UserStateChangedEvent, AgentStateChangedEvent
from livekit.agents.voice.io import PlaybackFinishedEvent
from livekit import api, rtc
from livekit.agents import (
tts,
Agent,
AgentSession,
JobContext,
llm,
voice,
WorkerOptions,
cli,
stt,
room_io,
RunContext,
RoomInputOptions,
RoomOutputOptions,
)
from livekit.plugins import silero, noise_cancellation
from livekit.agents import metrics, MetricsCollectedEvent, ConversationItemAddedEvent
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# Set SSL_CERT_FILE to use certifi's certificate bundle
os.environ["SSL_CERT_FILE"] = certifi.where()
load_dotenv(".env.local")
# =============================================================================
# FASTAPI APP
# =============================================================================
app = FastAPI(title="BMMI HR Voice AI", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/api/check-online")
def check_online():
print("Success")
return "Success-12"
# =============================================================================
# CUSTOM AGENT
# =============================================================================
class MyAgent(Agent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.conversation_history: list[tuple[float, str, str]] = []
self.sound_enabled = True # Track sound state
def log_transcription(self, timestamp: float, role: str, text: str):
"""Keep local conversation record."""
self.conversation_history.append((timestamp, role, text))
self.conversation_history.sort(key=lambda x: x[0])
logging.info(self.conversation_history)
class LoggingTTS(tts.TTS):
"""
A TTS wrapper that logs the text being synthesized.
"""
def __init__(self, tts_service: tts.TTS):
super().__init__(
capabilities=tts_service.capabilities,
sample_rate=tts_service.sample_rate,
num_channels=tts_service.num_channels,
)
self._tts = tts_service
async def synthesize(self, text: str) -> tts.SynthesizeStream:
logging.info(f"🎤 TTS synthesizing text: '{text}'")
return await self._tts.synthesize(text)
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[tts.SynthesizeStream]:
"""
Log text chunks as they are pushed to the TTS stream.
"""
async with self._tts.stream(**kwargs) as stream:
original_push_text = stream.push_text
def log_and_push(text: str):
logging.info(f"🎤 TTS streaming text chunk: '{text}'")
original_push_text(text)
stream.push_text = log_and_push
yield stream
# =============================================================================
# ENTRYPOINT
# =============================================================================
async def entrypoint(ctx: JobContext):
await ctx.connect()
participant = await ctx.wait_for_participant()
logging.info(f"👤 Participant connected: {participant.identity}")
# -- Setup Agent + LLM + TTS + STT --
agent = MyAgent(
instructions="You are a helpful assistant that answers company-related questions clearly."
)
last_spoken_info = {"text": "", "timestamp": 0.0}
async def on_full_sentence(text: str):
"""Called when a full sentence is ready for TTS."""
nonlocal last_spoken_info
timestamp = time.time()
last_spoken_info = {"text": text, "timestamp": timestamp}
agent.log_transcription(timestamp, "assistant", text)
session = AgentSession(
stt=deepgram.STT(
model="nova-2-general",
),
llm=groq.LLM(model="llama-3.1-8b-instant"),
tts=LoggingTTS(
tts_service=elevenlabs.TTS(
voice_id="pNInz6obpgDQGcFmaJgB", model="eleven_turbo_v2"
)
),
vad=silero.VAD.load(),
)
@session.on("agent_state_changed")
def on_agent_state_changed(ev: AgentStateChangedEvent):
nonlocal last_spoken_info
if ev.new_state == "speaking":
logging.info("🗣️ Agent started speaking")
# 🧠 NEW: Send final sentence payload 1 second after the agent starts speaking.
if last_spoken_info["text"]:
async def send_with_delay():
await asyncio.sleep(1) # Wait for 1 second
try:
payload = {
"type": "final_sentence",
"role": "assistant",
"text": last_spoken_info["text"],
"timestamp": last_spoken_info["timestamp"],
}
await ctx.room.local_participant.publish_data(
json.dumps(payload).encode("utf-8"), reliable=True
)
logging.info("✅ Sent final_sentence payload after 1s delay.")
except Exception as e:
logging.error(f"❌ Failed to send final sentence from on_agent_state_changed: {e}")
asyncio.create_task(send_with_delay())
# -------------------------------------------------------------------------
# 🗣️ Log conversation items
# -------------------------------------------------------------------------
@session.on("conversation_item_added")
def on_conversation_item(event: ConversationItemAddedEvent):
if event.item.role != "user":
return
text = event.item.text_content.lower()
if "goodbye" in text:
async def end_session():
logging.info("User said goodbye — interrupting agent and closing session")
# 1️⃣ Interrupt any ongoing or queued LLM/TTS
await session.interrupt()
# 2️⃣ Say your final message
await session.say("Goodbye, have a wonderful day!")
# 3️⃣ Close session cleanly
await asyncio.sleep(0.3) # let audio flush
await session.close()
asyncio.create_task(end_session())
# -------------------------------------------------------------------------
# 📩 Handle text-based interruptions
# -------------------------------------------------------------------------
@ctx.room.local_participant.register_rpc_method("sound_control")
async def handle_sound_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
sound_enabled = payload.get("soundEnabled")
if sound_enabled is True:
logging.info("🔊 User turned SOUND ON via RPC")
session.output.set_audio_enabled(True)
elif sound_enabled is False:
logging.info("🔇 User turned SOUND OFF via RPC")
session.output.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling sound_control RPC: {e}")
@ctx.room.local_participant.register_rpc_method("mic_control")
async def handle_mic_control(data: rtc.RpcInvocationData):
try:
payload = json.loads(data.payload)
mic_enabled = payload.get("micEnabled")
if mic_enabled is True:
logging.info("🎤 User turned MIC ON via RPC")
session.input.set_audio_enabled(True)
elif mic_enabled is False:
logging.info("🔇 User turned MIC OFF via RPC")
session.input.set_audio_enabled(False)
except Exception as e:
logging.error(f"Error handling mic_control RPC: {e}")
@ctx.room.on("data_received")
def on_data_received(data: rtc.DataPacket):
try:
payload = getattr(data, "payload", None) or getattr(data, "data", None)
if not payload:
return
raw_text = payload.decode("utf-8", errors="ignore") if isinstance(payload, bytes) else str(payload)
logging.info(f"💬 User Message: {raw_text}")
except Exception as e:
logging.error(f"Error handling data_received: {e}")
# -------------------------------------------------------------------------
# 🎬 Start the agent session
# -------------------------------------------------------------------------
await session.start(
agent=agent,
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(),
),
)
session.say("hello how are you")
@session.output.audio.on("playback_finished")
def _on_playback_finished(ev: PlaybackFinishedEvent):
if ev.interrupted:
logging.info("Agent was interrupted during playback")
interruption_payload = {
"type": "interrupted",
"message": "User interrupted the assistant.",
}
asyncio.create_task(
ctx.room.local_participant.publish_data(
json.dumps(interruption_payload).encode("utf-8"), reliable=True
))
# =============================================================================
# SERVER STARTUP
# =============================================================================
async def start_api_server(port: int):
logging.info(f"🚀 Starting FastAPI server on port {port}")
config = uvicorn.Config(app=app, host="0.0.0.0", port=port, log_level="info")
server = uvicorn.Server(config)
await server.serve()
def run_api_server(base_port):
asyncio.run(start_api_server(base_port))
def start_livekit_agent(port: int, agent_name: str):
logging.info(f"🤖 Starting LiveKit agent on port {port}")
options = WorkerOptions(entrypoint_fnc=entrypoint, port=port,agent_name="BMMI HR VoiceAI")
cli.run_app(options)
def main():
logging.info("Starting BMMI HR Voice AI system")
base_port = 5029
api_process = multiprocessing.Process(target=run_api_server, args=(base_port,))
api_process.start()
try:
start_livekit_agent(base_port + 1, "agent_name")
except KeyboardInterrupt:
logging.info("🛑 Shutdown signal received")
finally:
api_process.terminate()
api_process.join()
logging.info("✅ System shutdown complete")
if name == "__main__":
load_dotenv()
main().....is this code correct like when i am turning on the mic then livekit is saying input attaching and if turned off saying dethatched
t
tall-airline-16756
12/18/2025, 8:38 AMYou're close — your toggles are working as expected from the AgentSession side, but a couple small points and one recommended change will make behavior match your intention exactly.
What your current code does
• When you call
session.input.set_audio_enabled(True)session.input.set_audio_enabled(False)stt=deepgram.STT(...) Copy code
from contextlib import asynccontextmanager
from typing import AsyncIterator
from livekit.agents import stt, rtc
class ConditionalSTT(stt.STT):
def __init__(self, factory):
super().__init__(capabilities=None)
self._factory = factory
self._delegate = None
self._enabled = False
def enable(self):
self._enabled = True
if self._delegate is None:
self._delegate = self._factory()
def disable(self):
self._enabled = False
# optional: close delegate if it exposes close()
async def transcribe(self, audio_stream) -> AsyncIterator[stt.TranscriptionChunk]:
if not self._enabled or self._delegate is None:
if False:
yield
return
async for chunk in self._delegate.transcribe(audio_stream):
yield chunk
@asynccontextmanager
async def stream(self, **kwargs) -> AsyncIterator[stt.TranscriptionStream]:
if not self._enabled:
class _NoopStream:
def push_frame(self, frame: rtc.AudioFrame): pass
async def close(self): pass
yield _NoopStream()
return
if self._delegate is None:
self._delegate = self._factory()
async with self._delegate.stream(**kwargs) as s:
yield ssession.stt.enable()session.say("hello how are you")await session.say(...)await session.interrupt()s
sparse-napkin-34305
12/18/2025, 8:40 AMso in my code even when the mic is off is deepgram STT working?
t
tall-airline-16756
12/18/2025, 8:41 AMShort answer: Yes — in your current code Deepgram is instantiated at session creation, so it will try to connect and run even when you immediately call
session.input.set_audio_enabled(False)session.stt.enable(); session.input.set_audio_enabled(True)session.input.set_audio_enabled(False); session.stt.disable().enable()s
sparse-napkin-34305
12/18/2025, 9:18 AMis there a way to know if deepgram is being used
t
tall-airline-16756
12/18/2025, 9:18 AMYes — here are practical ways to know whether Deepgram is being used in your agent.
1) Watch logs (fastest)
• Look for Deepgram connection logs or the 401 you saw. When Deepgram is instantiated it will log attempts to open the websocket; absence of those logs means it hasn’t connected.
• Add an explicit log when you create/enable Deepgram:
◦ e.g. logging.info("Attaching Deepgram STT") right before instantiating.
2) Instrument the STT wrapper
• Use the ConditionalSTT/NoopSTT patterns we discussed and add logging inside:
◦ Log when
.enable()stream()transcribe()s
sparse-napkin-34305
12/18/2025, 9:24 AMwait so even if we do disable the input audio deepgram is still streaming?
t
tall-airline-16756
12/18/2025, 9:24 AMShort answer: Not necessarily — disabling input audio stops the agent from sending audio frames to the STT pipeline, but it does NOT prevent the Deepgram plugin from having already opened a websocket connection if you instantiated
deepgram.STT(...)stt=deepgram.STT(...)AgentSession.stttype(session.stt)getattr(session.stt, "_delegate", None)2 Views