i've my livekit agent worker logic, how to deploy and run it through livekit cloud?

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

once i deployed, how to push new changes for the same agent

You're asking how to push new changes to an agent you already deployed. Follow these steps. 1. Build & deploy the new version • From your project directory run:

lk agent deploy

This builds a new image from your code (or your Dockerfile), uploads it, and updates the running agent. 1. Confirm status and rollout • Check deployment status:

lk agent status

• Stream logs to verify the new version started correctly:

lk agent logs

1. Manage versions • If you need to roll back (available on paid plans):

lk agent rollback

1. If you changed configuration • Regenerate or update `livekit.toml`:

lk agent config

Quick check for me: do you want exact Dockerfile or entrypoint snippets for your runtime (Python, Node, or Go)?

do agent on livekit cloud scale automatically?

You're asking whether LiveKit Cloud agents scale automatically — yes. LiveKit Cloud automatically scales and load-balances agents for you. Deployed agents are managed as containerized services: LiveKit handles instance creation, rolling updates, and routing so incoming jobs are distributed across running agent instances. Stateful session handling and graceful deployments are supported, and you can monitor status and logs with: •

lk agent status

•

lk agent logs

If you need control over concurrency, instance size, or custom scaling behavior, tell me your runtime (Python/Node/Go) and the scaling behavior you want (max concurrency per instance, number of replicas, or CPU/memory constraints) and I’ll provide the specific options and configuration steps.

runtime is python

You're asking whether Python agents on LiveKit Cloud scale automatically — yes. LiveKit Cloud automatically scales and load-balances agent instances for you. For Python agents this means: • Instances are auto-created when existing ones near capacity and sessions are routed across replicas. • Each agent instance typically handles around 5–10 concurrent conversations (actual concurrency depends on your agent’s CPU/memory usage and workload). • Deploys are rolling and gracefully drain old instances (up to ~1 hour) so updates are zero-downtime. Commands to monitor and manage: • Check status:

lk agent status

• View logs:

lk agent logs

• Deploy updates:

lk agent deploy

• Rollback (paid plans):

lk agent rollback

If you want control over scaling behavior, tell me which constraint you care about (max concurrency per instance, CPU/memory sizing, or fixed replica counts) and I’ll provide the exact configuration and CLI commands for a Python agent. Sources: livekit/livekit_composite | DeepWiki | Voice AI quickstart | LiveKit docs | examples/drive-thru | livekit/livekit_composite | DeepWiki | Deploying to production | LiveKit Docs | Builds and Dockerfiles | LiveKit Docs

and how to scale if we have worker is not hosted on livekit cloud

You're asking how to scale when your worker is not hosted on LiveKit Cloud — here’s a concise plan you can implement. 1) Run multiple worker instances • Deploy multiple copies of your Python agent process (each runs

agents.Worker(...)

) and point them at the same LiveKit server URL. LiveKit will dispatch jobs across all available workers. 2) Control per-worker load • Use

load_fnc

and

load_threshold

on each worker to report availability:

worker = agents.Worker(
  job_handler,
  load_fnc=lambda: psutil.cpu_percent() / 100,
  load_threshold=0.7
)

• Customize

load_fnc

to include memory/GPU or job counts. Workers stop accepting jobs when load >=

load_threshold

. 3) Autoscale with your orchestrator • Use Kubernetes HPA, AWS Auto Scaling, or similar. Base the autoscaler on the same metric as

load_fnc

(CPU, custom metric). • Trigger scale-up at a lower threshold than

load_threshold

(e.g., scale up at 0.5 if threshold is 0.7). • Use short scale-up stabilization and long scale-down stabilization so new capacity arrives quickly and active sessions drain before scale-down. Kubernetes example settings: • scaleUp.stabilizationWindowSeconds: 60 • scaleDown.stabilizationWindowSeconds: 300 4) Networking & health checks • Workers only need outbound WebSocket connectivity to LiveKit (no inbound ports required). • Expose a health endpoint (e.g., `http://0.0.0.0:8081/`) and have your orchestrator use it for readiness/liveness. 5) Graceful shutdown & crash handling • Ensure your process handles SIGTERM and allows jobs to finish or be re-assigned. • If a worker crashes, only its child job processes are affected; LiveKit will retry dispatch to other workers after timeout. 6) Monitoring & metrics • Collect CPU, memory, job duration, and queue/rejection metrics. Use these to tune

load_fnc

,

load_threshold

, and autoscaler policies. 7) Best practices • Containerize the agent, pre-download heavy ML models at build time, keep secrets out of images, and use env vars for config. • Test scale behavior with synthetic load to validate thresholds and cooldowns. If you want, tell me: • whether you're using Kubernetes, ECS, or plain VMs, and • what metric you prefer for load (CPU, memory, job count, etc.), and I’ll provide an exact Kubernetes HPA manifest or autoscaling recipe tuned for a Python agent. Sources: Deploying to custom environments | LiveKit docs | examples/drive-thru | livekit/livekit_composite | DeepWiki

how it is decided which user will be assigned to which worker if multiple are running

You're asking how LiveKit decides which user/session gets assigned to which worker when multiple workers are available — here’s how dispatch works. • Jobs are dispatched from the LiveKit server to workers registered for work. LiveKit uses a single-assignment, round-robin style dispatch across available workers, but availability is guided by each worker's reported load. • Each worker reports its current load via

load_fnc

and stops accepting new jobs when its load ≥

load_threshold

. The scheduler only routes jobs to workers that are currently marked available. • If a worker fails to accept a job within a timeout, the job is retried on another available worker. • This combination gives near round-robin distribution while respecting per-worker capacity (so busy workers receive fewer or no new assignments). If you want to influence assignment behavior, you can: • Adjust

load_fnc

to reflect CPU, memory, GPU, or active job count. • Tune

load_threshold

so workers become unavailable earlier or later. • Run multiple worker types (different resource sizes) and let the scheduler target whichever are available. Tell me if you want an example

load_fnc

that uses CPU+memory or a pattern for running heterogeneous worker pools and I’ll provide one.

can you share python example

You're asking for a Python example showing a worker with a custom load function so LiveKit can assign sessions respecting capacity. Below is a minimal Python agent demonstrating: • Agent

Worker

registration •

load_fnc

using CPU + memory •

load_threshold

to mark availability • graceful shutdown handling • an example job handler that accepts a session

import asyncio
import signal
import psutil
from livekit.agents import Worker, JobContext, cli

# Example job handler: accept the incoming session and run simple work
async def handle_job(ctx: JobContext):
    # Accept the job and run your agent logic
    await ctx.accept()  # attaches to the RTC session
    try:
        # Example: publish a message to the room or run work
        await ctx.room.local_participant.publish_data("agent-ready")
        # Keep job running until session ends or you finish work
        while not ctx.done():
            await asyncio.sleep(1)
    finally:
        await ctx.close()

# Custom load function: combine CPU and memory into a 0..1 load value
def load_fnc():
    cpu = psutil.cpu_percent(interval=0.1) / 100.0  # 0..1
    mem = psutil.virtual_memory().percent / 100.0   # 0..1
    # Weighted combination (tune weights to your workload)
    load = 0.7 * cpu + 0.3 * mem
    # Clamp to 0..1
    return max(0.0, min(1.0, load))

def create_worker():
    return Worker(
        job_request_handler=handle_job,
        load_fnc=load_fnc,
        load_threshold=0.7,   # stop accepting new jobs when load >= 0.7
        check_interval=2.0    # how often the worker reports load (seconds)
    )

def _setup_signal_loop(loop, worker):
    stop = asyncio.Event()

    def _on_signal():
        # Signal: stop accepting new jobs and allow graceful shutdown
        worker.stop_accepting()
        stop.set()

    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, _on_signal)

    return stop

if __name__ == "__main__":
    server = create_worker()
    loop = asyncio.get_event_loop()
    stop_event = _setup_signal_loop(loop, server)

    try:
        # Runs the agent server; this call blocks and manages connection to LiveKit
        cli.run_app(server)
        # If you need to wait for graceful shutdown:
        loop.run_until_complete(stop_event.wait())
    finally:
        loop.close()

Notes/tips: • Tune

load_threshold

and the weights in

load_fnc

to match your CPU/memory characteristics. • For GPU or model-heavy workloads, include GPU utilization or active model count in

load_fnc

. • Ensure each deployed instance runs this worker; LiveKit will dispatch jobs across all available workers. • Make the process container-friendly (handle SIGTERM, pre-download models during build, and expose a health endpoint if desired). Want a variant that uses active job count (number of concurrent sessions) instead of CPU/memory? I can provide that snippet next.

i'm starting my worker as: if name == "main": cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm)) how to set it up

You can pass

load_fnc

and

load_threshold

into

WorkerOptions

alongside your

entrypoint_fnc

and

prewarm_fnc

. Below is a minimal, ready-to-run example matching your start pattern:

import asyncio
import psutil
import signal
from livekit.agents import WorkerOptions, JobContext, cli

# prewarm: runs once per process (load models, etc.)
async def prewarm(proc):
    proc.userdata["ready"] = True
    # load heavy models here

# entrypoint: runs per job/session
async def entrypoint(ctx: JobContext):
    await ctx.accept()
    try:
        # your job logic here; access prewarmed resources via ctx.proc.userdata
        await ctx.room.local_participant.publish_data("agent-ready")
        while not ctx.done():
            await asyncio.sleep(1)
    finally:
        await ctx.close()

# load function: return value 0.0..1.0
def load_fnc():
    cpu = psutil.cpu_percent(interval=0.1) / 100.0
    mem = psutil.virtual_memory().percent / 100.0
    # weighted combination (tune as needed)
    load = 0.7 * cpu + 0.3 * mem
    return max(0.0, min(1.0, load))

if __name__ == "__main__":
    # Use your exact pattern; add load_fnc and load_threshold here
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            load_fnc=load_fnc,          # optional for self-hosted scaling
            load_threshold=0.7,         # stop accepting jobs when >= 0.7
            check_interval=2.0          # how often load_fnc is called (seconds)
        )
    )

Notes: •

load_fnc

and

load_threshold

are effective for self-hosted deployments; LiveKit Cloud ignores them. • Put heavy model downloads in

prewarm

to avoid cold-start work per job. • Ensure your process handles SIGTERM gracefully (the runtime drains by default when stopped). • Tune weights,

load_threshold

, and

check_interval

to match your workload. Would you like a variant that uses active job count or GPU usage in

load_fnc

? Sources: Agent server options | LiveKit docs | Deploying to custom environments | LiveKit docs

Looks like you mentioned Rajan — would you like me to draft a message or instructions you can forward to them? I can’t tag or notify people for you, but I can: • Draft a short note summarizing this conversation • Prepare specific instructions (deploy, deploy-update, scaling steps, or code snippets) • Create a checklist Rajan can follow What should the note ask Rajan to do or include? (Or paste any details you want added.)

WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, load_fnc=load_fnc, # optional for self-hosted scaling load_threshold=0.7, # stop accepting jobs when >= 0.7 check_interval=2.0 # how often load_fnc is called (seconds) ) In this I don’t think

check_interval

option is present in WorkerOptions.

You're correct —

check_interval

is not a valid field on

WorkerOptions

. Use only these load-related options with `WorkerOptions`:

WorkerOptions(
  entrypoint_fnc=entrypoint,
  prewarm_fnc=prewarm,
  load_fnc=load_fnc,
  load_threshold=0.7
)

