what we write in prewarm function, will they be shared across all sessions ? is it ?

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

if 2 sip calls are going on, both will have separate prewarm resources right ?

Thanks — good follow-up. You're asking whether two simultaneous SIP calls will each get separate prewarm resources. Short answer: It depends on how your processes are running. • If both SIP calls are handled by the same process, they will share the process-level prewarm resources (the objects you stored in

proc.userdata

). The

prewarm

function runs once per process, so expensive resources loaded there (models, caches, DB clients) are reused by both calls in that process. • If the SIP calls are handled by different processes (for scaling or isolation), each process runs its own

prewarm

and will have its own copy of those resources. In that case the calls do not share the same in-memory prewarm objects. If you’d like, tell me: • whether your deployment uses multiple processes or a single process, and • what resources you plan to prewarm (e.g., VAD model, codec, DB connection). I can then recommend best practices for sharing vs. isolating resources and for keeping memory/CPU usage predictable.

see this is my code

from datetime import datetime
import logging
import json
from typing import Any, Dict

from livekit.agents.voice.agent_session import SessionConnectOptions
import pytz
from livekit import agents, api
from livekit.agents import (
    APIConnectOptions,
    Agent,
    AgentSession,
    AgentStateChangedEvent,
    AudioConfig,
    BackgroundAudioPlayer,
    BuiltinAudioClip,
    JobProcess,
    MetricsCollectedEvent,
    RoomInputOptions,
    UserInputTranscribedEvent,
    UserStateChangedEvent,
    metrics,
)
from livekit.agents import tts
from livekit.agents.voice import room_io
from livekit.plugins import google, openai, sarvam, silero, elevenlabs, noise_cancellation
from cellassist.agents.main_agent import MainAgent
from cellassist.agents.single_agent import SingleAgent
from cellassist.core.aws.s3 import S3SessionStorage
from cellassist.core.common import get_logger
from cellassist.core.common.silence_timeout_handler import SilenceTimeoutHandler

logger = get_logger(__name__)

# Disable Sarvam STT logs
logging.getLogger("livekit.plugins.sarvam").setLevel(logging.WARNING)
logging.getLogger("livekit.plugins.sarvam.log").setLevel(logging.WARNING)
logging.getLogger("livekit.plugins.sarvam.speechstream").setLevel(logging.WARNING)

# Reduce boto3/botocore debug noise
logging.getLogger("boto3").setLevel(logging.WARNING)
logging.getLogger("botocore").setLevel(logging.WARNING)
logging.getLogger("boto3.resources.factory").setLevel(logging.WARNING)

def prewarm(proc: JobProcess):
    proc.userdata["llm"] = openai.LLM(model="gpt-4.1")
    proc.userdata["stt"] = sarvam.STT(
        language="unknown",
        model="saaras:v3",
        # mode="transcribe",
        mode="codemix",
        sample_rate=8000
    )
    sarvam_tts = sarvam.TTS(
        target_language_code="en-IN",
        model="bulbul:v3",
        # speaker="simran",
        speaker="ishita",
        # speaker="shreya",
        pace=1.18,
        speech_sample_rate=24000,
    )

    sarvam_2_tts = sarvam.TTS(
        target_language_code="en-IN",
        model="bulbul:v3",
        # speaker="simran",
        speaker="ishita",
        # speaker="shreya",
        pace=1.18,
        speech_sample_rate=24000,
    )

    proc.userdata["tts"] = tts.FallbackAdapter(
        tts=[sarvam_tts, sarvam_2_tts],
        max_retry_per_tts=2,
        sample_rate=24000
    )

    proc.userdata["vad"] = silero.VAD.load(
        min_silence_duration=0.35,  # from 0.55
        prefix_padding_duration=0.5,
        activation_threshold=0.5,
    )

