{"message": "inference is slower than realtime", "level": "WARNING", "name": "livekit.plugins.silero", "room": "playground-4vKi-XoPr", "delay": 3.342645474999852, "pid": 11488, "job_id": "AJ_25uC5SKeyjyz", "timestamp": "2025-12-21T050829.108774+00:00"} how to solve this

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

i am using 16v cpu Preload (prewarm) the VAD model

Thanks — helpful details. You’re on 16 vCPU and already prewarming the VAD, so let’s narrow why inference is still slow. Please tell me: • Where the agent runs (cloud provider + exact instance type or on-prem VM/container image). • If it’s containerized: the CPU limits/requests and cgroup quota. • Current CPU usage when the warning occurs (total and per-core, if available). • Whether other heavy plugins/processes (STT, TTS, LLMs) run on the same host. • How many concurrent sessions/rooms the host serves. • The Silero VAD model variant you’re using (small/standard/large) and your VAD config (e.g.,

min_silence_duration

,

activation_threshold

). With those I can recommend specific steps (instance sizing, CPU affinity, process separation, model choice, or config tweaks) and an exact prewarm snippet if needed.

how can i cange vad

You're asking how to change the Silero VAD — I’ll show how to switch model variants, adjust sensitivity parameters, and update prewarm/usage so changes take effect. 1) Change model variant (choose smaller/faster vs larger/accurate) • When loading the model, specify the variant name used by your Silero package (example names:

silero_vad_small

,

silero_vad

,

silero_vad_large

— replace with the exact variant your code/library accepts). • Load the chosen variant in your prewarm function so it's reused across sessions. 2) Adjust VAD sensitivity/configuration • Key parameters to tune: ◦

activation_threshold

(increase to require louder speech to activate; decrease to be more sensitive) ◦

min_silence_duration

(shorter = faster re-triggering; longer = fewer false positives) ◦

frame_duration

/

window_size

(smaller frames reduce latency but increase CPU) • Apply these in the VAD instance or wrapper you create after loading the model. 3) Prewarm and reuse (example pattern) • Preload the model once per process and store it in

proc.userdata

, then pass it into sessions:

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

server.setup_fnc = prewarm

# Later, reuse in session creation:
session = AgentSession(
    vad=ctx.proc.userdata["vad"],
    # ...
)

4) Example: instantiate with config (pseudocode — adapt to your Silero API)

vad = silero.VAD.load(variant="silero_vad_small")
vad.activation_threshold = 0.6
vad.min_silence_duration = 0.2
vad.frame_duration = 0.02

5) If using a wrapper that processes audio frames, ensure you pass the config when creating that wrapper so every inference uses the tuned settings. 6) Testing suggestions • Start with a faster/smaller model variant plus slightly higher

activation_threshold

and shorter

frame_duration

to reduce CPU and latency. • Measure

delay

after each change and watch CPU usage. If you want, tell me: • which Silero package/API you call (exact import/function names), and • current VAD parameters or variant; I’ll provide an exact code snippet matching your API.

i am in previous version

Thanks — since you’re on a previous version, I need one detail to give correct instructions. Please paste: • the exact LiveKit Agents/SDK version (from your deps or `pip freeze`/`poetry.lock`) or a short snippet showing how you currently load/configure Silero VAD. With that I’ll provide the exact code/steps to change model variant and VAD parameters (and confirm prewarm behavior) for your version. Sources: Silero VAD plugin | LiveKit docs | VAD Metrics | LiveKit docs | Playing Audio | LiveKit docs | LiveKit turn detector plugin | LiveKit docs

"livekit-plugins-silero==1.2.6",

Thanks — with

livekit-plugins-silero==1.2.6

I can give concrete steps. Below are concise, version-compatible changes you can make to swap model variants, tune sensitivity, and ensure prewarm is used. 1) Update prewarm to load chosen variant