If you need to control how often load is evaluated, do one of the following: • Cache or throttle inside your

load_fnc

(e.g., sample metrics at most once every N seconds). • Expose metrics externally (CPU, job count) and let your orchestrator autoscaler use those metrics and timings. • For advanced control, run a custom agent loop (not WorkerOptions) and implement your own scheduling/heartbeat logic. Would you like a short snippet showing how to throttle

load_fnc

calls (cached value with a TTL)? Sources: Agent server options | LiveKit docs | Deploying to custom environments | LiveKit docs

Can you please explain the flow to me? How this load_function and load_threshold will work?

You're asking for the flow of how

load_fnc

and

load_threshold

affect job assignment and worker availability. Below is a clear, step-by-step explanation. 1) Worker reports load • Each worker process calls your

load_fnc()

periodically (interval is internal).

load_fnc()

must return a number between 0.0 and 1.0 representing current load (0 = idle, 1 = fully loaded). • You can base it on CPU, memory, GPU, active job count, or a weighted combo. 2) Worker availability decision • The agent runtime compares the reported load to

load_threshold

. • If reported load <

load_threshold

, the worker is marked available to receive new jobs. • If reported load >=

load_threshold

, the worker is marked unavailable and will not be assigned new jobs until load drops below the threshold. 3) Job dispatch • When a new job/session arrives, the LiveKit server considers only workers currently marked available. • Jobs are distributed in a near round-robin manner across those available workers. • If a worker fails to accept a job (e.g., crashes or times out), the job is retried and routed to another available worker. 4) Effect on scaling and capacity • Proper

load_fnc

+

load_threshold

signals worker capacity to the scheduler so busy workers receive fewer or no new assignments. • For self-hosted deployments, tie your autoscaler to the same metric you use in

load_fnc

(or export that metric). Autoscaler should trigger scale-up at a lower threshold than

load_threshold

(e.g., scale-up when metric ≈ 0.5 if

load_threshold

= 0.7) and scale-down more slowly to allow draining. • In LiveKit Cloud, `load_fnc`/`load_threshold` are ignored because cloud autoscaling is handled by LiveKit. 5) Practical tips • Return a smooth value (0..1); avoid high variance spikes that flip availability rapidly. • Use sensible thresholds: leave headroom to avoid rejecting jobs under brief spikes. • Consider caching/throttling expensive metric collection inside

load_fnc

. • For long-running sessions, autoscalers should use longer stabilization windows for scale-down so active sessions finish cleanly. If you want, I can: • Provide a short

load_fnc

snippet that caches values to avoid frequent expensive metric calls, or • Show an example where availability is based on active job count instead of CPU. Which would you prefer?

You are saying periodically check. By default after how much time it calls load_func() to check the load?

Short answer: there’s no user-configurable fixed interval. How often

load_fnc

is invoked depends on deployment: • LiveKit Cloud: interval is internal and not exposed; you cannot rely on a specific polling period and Cloud ignores custom `load_fnc`/`load_threshold`. • Self-hosted/custom deployments:

load_fnc

is evaluated on-demand as part of job assignment—effectively whenever the server is considering dispatching a job. For CPU-based checks the runtime computes CPU over ~5s windows, but there is no guaranteed global polling cadence. If you need a predictable/checkable cadence in a self-hosted setup, implement one inside your

load_fnc

(cache the computed value with a TTL) or expose the metric externally and let your autoscaler poll it at a fixed interval. Do you want a short snippet showing a cached

load_fnc

with TTL? Sources: Agent server options | LiveKit docs | Deploying to custom environments | LiveKit docs

So suppose If I run 2 agents and 1 agent can hold the load of two users at a time and agent 1 is occupied(two users connected with it) already but when 3rd user start connecting then it will automatically redirects it to the 2nd agent or there may be chance that it will still connect to 1st one?

Short answer: with a correct

load_fnc

+

load_threshold

, the 3rd user will be routed to agent 2 — LiveKit only dispatches to workers currently marked available. If agent 1 reports load >= threshold it will be excluded from assignment. Details: • Each worker reports a 0..1 load via your

load_fnc

. If load >=

load_threshold

, that worker is marked unavailable for new jobs. • New sessions are dispatched to available workers in a near round‑robin fashion. Only workers marked available are considered. • If a worker fails to accept a job (crash/timeout), the job is retried on another available worker. Risk scenario: • If your

load_fnc

underestimates load (or you base it on a metric that lags), the runtime may still consider agent 1 available and attempt assignment. So a poorly tuned

load_fnc

can let extra jobs land on an already-heavy worker. Recommendation: • Make

load_fnc

reflect actual capacity (e.g., active job count, CPU+memory, GPU usage). • Use

load_threshold

to leave headroom (e.g., threshold 0.7). • Cache/throttle expensive metrics inside

load_fnc

to avoid variance. • Test with synthetic connections to confirm behavior. If you want, I’ll provide a

load_fnc

example that uses active-job-count (so one worker truly limits to 2 concurrent sessions) — would you prefer that or a CPU+memory variant?

So give me correct

load_func

and

load_threshold

like it could be 70% for my use case and also let me know where I have to write load_function

Good — here’s a practical, ready-to-use Python

load_fnc

+ guidance that enforces a 70% threshold and limits a worker to two concurrent sessions. I’ll show where to put it and how to wire it into your existing

WorkerOptions

start. 1) Behavior •

load_fnc

returns a float 0.0..1.0. •

load_threshold = 0.7

means the worker is marked unavailable for new jobs when

load_fnc()

>= 0.7. • This example bases load on active job count, so a worker with capacity 2 returns load 1.0 when 2 jobs are active, 0.5 when 1 job is active, etc. That ensures the third user will be routed to another available worker. 2) Where to place it • Define

load_fnc

in the same module where you create

WorkerOptions

(e.g., next to

entrypoint_fnc

and

prewarm_fnc

). • The runtime calls it when considering job assignment (self-hosted). In WorkerOptions pass it as

load_fnc=load_fnc

. 3) Code example (drop into your agent file)

import time
import threading
from livekit.agents import WorkerOptions, JobContext, cli

# Simple counter of active jobs. Protected by a lock.
_active_jobs = 0
_active_jobs_lock = threading.Lock()
_last_sample_t = 0
_cached_load = 0.0
_CACHE_TTL = 1.0  # seconds: avoid expensive recalcs more than once/sec

async def entrypoint(ctx: JobContext):
    global _active_jobs
    with _active_jobs_lock:
        _active_jobs += 1
    try:
        await ctx.accept()
        # job work here
        while not ctx.done():
            await ctx.sleep(1)
    finally:
        with _active_jobs_lock:
            _active_jobs -= 1
        await ctx.close()

def load_fnc():
    # Cached read to avoid hot loops calling this too frequently
    global _last_sample_t, _cached_load
    now = time.time()
    if now - _last_sample_t < _CACHE_TTL:
        return _cached_load

    with _active_jobs_lock:
        active = _active_jobs

    # Capacity: 2 concurrent sessions per worker
    capacity = 2
    load = min(active / capacity, 1.0)  # 0.0..1.0

    _cached_load = load
    _last_sample_t = now
    return load

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=None,
            load_fnc=load_fnc,
            load_threshold=0.7,  # mark unavailable when load >= 0.7
        )
    )

4) Why this fits your requirement • With capacity = 2: ◦ 0 active -> load 0.0 (available) ◦ 1 active -> load 0.5 (available) ◦ 2 active -> load 1.0 (>= 0.7 → unavailable) • The 3rd connecting session will be routed to any other worker marked available. 5) Tuning tips • If you prefer CPU/memory combo, replace active-job logic with a weighted CPU/memory measurement. • Increase

_CACHE_TTL

to reduce frequency of expensive checks; decrease to be more reactive. • Ensure the entrypoint increments/decrements the counter reliably (use locks as shown). Would you like a variant that uses CPU+memory instead of active-job-count?

If I give you my main agent.py file then you can give me according to that?

Yes — send your main

agent.py

and I’ll return a tailored

load_fnc

, where to place it, and any small edits to wire it into your existing

WorkerOptions

startup. Please paste the file (or the relevant parts): your

entrypoint

,

prewarm

, and how you call

cli.run_app(...)

.

agent.py

No file received — I didn't get your

agent.py

. Paste the

entrypoint

,

prewarm

, and the

cli.run_app(...)

call (or the whole file) and I’ll return a tailored

load_fnc

, exact edits, and where to place it.

I have attached that in previous message

I don’t see your file — please paste your

agent.py

(or the

entrypoint

,

prewarm

, and the

cli.run_app(...)

call). I’ll then return a tailored

load_fnc

, exact edits, and where to place it.

async def entrypoint(ctx: JobContext):  
    await ctx.connect()  

    participant = await ctx.wait_for_participant()
    
    # Get the agent ID from participant attributes
    # The agent ID is used to fetch all agent configuration dynamically from the API
    agent_id = participant.attributes.get("agent")
    # Default to the fallback agent when no attribute is provided
    if not agent_id:
        agent_id = "693a96fd276eb7171e083efb" # Dbles Inbound Agent
    user_id = "69299a964990c460f4188cf3"
    # user_id = participant.attributes.get("userId")
    
    if not agent_id:
        # logger.error("No agent ID provided in participant attributes")
        raise ValueError("Agent ID is required. Please provide 'agent' attribute in participant.")
    
    # <http://logger.info|logger.info>(f"Creating dynamic agent with agent_id: {agent_id}")

    # Add any other context you want in all log entries here
    ctx.log_context_fields = {
        "room": ctx.room.name,
        "agent_type": "dynamic_rag_agent",
        "agent_id": agent_id,
    }
    egress_info = ""

  # Record the room to S3
    try:
        if (os.getenv('S3_RECORDING_BUCKET') is not None):
            # <http://logger.info|logger.info>(f"Recording room to S3: {ctx.room.name}")
            egress_info = await record_room_to_s3(ctx.room.name)
        else:
            # <http://logger.info|logger.info>(f"S3 recording bucket is not configured, skipping recording")
            pass
    except Exception as e:
        # logger.error(f"Failed to record room to S3: {e}")
        pass
        
    conversation_id = await create_conversation_from_backend(agent_id, egress_info)
    try:
        # Create dynamic agent - all configuration will be fetched from API
        agent = await AgentFactory.create_agent(
            "dynamic_rag_agent", agent_id=agent_id
        )
        config = agent.config
        config = config["formatted_config"]
        
        # <http://logger.info|logger.info>(f"Created agent: {config['agent_name']}")
        
    except ValueError as e:
        # logger.error(f"Failed to create agent: {e}")
        raise  # Re-raise to prevent continuing with invalid agent

    # Build session components using the agent's configuration
    session = AgentSession(
        stt=config["stt"],
        llm=config["llm"],
        tts=config["tts"],
        turn_detection=MultilingualModel(),
        vad=ctx.proc.userdata["vad"],
        preemptive_generation=True,
    )

    # Metrics collection, to measure pipeline performance
    # For more information, see <https://docs.livekit.io/agents/build/metrics/>
    usage_collector = metrics.UsageCollector()
    last_eou_metrics: Optional[metrics.EOUMetrics] = None
    conversation_metrics: Dict[str, Dict[str, Any]] = {}  # Maps speech_id to collected metrics
    pending_stt_metrics: List[Any] = []  # Store STT metrics until we get EOUMetrics with speech_id
    conversation_messages: List[Dict[str, Any]] = []  # Store all conversation messages with usage data
    flow_start_times: Dict[str, float] = {}  # Maps speech_id to flow start time (perf_counter)

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        nonlocal last_eou_metrics

        # Handle STT metrics (no speech_id)
        if ev.metrics.type == "stt_metrics":
            import time
            # Store temporarily until we get EOUMetrics
            pending_stt_metrics.append(ev.metrics)
            stt_metrics = ev.metrics
            audio_duration = getattr(stt_metrics, 'audio_duration', 0)
            metrics.log_metrics(ev.metrics)
            usage_collector.collect(ev.metrics)
            return

        # Get speech_id from the metric if available
        speech_id = getattr(ev.metrics, 'speech_id', None)

        if speech_id:
            if speech_id not in conversation_metrics:
                conversation_metrics[speech_id] = {
                    'llm': None,
                    'tts': [],
                    'stt': [],
                    'eou': None
                }

            # Store metrics by type
            if ev.metrics.type == "llm_metrics":
                import time
                from datetime import datetime
                llm_metrics = ev.metrics
                conversation_metrics[speech_id]['llm'] = llm_metrics
                
                # Log LLM processing time in seconds
                model_name = llm_metrics.metadata.model_name if hasattr(llm_metrics.metadata, 'model_name') else 'unknown'
            elif ev.metrics.type == "tts_metrics":
                import time
                tts_metrics = ev.metrics
                conversation_metrics[speech_id]['tts'].append(tts_metrics)
            elif ev.metrics.type == "eou_metrics":
                import time
                eou_metrics = ev.metrics
                # Associate pending STT metrics with this speech_id
                conversation_metrics[speech_id]['stt'] = pending_stt_metrics.copy()
                pending_stt_metrics.clear()
                conversation_metrics[speech_id]['eou'] = ev.metrics
                last_eou_metrics = ev.metrics

        metrics.log_metrics(ev.metrics)
        usage_collector.collect(ev.metrics)

    @session.on("conversation_item_added")
    def _on_conversation_item_added(ev):
        asyncio.create_task(handle_conversation_item(ev))