async def entrypoint(ctx: agents.JobContext):

    metadata_dict: Dict[str, Any] = json.loads(ctx.job.metadata) if ctx.job.metadata else {}
    <http://logger.info|logger.info>("Metadata: %s", metadata_dict)

    call_type = metadata_dict.get("call_type", "web_call")          # web_call or phone_call

    await ctx.connect()
    <http://logger.info|logger.info>("Connected to the room")

    if call_type == "phone_call":

        <http://logger.info|logger.info>("Phone call detected")

        <http://logger.info|logger.info>("Waiting for participant to connect...")
        participant = await ctx.wait_for_participant()
        <http://logger.info|logger.info>("Participant attributes: %s", participant.attributes)

        sip_number = participant.attributes.get("sip.phoneNumber")[3:]
        <http://logger.info|logger.info>("SIP number: %s", sip_number)

    else:
        <http://logger.info|logger.info>("Web call detected")
        sip_number = "9765851707"
    
    # Initialize S3 storage for session data
    s3_storage = S3SessionStorage()
    
    # Use room name as session ID for consistent folder naming
    session_id = ctx.room.name

    session = AgentSession(
        stt = ctx.proc.userdata["stt"],
        llm = ctx.proc.userdata["llm"],
        tts = ctx.proc.userdata["tts"],
        vad = ctx.proc.userdata["vad"],
        # tts=tts_with_fallback,  # Use fallback adapter instead of sarvam.TTS directly
        
        turn_detection="vad",
        # turn_detection="stt",

        user_away_timeout=10.0,

        userdata={
            "sip_number": sip_number
        },
        max_tool_steps=15,
        tts_text_transforms=["filter_markdown"],

        conn_options=SessionConnectOptions(
            tts_conn_options=APIConnectOptions(max_retry=5, timeout=180),
        ),
        # preemptive_generation=True
    )

    current_datetime_str = (
        datetime.now()
        .astimezone(pytz.timezone("Asia/Kolkata"))
        .strftime("%Y-%m-%d %A %H:%M %p")
    )

    # Pass current_datetime to session for AgentTasks (e.g., slot selection)
    session.userdata["current_datetime"] = current_datetime_str

    if call_type == "web_call":
        <http://logger.info|logger.info>("Waiting for participant to connect...")
        participant = await ctx.wait_for_participant()
        <http://logger.info|logger.info>("Participant attributes: %s", participant.attributes)

    await session.start(
        room=ctx.room,
        agent=SingleAgent(
            current_datetime=current_datetime_str,
            sip_number=sip_number,
        ),
    )


    @session.on("user_state_changed")
    def _on_user_state_changed(ev: UserStateChangedEvent):
        <http://logger.info|logger.info>(f"User state changed: {ev.new_state}")

    @session.on("agent_state_changed")
    def _on_agent_state_changed(ev: AgentStateChangedEvent):
        <http://logger.info|logger.info>(f"Agent state changed: {ev.new_state}")

    @session.on("user_input_transcribed")
    def _on_user_input_transcribed(ev: UserInputTranscribedEvent):
        tag = "FINAL" if getattr(ev, "is_final", False) else "INTERIM"
        <http://logger.info|logger.info>(f"STT {tag} ts={ev.created_at} sid={getattr(ev,'sid',None)} text={ev.transcript!r}")

    usage_collector = metrics.UsageCollector()

    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent):
        # with open("metrics.jsonl", "a") as f:
        #     f.write(ev.metrics.model_dump_json() + "\n")

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

    async def log_usage():
        summary = usage_collector.get_summary()
        <http://logger.info|logger.info>(f"Usage: {summary}")

    # At shutdown, generate and log the summary from the usage collector
    ctx.add_shutdown_callback(log_usage)

    # Set up shutdown callbacks to save data
    async def save_session_data():
        """Save transcript and metrics to S3 on session end."""
        try:
            <http://s3_storage.logger.info|s3_storage.logger.info>(f"Saving session data for {session_id}...")
            
            # Get conversation history
            transcript_data = {
                "session_id": session_id,
                "room_name": ctx.room.name,
                "conversation_history": session.history.to_dict(),
                "session_start": current_datetime_str,
                "session_end": datetime.now().astimezone(pytz.timezone("Asia/Kolkata")).strftime("%Y-%m-%d %A %H:%M %p"),
                "saved_at": datetime.now().isoformat(),
                # "recording_egress_id": recording_egress_id
            }
            # <http://logger.info|logger.info>(f"Transcript data: {transcript_data}")

            # Save transcript
            transcript_path = s3_storage.save_session_transcript(session_id, transcript_data)

        except Exception as e:
            s3_storage.logger.error(f"Failed to save session data: {e}")

    # Add shutdown callback
    ctx.add_shutdown_callback(save_session_data)

Thanks — this is helpful context. I’m responding to your question about whether two simultaneous SIP calls will have separate prewarm resources, and I’ll reference your code. Short answer • Resources created in

prewarm

(like

proc.userdata["llm"]

,

["stt"]

,

["tts"]

,

["vad"]

) are shared by all sessions that run inside the same process. • If the two SIP calls are handled by two different processes, each process runs its own

prewarm

and will have its own copy of those resources. So “separate” depends on whether your calls land in different processes. How this applies to your code • You load LLM, STT, TTS, and VAD into

proc.userdata

in

prewarm

. Any AgentSession you create using

ctx.proc.userdata[...]

(as you do) will reuse those same instances when sessions run in the same process. • If your runtime spawns multiple processes (for scaling, container replicas, worker pools, etc.), each process will run

prewarm

independently and hold its own instances. Two calls handled by different processes will therefore not share those in-memory objects. Practical implications & recommendations • Reuse is good for expensive one-time initialization: your VAD/model loads reduce latency and memory churn when calls are served by the same process. • Watch memory/CPU if you have large models: multiple processes each holding copies can increase resource usage linearly with process count. • If you need strict per-call isolation for a particular resource, don’t store it in `proc.userdata`; instead initialize it in

entrypoint

(session-local) or wrap the resource so each session gets its own instance. • If you need cross-process sharing (one canonical model instance), consider an external service (model server, shared cache, or managed inference endpoint) rather than in-process sharing. If you want, tell me: • how you deploy workers (single process, multiple processes, Kubernetes replicas, etc.), and • which resources you’re most concerned about (LLM, STT, TTS, VAD), and I’ll suggest a specific architecture and memory/latency trade-offs for your setup.

livekit will generate separate process for every call . this is default behaviour i guess

Thanks — you're asking whether LiveKit spawns a separate process per call and whether that means each call gets its own prewarm resources. Short answer: • Yes — when each call runs in its own process, each process runs

prewarm

independently, so each call/process gets its own copy of the prewarmed resources you set on

proc.userdata

. What that means for your code: • In your

prewarm

, you load

llm

,

stt

,

tts

, and

vad

into

proc.userdata

. If LiveKit runs a separate process per call (your stated default), then those resources will not be shared across calls — each call/process will have its own instances. • If instead multiple calls ran inside the same process, they would share the same

proc.userdata

instances and reuse those resources. Practical implications and quick recommendations: • Isolation: Per-call processes give strong isolation (no accidental cross-call state) at the cost of higher memory/CPU because each process holds its own model instances. • Cost/Performance: If models are large and you want to save memory, consider using a shared external service or model server that processes can call, rather than loading the full model in-process per call. • If you truly want per-call instances (for isolation or thread-safety), your current design (relying on process-per-call) already achieves that. If you’d like, tell me: • whether you want to optimize memory usage or keep strict isolation, and I can suggest a concrete pattern (e.g., model server vs in-process preload vs session-local init) tailored to your needs.