# server setup (run once per process)
def prewarm(proc: JobProcess):
    # choose a smaller/faster variant if available, e.g. "silero_vad_small"
    proc.userdata["vad"] = silero.VAD.load(variant="silero_vad_small")
server.setup_fnc = prewarm

2) Reuse the prewarmed instance in sessions

# when creating AgentSession / session wrapper
session = AgentSession(
    vad=ctx.proc.userdata["vad"],
    # other args...
)

3) Tune VAD parameters (apply after load)

vad = ctx.proc.userdata["vad"]
# example adjustments — tweak values to trade sensitivity vs CPU
vad.activation_threshold = 0.6        # increase to require louder speech
vad.min_silence_duration = 0.2        # shorter to re-trigger faster
vad.frame_duration = 0.02             # smaller frames = lower latency but more CPU

4) If the plugin API requires config at load time, pass params there

proc.userdata["vad"] = silero.VAD.load(
    variant="silero_vad_small",
    activation_threshold=0.6,
    min_silence_duration=0.2,
    frame_duration=0.02,
)

5) Quick tuning steps to follow • Start with

variant="silero_vad_small"

,

activation_threshold=0.6

,

frame_duration=0.02

. • Measure

delay

and CPU usage. • If CPU still high, increase

activation_threshold

or switch to even smaller variant. • If too insensitive, lower

activation_threshold

slightly. 6) If you still see "inference is slower than realtime" • Confirm the VAD instance is reused (not reloaded per session). • Check for other CPU-heavy plugins (STT/TTS/LLM) colocated; move them to separate hosts or increase cores. • Consider using a CPU‑optimized instance type if contention remains. If you want, paste the small snippet you currently use to load/create the VAD (from your process setup or session code) and I’ll produce the exact changed code for 1.2.6.

i want to know what quee service livekit using

Good question — do you mean the queue used for egress/job processing or the one used for webhooks? • For egress and inter-service job processing, LiveKit uses Redis Pub/Sub. • For webhooks, LiveKit uses a custom per-resource in-memory queue (

resourceQueue

) with configurable depth, age-based dropping, and retry/backoff logic. Tell me which component you’re focused on and I’ll provide more details (configuration keys, sizing guidance, or how to inspect/monitor the queue). Sources: Universal Egress | examples/survey | livekit/livekit_composite | DeepWiki

like if some one call stt tts llm worker if thier any quee services added in server

Short answer: LiveKit does not require an external queue service for STT/TTS/LLM worker jobs — it uses its internal job/dispatch system and per-agent queues. Details you can act on: • Job dispatch: The LiveKit server maintains a built‑in job queue that dispatches conversation/session jobs to registered agent workers. Workers with the same

agent_name

are load‑balanced automatically. • Agent internals: Each agent implements internal queues (e.g., `speechQueue`/priority queue for speech output and buffers for streaming STT) to manage ordering, interruptions, and media buffering. • No external dependency by default: You don’t need Redis/Kafka/RabbitMQ for STT/TTS/LLM in the Agents framework — horizontal scaling is achieved by adding more agent workers which pull jobs from the server queue. • Where Redis is used: Redis Pub/Sub is used for egress and some inter-service egress coordination, not as the default queue for STT/TTS/LLM job dispatch. • Scaling/ops notes: ◦ If you run heavy STT/TTS/LLM workloads and want persistent, durable background tasking, you can add your own external queue system and integrate it outside the default agent flow. ◦ To reduce contention, isolate CPU‑heavy plugins (TTS/LLM) on separate hosts or add more agent workers so the built‑in dispatcher can spread load. If you want, tell me whether you’re seeing job backlogs or CPU contention; I’ll recommend concrete changes (worker counts, process separation, or integrating an external queue) for your setup. Sources: Agents Framework | livekit/livekit_composite | DeepWiki | examples/voice_agents | livekit/livekit_composite | DeepWiki | Speech-to-text (STT) models | LiveKit Docs | STT Metrics | LiveKit docs