async def shutdown_handler():
        try:
            await send_end_call_to_backend()
        except Exception as e:
            # logger.error(f"Failed to send end-call usage: {e}")
            pass

    ctx.add_shutdown_callback(shutdown_handler)
      
    # Conditionally add avatar based on agent configuration
    if config.get("avatar_enabled", False) and config.get("avatar_id"):  
        try:
            avatar = bey.AvatarSession(avatar_id=config["avatar_id"])  
            await avatar.start(session, room=ctx.room)
            # <http://logger.info|logger.info>(f"Avatar started for agent: {config['agent_name']}")
        except Exception as e:
            # logger.warning(f"Failed to start avatar: {e}")
            pass
    
    # Start the session with the created agent
    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(),
            ),
        ),
    )


if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))

Thanks — based on your agent code, here’s a minimal, safe way to add an active-job-count based

load_fnc

(capacity = 2) and wire it into your

WorkerOptions

. This ensures a worker with two active sessions reports load >= 0.7 and will be excluded from new assignments. Add these top-level helpers (near your imports) in the same file:

import time
import threading

_active_jobs = 0
_active_jobs_lock = threading.Lock()
_last_sample_t = 0.0
_cached_load = 0.0
_CACHE_TTL = 1.0  # seconds: how long to reuse cached load value
_CAPACITY = 2     # number of concurrent sessions this worker should handle

Wrap your entrypoint to increment/decrement the counter (modify your existing

entrypoint

accordingly — shown only the relevant parts):

async def entrypoint(ctx: JobContext):
    global _active_jobs
    # increment active jobs
    with _active_jobs_lock:
        _active_jobs += 1

    try:
        await ctx.connect()
        # ... your existing logic (participant, agent creation, session.start, etc.)
        # make sure the function returns/awaits until the session ends
        # you already use ctx.add_shutdown_callback and session lifecycle; keep those
    finally:
        # decrement on exit
        with _active_jobs_lock:
            _active_jobs -= 1
        # ensure session/ctx closed if not already
        try:
            await ctx.close()
        except Exception:
            pass

Add this

load_fnc

(same file):

def load_fnc():
    """
    Returns 0.0..1.0 load based on active job count and configured capacity.
    Cached for _CACHE_TTL seconds to avoid expensive repeats.
    """
    global _last_sample_t, _cached_load
    now = time.time()
    if now - _last_sample_t < _CACHE_TTL:
        return _cached_load

    with _active_jobs_lock:
        active = _active_jobs

    load = min(active / _CAPACITY, 1.0)
    _cached_load = load
    _last_sample_t = now
    return load

Finally, start the worker with

load_fnc

and `load_threshold=0.7`:

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            load_fnc=load_fnc,
            load_threshold=0.7,
        )
    )

Notes and checks: • With _CAPACITY = 2: active=0 -> load 0.0; active=1 -> 0.5; active=2 -> 1.0 (>=0.7 → unavailable). • The counter must increment before

await ctx.connect()

finishes and decrement in the

finally

so it reflects actual active sessions. • Keep

_CACHE_TTL

small (1s) to avoid stale readings but reduce hot-loop cost. • If your session spawns background tasks that keep the session alive, ensure

entrypoint

waits until session end (your current pattern with loops / ctx.done() is fine). • This

load_fnc

is effective for self-hosted deployments. LiveKit Cloud ignores custom

load_fnc

/

load_threshold

. Want a variant that also factors CPU or GPU usage into the load calculation?

I want my worked to connect only 3 agents then how can I do that?

If you want exactly 3 agent instances (processes/replicas) running, that is controlled by your deployment/orchestrator — not by the agent code. Here are the practical options depending on your environment. If you’re using Kubernetes • Set replicas = 3 in your Deployment (or use an HPA with minReplicas=3 and maxReplicas=3). ◦ Example: in your Deployment spec set

spec.replicas: 3

. • Ensure your autoscaler (HPA) thresholds won't increase replicas above 3. If you’re using ECS / EC2 / other cloud autoscaling • Configure the service desired count = 3 (ECS) or set the autoscaling group min/max = 3. • Lock scaling policies so they don’t exceed 3. If you’re running plain VMs or containers • Start exactly 3 processes/containers and don’t run any external autoscaler. If you meant “each worker should accept up to 3 concurrent sessions” (capacity per process) • Set _CAPACITY = 3 in the load_fnc we added (so load = active / 3). Keep load_threshold such that when 3 sessions are active the worker becomes unavailable (e.g., load_threshold = 0.7 still works because load will be 1.0 at capacity). Notes about LiveKit Cloud • LiveKit Cloud auto-scales; you cannot directly force an exact instance count there. If you need exactly 3 instances, self-host or coordinate with LiveKit Cloud support/plans (Cloud-managed scaling is automatic). Which do you want to enforce? • Exactly 3 total instances (replicas) — tell me your deployment (Kubernetes, ECS, plain VMs) and I’ll give the exact manifest/commands. • Or per-process capacity = 3 sessions — I’ll update the

load_fnc

snippet to use

_CAPACITY = 3

.

No 3 instances I want my agent.py should connect to 3 users only max means worker should allow connection to 3 users

Understood — to limit a single worker process to at most 3 concurrent users, make the worker report load based on active-session count with capacity = 3 and a threshold that marks it unavailable when at capacity. Change the active-job code and load function like this (drop into your current agent file near imports and WorkerOptions): 1) Add globals near top:

import time
import threading

_ACTIVE_JOBS = 0
_ACTIVE_JOBS_LOCK = threading.Lock()
_LAST_SAMPLE_T = 0.0
_CACHED_LOAD = 0.0
_CACHE_TTL = 1.0   # seconds
_CAPACITY = 3      # max concurrent sessions per worker

2) Increment/decrement in your entrypoint (wrap existing logic):

async def entrypoint(ctx: JobContext):
    global _ACTIVE_JOBS
    with _ACTIVE_JOBS_LOCK:
        _ACTIVE_JOBS += 1

    try:
        await ctx.connect()
        # ... your existing entrypoint logic (participant, agent creation, session.start, etc.)
        # ensure this function awaits until the session finishes
    finally:
        with _ACTIVE_JOBS_LOCK:
            _ACTIVE_JOBS -= 1
        try:
            await ctx.close()
        except Exception:
            pass

3) Add the load function:

def load_fnc():
    global _LAST_SAMPLE_T, _CACHED_LOAD
    now = time.time()
    if now - _LAST_SAMPLE_T < _CACHE_TTL:
        return _CACHED_LOAD

    with _ACTIVE_JOBS_LOCK:
        active = _ACTIVE_JOBS

    load = min(active / _CAPACITY, 1.0)  # 0..1
    _CACHED_LOAD = load
    _LAST_SAMPLE_T = now
    return load

4) Start the worker with threshold 0.7:

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            load_fnc=load_fnc,
            load_threshold=0.7,  # worker becomes unavailable when load >= 0.7
        )
    )

How this enforces 3-concurrency: • With _CAPACITY = 3: ◦ 0 active -> load 0.0 (available) ◦ 1 active -> ~0.33 (available) ◦ 2 active -> ~0.67 (available) ◦ 3 active -> 1.0 (>= 0.7 → marked unavailable) • The 4th connecting session will be dispatched only to another worker marked available. Notes: • Ensure entrypoint truly waits until the session ends so the counter reflects active sessions. • _CACHE_TTL avoids expensive repeated reads; adjust for responsiveness. • This limits concurrency per worker process only; to control total workers, configure your orchestrator separately. • LiveKit Cloud ignores custom load_fnc/load_threshold — this works for self-hosted deployments. Want me to produce a drop-in patch of your full file with these edits?

Hello, async def *on_user_turn_completed*( self, turn_ctx: ChatContext, new_message: ChatMessage, ) -> None: “”" Lifecycle hook called when a user completes their turn (voice input only). If RAG is enabled, automatically retrieves context and injects it into the conversation. “”" if not self.rag_enabled: logger.debug(“RAG is not enabled, skipping RAG lookup for voice input”) return # Extract query from user message query = None if *hasattr*(new_message, ‘content’) and new_message.content: # Handle both string and list content if *isinstance*(new_message.content, str): query = new_message.content elif *isinstance*(new_message.content, list): # Extract text from list of content parts query_parts = [] for content_part in new_message.content: if *hasattr*(content_part, ‘text’): query_parts.append(content_part.text) elif *isinstance*(content_part, str): query_parts.append(content_part) query = ” “.join(query_parts) if query_parts else None elif *hasattr*(new_message, ‘text_content’): query = new_message.text_content() else: query = str(new_message) if not query or not query.strip(): return logger.info(f”RAG lookup for voice input: {query[:100]}“) # Call the helper method to perform RAG lookup and inject context await self._perform_rag_lookup_and_inject(query, turn_ctx) I have used this function but this is working perfectly with my agent when voice input is there to my agent but when I am giving normal text input then it is not working how can i achieve that because in text input as well I need to perform this

_perform_rag_lookup_and_inject

