If yes can you give me an exemple in python
is it...
# ask-aig
glamorous-telephone-35069
03/04/2026, 4:20 PMIf yes can you give me an exemple in python
is it possible to replace sst with an audio-LLMs and directly use it to receive audio and execute the prompt?
t
tall-airline-16756
03/04/2026, 4:21 PMThanks for reaching out, Michael. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
03/04/2026, 4:21 PMThanks — I’ll help with a Python example and a short explanation. You asked whether you can replace STT with an audio-LLM that accepts raw audio and executes prompts directly — yes, you can. LiveKit supports realtime multimodal models via the
RealtimeModel Copy code
import asyncio
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins.openai import realtime
# Configure the realtime model to accept audio and produce audio/text
realtime_model = realtime.RealtimeModel(
voice="alloy", # model TTS voice (audio out)
temperature=0.2,
instructions="You are a helpful assistant. Answer concisely.",
# modalities default to audio+text for audio-LLMs that support it
)
async def entrypoint(ctx: JobContext):
# connect to the LiveKit room and subscribe only to audio
await ctx.connect(auto_subscribe=ctx.AutoSubscribe.AUDIO_ONLY)
# create an AgentSession which will route incoming audio to the realtime model
session = AgentSession(llm=realtime_model)
# start handling the room (this wires audio in → model and model audio/text → room)
session.start(ctx.room)
# Optional: send an initial spoken prompt from the model (TTS)
await session.generate_reply("Hello — I'm listening. Ask me anything.")
# Keep the worker running while the room is active
await ctx.wait_for_room_end()
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint)) Copy code
realtime_model = realtime.RealtimeModel(
modalities=["text"], # force text responses only
instructions="Answer in short text form."
)
# Then run TTS (your chosen engine) on the returned text and play into the room.g
glamorous-telephone-35069
03/04/2026, 4:22 PMJe prends dans mon cas qwen3 Omni pour audio text et qwentts
t
tall-airline-16756
03/04/2026, 4:23 PMSuper — merci pour la précision. Vous voulez utiliser Qwen3 Omni pour la reconnaissance vocale (audio → texte) et QwenTTS pour la synthèse vocale (texte → audio). Voici comment vous pouvez procéder avec LiveKit et des exemples/conseils en Python.
Résumé rapide : Qwen3 Omni et QwenTTS ne figurent pas parmi les plugins TTS/STT fournis en standard. Vous pouvez soit
• implémenter des nœuds personnalisés dans le pipeline Agent (intégrer directement les SDK/API Qwen), soit
• traiter Qwen3 Omni/QwenTTS en externe et injecter/récupérer l’audio depuis LiveKit (flux entrant/sortant).
Avant d’entrer dans un exemple, quelques questions pour adapter la solution :
• Avez-vous des SDK Python ou des endpoints HTTP prêts pour Qwen3 Omni et QwenTTS ?
• Souhaitez-vous que la reconnaissance soit en temps réel (streaming) ou en segments (upload de petits fichiers) ?
• Voulez-vous que la réponse vocale sorte directement dans la salle LiveKit (playback en temps réel) ou soit fournie comme fichier à jouer après génération ?
Si vous n’avez pas de contrainte différente, voici deux approches avec exemples succincts en Python.
Option A — Intégration via nœuds personnalisés (pipeline Agent)
• Idée : étendre/implémenter
stt_nodetts_node Copy code
from livekit.agents import Agent, ModelSettings
from typing import AsyncIterable, Optional
import asyncio
class QwenAgent(Agent):
async def stt_node(self, audio_frames: AsyncIterable, model_settings: ModelSettings) -> Optional[AsyncIterable]:
# 1) Convertir audio_frames en chunks/support attendu par Qwen3 Omni
# 2) Stream/POST les chunks à Qwen3 Omni SDK/endpoint
# 3) Yield/retourner des événements de parole/text à la pipeline LiveKit
pass
async def tts_node(self, text_stream: AsyncIterable[str], model_settings: ModelSettings) -> AsyncIterable:
# 1) Appeler QwenTTS pour chaque texte reçu
# 2) Récupérer l'audio généré (PCM/opus etc.)
# 3) Convertir en AudioFrames et yield vers LiveKit pour playback
passstt_nodetts_nodeg
glamorous-telephone-35069
03/04/2026, 4:31 PMJ utilise vllm donc openai compatible et en streaming
Je veux le
Moins de la latence
t
tall-airline-16756
03/04/2026, 4:32 PMParfait — merci pour la précision. Vous utilisez vLLM (OpenAI‑compatible) en streaming et vous voulez la latence la plus faible possible en combinant Qwen3 Omni (audio→texte) et QwenTTS (texte→audio). Voici une solution pratique, axée streaming et faible latence, avec un exemple Python adapté à LiveKit.
Résumé rapide
• Utilisez vLLM (Qwen3 Omni) exposé via une API OpenAI-compatible pour l’inférence LLM en streaming.
• Pour la latence la plus faible, faites la reconnaissance vocale en streaming avec un STT ultra‑faible latence (Deepgram/Flux ou un endpoint Whisper‑style si vous l’avez), envoyez les textes en streaming au vLLM, puis synthétisez la sortie immédiatement via QwenTTS si celui-ci propose un endpoint streaming compatible.
• Si QwenTTS n’expose pas d’API de streaming compatible, enveloppez-le dans un microservice qui accepte du text-stream et renvoie de l’audio chunké (PCM/opus) pour publish immédiat dans la room LiveKit.
Avant de coder — deux vérifications rapides
1. Qwen3 Omni (vLLM) est bien accessible via un endpoint OpenAI‑compatible (base_url) en streaming ?
2. QwenTTS propose un endpoint de synthèse en streaming (websocket / SSE / chunked HTTP) ou seulement génération batch (fichiers) ?
Répondez oui/non à ces deux points et j’adapte l’exemple.
Exemple de flux minimal (architecture)
• Participant → LiveKit audio track (webrtc)
• LiveKit Agent (subscriber) → stream audio frames vers STT streaming
• STT streaming → génère tokens/text partials → forward vers vLLM (streaming)
• vLLM génère réponse token par token (stream) → immédiatement envoyé au TTS streaming
• TTS streaming → audio chunks → publish comme track dans la room (playback immédiat)
Exemple Python (schéma concret, streaming, faible latence)
• Hypothèses : vous avez
◦ vLLM OpenAI-compatible à http://localhost:8000/v1
◦ Un STT streaming compatible (ici on montre Deepgram en exemple)
◦ QwenTTS exposé via un endpoint streaming ou via un wrapper que vous fournirez
• L’exemple montre la logique : recevoir partials STT → envoyer au LLM stream → pipe tokens vers TTS stream → publish audio chunks.
Copy code
import asyncio
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import openai, deepgram # exemples de plugins existants
import aiohttp
# Configuration endpoints
VLLM_BASE = "<http://localhost:8000/v1>" # vLLM OpenAI-compatible
QWENTTS_BASE = "<http://localhost:9000/tts_stream>" # endpoint streaming TTS (wrapper si besoin)
# LLM config pointant vers vLLM (OpenAI-compatible)
qwen_llm = openai.LLM(
model="qwen3-omni",
base_url=VLLM_BASE,
api_key=None, # si vous avez token, mettez-le
stream=True # indique qu'on utilisera streaming
)
# STT low-latency (ex: Deepgram flux). Utilisez votre plugin/config STT réel.
stt = deepgram.STT(model="flux")
async def tts_stream_to_livekit(session, text_stream_async_iterable):
"""
Exemple de wrapper simple pour envoyer texte en streaming à QwenTTS
et recevoir des chunks audio (PCM/opus) renvoyés vers LiveKit.
Implémentez votre conversion vers AudioFrames selon vos besoins.
"""
async with aiohttp.ClientSession() as http:
# Exemple POST streaming (adapté à votre API)
async with <http://http.post|http.post>(QWENTTS_BASE, data=text_stream_generator(text_stream_async_iterable)) as resp:
async for chunk in resp.content.iter_chunked(4096):
# Convertir chunk -> AudioFrame et publier dans la room
# session.publish_audio_chunk(chunk) # pseudo-fonction à implémenter
pass
def text_stream_generator(text_iterable):
"""Convertit un AsyncIterable[str] en bytes pour l'endpoint TTS streaming."""
async def _gen():
async for t in text_iterable:
yield t.encode("utf-8")
return _gen()
async def entrypoint(ctx: JobContext):
await ctx.connect(auto_subscribe=ctx.AutoSubscribe.AUDIO_ONLY)
# Crée la session agent avec STT + LLM. On laisse TTS en externe pour contrôle.
agent_session = AgentSession(stt=stt, llm=qwen_llm, preemptive_generation=True)
agent_session.start(ctx.room)
# Exemple: si vLLM renvoie stream text partials, vous devez les récupérer
# et forwarder vers tts_stream_to_livekit pour synthèse immédiate.
# L'API AgentSession/LLM expose callbacks/events pour partials -> abonnez-vous.
# Pseudo-code d'abonnement :
async def on_llm_partial(token: str):
# buffer tokens / or forward token-to-tts immediately
await send_token_to_tts(token) # implémentez streaming vers QwenTTS
# Assurez-vous d'attacher le callback sur l'objet LLM/session (pseudo)
qwen_llm.on_partial = on_llm_partial
await ctx.wait_for_room_end()
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))3 Views