yes i want to add queeue so my stt tts llm worker can handle realtime huge traffice

Great — adding an external queue is the right move for high-throughput STT/TTS/LLM. I can help design that. Quick question so I give an exact plan and code: which stack and environment are you using? Please tell me: • Language/runtime for workers and agent (Python / Node.js / other) • Cloud provider or on-prem (AWS/GCP/Azure/K8s) • Preferred queue tech (Redis Streams/Celery, RabbitMQ, SQS, Kafka) or open to suggestions • Expected traffic (concurrent sessions or requests/sec) and latency SLO (e.g., <500ms) • Whether you need durable persistent queueing or best-effort ephemeral I’ll return a concise architecture diagram, recommended queue choice, worker autoscaling rules, and a code template for publishing tasks from the agent and consuming results back into LiveKit. Sources: Agents Framework | livekit/livekit_composite | DeepWiki | examples/voice_agents | livekit/livekit_composite | DeepWiki | Custom agent deployments | LiveKit docs

Language/runtime for workers and agent Python. cloud -> aws , queue-> SQS, Kafka, 100concurrent sessions or requests, durable persistent queueing

Great — since you’re on Python + AWS and want durable queueing for ~100 concurrent sessions, here’s a concise, actionable plan with two options (SQS or Kafka) and code templates. Architecture (high level) • LiveKit Agent (real-time, low-latency) receives audio, does VAD & minimal buffering, then publishes tasks to the queue. • Worker pools (STT / LLM / TTS) consume from the queue, perform inference, and post results back to the agent via a result queue or direct API call/websocket. • Use separate topics/queues per stage (stt-input → stt-output → llm-input → llm-output → tts-input → tts-output) for clarity and scaling. Option A — AWS SQS (durable, easy managed scaling) • Use SQS FIFO if ordering per-session is required; otherwise use standard queues for throughput. • Agent publishes STT jobs to SQS; STT workers long-poll, process, then send result to an output queue or directly call the agent’s callback endpoint. • Make processing idempotent and delete SQS messages only after success. Use visibility timeouts and dead-letter queues (DLQ) for failures. Agent publish (SQS example)

import boto3, json
sqs = boto3.client('sqs', region_name='us-east-1')

def publish_stt_job(queue_url, job):
    sqs.send_message(QueueUrl=queue_url, MessageBody=json.dumps(job))

Worker consume (SQS, long polling)

def poll_and_process(queue_url):
    resp = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=10, WaitTimeSeconds=20)
    for m in resp.get('Messages', []):
        job = json.loads(m['Body'])
        try:
            result = process_stt(job)    # do STT
            publish_result(result_queue_url, result)
            sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=m['ReceiptHandle'])
        except Exception:
            # leave for retry or move to DLQ
            pass

Autoscaling (SQS) • Use AWS Application Auto Scaling (ECS/EKS HPA) based on SQS queue depth (messages visible). • Scale by worker count until average processing latency meets SLO. Option B — Kafka (high throughput, ordering, low-latency) • Use Amazon MSK for managed Kafka. Use topics for each pipeline stage. • Producers: agents produce to topic; Consumers: worker groups consume with consumer groups for parallelism. • Commit offsets after successful processing; use partitioning by session ID for ordering. Agent produce (confluent-kafka)

from confluent_kafka import Producer
p = Producer({'bootstrap.servers':'broker:9092'})

def publish_stt(topic, job, key):
    p.produce(topic, key=key, value=json.dumps(job).encode('utf-8'))
    p.flush()

Worker consume (confluent-kafka)

from confluent_kafka import Consumer
c = Consumer({...,'group.id':'stt-workers','auto.offset.reset':'earliest'})
c.subscribe(['stt-input'])
while True:
    msg = c.poll(1.0)
    if msg is None: continue
    job = json.loads(msg.value().decode('utf-8'))
    result = process_stt(job)
    produce('stt-output', result, key=job['session_id'])
    c.commit(msg)