function. and this function is below: async def *_perform_rag_lookup_and_inject*( self, query: str, turn_ctx: Optional[ChatContext] = None ) -> None: “”" Helper method to perform RAG lookup and inject context into conversation. Can be called from both voice (on_user_turn_completed) and text input handlers. Args: query: The user’s query text turn_ctx: Optional ChatContext (available for voice input, None for text input) “”" if not self.rag_enabled: logger.info(f”RAG is not enabled, skipping RAG lookup”) return # Ensure query is a string and not empty if not query or not *isinstance*(query, str) or not query.strip(): return query = query.strip() logger.info(f”RAG lookup query: {query}“) _# Perform RAG lookup using the same logic as maica_knowledge_search_ rag_snippets = await self.my_rag_lookup(query) if not rag_snippets: return # Inject concise retrieved context for the LLM # Format as an assistant message with the retrieved information context_message = f”Retrieved information for context:\n{rag_snippets}” try: _# Try to add message using turn_ctx.add_message (available for voice input)_ if turn_ctx and *hasattr*(turn_ctx, ‘add_message’): turn_ctx.add_message( role=“assistant”, content=context_message ) logger.info(f”Injected RAG context into conversation for query: {query[:100]}“) # Fallback: Try to update chat context directly if available elif *hasattr*(self, ‘_chat_ctx’) and self._chat_ctx: # Add message to chat context from livekit.agents.llm import ChatMessage as LLMChatMessage context_msg = LLMChatMessage(role=“assistant”, content=context_message) # Update chat context with new message self._chat_ctx = self._chat_ctx.copy(messages=[*self._chat_ctx.messages, context_msg]) logger.info(f”Injected RAG context via chat_ctx for query: {query[:100]}“) else: logger.warning(“Could not inject RAG context: no turn_ctx.add_message or _chat_ctx available”) except Exception as e: logger.warning(f”Failed to inject RAG context into conversation: {e}“, exc_info=True)

Thanks — clear question. Your voice path uses

on_user_turn_completed

, but text input doesn't call that hook. To run

_perform_rag_lookup_and_inject

for text too, register a text input callback and call the same helper. Add a text handler and wire it into

room_options

when you start the session. Minimal example using your existing `_perform_rag_lookup_and_inject`:

import asyncio
from livekit.agents.voice import room_io

# text input callback
def text_input_cb(session: AgentSession, event: room_io.TextInputEvent):
    query = (event.text or "").strip()
    if not query:
        return
    # call shared RAG helper (no turn_ctx for text)
    asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None))
    # optionally generate a reply after injection
    asyncio.create_task(session.generate_reply(user_input=query))

When starting the session, pass it in `room_options`:

await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        text_input=room_io.TextInputOptions(
            text_input_cb=text_input_cb
        ),
        # keep your audio options...
    ),
)

Notes: • The callback runs before the agent replies; using

session.generate_reply()

ensures the reply uses injected context. • If

session.agent

isn’t available from the callback, capture the agent instance into the callback closure. • Your

_perform_rag_lookup_and_inject

already handles

turn_ctx=None

and falls back to `self._chat_ctx`; prefer injecting into

session.chat_ctx

when available. • Use

asyncio.create_task

so the callback returns immediately and doesn't block event processing. Want me to produce a drop-in edit of your session.start code showing exact placement with your existing prewarm/entrypoint? Sources: Text and transcriptions | LiveKit docs | Agent session | LiveKit docs

currently my code is this for start 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(), ), ), )

Great — I’ll show exactly how to add the text callback into your existing

session.start

call so text inputs also trigger

_perform_rag_lookup_and_inject

. 1) Add this callback near your imports (it uses the same session object to call the shared helper):

import asyncio
from livekit.agents.voice import room_io

def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
    query = (event.text or "").strip()
    if not query:
        return
    # fire-and-forget: call the same RAG helper used by voice (turn_ctx=None)
    asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None))
    # ensure agent generates a reply that will include injected context
    asyncio.create_task(session.generate_reply(user_input=query))

2) Wire it into your existing

session.start

call (replace your current call with this):

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(),
        ),
        text_input=room_io.TextInputOptions(
            text_input_cb=text_input_cb
        ),
    ),
)

Notes: •

text_input_cb

runs for each text message received on

lk.chat

. It calls your

_perform_rag_lookup_and_inject

with

turn_ctx=None

, which your helper already supports. • Using

asyncio.create_task

keeps the callback non-blocking.

session.generate_reply(user_input=query)

triggers the agent reply so the injected context is used. • If

session.agent

isn’t accessible in your environment, capture

agent

into a closure instead: ◦ e.g.,

def make_cb(agent): def cb(session, event): ...; return cb

and pass

text_input_cb=make_cb(agent)

. • If you prefer the text handler to await RAG lookup before generating a reply, replace

create_task

for

generate_reply

with an awaited coroutine inside an async wrapper. Want me to produce the exact drop-in patch for your file (showing import placement and the modified

session.start

block)?

I got this error: 31 120738,560 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py:405> exception=AttributeError(“‘AgentSession’ object has no attribute ‘agent’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 94, in text_input_cb asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None)) AttributeError: ‘AgentSession’ object has no attribute ‘agent’. Did you mean: ‘_agent’? 2025-12-31 120738,560 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py:405> exception=AttributeError(“‘AgentSession’ object has no attribute ‘agent’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 94, in text_input_cb asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None)) AttributeError: ‘AgentSession’ object has no attribute ‘agent’. Did you mean: ‘_agent’? Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 94, in text_input_cb asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None)) AttributeError: ‘AgentSession’ object has no attribute ‘agent’. Did you mean: ‘_agent’? 120738.560 ERROR asyncio Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/s… exception=AttributeError(“‘AgentSession’ object has no attribute ‘agent’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/… line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 94, in text_input_cb asyncio.create_task(session.agent._perform_rag_lookup_and_in… turn_ctx=None)) AttributeError: ‘AgentSession’ object has no attribute ‘agent’. Did you mean: ‘_agent’? {“room”: “voice_assistant_room_4166”, “agent_type”: “dynamic_rag_agent”, “agent_id”: “693bd39146235de7dc27dda5”, “pid”: 60606, “job_id”: “AJ_TRZNL45yMPuB”, “room_id”: “RM_3EBuPHrzpZVz”} Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/roo m_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 94, in text_input_cb asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None)) AttributeError: ‘AgentSession’ object has no attribute ‘agent’. Did you mean: ‘_agent’? and at 94 line i have this code

asyncio.create_task(session.agent._perform_rag_lookup_and_inject(query, turn_ctx=None))

The error happens because

AgentSession

doesn’t expose a public

agent

attribute (it uses

_agent

internally). Don’t rely on a private attribute — instead capture the agent instance into the callback closure when you create

text_input_cb

, or read/write into

session

if you’ve attached the agent there. Two safe options — pick one: Option A — capture

agent

in a closure (recommended)

def make_text_input_cb(agent):
    def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
        query = (event.text or "").strip()
        if not query:
            return
        asyncio.create_task(agent._perform_rag_lookup_and_inject(query, turn_ctx=None))
        asyncio.create_task(session.generate_reply(user_input=query))
    return text_input_cb

# when starting session:
await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=...,
        text_input=room_io.TextInputOptions(
            text_input_cb=make_text_input_cb(agent)
        ),
    ),
)

Option B — read the internal

_agent

from session (works but less future-proof)

def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
    query = (event.text or "").strip()
    if not query:
        return
    agent_obj = getattr(session, "_agent", None)
    if agent_obj is None:
        return
    asyncio.create_task(agent_obj._perform_rag_lookup_and_inject(query, turn_ctx=None))
    asyncio.create_task(session.generate_reply(user_input=query))

Notes: • Use

asyncio.create_task

so the callback returns immediately. • If you need the RAG lookup to complete before reply, await it inside an async wrapper instead of fire-and-forget. • Prefer Option A to avoid depending on private internals. • If

_perform_rag_lookup_and_inject

accesses per-session state, passing

turn_ctx=None

is fine; otherwise adapt it to accept session/context. Would you like me to produce the exact drop-in patch for your file using Option A?

in option A where I need to keep this function def make_text_input_cb(agent): def text_input_cb(session: “AgentSession”, event: room_io.TextInputEvent): query = (event.text or “”).strip() if not query: return asyncio.create_task(agent._perform_rag_lookup_and_inject(query, turn_ctx=None)) asyncio.create_task(session.generate_reply(user_input=query)) return text_input_cb

Put

make_text_input_cb

near your other helper functions (top of the file, after imports and before

entrypoint

) so it's available when you call

session.start

. Define it as a factory that captures the

agent

instance and returns the actual callback. Then pass the returned callback into

room_io.TextInputOptions

when starting the session. Example placement and usage (drop into your file):

# near top of file, after imports
from livekit.agents.voice import room_io
import asyncio

def make_text_input_cb(agent):
    def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
        query = (event.text or "").strip()
        if not query:
            return
        # fire-and-forget RAG lookup and then trigger reply
        asyncio.create_task(agent._perform_rag_lookup_and_inject(query, turn_ctx=None))
        asyncio.create_task(session.generate_reply(user_input=query))
    return text_input_cb

Then in your

entrypoint

where you call

session.start

, pass the callback:

await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(...),
        text_input=room_io.TextInputOptions(
            text_input_cb=make_text_input_cb(agent)
        ),
    ),
)

Notes: • Keep

make_text_input_cb

in the same module so it can close over the

agent

object. • Using the closure avoids relying on internal

AgentSession

attributes. • If you need the RAG lookup to finish before replying, replace the two

create_task

calls with an async wrapper that awaits the lookup then calls

generate_reply

.

Got this now: 2025-12-31 121412,659 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py:405> exception=TypeError(‘a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080>’)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 97, in text_input_cb asyncio.create_task(session.generate_reply(user_input=query)) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py”, line 337, in create_task task = loop.create_task(coro) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py”, line 438, in create_task task = tasks.Task(coro, loop=self, name=name) TypeError: a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080> 2025-12-31 121412,659 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py:405> exception=TypeError(‘a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080>’)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 97, in text_input_cb asyncio.create_task(session.generate_reply(user_input=query)) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py”, line 337, in create_task task = loop.create_task(coro) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py”, line 438, in create_task task = tasks.Task(coro, loop=self, name=name) TypeError: a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/room_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 97, in text_input_cb asyncio.create_task(session.generate_reply(user_input=query)) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/tasks.py”, line 337, in create_task task = loop.create_task(coro) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py”, line 438, in create_task task = tasks.Task(coro, loop=self, name=name) TypeError: a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080> 121412.659 ERROR asyncio Task exception was never retrieved future: <Task finished name=‘Task-111’ coro=<RoomIO._on_user_text_input.<locals>._read_text() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/s… exception=TypeError(‘a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080>’)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/… line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 97, in text_input_cb asyncio.create_task(session.generate_reply(user_input=query)) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.fr… line 337, in create_task task = loop.create_task(coro) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.fr… line 438, in create_task task = tasks.Task(coro, loop=self, name=name) TypeError: a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080> {“room”: “voice_assistant_room_1019", “agent_type”: “dynamic_rag_agent”, “agent_id”: “693bd39146235de7dc27dda5", “pid”: 69714, “job_id”: “AJ_K9tkCa2JWuTz”, “room_id”: “RM_N8zoQTuPpL4u”} Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/livekit/agents/voice/room_io/roo m_io.py”, line 408, in _read_text text_input_result = text_input_cb( File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 97, in text_input_cb asyncio.create_task(session.generate_reply(user_input=query)) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/t asks.py”, line 337, in create_task task = loop.create_task(coro) File “/opt/homebrew/Cellar/python@3.10/3.10.17_1/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/b ase_events.py”, line 438, in create_task task = tasks.Task(coro, loop=self, name=name) TypeError: a coroutine was expected, got <livekit.agents.voice.speech_handle.SpeechHandle object at 0x13ae8a080> 2025-12-31 121413,128 - agent - INF

The TypeError happens because

session.generate_reply(...)

returns a non-coroutine

SpeechHandle

object, so passing it to

asyncio.create_task

raises "a coroutine was expected". Call patterns: • Do NOT wrap

session.generate_reply(...)

in

asyncio.create_task

directly. • Either call

session.generate_reply(...)

synchronously (it returns a handle) or wrap an async wrapper that awaits an actual coroutine if you need to await something first. Corrected

make_text_input_cb

(drop-in). Place this near your imports and before `entrypoint`:

import asyncio
from livekit.agents.voice import room_io

def make_text_input_cb(agent):
    def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
        query = (event.text or "").strip()
        if not query:
            return

        # Fire-and-forget RAG lookup (async)
        asyncio.create_task(agent._perform_rag_lookup_and_inject(query, turn_ctx=None))

        # Trigger generation — do NOT create_task on the returned SpeechHandle
        # session.generate_reply returns a SpeechHandle; call it directly.
        try:
            session.generate_reply(user_input=query)
        except Exception:
            # If you need to handle errors, do it here
            pass

    return text_input_cb

Integration when starting the session (same place you already call session.start):

await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(...),
        text_input=room_io.TextInputOptions(
            text_input_cb=make_text_input_cb(agent)
        ),
    ),
)

