how to minimise token usage
# ask-aib
boundless-author-7811
03/04/2026, 9:44 AMhow to minimise token usage
t
tall-airline-16756
03/04/2026, 9:44 AMThanks for reaching out, Utkarsh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
03/04/2026, 9:44 AMThanks — I’m responding to your question about how to minimize token usage (in LiveKit Agents when using LLMs). Below are practical, actionable steps you can apply right away.
1) Choose smaller / efficient models
• Use lighter models when full capability isn’t needed (for example, choose a “mini” or “nano” variant instead of the full-size model).
• Consider cost-effective models for simple tasks to reduce both prompt and completion tokens.
2) Limit output length
• Set
max_completion_tokensb
boundless-author-7811
03/04/2026, 9:49 AM Copy code
async def _create_agent_and_session(self):
<http://self.logger.info|self.logger.info>("[%s] 🤖 Starting customer session...", self.room_name)
# Get audio configuration for noisy phone environments
audio_config = self.audio_config_service.get_noisy_environment_config()
# Create Silero VAD for local speech detection (filters noise before OpenAI)
self.silero_vad = self.audio_config_service.create_silero_vad(audio_config)
# Get turn detection config for OpenAI semantic VAD
turn_detection = self.audio_config_service.get_turn_detection_config(audio_config)
# Create Model with semantic VAD for noise resilience
model = self.model_config.create_model(self.agent_config, turn_detection)
# Create agent - tools come from agent_config.tools (ToolSchema objects)
agent = GenericAgent(
agent_config=self.agent_config,
customer_data=self.customer_data,
on_transfer_requested=None # PHASE 2: Supervisor transfer disabled
)
# Load dynamic tools (agent handles this internally)
await agent.initialize_dynamic_tools()
<http://self.logger.info|self.logger.info>("✅ Agent ready with %d tools", len(agent.tools))
# Create Session with Silero VAD and tuned interruption settings
self.agent_session = AgentSession(
llm=model,
vad=self.silero_vad, # Silero VAD for noise filtering
min_interruption_duration=audio_config.min_interruption_duration,
min_interruption_words=audio_config.min_interruption_words,
)
# Setup Metrics Collector
self.usage_collector = metrics.UsageCollector()
@self.agent_session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
metrics.log_metrics(ev.metrics)
self.usage_collector.collect(ev.metrics)
# Store metadata for tools and shutdown
self.agent_session.metadata = {
"sip_trunk_id": self.call_context.trunk_id,
"customer_id": self.customer_data.customer_id,
"customer_name": self.customer_data.customer_name,
"campaign_id": self.customer_data.campaign_id,
"call_context": self.call_context
}
<http://self.logger.info|self.logger.info>(
"[%s] ✅ Agent session created with Silero VAD (threshold=%.2f) and Semantic VAD",
self.room_name,
audio_config.silero.activation_threshold
)
return agent
def get_usage_collector(self) -> Optional[metrics.UsageCollector]:
"""Get the usage collector for metrics reporting"""
return self.usage_collectort
tall-airline-16756
03/04/2026, 9:49 AMThanks — I’m responding to your question about reducing total token usage (I reviewed your agent/session code and the RealtimeModel metrics you pasted). You already have useful pieces in place (Silero VAD, semantic VAD, metrics collector). Below are focused, practical changes you can try right away to lower the input token sum and overall usage.
1) Reduce prompt/context size
• Trim or remove non-essential fields from session metadata and tool inputs before sending to the model. Only include the smallest identifiers needed (e.g.,
customer_idcustomer_namecall_contextmax_completion_tokensprompt_tokenscompletion_tokensself.agent_session.metadataget_noisy_environment_config()min_interruption_durationmin_interruption_wordsextra_kwargsmax_completion_tokensmin_interruption_*max_completion_tokensb
boundless-author-7811
03/04/2026, 9:53 AMcan multi agent approach solve this issue? Let say I have will have initial_greeting_agent, then name_confirmation_agent and then other agents so on based on condition, as the converstaion proceeds the agents gets loaded. Does this approach makes sense, if so, how should we do it?
t
tall-airline-16756
03/04/2026, 9:54 AMThanks — I’m responding to your question about using a multi‑agent approach to reduce token usage (I reviewed your code, metrics, and your proposed flow:
initial_greeting_agentname_confirmation_agent@function_tool@function_toolcaller_namecustomer_idcontext.session.userdatasystemmax_completion_tokenstransfer_to_name_confirmation()userdata.raw_nameNameConfirmationAgentuserdata.caller_nameuserdata.name_confirmed = Truefunction_tooluserdatafunction_toolb
boundless-author-7811
03/04/2026, 10:04 AMsame AgentSession, i prioritize latency, accuracy and then token cost
t
tall-airline-16756
03/04/2026, 10:04 AMGreat — thanks for confirming you want to keep everything in the same AgentSession and that your priority order is latency, accuracy, then token cost. That’s a sensible choice: it keeps state simple and lets you optimize handoffs for low latency while still reducing token bloat.
Below is a concise, practical plan and a small, ready-to-adapt code pattern you can drop into your app. It shows:
• same AgentSession usage,
• deterministic handoffs via
@function_tooluserdatasession.userdata@function_tool Copy code
from livekit.agents import Agent, AgentSession, function_tool, RunContext
from dataclasses import dataclass, field
from typing import List
@dataclass
class SessionState:
caller_name: str = None
name_confirmed: bool = False
history_summary: str = "" # compact summary if needed
# add other minimal flags/ids you need
def create_compact_context(state: SessionState, recent_msgs: List[dict] = None):
# Returns a small ChatContext for the incoming agent
ctx = ChatContext()
# 1-2 sentence system summary of persisted state
summary = "Session state: "
if state.name_confirmed:
summary += f"name={state.caller_name} (confirmed)."
elif state.caller_name:
summary += f"name={state.caller_name} (unconfirmed)."
else:
summary += "no name known."
if state.history_summary:
summary += " " + state.history_summary
ctx.add_message("system", summary[:400]) # cap length to keep tokens low
# optional: add last 1-2 user messages only when needed for context
if recent_msgs:
for m in recent_msgs[-2:]:
ctx.add_message(m["role"], m["content"][:800]) # cap each message size
return ctx
class GreetingAgent(Agent):
def __init__(self):
super().__init__(instructions="Greet user briefly and ask for name.")
@function_tool()
async def to_name_confirm(self, context: RunContext, raw_name: str = None):
# Save the raw name into persistent userdata
context.session.userdata.caller_name = raw_name
context.session.userdata.name_confirmed = False
# Prepare next agent with compact context
next_agent = NameConfirmationAgent()
next_agent.chat_ctx = create_compact_context(context.session.userdata)
return next_agent, "Thanks — can you confirm the spelling of your name?"
class NameConfirmationAgent(Agent):
def __init__(self):
super().__init__(
instructions="Confirm the caller's name precisely; ask once then store."
)
@function_tool()
async def confirm_and_proceed(self, context: RunContext, confirmed_name: str = None):
# Confirmed name stored centrally
context.session.userdata.caller_name = confirmed_name
context.session.userdata.name_confirmed = True
# Minimal summary to pass forward; we intentionally avoid full transcript
context.session.userdata.history_summary = f"Caller name confirmed: {confirmed_name}."
# Choose next agent (specialized). Use a lightweight model if the task is simple.
next_agent = TaskAgent() # replace with your domain agent
next_agent.chat_ctx = create_compact_context(context.session.userdata)
return next_agent, f"Thanks, {confirmed_name}. How can I help you today?"
class TaskAgent(Agent):
def __init__(self):
super().__init__(instructions="Handle the user's request with high accuracy.")
# Optionally override LLM model for heavier reasoning here if needed
async def on_enter(self):
# Use userdata instead of re-sending the whole dialogue
name = self.session.userdata.caller_name
# Generate a targeted prompt using name + short summary
await self.session.generate_reply(
instructions=f"User {name} (confirmed). Proceed to handle their request concisely."
)
# Creating the AgentSession (single session for the whole call)
session = AgentSession[SessionState](
userdata=SessionState(),
stt="deepgram/your-stt",
llm="openai/your-default-llm",
tts="your-tts",
preemptive_generation=True, # helps latency
# other session tuning...
)
# Start session with GreetingAgent:
initial_agent = GreetingAgent()
await session.start(agent=initial_agent, room=your_room)min_interruption_wordsuserdatahistory_summarymax_completion_tokenshistory_summaryb
boundless-author-7811
03/04/2026, 10:12 AMshould we use langchain for this?
t
tall-airline-16756
03/04/2026, 10:12 AMShort answer: No — don’t reach for LangChain as your default. Use LiveKit’s native multi‑agent handoffs unless you specifically need LangChain’s stateful planning features.
Why LiveKit native is the better fit for your priorities (latency → accuracy → token cost)
• Lower latency: LiveKit’s agent handoffs via
@function_toolsession.userdata@function_toolsession.userdatab
boundless-author-7811
03/04/2026, 10:43 AMor can we simply update the agents instructions, instead of multi agent approach?
t
tall-airline-16756
03/04/2026, 10:44 AMGood question — yes, you can often get meaningful token savings by tightening a single agent’s instructions, and sometimes that’s the simplest first step. Below I’ll compare both approaches quickly and give concrete, actionable changes you can make if you choose to keep one agent.
Short summary
• Updating instructions can reduce tokens quickly with minimal architecture changes.
• Multi‑agent is still the stronger long‑term solution if you expect long sessions, many distinct phases, or need to use different models per phase.
• Since you prioritize latency → accuracy → cost, start with instruction + context compaction; move to multi‑agent only if token/complexity still grows or you need model specialization.
When updating a single agent makes sense
• Session lengths are short or medium and don’t accumulate huge history.
• Tasks are fairly linear and don’t require frequent role-switching or different models.
• You want minimal engineering changes and fast wins.
Concrete, minimal-change steps to reduce tokens by updating instructions (apply these first)
1) Make the system prompt extremely concise and prescriptive
• Replace long persona/boilerplate with a tiny instruction: “You are a concise, accurate voice agent. Reply in 1–2 short sentences unless the user requests more detail.”
• If some behavior is needed rarely, keep that logic out of the system prompt and call it only when required (see tool gating).
2) Add explicit brevity & format constraints in turn-level prompts
• For confirmations: “Answer in <= 20 tokens. Only confirm or ask to repeat.”
• For routing: “Return one of these tokens only: BILLING, TECH, SCHEDULING.”
3) Stop resending static context every turn
• Move static instructions, tool descriptions, and persona to a cached system message that is not re-sent every turn. Only send dynamic variables (new utterance, last action).
4) Use summarization instead of full history
• Keep a rolling one-line summary in
session.userdata.history_summarymax_completion_tokenshistory_summaryb
boundless-author-7811
03/04/2026, 11:03 AM Copy code
#CORE SYSTEM IDENTITY (EXPLICIT FEMALE VOICE)
You are Ritika, a senior female collection specialist at {bank} bank ABC XYZ. Be warm, firm, and solution-focused while staying respectful and compliant.
## **LANGUAGE GUARDRAIL (ABSOLUTE - NEVER BREAK):**
**RULE:** Once customer requests English (or any language), you are FORBIDDEN from speaking ANY word in Hindi/Hinglish for the REST OF THE CALL.
**ENFORCEMENT:** Before speaking EVERY response, check: "Is customer speaking English?" If YES, translate ALL Hindi words to English.
## **INDIAN VOICE & TONE (FEMALE DELIVERY):**
* **Gender (CRITICAL):** You are **FEMALE**. Always use feminine verbs (**"Kar rahi hoon"**, **"Samajh sakti hoon"**, **"Main baat kar rahi thi"**) and describe actions femininely ("Main guide kar rahi hoon").
* **Script Output:** Write Hindi/Hinglish in **ROMAN SCRIPT** only (e.g., "Main check karti hoon"), never Devanagari.
* **LANGUAGE (CRITICAL):** Once the customer's language is clear (Hindi/Hinglish, English or any other Customer Mentioned Indian Language), then **IMMEDIATELY switch and STAY in that language for ALL responses**.
* **TRANSLATION RULE (ABSOLUTE PRIORITY):** If customer is speaking English, you MUST translate ALL Hindi/Hinglish responses in this script to English before speaking. If customer is speaking Tamil, translate to Tamil. Match the customer's language for EVERY response.
* **LANGUAGE LOCK:** Once you switch to a language, you MUST continue in that language. Do NOT mix languages.
- Start in the opening language; then lock based on the customer's first meaningful reply.
- Do NOT switch languages due to short/neutral replies like "ok", "yes", "haan", "ji", "hmm", "hello","no".
* **SWITCH ONLY IF (CRITICAL):**The customer explicitly asks ("Tamil me bolo", "Telugu" me bolo","English" me bolo) OR they ask to speak any Indian languages.
* **No Translation verbatim:** Do not translate English idioms literally into Hindi. Use local phrasing (e.g., "How can I help you?" becomes "Bataiye, main aapki kaise madad kar sakti hoon?").
* **Delivery & Syntax:** Speak natural Hinglish with Indian fillers (**"Accha"**, **"Theek hai"**, **"Ji"**, **"Matlab"**) and sentence tags ("Right?", "No?").
* **Pacing & Intonation:** Slow down with natural pauses; use a soft rising tone for questions.
* **Tone:** Use respectful address ("ji", "aap") and warm phrases ("Is everything fine?", "Tell me freely").
**BREVITY ENFORCER:**
- Maximum 1-2 sentences per response
- Customer acknowledgment ("samajh gaya", "theek hai", "ok", "got it","hmmm") = MOVE FORWARD, don't repeat
- Be direct and action-focused
- Avoid long explanations
**ACKNOWLEDGMENT DETECTION:**
- If the user says "Samajh gaya", "Theek hai", "Ok", "Got it", or "Hmm" while you are explaining a policy: STOP explaining. They heard you.
- IMMEDIATELY move to the next logical step (e.g., asking for the payment time).
## **NUMBERS RULE (CRITICAL)**
- Do NOT introduce any new numbers (amounts, time, dates, counts, caps) unless the customer explicitly said them OR they are provided in the customer data shown in this prompt.
- If you don't have the exact number, say you can't confirm the exact figure and ask them to check the app.
## **ZERO-TRUST VERIFICATION PROTOCOL (MANDATORY)**
You are FORBIDDEN from revealing **amount**, **due date**, or **account details** until the user confirms identity.
**Trap Scenarios & Responses (before verification):**
1. **"Paisa lene aaye ho na? Le lo."** → "Ji sir, payment collect karna hai, par pehle confirm kar dijiye—kya meri baat {customer_name} se ho rahi hai?"
2. **"I know who you are, just tell me the balance."** → "I understand sir, but as per bank policy, I cannot open the account file without verifying your name first. Are you {customer_name}?"
3. **"Main uska bhai/papa/wife hoon, mujhe bata do."** → "Sorry sir, I can only discuss this with {customer_name} ji. Please ask them to call us back." → [END CALL]
**ONE-TIME VERIFICATION:** Once customer confirms identity (says "Yes", "Ha", "Haan", "Ji"), consider them VERIFIED for the rest of the call. Do NOT ask "Are you {customer_name}?" again. Answer all subsequent questions directly without re-verification.
## **CRITICAL SITUATION PROTOCOLS - HIGHEST PRIORITY**
Personal Questions: If asked "Are you a robot?", "Are you male/female?", or offered a "deal" to pay if you reveal your gender:
Action: Answer directly and briefly as a female agent, then immediately pivot to payment.
Response: "Ji, main female agent Ritika baat kar rahi hoon XYZ se."
### **DEATH/SUICIDE REPORTING:**
**If customer mentions death, suicide, "passed away", "mar gaya", "suicide kar liya":**
- Response (pick based on user's LAST sentence language):
- Hinglish: "Yeh sunkar mujhe bahut dukh hua. Is mushkil waqt mein humari poori sympathy aapke saath hai."
- English: "I'm very sorry for your loss. My sincere condolences to your family."
- Actions: Use death_reported tool, end call immediately, NEVER continue collection discussion
### **CALL TERMINATION PROTOCOL:**
**Customer asks to end call:** first request = brief retention, second request = end call immediately via handle_call_termination_request.
**Immediate termination:** threats, sexual harassment, extreme abuse, discriminatory slurs.
**Before ending call : Follow the professional closure rule.**
### **AUTOMATIC CALL TERMINATION**
**Same ineligible request 3x:** (waiver/settlement/manager/EMI when ineligible)
- 1st: Explain unavailability + alternatives
- 2nd: "I already explained this isn't available."
- 3rd: Call repeated_ineligible_request tool → [END CALL]
**NOT ineligible:** "no"/"I can't"/"no money"/CIBIL questions/disputes [END CALL].
**Tool usage restriction (MANDATORY):** Only call **repeated_ineligible_request** when the customer explicitly asks for **waiver / settlement / manager**, OR asks for **EMI when EMI is not eligible**. Do NOT call it for payment refusal, CIBIL questions, disputes, "no money", or vague complaints.
**CRITICAL (Avoid wrong termination):** Do NOT treat a customer saying "no" / "I can't" / "I don't have money" / asking a CIBIL question as an ineligible request. Only apply this rule when the customer explicitly asks for an unavailable option (e.g., waiver/settlement/EMI when not eligible/manager) repeatedly.
# VOICEMAIL DETECTION (CHECK FIRST)
**Triggers:** "leave message", beep/tone, robotic voice, "not available", 5+ sec silence
**Action:**
"Hi {customer_name}, main Ritika {bank} XYZ se. Aapke account ke baare mein call kiya tha. Please call back."→ end_call IMMEDIATELY
**DO NOT wait, DO NOT say "Hello", DO NOT continue conversation**
## **ANTI-LOOP & REPETITION CHECK (CRITICAL):**
**Before EVERY response:**
1. **Check last 2 responses (cross-response):**
- Already asked this question? → Change approach or move forward
- Already stated payment amount/CIBIL? → Ask action question instead
- Customer said "Samajh gaya"/"Theek hai"/"Ok"? → MOVE FORWARD
2. **Check CURRENT response (intra-response):**
- Same sentence appears twice? → DELETE duplicates
- Amount/date mentioned twice? → Keep only first mention
**Rule:** Each sentence/fact appears ONCE per response. Never repeat across responses.
**Example:**
❌ "Rahul, Rs.19,344 due hai. Rahul, Rs.19,344 due hai."
✅ "Rahul, Rs.19,344 payment 2 February ko due hai. Kya resolve kar sakte hain?"
## **STRUCTURED CALL FLOW**
### **CURRENT CUSTOMER SNAPSHOT**
- Customer: {customer_name}
- Outstanding: {outstanding_balance}
- Minimum due: {minimum_due_amount}
- Due: {due_date}
- Days past due: {days_past_due}
- EMI eligible: {emi_eligible}
- EMI interest rate: {emi_interest_rate}
- EMI options: {emi_tenure_options}
- Waiver eligible: {waiver_eligible}
- Waiver amount: {waiver_amount}
### **PHASE 1: CALL INITIATION & VERIFICATION**
**INTRODUCTION DELIVERY RULE (STRICT):**
1. **Deliver the FULL opening statement without stopping** - do NOT pause or stop for any interruption during intro.
2. **Only after intro is complete**, listen for customer response and proceed to verification.
**Opening (choose based on language):**
** "Hello, main Ritika baat kar rahi hoon {bank} bank XYZ Credit Card se. Kya meri baat {customer_name} ji se ho rahi hai?"
**WAIT FOR VERIFICATION - DO NOT PROCEED WITHOUT IT**
**Verification Responses:**
- Customer says: Yes / Haan / Ji / Boliye / Bol raha hu / Avunandi → verification_status = VERIFIED → Proceed to Phase 2
- Customer asks "Kiska payment?" or "Kis baare mein?" → "Pehle confirm karein - kya aap {customer_name} hain?"
- No / Wrong person → "Kya {customer_name} available hain?" → If no: record_call_outcome(status='wrong_number')
- 3rd party → "Sorry, main sirf {customer_name} ji se discuss kar sakti hoon." → end_call
- Refuses/Dodges (TWO-STRIKE RULE):
- Strike 1: "Security ke liye details batane se pehle confirm karna jaroori hoga—kya aap {customer_name} hain?"
- Strike 2: "Verification ke bina main details share nahi kar sakti." → end_call
**FORBIDDEN until VERIFIED:** You CANNOT say {outstanding_balance}, {due_date}, or any account details.
## PHASE 2: ACCOUNT DISCLOSURE (ONLY AFTER VERIFIED)
**After verification confirmed:**
"{customer_name} ji, aapka {outstanding_balance} payment {due_date} se due hai. Kya aaj resolve kar sakte hain?"
## PHASE 3: CUSTOMER RESPONSES (PRIORITY ORDER)
### PRIORITY 0: REPEAT REQUEST (HIGHEST)
**Triggers:** "Kya bola?", "Repeat please", "Phir se bolo", "Awaaz nahi aayi", "Sunai nahi diya"
**Action:**
1. Identify the KEY POINT of your last response (not the entire response)
2. Repeat ONLY that key point in 1 simple sentence
3. Speak slower and clearer
### PRIORITY 1: PAYMENT CONFIRMED
**Triggers:** "paid", "payment done", "already paid", "maine kar diya", "amount bhara", "jama kar diya"
**ATOMIC SEQUENCE (DO NOT DEVIATE):**
1. Say: "Thanks for the update. Kis medium se payment kiya - UPI, app, ya bank transfer?"
2. **WAIT** for customer response
3. Say: "Perfect! System mein reflect hone mein time lagta hai. Have a great day."
4. Call: record_call_outcome(status='payment_claimed')
5. Call: end_call
6. **HARD STOP - Generate ZERO text after this point**t
tall-airline-16756
03/04/2026, 11:03 AMThanks — I’m responding to your question about whether you can just update the single agent’s instructions (you pasted a long, complex system prompt) instead of moving to multi‑agent. Short answer: yes — you can often get large wins by compacting and restructuring that prompt, and that’s the lowest‑effort first step. Below I’ll (1) highlight the main risks of your current prompt, (2) give concrete, safe compaction strategies that preserve accuracy/compliance, and (3) provide a short compacted example and next steps you can use immediately.
Why compacting the single agent prompt is a good first step
• Fast to implement with no architecture change.
• Keeps latency low (no extra handoffs).
• Good when you prioritize latency and accuracy over absolute token cost.
• If done correctly (structured state + explicit verification rules), it can achieve most of the benefits of a multi‑agent system while remaining simple.
Main risks in the current long prompt
• Large, repeated system text increases input_tokens every turn if sent each request.
• Complex conditional rules (language locks, multiple critical protocols) are easy for an LLM to mishandle if prompts repeat redundant context every turn.
• Verbose rules raise the chance of token overshoot and slow inference.
• If static content is re-sent every turn, caching may not be applied effectively.
How to compact safely while preserving accuracy/compliance
1. Move static, rarely changing rules into cached system messages
◦ Put immutable policies (ZERO-TRUST VERIFICATION, NUMBERS RULE, CRITICAL SITUATION PROTOCOLS, VOICEMAIL DETECTION, TERMINATION rules) into a cached system message that you send once or mark cacheable. Do not re-send every turn.
1. Keep per-turn system prompt minimal and actionable
◦ Per turn send only what the model absolutely needs: tiny system summary + the immediate user utterance.
◦ Example per-turn system line: “Ritika, female collections agent. Confirm identity first. Max 1–2 short sentences. Follow cached policies.”
1. Replace long procedural prose with bulletized, machine-readable instructions
◦ Use short numbered rules with direct guards (e.g., RULE: Do not speak Hindi if user switched to English). LLMs follow concise guarded tokens more reliably than long paragraphs.
1. Use structured flags in session.userdata rather than re-sent narrative
◦ Persist: name_confirmed, customer_name, language_locked, last_action, critical_flag. Downstream prompts use these short flags instead of verbose history.
1. Move examples and long trap-scenarios to external docs & only send an ID
◦ If you need the LLM to reference “trap scenarios,” send a small token like
policy_v2_trap_handlingmax_completion_tokenshistory_summarypolicy_v2VERIFIEDmax_completion_tokenspolicy_v2policy_v2b
boundless-author-7811
03/04/2026, 11:12 AMompress your full prompt into a cached policy + short per-turn prompt and return both,
t
tall-airline-16756
03/04/2026, 11:13 AMThanks — I compressed your full prompt into two pieces you can use immediately:
• Cached policy (send once and store server-side under a key like
policy_v2language_lockedname_confirmedcustomer_namehistory_summarymax_completion_tokenspolicy_v2repeated_ineligible_requestdeath_reportedpolicy_v2policy_v2policy_v2history_summaryb
boundless-author-7811
03/04/2026, 11:32 AMhow should i refactor and use the updated prompt for that it works correctly
t
tall-airline-16756
03/04/2026, 11:33 AMThanks — I’ll walk you through a clear, step‑by‑step refactor so the cached policy + short per‑turn prompt works reliably in your LiveKit agent flow. I’ll keep each step short and actionable and include a practical code pattern you can drop into your existing service.
High‑level plan
1. Store the long policy once (server-side) as
policy_v2session.userdatapolicy_v2 Copy code
# policy_store.py
POLICIES = {}
def store_policy(key: str, text: str):
POLICIES[key] = text
def get_policy(key: str) -> str:
return POLICIES[key] Copy code
store_policy("policy_v2", "<<< full long policy text >>>") Copy code
def prechecks(asr_text: str, session):
# voicemail heuristics
if is_voicemail(asr_text) or session.silence_seconds > 5:
play_voicemail_and_end(session)
return {"action": "handled_voicemail"}
# language detection (first meaningful reply)
if session.userdata.language_locked == "none":
lang = detect_language(asr_text)
if lang in ["en","hi","ta",...]:
session.userdata.language_locked = lang
# verification detection
if looks_like_verification(asr_text):
session.userdata.name_confirmed = True
session.userdata.verification_attempts += 1
return {"action": "verified"}
# trap detection
if looks_like_trap(asr_text):
return {"action": "trap_response", "reply": canned_trap_reply(session)}
return {"action": "continue"}policy_v2 Copy code
def build_payload(session, user_asr):
system = f"Ritika — female collections specialist. Follow cached policy `policy_v2`. Verify identity before revealing account details. Language_lock={session.userdata.language_locked}. name_confirmed={session.userdata.name_confirmed}. Reply in 1–2 short sentences. Do not introduce new numbers. Max 40 tokens."
# include protected metadata only if verified
metadata = {}
if session.userdata.name_confirmed:
metadata.update({
"customer_name": session.metadata.get("customer_name"),
"outstanding_balance": session.metadata.get("outstanding_balance"),
"due_date": session.metadata.get("due_date"),
})
payload = {
"system": system,
"user_message": user_asr,
"flags": {
"language_locked": session.userdata.language_locked,
"name_confirmed": session.userdata.name_confirmed,
"verification_attempts": session.userdata.verification_attempts
},
"metadata": metadata
}
return payload Copy code
class SessionUserData:
def __init__(self):
self.language_locked = "none"
self.name_confirmed = False
self.customer_name = None
self.verification_attempts = 0
self.history_summary = "" Copy code
def update_history_summary(session, event: str):
session.userdata.history_summary = (session.userdata.history_summary + " | " + event).strip()[:400]max_completion_tokens Copy code
@session.on("asr_result")
async def handle_asr(asr_text, session):
pre = prechecks(asr_text, session)
if pre["action"] == "handled_voicemail":
return
if pre["action"] == "trap_response":
await session.send_reply(pre["reply"])
return
if pre["action"] == "verified":
# optionally send a short confirmation reply or allow disclosure
await session.send_reply("Thank you — verification complete.")
return
payload = build_payload(session, asr_text)
# choose max tokens based on flags or role
max_toks = 30 if not session.userdata.name_confirmed else 60
response = await call_llm(payload, max_completion_tokens=max_toks)
# post-process checks (dedupe, anti-loop)
resp = dedupe_and_apply_policy(response, session)
await session.send_reply(resp)policy_v2max_completion_tokensself.agent_sessionself.usage_collectorself.agent_configb
boundless-author-7811
03/04/2026, 12:27 PMdoes livekit dispatch request need trunk_id>
t
tall-airline-16756
03/04/2026, 12:27 PMShort answer: No — LiveKit agent dispatch requests do not require a
trunk_idtrunk_idtrunk_id Copy code
await lkapi.agent_dispatch.create_dispatch(
api.CreateAgentDispatchRequest(
agent_name="test-agent",
room="my-room",
metadata='{"user_id":"12345"}'
)
) Copy code
await lkapi.sip.create_sip_participant(
api.CreateSIPParticipantRequest(
room_name="my-room",
sip_trunk_id="ST_outbound_abc",
sip_call_to="+1234567890",
participant_identity="caller"
)
)b
boundless-author-7811
03/04/2026, 12:29 PMwe are dispatching for sip participant but right now we are passing the trunk on scheduler which is hampering our design, we want to know whether we can dispatch the job from scheduler but inside the create sip participant in orchestrator we choose the sip trunk
t
tall-airline-16756
03/04/2026, 12:29 PMYes — you can dispatch the job from your scheduler without a
trunk_idcreate_sip_participantcreate_sip_participant(..., sip_trunk_id=chosen_trunk, sip_call_to=phone_number, ...) Copy code
# scheduler -> creates job without trunk_id
scheduler.create_dispatch(agent_name="outbound-agent", room="r123", metadata={
"phone_number":"+91xxxx", "region":"IN", "priority":"standard", "campaign":"CAMP1"
})
# orchestrator / agent
def choose_trunk(metadata):
if metadata["priority"] == "premium":
return os.getenv("PREMIUM_SIP_TRUNK")
if metadata["region"] == "IN":
return os.getenv("IN_SIP_TRUNK")
return os.getenv("DEFAULT_SIP_TRUNK")
async def make_call(job):
phone = job.metadata["phone_number"]
trunk = choose_trunk(job.metadata)
await api.sip.create_sip_participant(
room_name=job.room,
sip_trunk_id=trunk,
sip_call_to=phone,
participant_identity=f"caller-{job.id}"
)b
boundless-author-7811
03/04/2026, 12:57 PMif all AgentServers are down, what will happen to the dispatched job?
t
tall-airline-16756
03/04/2026, 12:57 PMThanks — I’m responding to your question about what happens to a dispatched job if all AgentServers are down. Short answer: the dispatch will not be processed and you must handle retries/fallbacks on the client/orchestrator side.
What happens
• Job assignment will fail because there are no AgentServers available to accept the job. The dispatch request will be rejected or time out.
• LiveKit does not automatically queue and replay dispatches when all AgentServers are offline. The system requires an available AgentServer to accept and run the job.
• When AgentServers come back online, they do not automatically resume previously failed dispatch attempts — new dispatch requests must be issued.
Recommended mitigations (practical, prioritized)
1. Add redundancy (high impact)
◦ Run multiple AgentServers for each agent_name across zones/instances so a single outage won’t exhaust capacity.
1. Implement retries with backoff in your scheduler (medium impact)
◦ On dispatch failure, retry with exponential backoff and a max retry window. Log failures and surface alerts after N retries.
1. Fallback handling in scheduler/orchestrator (high impact)
◦ If dispatch fails repeatedly, mark the job as “failed” and run fallback flows: notify an operator, enqueue for manual processing, or route to an alternative channel (outbound SIP queue, SMS, or callback list).
1. Health checks & auto-scaling (high impact)
◦ Monitor AgentServer health and register/unregister capacity. Auto-scale AgentServers when utilization crosses thresholds so capacity is available.
1. Graceful degradation / user messaging (low effort)
◦ If the job represents a customer-facing call, notify the customer of temporary delay and optionally schedule a retry window.
1. Audit & observability (critical)
◦ Log dispatch attempts, failures, chosen retry strategy, and final outcome. Track metrics (failure rate, mean time to recover) and create alerts for sustained outages.
Implementation ideas / snippets
• Retry with backoff (pseudo):
Copy code
def dispatch_with_retries(dispatch_payload, max_retries=5):
delay = 1
for attempt in range(max_retries):
try:
return lkapi.agent_dispatch.create_dispatch(dispatch_payload)
except TransientDispatchError:
time.sleep(delay)
delay *= 2
# after retries fail
mark_job_failed(dispatch_payload)
alert_ops(dispatch_payload)
return None10 Views