Result delivery patterns • Result queue: workers publish outputs to an output queue/topic the agent subscribes to and then the agent injects audio/text back into the LiveKit session. • Direct callback: worker calls a secured agent endpoint (webhook) with result (lower latency but requires reachable agent). • Hybrid: use output queue for durability and fast direct callback when possible. Operational recommendations • Separate CPU‑heavy workloads: run STT/LLM/TTS in separate ASG/ECS services or pods with tuned CPU/RAM. • Prewarm models (you already do this); use instance types with sustained CPU (c6i/c7i) for workers doing inference. • Use per-session partition key (session_id) for ordering when needed. • Monitor metrics: queue depth, worker processing time, Time To First Token, and LiveKit delay metric. Alert on queue depth and processing latency. • Implement retries, exponential backoff, and DLQs. Ensure idempotency (store job IDs). • For ~100 concurrent sessions, start with 4–8 workers per service (stt/llm/tts) and autoscale based on queue depth and observed latency. Next step Which queue do you prefer for the first implementation — SQS (managed, simpler) or Kafka/MSK (higher throughput, ordering)? I’ll provide a deploy-ready example (ECS task definitions or EKS manifests), autoscaling rules, and a concrete code flow connecting LiveKit Agent → queue → workers → agent for your choice. Sources: Survey Calling Agent | LiveKit docs | Repeater | LiveKit docs | Silero VAD plugin | LiveKit docs | examples/survey | livekit/livekit_composite | DeepWiki | Deploy and scale agents on LiveKit Cloud

now i got code sqs but how to apply to agent pipeline

Good — below is a concise, practical recipe to wire your SQS queue into the LiveKit Python agent pipeline so STT/TTS/LLM work is offloaded and results are injected back into sessions with low blocking. Overview (flow) • Agent: keeps real-time duties (VAD, buffering) and publishes jobs to SQS. • Worker(s): consume SQS, run STT/LLM/TTS, publish results to a results SQS queue. • Agent: background task polls results queue and injects events into the session pipeline. 1) Environment • Set AWS creds and queue URLs in env: ◦ AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION ◦ SQS_STT_IN_URL, SQS_STT_OUT_URL (and similar for llm/tts if separate) 2) Publish from pipeline node (stt_node example) • In your Agent subclass override the stt node to push audio frames as messages and return a generator that yields interim/final results when the result poller injects them. Example pattern (conceptual):

import asyncio, base64, json, os, boto3
from livekit.agents import Agent, stt

sqs = boto3.client('sqs', region_name=os.getenv('AWS_REGION'))

class SQSAwareAgent(Agent):
    async def stt_node(self, audio: AsyncIterable, model_settings):
        job_id = generate_uuid()
        in_queue = os.getenv('SQS_STT_IN_URL')
        out_queue = os.getenv('SQS_STT_OUT_URL')

        # send frames asynchronously so pipeline isn't blocked
        async def send_frames():
            async for frame in audio:
                payload = {
                    "job_id": job_id,
                    "session_id": self.session.session_id,
                    "sample_rate": frame.sample_rate,
                    "data_b64": base64.b64encode(frame.data).decode('utf-8')
                }
                sqs.send_message(QueueUrl=in_queue, MessageBody=json.dumps(payload))
            # EOF marker
            sqs.send_message(QueueUrl=in_queue, MessageBody=json.dumps({"job_id": job_id, "eof": True}))

        asyncio.create_task(send_frames())

        # Return an async generator that yields SpeechEvent objects as results are injected
        queue = asyncio.Queue()

        # register a local handler keyed by job_id so the poller can put results here
        self._register_result_queue(job_id, queue)

        async def result_generator():
            while True:
                evt = await queue.get()
                if evt == "DONE":
                    break
                yield evt  # should be stt.SpeechEvent or compatible object
            self._unregister_result_queue(job_id)

        return result_generator()