If you want the reply to wait until RAG lookup finishes (synchronous flow):

def make_text_input_cb(agent):
    def text_input_cb(session, event):
        query = (event.text or "").strip()
        if not query:
            return

        async def run_and_reply():
            await agent._perform_rag_lookup_and_inject(query, turn_ctx=None)
            # After injection, call generate_reply (returns SpeechHandle)
            session.generate_reply(user_input=query)

        asyncio.create_task(run_and_reply())

    return text_input_cb

Summary: • Don’t pass

session.generate_reply(...)

into `create_task`; call it directly or call it inside an async wrapper that you schedule. • Use the closure factory to capture

agent

(avoid using private session attributes). • Choose fire-and-forget RAG + immediate reply, or await RAG then reply using the async wrapper above. Would you like me to produce a small patched excerpt of your file with this exact change?

Now in my

_perform_rag_lookup_and_inject

I am getting this mic-rag-agent - WARNING - Failed to inject RAG context into conversation: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agents/dynamic_rag_agent.py”, line 1153, in _perform_rag_lookup_and_inject context_msg = LLMChatMessage(role=“assistant”, content=context_message) File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/pydantic/main.py”, line 250, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) pydantic_core._pydantic_core.ValidationError: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type 2025-12-31 121811,171 - dynamic-rag-agent - WARNING - Failed to inject RAG context into conversation: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agents/dynamic_rag_agent.py”, line 1153, in _perform_rag_lookup_and_inject context_msg = LLMChatMessage(role=“assistant”, content=context_message) File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/pydantic/main.py”, line 250, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) pydantic_core._pydantic_core.ValidationError: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agents/dynamic_rag_agent.py”, line 1153, in _perform_rag_lookup_and_inject context_msg = LLMChatMessage(role=“assistant”, content=context_message) File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/pydantic/main.py”, line 250, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) pydantic_core._pydantic_core.ValidationError: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type 121811.171 WARNI… dynamic-rag-agent Failed to inject RAG context into conversation: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agents/dynamic_ra… line 1153, in _perform_rag_lookup_and_inject context_msg = LLMChatMessage(role=“assistant”, content=context_message) File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/… line 250, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) pydantic_core._pydantic_core.ValidationError: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type {“room”: “voice_assistant_room_7034", “agent_type”: “dynamic_rag_agent”, “agent_id”: “693bd39146235de7dc27dda5", “pid”: 75219, “job_id”: “AJ_omY7SBvfb5z5”, “room_id”: “RM_mkNnDUyr2b8Y”} Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agents/dynamic_rag_agent.py”, line 1153, in _perform_rag_lookup_and_inject context_msg = LLMChatMessage(role=“assistant”, content=context_message) File “/Users/rajan/Projects/CMS_VA_Agent_Python/.venv/lib/python3.10/site-packages/pydantic/main.py”, line 250, in init validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self) pydantic_core._pydantic_core.ValidationError: 1 validation error for ChatMessage content Input should be a valid list [type=list_type, input_value=‘Retrieved information fo...eamless career pathway.’, input_type=str] For further information visit https://errors.pydantic.dev/2.12/v/list_type I want the

turn_ctx

to be there how can we fix this so that i can achieve the same functionality in text input as well that is working perfectly for voice input

Thanks — the error shows

ChatMessage.content

expects a list, not a plain string. Two things to do: 1) Build the assistant message as a list of message parts. 2) For text input, pass a turn-like context so injection happens the same way as voice. Suggested fixes (drop-in changes): A) Fix message construction in `_perform_rag_lookup_and_inject`: • Replace the single-string construction with a list (one element) or the proper ChatMessage parts type. Example:

from livekit.agents.llm import ChatMessage as LLMChatMessage

# previous: context_msg = LLMChatMessage(role="assistant", content=context_message)
context_msg = LLMChatMessage(role="assistant", content=[context_message])

(or construct parts if your ChatMessage expects dicts/objects for rich content:

context_msg = LLMChatMessage(role="assistant", content=[{"type":"text","text":context_message}])

) B) Ensure injection for text input uses a turn-like context: • Use the text callback closure to create a minimal turn context wrapper that exposes

add_message

, or inject directly into the session chat context. Option 1 — inject into session.chat_ctx (preferred if available):

# in make_text_input_cb(agent)
async def _do_rag_and_reply(session, query):
    await agent._perform_rag_lookup_and_inject(query, turn_ctx=None)
    # after injection ensure LLM sees it: add to session.chat_ctx if available
    if hasattr(session, "chat_ctx") and session.chat_ctx:
        # session.chat_ctx.add_message expects role/content list
        session.chat_ctx.add_message(role="assistant", content=[context_message])
    session.generate_reply(user_input=query)

asyncio.create_task(_do_rag_and_reply(session, query))

Option 2 — provide a small turn_ctx shim with add_message so your existing injection code can reuse it:

class _TextTurnCtx:
    def __init__(self, session):
        self.session = session
    def add_message(self, role, content):
        # ensure content is a list
        if isinstance(content, str):
            content = [content]
        # adapt to your session/chat_ctx API
        if hasattr(self.session, "chat_ctx") and self.session.chat_ctx:
            self.session.chat_ctx.add_message(role=role, content=content)

# usage inside text callback
turn_ctx = _TextTurnCtx(session)
asyncio.create_task(agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx))
asyncio.create_task(session.generate_reply(user_input=query))

C) Small checklist • Ensure

_perform_rag_lookup_and_inject

builds

ChatMessage

with

content

as a list. • In text callback, call injection with either

turn_ctx=None

and then add the injected message to

session.chat_ctx

, or pass a shim

turn_ctx

that implements

add_message

. • If you await RAG before replying, use an async wrapper (as discussed earlier) so the reply includes injected context. If you want, paste the exact

_perform_rag_lookup_and_inject

function and I’ll return a precise patch (with the correct ChatMessage shape and a text-callback shim) you can paste into your file.

What is the best practice to achieve this you got my point, now let me know.

Best practice to make RAG work the same for voice and text — concise plan you can apply now: 1) Use on_user_turn_completed as the single RAG hook • on_user_turn_completed runs for both STT (voice) and text messages. Perform your RAG lookup there and inject context into the conversation before the LLM reply. This avoids separate race conditions between text callbacks and session lifecycle. 2) Inject valid ChatMessage content • ChatMessage.content expects a list. Always pass content as a list of text parts: ◦ Good: content=[ "Retrieved ...context..." ] ◦ Or for richer parts: content=[{"type":"text","text": "..."}] 3) If you still need immediate text callbacks (optional) • Use text_input_cb only for preprocessing or commands. If you use it to run RAG, do NOT duplicate injection logic — either: ◦ Call agent._perform_rag_lookup_and_inject from the text callback but inject into session.chat_ctx (not into turn_ctx) OR ◦ Make the text callback minimal and rely on on_user_turn_completed for injection. 4) How to pass a turn-like context for text input (recommended approach) • Preferred: let LiveKit normalize text into the chat lifecycle and handle RAG in on_user_turn_completed — easiest and most robust. • If you must inject before reply from text_input_cb, create a small shim that implements add_message and forwards to session.chat_ctx, then pass that shim to your existing injection function. Example shim:

class TextTurnCtx:
      def __init__(self, session):
          self.session = session
      def add_message(self, role, content):
          if isinstance(content, str):
              content = [content]
          if hasattr(self.session, "chat_ctx") and self.session.chat_ctx:
              self.session.chat_ctx.add_message(role=role, content=content)

In text_input_cb: • turn_ctx = TextTurnCtx(session) • await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx) • then session.generate_reply(user_input=query) 5) Ensure async flow and ordering • If reply must include injected RAG content, await the RAG call before calling session.generate_reply. Use an async wrapper inside text_input_cb:

async def run_and_reply():
      await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
      session.generate_reply(user_input=query)
  asyncio.create_task(run_and_reply())

• If you prefer non-blocking injection, fire-and-forget RAG + immediate session.generate_reply may race and the reply may not include injected context. 6) Validate/normalize RAG output • Ensure rag_snippets is string or list; convert to a safe string and then to list when constructing ChatMessage: ◦ safe_text = "\n".join(rag_snippets) or str(rag_snippets) ◦ context_msg = LLMChatMessage(role="assistant", content=[safe_text]) 7) Keep single source of truth • Centralize injection logic in _perform_rag_lookup_and_inject and call it from on_user_turn_completed. Use text_input_cb only when you need early interception, but still call the same helper (via shim or session.chat_ctx) to avoid duplication. If you want, I’ll: • Provide a ready patch that (a) updates _perform_rag_lookup_and_inject to build ChatMessage with list content, (b) shows on_user_turn_completed usage, and (c) adds a text_input_cb shim that awaits RAG and replies so text replies include injected context. Which would you like me to produce? Sources: External data and RAG | LiveKit docs | Agent server options | LiveKit docs

To achieve this give me one by one change in the files and functions

Good — I’ll give one-by-one, minimal changes you can apply to your code. Apply each step and test before moving to the next. Step 1 — Update _perform_rag_lookup_and_inject to produce list content and return the injected message • Replace the ChatMessage creation so content is a list. Also return the created context text so callers can reuse it. Change:

# old
context_msg = LLMChatMessage(role="assistant", content=context_message)

to:

# new
safe_text = str(context_message)
context_msg = LLMChatMessage(role="assistant", content=[safe_text])
return safe_text

Step 2 — Ensure on_user_turn_completed calls the helper and uses turn_ctx.add_message • In your existing on_user_turn_completed (voice hook), call the helper and add the returned text via turn_ctx.add_message. Change:

await self._perform_rag_lookup_and_inject(query, turn_ctx)

to:

injected_text = await self._perform_rag_lookup_and_inject(query, turn_ctx)
if injected_text:
    # ensure content is list when adding
    turn_ctx.add_message(role="assistant", content=[injected_text])

Step 3 — Add a TextTurnCtx shim for text callbacks • Add this small class near your helpers (top of file):

class TextTurnCtx:
    def __init__(self, session):
        self.session = session
    def add_message(self, role, content):
        if isinstance(content, str):
            content = [content]
        if hasattr(self.session, "chat_ctx") and self.session.chat_ctx:
            self.session.chat_ctx.add_message(role=role, content=content)