3) Poll results and inject into session • Run a background task on session start that long-polls SQS_STT_OUT_URL and routes messages to the correct

job_id

queue (registered above). Delete messages after successful routing. Concept:

async def poll_stt_results(self):
    out_queue = os.getenv('SQS_STT_OUT_URL')
    while True:
        resp = sqs.receive_message(QueueUrl=out_queue, MaxNumberOfMessages=10, WaitTimeSeconds=20)
        for m in resp.get('Messages', []):
            body = json.loads(m['Body'])
            job_id = body.get('job_id')
            transcript = body.get('transcript')
            if job_id and transcript:
                q = self._get_result_queue(job_id)
                if q:
                    evt = stt.SpeechEvent(type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                                          alternatives=[stt.SpeechAlternative(text=transcript)])
                    await q.put(evt)
                # optionally signal DONE
                if body.get('done'):
                    q = self._get_result_queue(job_id)
                    if q: await q.put("DONE")
            sqs.delete_message(QueueUrl=out_queue, ReceiptHandle=m['ReceiptHandle'])
        await asyncio.sleep(0.1)

• Start poller from

on_enter

or session entry:

async def on_enter(self):
    asyncio.create_task(self.poll_stt_results())

4) Worker behavior (high level) • Consume SQS_STT_IN_URL, assemble audio frames until EOF, run STT, then send result to SQS_STT_OUT_URL with { job_id, transcript, done: true }. • Make processing idempotent and name messages by job_id. 5) Correlation & ordering • Use job_id and session_id for correlation. • If you need strict per-session ordering, use FIFO queues (SQS FIFO) with message group ID = session_id. 6) Reliability & operational best practices • Long-poll SQS (WaitTimeSeconds=20) to reduce empty responses. • Use visibility timeout larger than expected processing time; worker should extend visibility for long jobs. • Use DLQs for poison messages. • Make message processing idempotent and include a request_id/job_id. • Monitor queue depth, worker latency; autoscale workers on visible message count. • For lower latency on outputs, consider direct callback from worker to the agent (if reachable) plus the durable output queue as fallback. 7) Performance tips for real-time feel • Keep agent-side processing minimal: VAD + chunking + publish frame quickly. • Use partial results: workers can publish interim transcripts to out-queue frequently so agent can stream partials. • Prewarm models on worker nodes and provision CPU-optimized instances for inference. 8) Security • Use IAM least-privilege for SQS access. • Sign/encrypt payloads if needed. Include authentication tokens in messages if workers need to call back agent endpoints. If you want, I can: • Produce a ready-to-run example repo file (full Agent class + poller + worker example) tailored to your current code. • Provide ECS/EKS task definitions and autoscaling rules for workers based on SQS queue depth. Which next step do you want — full code example or deployment autoscaling config? Sources: Pipeline nodes and hooks | LiveKit docs

but how to add in llm on entry or where?

Great — below is a complete, practical example you can adapt. It includes: • A LiveKit Python Agent with SQS-integrated

stt_node

,

llm_node

, and

tts_node

. • A simple worker example that consumes SQS STT/LLM/TTS input queues and posts results to output queues. • AWS deployment/autoscaling guidance with example ECS task/service and autoscaling config using SQS-visible-message metrics. Keep values like queue URLs, AWS region, and LiveKit session identifiers in env vars. 1) Agent code (agent.py)

# agent.py
import os
import asyncio
import json
import base64
import uuid
from typing import AsyncIterable
import boto3

from livekit.agents import Agent, stt as stt_module, llm as llm_module, tts as tts_module

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
SQS_STT_IN = os.getenv("SQS_STT_IN_URL")
SQS_STT_OUT = os.getenv("SQS_STT_OUT_URL")
SQS_LLM_IN = os.getenv("SQS_LLM_IN_URL")
SQS_LLM_OUT = os.getenv("SQS_LLM_OUT_URL")
SQS_TTS_IN = os.getenv("SQS_TTS_IN_URL")
SQS_TTS_OUT = os.getenv("SQS_TTS_OUT_URL")

sqs = boto3.client("sqs", region_name=AWS_REGION)


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

        # maps job_id -> asyncio.Queue for results
        self._result_queues = {}
        self._pollers_started = False

    def _register_result_queue(self, job_id: str):
        q = asyncio.Queue()
        self._result_queues[job_id] = q
        return q

    def _unregister_result_queue(self, job_id: str):
        self._result_queues.pop(job_id, None)

    def _get_result_queue(self, job_id: str):
        return self._result_queues.get(job_id)

    async def on_enter(self):
        # Start poller tasks once per agent/session
        if not self._pollers_started:
            self._pollers_started = True
            asyncio.create_task(self._poll_out_queue(SQS_STT_OUT, self._handle_stt_out))
            asyncio.create_task(self._poll_out_queue(SQS_LLM_OUT, self._handle_llm_out))
            asyncio.create_task(self._poll_out_queue(SQS_TTS_OUT, self._handle_tts_out))

    async def _poll_out_queue(self, queue_url, handler_fn):
        # Long-poll SQS for results and dispatch to registered queues
        while True:
            try:
                resp = sqs.receive_message(
                    QueueUrl=queue_url,
                    MaxNumberOfMessages=10,
                    WaitTimeSeconds=20,
                )
                for m in resp.get("Messages", []):
                    body = json.loads(m["Body"])
                    await handler_fn(body)
                    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=m["ReceiptHandle"])
            except Exception as e:
                # log and continue
                print("SQS poll error", e)
                await asyncio.sleep(1)

    async def _handle_stt_out(self, body):
        job_id = body.get("job_id")
        q = self._get_result_queue(job_id)
        if not q:
            return
        # create a SpeechEvent-like object: adapt to your stt module types
        evt = stt_module.SpeechEvent(
            type=stt_module.SpeechEventType.FINAL_TRANSCRIPT,
            alternatives=[stt_module.SpeechAlternative(text=body.get("transcript", ""))],
        )
        await q.put(evt)
        if body.get("done"):
            await q.put("DONE")

    async def _handle_llm_out(self, body):
        job_id = body.get("job_id")
        q = self._get_result_queue(job_id)
        if not q:
            return
        # llm chunk can be plain string chunks or ChatChunk objects
        await q.put(body.get("response", ""))
        if body.get("done"):
            await q.put("DONE")

    async def _handle_tts_out(self, body):
        # For TTS, body may contain base64 audio; handle as needed
        job_id = body.get("job_id")
        q = self._get_result_queue(job_id)
        if not q:
            return
        await q.put(body)  # worker may send {job_id, audio_b64, done: true}

    # STT node: stream frames to SQS, return an async generator that yields SpeechEvent(s)
    async def stt_node(self, audio: AsyncIterable, model_settings):
        job_id = str(uuid.uuid4())
        in_url = SQS_STT_IN
        q = self._register_result_queue(job_id)

        async def send_frames():
            async for frame in audio:
                payload = {
                    "job_id": job_id,
                    "session_id": getattr(self, "session_id", None),
                    "sample_rate": getattr(frame, "sample_rate", 16000),
                    "data_b64": base64.b64encode(frame.data).decode("utf-8"),
                }
                sqs.send_message(QueueUrl=in_url, MessageBody=json.dumps(payload))
            # EOF
            sqs.send_message(QueueUrl=in_url, MessageBody=json.dumps({"job_id": job_id, "eof": True}))

        asyncio.create_task(send_frames())

        async def result_gen():
            while True:
                item = await q.get()
                if item == "DONE":
                    break
                yield item
            self._unregister_result_queue(job_id)

        return result_gen()

    # LLM node: publish chat context to SQS and yield response stream from out queue
    async def llm_node(self, chat_ctx: llm_module.ChatContext, tools, model_settings=None):
        job_id = str(uuid.uuid4())
        in_url = SQS_LLM_IN
        q = self._register_result_queue(job_id)

        # publish request
        payload = {
            "job_id": job_id,
            "session_id": getattr(self, "session_id", None),
            "chat_ctx": chat_ctx.model_dump() if hasattr(chat_ctx, "model_dump") else str(chat_ctx),
        }
        sqs.send_message(QueueUrl=in_url, MessageBody=json.dumps(payload))

        async def response_gen():
            # wait for chunks placed into q by poller
            while True:
                item = await q.get()
                if item == "DONE":
                    break
                # yield chunk (string) to LLM consumer in the pipeline
                yield item
            self._unregister_result_queue(job_id)

        return response_gen()

    # TTS node: send text to TTS queue and wait for audio back
    async def tts_node(self, text_stream: AsyncIterable, voice_settings, model_settings=None):
        # collect text into one request (or stream multiple chunks)
        job_id = str(uuid.uuid4())
        in_url = SQS_TTS_IN
        q = self._register_result_queue(job_id)

        # collect full text (simple approach)
        texts = []
        async for t in text_stream:
            texts.append(t)
        payload = {"job_id": job_id, "session_id": getattr(self, "session_id", None), "text": "".join(texts)}
        sqs.send_message(QueueUrl=in_url, MessageBody=json.dumps(payload))

        async def audio_gen():
            while True:
                item = await q.get()
                if item == "DONE":
                    break
                # item expected to contain audio_b64; return raw bytes or appropriate object
                audio_b64 = item.get("audio_b64")
                audio = base64.b64decode(audio_b64)
                yield audio
            self._unregister_result_queue(job_id)

        return audio_gen()