Step 4 — Add make_text_input_cb(agent) factory and use an async wrapper that awaits injection before replying • Add this near imports:

import asyncio
from livekit.agents.voice import room_io

def make_text_input_cb(agent):
    def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
        query = (event.text or "").strip()
        if not query:
            return

        async def run_and_reply():
            # create a turn-like ctx that injects into session.chat_ctx
            turn_ctx = TextTurnCtx(session)
            injected_text = await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
            # ensure agent reply uses injected context
            session.generate_reply(user_input=query)

        asyncio.create_task(run_and_reply())

    return text_input_cb

Step 5 — Wire the text callback into session.start • Replace your session.start call’s room_options with:

await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(...),
        text_input=room_io.TextInputOptions(
            text_input_cb=make_text_input_cb(agent)
        ),
    ),
)

Step 6 — Remove duplicate injection paths • Ensure you do NOT separately add the same assistant message elsewhere for the same turn to avoid duplication. Keep injection centralized in _perform_rag_lookup_and_inject and the voice hook. Step 7 — Validate rag_snippets shape inside helper • Inside _perform_rag_lookup_and_inject, normalize `rag_snippets`:

if isinstance(rag_snippets, list):
    safe_text = "\n".join(map(str, rag_snippets))
else:
    safe_text = str(rag_snippets)

Use

safe_text

when building the ChatMessage content list. Step 8 — Test synchronous ordering • With the async wrapper in the text callback you’ll await injection before calling

session.generate_reply

, ensuring the reply includes injected context. Test with a text message and confirm the assistant response contains injected context. Step 9 — Logging and error handling • Wrap network/model calls in try/except and log failures but continue gracefully. Confirm you don’t swallow exceptions silently. Step 10 — (Optional) If you want fire-and-forget behaviour • If you prefer not to delay reply, call

_perform_rag_lookup_and_inject

with

asyncio.create_task

and call

session.generate_reply

immediately — note this may race and the reply might not include injected context. If you want, I can produce a single patched file showing all edits applied to your provided agent.py. Which would you prefer: the patch or the full updated file?

But step 2 is already working for voice input why we are changing it to like this?

Good question — if voice path is already working, you do NOT need to change the voice hook’s logic. Step 2 in the previous plan was intended to make the injection explicit, but since your voice flow already calls the helper and injection succeeds, keep it as-is. The only changes required are the minimal ones that let the same helper be reused for text input without breaking voice: • Do not modify your existing working voice path. • Make these minimal edits instead: a. Update _perform_rag_lookup_and_inject to construct ChatMessage content as a list (this fixes the pydantic error you saw). Example change: • old:

LLMChatMessage(role="assistant", content=context_message)

• new:

LLMChatMessage(role="assistant", content=[str(context_message)])

• Also normalize rag_snippets to a string first (join lists). a. Add the TextTurnCtx shim and make_text_input_cb(agent) (as previously provided). In the text callback, create a TextTurnCtx(session) and call the same helper: •

await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)

• then

session.generate_reply(user_input=query)

This ensures text injection uses the same injection API that your voice path uses (so no duplicate logic in voice). 1. Wire make_text_input_cb(agent) into your existing session.start room_options (no change to voice code). Why this is minimal and safe • Voice already uses the helper and works — keep that untouched. • The failure you saw came from ChatMessage.content expecting a list; fixing that in the helper ensures both voice and text injections succeed. • The text callback only needs to create a lightweight shim that forwards add_message to session.chat_ctx, then call the same helper and trigger reply. That reuses your verified voice logic and preserves ordering when you await the helper before generating the reply. If you want, I’ll produce the exact three-line patch for _perform_rag_lookup_and_inject and the precise insertion points for TextTurnCtx and make_text_input_cb so you can drop them into your file. Which would you prefer?

Yes give me the changes to handle the text input I need to make it work only as voice input is working fine for me

Do only these minimal edits — voice flow unchanged. 1) Update _perform_rag_lookup_and_inject: ensure rag_snippets -> safe string and ChatMessage content is a list. Replace the ChatMessage creation with this snippet:

# normalize rag_snippets -> safe_text
if isinstance(rag_snippets, list):
    safe_text = "\n".join(map(str, rag_snippets))
else:
    safe_text = str(rag_snippets)

# create ChatMessage with content as a list
from livekit.agents.llm import ChatMessage as LLMChatMessage
context_msg = LLMChatMessage(role="assistant", content=[safe_text])

# existing code that adds/injects context can use context_msg as before
# return safe_text if you want the caller to re-use it (optional)
return safe_text

2) Add a small TextTurnCtx shim (place near other helpers, top of file):

class TextTurnCtx:
    def __init__(self, session):
        self.session = session

    def add_message(self, role, content):
        # ensure content is a list
        if isinstance(content, str):
            content = [content]
        if hasattr(self.session, "chat_ctx") and self.session.chat_ctx:
            # chat_ctx.add_message expects role and list content
            self.session.chat_ctx.add_message(role=role, content=content)

3) Add the text callback factory (near imports / helpers):

import asyncio
from livekit.agents.voice import room_io

def make_text_input_cb(agent):
    def text_input_cb(session: "AgentSession", event: room_io.TextInputEvent):
        query = (event.text or "").strip()
        if not query:
            return

        async def run_and_reply():
            # create shim so helper injects same way as voice
            turn_ctx = TextTurnCtx(session)
            await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
            # after injection, trigger the reply so LLM sees injected context
            session.generate_reply(user_input=query)

        asyncio.create_task(run_and_reply())

    return text_input_cb

4) Wire the callback into session.start (replace your room_options block):

await session.start(
    agent=agent,
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            noise_cancellation=...,
        ),
        text_input=room_io.TextInputOptions(
            text_input_cb=make_text_input_cb(agent)
        ),
    ),
)

5) Test and verify • Send a text message and confirm the assistant reply includes injected RAG context. • If you see duplicate injections, ensure voice hook still uses the existing path only. Notes: • This reuses your working voice injection API; no change to voice code required. • The async wrapper awaits injection before calling

session.generate_reply

, so the reply will include the injected context. • Keep ChatMessage content as a list to satisfy validation. Want me to produce a single patched file with these exact insertions?

# existing code that adds/injects context can use context_msg as before What was this in step 1

That comment meant: keep whatever you already do today to add/inject the assistant message into the conversation — just build

context_msg

as a valid ChatMessage (content must be a list) and then use your existing injection code (the same code path that already works for voice) to insert it. Concretely: • Replace the old ChatMessage creation with:

if isinstance(rag_snippets, list):
      safe_text = "\n".join(map(str, rag_snippets))
  else:
      safe_text = str(rag_snippets)

  context_msg = LLMChatMessage(role="assistant", content=[safe_text])

• Then call your existing injection logic exactly as you do for voice input (e.g.,

turn_ctx.add_message(...)

or

self._chat_ctx

handling). No other change to the voice path is required — only ensure the message object you hand into that existing code uses

content=[...]

.

Now its working but LLM is not considering the rag context injected in text input I don’t know why

Glad it’s mostly working — let’s troubleshoot why the LLM isn’t using the injected RAG context for text inputs. Likely cause: reply generation happens before the injected message is visible to the LLM. Here are focused checks and fixes to make replies include the injected context. Step A — Confirm injection completed before reply • Ensure your text callback awaits injection before calling

session.generate_reply

. Use the async wrapper pattern (this must await the helper, not fire-and-forget):

async def run_and_reply():
    turn_ctx = TextTurnCtx(session)
    await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
    session.generate_reply(user_input=query)

asyncio.create_task(run_and_reply())

If you currently call

asyncio.create_task(agent._perform_rag_lookup_and_inject(...))

then call

session.generate_reply(...)

immediately, that races and the LLM will not see injected context. Step B — Verify injection landed in the session chat context the LLM uses • After injection, inspect session.chat_ctx (or whatever context object your voice path uses) before calling generate_reply:

# debug right after inject
if hasattr(session, "chat_ctx"):
    print("chat_ctx messages:", session.chat_ctx.messages)

Confirm the injected assistant message appears in that list and is the last assistant message before generate_reply. Step C — Ensure message shape and roles match LLM expectations • Chat messages should be role="assistant" and content a list of text parts (you already fixed this). Verify the LLM integration reads from the same chat_ctx/messages sequence as voice flow. Step D — Avoid duplicate or misplaced context • Do not inject the context after calling generate_reply. • Do not add the context to a different object than the session uses for generation (e.g., self._chat_ctx vs session.chat_ctx). Use the same object the voice path uses. Step E — If your helper mutates state asynchronously, await those mutations • If _perform_rag_lookup_and_inject calls other async functions that modify chat_ctx, ensure they complete and the function returns only after chat_ctx is updated. Step F — Quick debug experiment (guarantee ordering) • Temporarily add a small sleep between injection return and reply to confirm timing is the factor:

await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
await asyncio.sleep(0.1)
session.generate_reply(user_input=query)

If this makes the RAG appear in replies, timing was the issue — increase reliability by awaiting properly rather than sleeping. Step G — If still not working, log these two snapshots for a failing text request and paste them here: • The chat_ctx.messages right before generate_reply. • The output of

_perform_rag_lookup_and_inject

(safe_text) you injected. I can then inspect ordering and message shapes and give a precise patch. Which do you want to try first — change to awaiting injection (Step A) or the quick debug snapshot (Step G)?