Notes: • Adapt

SpeechEvent

,

ChatContext

, and TTS audio yield shapes to match your LiveKit library version; above uses conceptual patterns. • Use

job_id

and

session_id

for correlation. FIFO ordering: use SQS FIFO with MessageGroupId=session_id. 2) Worker example (worker.py)

# worker.py
import os, json, base64, time
import boto3
from some_stt_lib import do_stt  # replace with your STT impl
from some_llm_lib import do_llm
from some_tts_lib import do_tts

AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
SQS_STT_IN = os.getenv("SQS_STT_IN_URL")
SQS_STT_OUT = os.getenv("SQS_STT_OUT_URL")
SQS_LLM_IN = os.getenv("SQS_LLM_IN_URL")
SQS_LLM_OUT = os.getenv("SQS_LLM_OUT_URL")
SQS_TTS_IN = os.getenv("SQS_TTS_IN_URL")
SQS_TTS_OUT = os.getenv("SQS_TTS_OUT_URL")

sqs = boto3.client("sqs", region_name=AWS_REGION)


def poll_process(queue_url, handler):
    while True:
        resp = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=10, WaitTimeSeconds=20)
        for m in resp.get("Messages", []):
            body = json.loads(m["Body"])
            try:
                handler(body)
                sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=m["ReceiptHandle"])
            except Exception as e:
                print("handler error", e)
        time.sleep(0.1)


# STT handler: assemble frames until EOF then run STT
def stt_handler_factory():
    # simple in-memory assembler keyed by job_id
    buffers = {}

    def handle(msg):
        job_id = msg["job_id"]
        if job_id not in buffers:
            buffers[job_id] = []
        if msg.get("eof"):
            # assemble PCM from base64 frames
            frames = buffers.pop(job_id, [])
            pcm = b"".join(base64.b64decode(f) for f in frames)
            transcript = do_stt(pcm)  # replace with real STT call
            out = {"job_id": job_id, "transcript": transcript, "done": True}
            sqs.send_message(QueueUrl=SQS_STT_OUT, MessageBody=json.dumps(out))
        else:
            buffers[job_id].append(msg["data_b64"])

    return handle