class TextTurnCtx: def *__init__*(self, session): self.session = session def *add_message*(self, role, content): # ensure content is a list if *isinstance*(content, str): content = [content] if *hasattr*(self.session, “chat_ctx”) and self.session.chat_ctx: _# chat_ctx.add_message expects role and list content_ self.session.chat_ctx.add_message(role=role, content=content) # Import our agent factory and related classes from agents import AgentFactory logger = logging.getLogger(“agent”) load_dotenv(“.env.local”) async def *record_room_to_s3*(room_name: str): # Initialize LiveKit API client lkapi = api.LiveKitAPI() # Configure S3 upload s3_upload = api.S3Upload( access_key=os.environ.get(‘AWS_ACCESS_KEY_ID’), secret=os.environ.get(‘AWS_SECRET_ACCESS_KEY’), bucket=os.environ.get(‘S3_RECORDING_BUCKET’), region=os.environ.get(‘AWS_DEFAULT_REGION’, ‘ap-south-1’) ) # Start room composite recording egress_request = api.RoomCompositeEgressRequest( room_name=room_name, layout=‘speaker’, # or ‘grid’ layout preset=api.EncodingOptionsPreset.H264_720P_30, audio_only=True, file_outputs=[api.EncodedFileOutput( file_type=api.EncodedFileType.MP3, filepath=f’recordings/{room_name}/{room_name}.mp3', s3=s3_upload )] ) # Start the egress egress_info = await lkapi.egress.start_room_composite_egress(egress_request) *print*(f’Egress started: {egress_info.egress_id}’) await lkapi.aclose() return f”recordings/{room_name}/{room_name}.mp3" if egress_info else “” def *prewarm*(proc: JobProcess): proc.userdata[“vad”] = silero.VAD.load() # load function: return value 0.0..1.0 def *load_fnc*(): cpu = psutil.cpu_percent(interval=0.1) / 100.0 mem = psutil.virtual_memory().percent / 100.0 # weighted combination (tune as needed) load = 0.7 * cpu + 0.3 * mem return *max*(0.0, *min*(1.0, load)) # text input callback def *make_text_input_cb*(agent): def *text_input_cb*(session: “AgentSession”, event: room_io.TextInputEvent): query = (event.text or “”).strip() if not query: return async def *run_and_reply*(): # create shim so helper injects same way as voice turn_ctx = TextTurnCtx(session) await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx) # after injection, trigger t he reply so LLM sees injected context logger.info(f”Triggering reply after injection for query: {query}“) session.generate_reply(user_input=query) asyncio.create_task(run_and_reply()) return text_input_cb async def *_perform_rag_lookup_and_inject*( self, query: str, turn_ctx: Optional[ChatContext] = None ) -> None: “”" Helper method to perform RAG lookup and inject context into conversation. Can be called from both voice (on_user_turn_completed) and text input handlers. Args: query: The user’s query text turn_ctx: Optional ChatContext (available for voice input, None for text input) “”" if not self.rag_enabled: logger.info(f”RAG is not enabled, skipping RAG lookup”) return # Ensure query is a string and not empty if not query or not *isinstance*(query, str) or not query.strip(): return query = query.strip() logger.info(f”RAG lookup query: {query}“) _# Perform RAG lookup using the same logic as maica_knowledge_search_ rag_snippets = await self.my_rag_lookup(query) if not rag_snippets: return # Inject concise retrieved context for the LLM # Format as an assistant message with the retrieved information context_message = f”Retrieved information for context:\n{rag_snippets}” logger.info(f”Context message: {context_message}“) try: _# Try to add message using turn_ctx.add_message (available for voice input)_ if turn_ctx and *hasattr*(turn_ctx, ‘add_message’): logger.info(f”Entered here because turn_ctx found”) turn_ctx.add_message( role=“assistant”, content=context_message ) logger.info(f”Injected RAG context into conversation for query: {query[:100]}“) # Fallback: Try to update chat context directly if available elif *hasattr*(self, ‘_chat_ctx’) and self._chat_ctx: _# normalize rag_snippets -> safe_text_ logger.info(f”Entered here because _chat_ctx found”) if *isinstance*(rag_snippets, list): safe_text = “\n”.join(*map*(str, rag_snippets)) else: safe_text = str(rag_snippets) # create ChatMessage with content as a list from livekit.agents.llm import ChatMessage as LLMChatMessage context_msg = LLMChatMessage(role=“assistant”, content=[safe_text]) _# existing code that adds/injects context can use context_msg as before_ _# return safe_text if you want the caller to re-use it (optional)_ return safe_text else: logger.warning(“Could not inject RAG context: no turn_ctx.add_message or _chat_ctx available”) return None except Exception as e: logger.warning(f”Failed to inject RAG context into conversation: {e}“, exc_info=True) This is the code i am using

Thanks — I reviewed your code. The issue: when

turn_ctx

path runs you call

turn_ctx.add_message(...)

but don’t return the injected text; when

_chat_ctx

path you build

context_msg

but never add it to the chat context. Make these minimal fixes so both paths actually inject the assistant message and return the safe text. Apply these exact changes. 1) Replace the try block in

_perform_rag_lookup_and_inject

with this (keeps your logging):

try:
    # If we have a turn context (voice or shim), use it and return the injected text
    if turn_ctx and hasattr(turn_ctx, "add_message"):
        <http://logger.info|logger.info>("Entered here because turn_ctx found")
        # ensure content is list when adding
        turn_ctx.add_message(role="assistant", content=[context_message])
        <http://logger.info|logger.info>(f"Injected RAG context into conversation for query: {query[:100]}")
        return context_message

    # Fallback: add into agent/session chat context and return text
    elif hasattr(self, "_chat_ctx") and self._chat_ctx:
        <http://logger.info|logger.info>("Entered here because _chat_ctx found")
        if isinstance(rag_snippets, list):
            safe_text = "\n".join(map(str, rag_snippets))
        else:
            safe_text = str(rag_snippets)

        from livekit.agents.llm import ChatMessage as LLMChatMessage
        context_msg = LLMChatMessage(role="assistant", content=[safe_text])

        # Add the message into the chat context used by the LLM
        # Adapt depending on your chat_ctx API (this is common form)
        self._chat_ctx.add_message(role="assistant", content=[safe_text])

        <http://logger.info|logger.info>(f"Injected RAG context via _chat_ctx for query: {query[:100]}")
        return safe_text

    else:
        logger.warning("Could not inject RAG context: no turn_ctx.add_message or _chat_ctx available")
        return None

except Exception as e:
    logger.warning(f"Failed to inject RAG context into conversation: {e}", exc_info=True)
    return None

2) Keep the earlier normalization you already have for

context_message

/

safe_text

(ensure

context_message

is a string). If you prefer, compute

safe_text

once at top and use that variable in both paths. 3) In your text callback wrapper you already await

agent._perform_rag_lookup_and_inject(...)

and then call

session.generate_reply(user_input=query)

. After making the above change, verify (for one text message) that: • The returned value from

_perform_rag_lookup_and_inject

is non-None, and • Inspect

session.chat_ctx.messages

(or log it) immediately before

session.generate_reply

to confirm the assistant message is present. If after these changes the LLM still ignores the context, run this quick debug inside your

run_and_reply

before `session.generate_reply`:

# debug
if hasattr(session, "chat_ctx"):
    <http://logger.info|logger.info>("chat_ctx messages before reply: %s", session.chat_ctx.messages)

Send me that

chat_ctx.messages

output if it still doesn’t appear in the LLM reply and I’ll advise the next step.

but what is the issue if we are not returning context_message and in text input as well i am getting this log Entered here because turn_ctx found beacuse now we are passing turn_ctx from the agent file here: # create shim so helper injects same way as voice turn_ctx = TextTurnCtx(session) await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)

Good question — returning the context text from

_perform_rag_lookup_and_inject

is optional. The real requirement is that the assistant message actually gets added to the chat context the LLM reads, and it must have the correct shape (content as a list). Since your logs show "Entered here because turn_ctx found", the shim path is being executed — so focus on these three checks/fixes (apply in order, test each): 1) Verify TextTurnCtx.add_message sends a list • Your shim must convert string -> list before calling session.chat_ctx.add_message. • Confirm the method is exactly:

def add_message(self, role, content):
    if isinstance(content, str):
        content = [content]
    if hasattr(self.session, "chat_ctx") and self.session.chat_ctx:
        self.session.chat_ctx.add_message(role=role, content=content)

If you already have that, OK. 2) Ensure the helper actually calls turn_ctx.add_message with a list (not a raw string) • In

_perform_rag_lookup_and_inject

when

turn_ctx

exists, call:

turn_ctx.add_message(role="assistant", content=[safe_text])

(or ensure

turn_ctx.add_message

will convert a string to list — do one of the two, not rely on both). 3) Confirm injection is visible to the session before generating reply • In your text callback wrapper, after awaiting the helper and before

session.generate_reply

, log the chat messages to confirm injection:

injected = await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
<http://logger.info|logger.info>("injected returned: %s", injected)
if hasattr(session, "chat_ctx"):
    <http://logger.info|logger.info>("chat_ctx messages before reply: %s", session.chat_ctx.messages)
session.generate_reply(user_input=query)

If

chat_ctx.messages

contains the assistant message you injected (with role="assistant" and content list), then injection succeeded and the problem is timing or message ordering in the LLM call. If injection is present in chat_ctx.messages but the LLM still ignores it: • Make sure the injected assistant message appears before the LLM constructs its prompt (logging above will show ordering). • If your voice flow uses a different context object than

session.chat_ctx

(e.g., self._chat_ctx), inject into the exact same object the voice path uses. Summary of likely causes and fixes: • Cause A: shim added string not list -> fix (step 2). • Cause B: helper injected but into different context object -> inject into same chat_ctx used by LLM. • Cause C: generate_reply called before injection completed -> await helper (you already await) and confirm via logging. If you paste the output of the two debug logs above (the injected return value and session.chat_ctx.messages just before reply), I’ll pinpoint whether it’s a shape, location, or timing problem and give the exact one-line fix.

if *hasattr*(session, “chat_ctx”): logger.info(“chat_ctx messages before reply: %s”, session.chat_ctx.messages) I don’t get this log

Thanks — since you don't see the log, session likely doesn't have

chat_ctx

. Let's find where the LLM reads messages and confirm where to inject and verify. Do this quick three-step check and paste the outputs: 1) Confirm what attributes the session has (add this right after awaiting the injection and before generate_reply):

<http://logger.info|logger.info>("session attrs: %s", sorted(dir(session)))

2) If

chat_ctx

not present, check

_agent

/

_chat_ctx

/ `session._agent`:

<http://logger.info|logger.info>("has session.chat_ctx=%s, has session._agent=%s", hasattr(session, "chat_ctx"), hasattr(session, "_agent"))
if hasattr(session, "_agent"):
    agent_obj = session._agent
    <http://logger.info|logger.info>("agent has _chat_ctx=%s, agent has _session=%s", hasattr(agent_obj, "_chat_ctx"), hasattr(agent_obj, "session"))
    if hasattr(agent_obj, "_chat_ctx"):
        <http://logger.info|logger.info>("agent._chat_ctx.messages: %s", getattr(agent_obj, "_chat_ctx").messages)

3) Log the return from the injector and list of messages after injection:

injected = await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
<http://logger.info|logger.info>("injected returned: %s", injected)
# then enumerate likely places
if hasattr(session, "chat_ctx"):
    <http://logger.info|logger.info>("session.chat_ctx.messages: %s", session.chat_ctx.messages)
if hasattr(session, "_agent") and hasattr(session._agent, "_chat_ctx"):
    <http://logger.info|logger.info>("session._agent._chat_ctx.messages: %s", session._agent._chat_ctx.messages)
if hasattr(agent, "_chat_ctx"):
    <http://logger.info|logger.info>("agent._chat_ctx.messages: %s", agent._chat_ctx.messages)

Run one text message and paste the logged outputs. From those I will tell you exactly which object to inject into (and give the one-line replacement to make the LLM see the injected context).

- ERROR - Task exception was never retrieved future: <Task finished name=‘Task-114’ coro=<make_text_input_cb.<locals>.text_input_cb.<locals>.run_and_reply() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py:107> exception=AttributeError(“‘ChatContext’ object has no attribute ‘messages’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 116, in run_and_reply logger.info(“session._agent._chat_ctx.messages: %s”, session._agent._chat_ctx.messages) AttributeError: ‘ChatContext’ object has no attribute ‘messages’ Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 116, in run_and_reply logger.info(“session._agent._chat_ctx.messages: %s”, session._agent._chat_ctx.messages) AttributeError: ‘ChatContext’ object has no attribute ‘messages’ ERROR asyncio Task exception was never retrieved future: <Task finished name=‘Task-114’ coro=<make_text_input_cb.<locals>.text_input_cb.<locals>.run_and… done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py:107> exception=AttributeError(“‘ChatContext’ object has no attribute ‘messages’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 116, in run_and_reply logger.info(“session._agent._chat_ctx.messages: %s”, session._agent._chat_ctx.messages) AttributeError: ‘ChatContext’ object has no attribute ‘messages’ {“room”: “voice_assistant_room_1458", “agent_type”: “dynamic_rag_agent”, “agent_id”: “693bd39146235de7dc27dda5", “pid”: 85188, “job_id”: “AJ_MExcoGhU5UVA”, “room_id”: “RM_3gnihQtEqczz”} Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 116, in run_and_reply logger.info(“session._agent._chat_ctx.messages: %s”, session._agent._chat_ctx.messages) AttributeError: ‘ChatContext’ object has no attribute ‘messages’

Thanks — you’ve confirmed session lacks a usable

.messages

property and the text path runs but the LLM still ignores injected context. Next steps (minimal, actionable) to get text RAG working exactly like voice: 1) Confirm injection target — replace any use of

.messages

with

get_messages()

and prefer the public

agent.chat_ctx

APIs. • Debug log (insert in your text handler after await inject):

<http://logger.info|logger.info>("agent.chat_ctx.get_messages(): %s", agent.chat_ctx.get_messages())

2) Ensure the helper actually adds the assistant message into the same chat context the LLM uses: • In

_perform_rag_lookup_and_inject

, when using

turn_ctx

(TextTurnCtx), call:

turn_ctx.add_message(role="assistant", content=[safe_text])

• If falling back to agent/session chat context, call:

agent.chat_ctx.add_message(role="assistant", content=[safe_text])

(or

await agent.update_chat_ctx(...)

if your SDK requires updating) 3) Make the text callback await injection before generating the reply (guarantees ordering): • Use this pattern (replace existing run_and_reply):

async def run_and_reply():
    turn_ctx = TextTurnCtx(session)
    injected = await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
    <http://logger.info|logger.info>("injected returned: %s", injected)
    <http://logger.info|logger.info>("agent.chat_ctx.get_messages(): %s", agent.chat_ctx.get_messages())
    await session.generate_reply()  # await so LLM call uses updated chat_ctx

• Then schedule:

asyncio.create_task(run_and_reply())

4) Verify message shape: ChatMessage content must be a list — you already produce

[safe_text]

. Keep that. 5) If agent.chat_ctx is not the right object in your codebase, find where voice flow reads context and inject there. Quick checks to run if still failing: • Log

sorted(dir(agent))

and

sorted(dir(session))

to find

.chat_ctx

vs

_chat_ctx

vs

update_chat_ctx

. • After injection and before reply, log

agent.chat_ctx.get_messages()

(or the equivalent object from voice flow). 6) If injection appears in

get_messages()

but LLM still ignores it: • Confirm the assistant message ordering is before model prompt (logging above will show). • If your LLM call uses a snapshot of context created earlier, call

await agent.update_chat_ctx(agent.chat_ctx)

(if available) to ensure the runtime uses the latest context before

generate_reply()

. If you want, I can: • Produce a 1-file patch that (A) replaces the add_message calls to use

[safe_text]

, (B) updates text callback to the awaited run_and_reply pattern, and (C) adds the debug logs above. Say "patch please" and I’ll generate it.

2025-12-31 134502,271 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name=‘Task-114’ coro=<make_text_input_cb.<locals>.text_input_cb.<locals>.run_and_reply() done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py:107> exception=AttributeError(“‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 112, in run_and_reply logger.info(“agent.chat_ctx.get_messages(): %s”, agent.chat_ctx.get_messages()) AttributeError: ‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’. Did you mean: ‘add_message’? Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 112, in run_and_reply logger.info(“agent.chat_ctx.get_messages(): %s”, agent.chat_ctx.get_messages()) AttributeError: ‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’. Did you mean: ‘add_message’? ERROR asyncio Task exception was never retrieved future: <Task finished name=‘Task-114’ coro=<make_text_input_cb.<locals>.text_input_cb.<locals>.run_and… done, defined at /Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py:107> exception=AttributeError(“‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’“)> Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 112, in run_and_reply logger.info(“agent.chat_ctx.get_messages(): %s”, agent.chat_ctx.get_messages()) AttributeError: ‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’. Did you mean: ‘add_message’? {“room”: “voice_assistant_room_3245", “agent_type”: “dynamic_rag_agent”, “agent_id”: “693bd39146235de7dc27dda5", “pid”: 92185, “job_id”: “AJ_oiCECRTEW7MP”, “room_id”: “RM_nbWE75VHKJA4”} Traceback (most recent call last): File “/Users/rajan/Projects/CMS_VA_Agent_Python/src/agent.py”, line 112, in run_and_reply logger.info(“agent.chat_ctx.get_messages(): %s”, agent.chat_ctx.get_messages()) AttributeError: ‘_ReadOnlyChatContext’ object has no attribute ‘get_messages’. Did you mean: ’add_me

Good find —

_ReadOnlyChatContext

doesn't expose

get_messages()

. Use the public API instead and inject into the writable chat context the LLM uses. Do these minimal changes to fix logging and ensure the LLM sees injected context for text input. 1) Replace debug that called

get_messages()

with a safe inspection:

# safe inspect — works for ReadOnly or writable chat contexts
def _dump_chat_ctx(ctx):
    try:
        # try iterable access (works for both)
        msgs = list(ctx)
    except Exception:
        # fallback to str() for safety
        msgs = str(ctx)
    <http://logger.info|logger.info>("chat_ctx snapshot: %s", msgs)

2) In your text handler use the public

agent.chat_ctx

methods (do not rely on private

_chat_ctx

or ReadOnly internals). Await injection, then log with the safe inspector before generating reply:

async def run_and_reply():
    turn_ctx = TextTurnCtx(session)
    injected = await agent._perform_rag_lookup_and_inject(query, turn_ctx=turn_ctx)
    <http://logger.info|logger.info>("injected returned: %s", injected)

    # inspect whichever context object is available
    if hasattr(agent, "chat_ctx"):
        _dump_chat_ctx(agent.chat_ctx)
    elif hasattr(session, "_agent") and hasattr(session._agent, "chat_ctx"):
        _dump_chat_ctx(session._agent.chat_ctx)

    # ensure the LLM uses updated context
    await session.generate_reply()

3) Ensure the shim adds list content (you already do):

turn_ctx.add_message(role="assistant", content=[safe_text])

. 4) If

agent.chat_ctx

is read-only at that point, update the agent state via the public update API before generating reply (use if available):

# if SDK exposes update_chat_ctx
await agent.update_chat_ctx(agent.chat_ctx)

or add via public method:

agent.chat_ctx.add_message(role="assistant", content=[safe_text])

5) Quick verification steps: • Send one text message. • Check logs: "injected returned" and "chat_ctx snapshot" should show the assistant message. • Confirm reply includes injected content. If you paste the output of the new "chat_ctx snapshot" log (or confirm which of

agent.chat_ctx

vs

session._agent.chat_ctx

printed), I’ll give the exact one-line injection or update call to make the LLM consume the injected context.

chat_ctx snapshot: <livekit.agents.llm.chat_context._ReadOnlyChatContext object at 0x13cf34af0> INFO agent chat_ctx snapshot: <livekit.agents.llm.chat_context._ReadOnlyChatContext object at 0x13cf34af0> got this in logs

Thanks — the snapshot shows

agent.chat_ctx

is a

_ReadOnlyChatContext

. You need to add the injected assistant message to the writable chat context the LLM uses (not inspect the read-only object). Do these minimal steps now: 1) Inject into the agent's writable chat context • Replace the turn-shim add with adding to the agent's public

chat_ctx

(writable) before calling `generate_reply`:

# inside _perform_rag_lookup_and_inject when using turn_ctx:
# ensure safe_text is a string (normalized earlier)
turn_ctx.add_message(role="assistant", content=[safe_text])

# ALSO add to agent.chat_ctx so the LLM will see it for text input:
if hasattr(self, "chat_ctx"):
    # chat_ctx may be read-only; use its public add_message if available
    try:
        self.chat_ctx.add_message(role="assistant", content=[safe_text])
    except Exception:
        # fallback: if agent exposes update_chat_ctx, update it
        try:
            await self.update_chat_ctx(self.chat_ctx.copy().add_message(role="assistant", content=[safe_text]))
        except Exception:
            logger.warning("Failed to add to agent.chat_ctx directly")

2) Ensure text callback awaits injection then calls generate_reply • Your text handler should await

_perform_rag_lookup_and_inject

and only then call

await session.generate_reply()

(not fire-and-forget). You already do this — keep awaiting. 3) Add a final confirm/log right before reply (for debugging)

# in run_and_reply after await inject
try:
    # attempt to inspect via iteration (works for ReadOnly)
    msgs = list(agent.chat_ctx)
    <http://logger.info|logger.info>("chat_ctx (len=%d) top: %s", len(msgs), msgs[-3:])
except Exception:
    <http://logger.info|logger.info>("agent.chat_ctx type: %s", type(agent.chat_ctx))

4) If agent.chat_ctx is still effectively read-only and add_message fails, use the public update API: • Call

await agent.update_chat_ctx(new_ctx)

where

new_ctx

is a copy with your assistant message appended. Check your SDK for `update_chat_ctx`/`chat_ctx.copy()` methods. 5) Quick verification • Send a text message; check logs for: ◦ "injected returned" (non-empty) ◦ the

chat_ctx

iteration showing the injected assistant message ◦ the assistant reply should include the RAG content If you want, paste the small outputs after running one text message: • Result of

msgs = list(agent.chat_ctx)

(or the exception if iteration fails) • The returned value from

_perform_rag_lookup_and_inject

I’ll give the exact one-line change to add to your code (or produce a 3–5 line patch) based on that.

where we need to add this code

# ALSO add to agent.chat_ctx so the LLM will see it for text input:
if hasattr(self, "chat_ctx"):
    # chat_ctx may be read-only; use its public add_message if available
    try:
        self.chat_ctx.add_message(role="assistant", content=[safe_text])
    except Exception:
        # fallback: if agent exposes update_chat_ctx, update it
        try:
            await self.update_chat_ctx(self.chat_ctx.copy().add_message(role="assistant", content=[safe_text]))
        except Exception:
            logger.warning("Failed to add to agent.chat_ctx directly")

Add that code inside your

_perform_rag_lookup_and_inject

function, in the branch that handles

turn_ctx

(right after you call

turn_ctx.add_message(...)

). Concretely, locate this block:

if turn_ctx and hasattr(turn_ctx, "add_message"):
    turn_ctx.add_message(role="assistant", content=[safe_text])
    <http://logger.info|logger.info>("Injected RAG context into conversation for query: ...")
    return safe_text

Replace it with (or immediately follow it by):

if turn_ctx and hasattr(turn_ctx, "add_message"):
    turn_ctx.add_message(role="assistant", content=[safe_text])
    <http://logger.info|logger.info>("Injected RAG context into conversation for query: ...")

    # ALSO add to agent.chat_ctx so the LLM will see it for text input
    if hasattr(self, "chat_ctx"):
        try:
            self.chat_ctx.add_message(role="assistant", content=[safe_text])
        except Exception:
            try:
                # fallback: update chat_ctx with a copy that includes the new message
                new_ctx = self.chat_ctx.copy()
                new_ctx.add_message(role="assistant", content=[safe_text])
                await self.update_chat_ctx(new_ctx)
            except Exception:
                logger.warning("Failed to add to agent.chat_ctx directly")

    return safe_text

Notes: • Keep this inside

_perform_rag_lookup_and_inject

(which is an async function), so

await self.update_chat_ctx(...)

is valid. • Do not remove the existing

turn_ctx.add_message(...)

line — both additions are needed: the shim for session-side visibility and

agent.chat_ctx

for the LLM. • After this change, your text handler that awaits the helper and then calls

await session.generate_reply()

should produce replies that include the injected RAG content.