def llm_handler(msg):
    job_id = msg["job_id"]
    chat_ctx = msg.get("chat_ctx")
    # call LLM (can stream partials; here we send final)
    response = do_llm(chat_ctx)
    out = {"job_id": job_id, "response": response, "done": True}
    sqs.send_message(QueueUrl=SQS_LLM_OUT, MessageBody=json.dumps(out))


def tts_handler(msg):
    job_id = msg["job_id"]
    text = msg.get("text")
    audio_bytes = do_tts(text)  # returns raw pcm/wav bytes
    out = {"job_id": job_id, "audio_b64": base64.b64encode(audio_bytes).decode("utf-8"), "done": True}
    sqs.send_message(QueueUrl=SQS_TTS_OUT, MessageBody=json.dumps(out))


if __name__ == "__main__":
    import threading
    threading.Thread(target=poll_process, args=(SQS_STT_IN, stt_handler_factory())).start()
    threading.Thread(target=poll_process, args=(SQS_LLM_IN, llm_handler)).start()
    threading.Thread(target=poll_process, args=(SQS_TTS_IN, tts_handler)).start()
    # Keep main alive
    while True:
        time.sleep(60)

3) Deploy & Autoscale (AWS ECS + Application Auto Scaling) • Run workers as ECS services (one service per role: stt-worker, llm-worker, tts-worker). • Use CloudWatch metric

ApproximateNumberOfMessagesVisible

for the SQS queue to drive scaling. Example (conceptual) steps: • Create ECS Task Definition for stt-worker with CPU/memory sufficient for models. • Create ECS Service (Fargate or EC2). • Create CloudWatch Alarms on SQS metric: ◦ High alarm: visible messages > 50 → scale out (increase desired count) ◦ Low alarm: visible messages < 10 → scale in • Configure Application Auto Scaling policy attached to ECS service to change desired count on alarm. Example AWS CLI commands (outline) 1. Create target tracking policy (pseudocode):

aws application-autoscaling register-scalable-target --service-namespace ecs --resource-id service/cluster/stt-service --scalable-dimension ecs:service:DesiredCount --min-capacity 2 --max-capacity 50

aws application-autoscaling put-scaling-policy --service-namespace ecs --resource-id service/cluster/stt-service --scalable-dimension ecs:service:DesiredCount --policy-name scale-on-sqs --policy-type TargetTrackingScaling --target-tracking-scaling-policy-configuration file://policy.json

policy.json example:

{
  "TargetValue": 10,
  "PredefinedMetricSpecification": {
    "PredefinedMetricType": "SQSQueueMetrics",
    "ResourceLabel": "123456789012/queue-name"
  },
  "ScaleOutCooldown": 60,
  "ScaleInCooldown": 300
}

• Alternatively, use CloudWatch alarms + step-scaling policies reacting to

ApproximateNumberOfMessagesVisible

. Autoscaling tips: • Start with min 2 workers per role; set sensible max (e.g., 50). • Tune

TargetValue

to desired messages-per-worker. E.g., if one worker handles ~5 messages/sec, set target so ECS scales to keep queue depth per worker low. • Use FIFO queues (MessageGroupId=session_id) if ordering is critical; each group is consumed by one consumer at a time (affects parallelism). 4) Operational considerations & tuning • VisibilityTimeout must be > expected processing time; extend visibility for long LLM jobs. • Use DLQ for failed messages. • Implement idempotency (job_id) so retries do not duplicate effects. • For lower-latency result delivery, workers can POST results directly to an agent callback endpoint (authenticated) while still writing durable copy to SQS. • Prewarm models on worker startup to reduce first-request latency. • Monitor LiveKit delay metric and SQS queue depth. Autoscale before queue grows large. Next steps I can provide • A ready-to-run repo with these files adapted to your exact LiveKit SDK types. • ECS task definition JSON and a full CloudFormation/Terraform template to provision queues, worker services, DLQs, and autoscaling policies. Which next step do you prefer — code repo or infra templates (CloudFormation/Terraform)?