blue-stone-74148
03/14/2026, 2:37 PMtall-airline-16756
03/14/2026, 2:37 PMtall-airline-16756
03/14/2026, 2:37 PMblue-stone-74148
03/14/2026, 2:38 PMtall-airline-16756
03/14/2026, 2:39 PM# agent.py
import asyncio
import logging
from dotenv import load_dotenv
from livekit.agents import AgentServer, AgentSession, Agent, JobContext
from livekit.agents.tools import function_tool
from livekit.plugins import silero
load_dotenv()
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger("multilingual_agent")
server = AgentServer()
# Optional: preload VAD/other heavy models to speed cold starts
def prewarm(proc):
proc.userdata["vad"] = silero.VAD.load()
server.setup_fnc = prewarm
class VoiceAssistant(Agent):
def __init__(self):
super().__init__(
instructions=(
"You are a helpful multilingual voice assistant for an e-commerce service. "
"You can speak Hindi and English and handle code-switched (Hinglish) input."
)
)
@function_tool
async def check_order_status(self, order_id: str) -> str:
# TODO: replace with real API call to your order system
# Simulate lookup
return f"Order {order_id} is out for delivery and will arrive tomorrow."
@function_tool
async def cancel_order(self, order_id: str) -> str:
# TODO: call your cancel API and handle errors
return f"Order {order_id} has been canceled successfully."
async def on_enter(self):
# welcome the caller
await self.session.generate_reply(instructions="Welcome! How can I help you today?")
@server.rtc_session(agent_name="multilingual-agent")
async def entrypoint(ctx: JobContext):
# ctx.room is the LiveKit room for this call
<http://logger.info|logger.info>("Starting agent session for room: %s", ctx.room.name)
session = AgentSession(
stt="deepgram/nova-3:multi",
llm="openai/gpt-4.1-mini",
tts="cartesia/sonic-3",
vad=ctx.proc.userdata.get("vad"),
)
await session.start(room=ctx.room, agent=VoiceAssistant())
await ctx.connect() # keep session alive while connected
<http://logger.info|logger.info>("Agent session ended for room: %s", ctx.room.name)
if __name__ == '__main__':
server.run()
Step 3 — Run the agent server locally
• python agent.py
• The server should prewarm VAD and listen for LiveKit session connections.
Step 4 — Configure LiveKit inbound trunk and Twilio
• In LiveKit dashboard, create an inbound SIP trunk and add a dispatch rule to route incoming calls to a room/agent (use the agent name configured above).
• In Twilio, create a SIP or SIP Trunk configuration (or use Twilio Media Streams) to send calls to the LiveKit SIP endpoint or phone number forwarding LiveKit.
• For Sources: LiveKit Phone Numbers | Connect voice agents to phone lines | LiveKit | Creating SIP Inbound Trunks and Dispatch Rules with Python SDK | Agent session | LiveKit Documentation | An open source stack for real-time multimodal AI | LiveKit | examples/bank-ivr | livekit/livekit_composite | DeepWikiblue-stone-74148
03/14/2026, 2:41 PMtall-airline-16756
03/14/2026, 2:41 PM# local_agent.py
import os
import queue
import threading
import json
import time
import datetime
from pathlib import Path
from dotenv import load_dotenv
import sounddevice as sd
import soundfile as sf
import pyttsx3
import requests # optional for cloud STT/LLM
# pip install sounddevice soundfile pyttsx3 requests python-dotenv
load_dotenv()
OUTPUT_JSON = Path("conversation_log.json")
SAMPLE_RATE = 16000
CHANNELS = 1
CHUNK_SECONDS = 4 # record 4s chunks; adjust for VAD or push-to-talk
# Conversation structure
conversation = []
# Local TTS engine (fallback)
tts_engine = pyttsx3.init()
tts_engine.setProperty("rate", 150)
def save_conversation():
with open(OUTPUT_JSON, "w", encoding="utf-8") as f:
json.dump(conversation, f, ensure_ascii=False, indent=2)
print(f"Saved conversation to {OUTPUT_JSON}")
def record_chunk(duration=CHUNK_SECONDS):
print(f"Recording for {duration}s...")
recording = sd.rec(int(duration * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=CHANNELS, dtype='int16')
sd.wait()
return recording.flatten()
def write_wav(data, filename="chunk.wav"):
sf.write(filename, data, SAMPLE_RATE, subtype="PCM_16")
return filename
# STT: try Deepgram (cloud) if API key present, otherwise use a naive local stub
def transcribe_audio(wav_path):
dg_key = os.getenv("DEEPGRAM_API_KEY")
if dg_key:
print("Using Deepgram STT")
with open(wav_path, "rb") as f:
r = <http://requests.post|requests.post>(
"<https://api.deepgram.com/v1/listen?language=auto>",
headers={"Authorization": f"Token {dg_key}"},
data=f,
)
if r.ok:
resp = r.json()
# deepgram returns alternatives; adjust depending on response shape
transcript = resp.get("results", {}).get("channels", [])[0].get("alternatives", [])[0].get("transcript", "")
lang = resp.get("metadata", {}).get("language", "unknown")
return transcript.strip(), lang
else:
print("Deepgram error:", r.text)
return "", "unknown"
else:
# Local fallback stub: uses simple whisper api or placeholder
print("No Deepgram key found — using local placeholder STT")
# For quick testing, ask user to type the transcript instead of real STT
typed = input("Type what you said (local test mode): ").strip()
return typed, "und"
# LLM: simple rule-based or placeholder; swap with OpenAI API if you have a key
def llm_respond(transcript, language="und"):
openai_key = os.getenv("OPENAI_API_KEY")
if openai_key:
print("Calling OpenAI (placeholder) — implement real call")
# implement OpenAI chat completion function call here
# For now return an echo
return f"I heard: {transcript}"
else:
# Very simple local logic for testing
if "order" in transcript.lower():
return "I can check your order. What is the order id?"
if "cancel" in transcript.lower():
return "To cancel, please confirm the order id."
if transcript.strip() == "":
return "Sorry, I didn't catch that. Could you repeat?"
return f"You said: {transcript}"
# TTS: play assistant reply
def speak_text(text, language="en"):
print("Assistant:", text)
# For local: pyttsx3
tts_engine.say(text)
tts_engine.runAndWait()
def append_turn(role, content, language):
conversation.append({
"role": role,
"content": content,
"language": language,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z"
})
def main_loop():
try:
print("Starting local agent. Press Ctrl+C to exit.")
while True:
data = record_chunk()
wav = write_wav(data, "chunk.wav")
transcript, lang = transcribe_audio(wav)
append_turn("user", transcript, lang)
reply = llm_respond(transcript, language=lang)
append_turn("assistant", reply, lang)
speak_text(reply, language=lang)
# short pause
time.sleep(0.2)
except KeyboardInterrupt:
print("Shutting down.")
save_conversation()
if __name__ == "__main__":
main_loop()
How to use
• Run: python local_agent.py
• Speak into your mic (or type in the prompt if no Deepgram key)
• Each user/assistant turn is appended to conversation and on exit saved to conversation_log.json
• Edit CHUNK_SECONDS or replace the naive record_chunk with VAD or push-to-talk to get better UX
Custom JSON formatting
• The script appends objects with keys: role, content, language, timestamp.
• If you want a different structure (e.g., including turn_id, tool_calls, confidence), change append_turn() to include those fields.
• Example: add tool metadata:
◦ append_turn("assistant", reply, lang) -> include {"tools_invoked": [...], "tool_outputs": {...}}
Next steps I can help with
• Swap the simple placeholders for real Deepgram / OpenAI API calls and provide exact request examples.
• Replace audio chunks with VAD so the system captures real conversational turns (better for natural interaction).
• Add function/tool calling: show how LLM function-calling results trigger your order/cancel API and are logged in JSON.
• Change TTS from pyttsx3 to a higher-quality cloud TTS and show how to stream audio back.
• Show how to save richer logs (timestamps per segment, STT confidences, audio file references) in JSON.
Tell me:
• Do you want cloud STT/LLM/TTS examples (I’ll include exact API request code), or keep everything local for now?
• Any specific JSON schema you want (I’ll provide the schema and update the script)? Sources: Build a multilingual voice agent that automatically switches languages | Li...blue-stone-74148
03/14/2026, 2:43 PMtall-airline-16756
03/14/2026, 2:44 PM# agent_console.py
import os
import json
import asyncio
import datetime
from pathlib import Path
from dotenv import load_dotenv
from livekit.agents import AgentServer, AgentSession, Agent, JobContext, cli
from livekit.agents.tools import function_tool
from livekit.plugins import silero, deepgram, inference
# Load environment
load_dotenv()
OUTPUT_DIR = Path("console-recordings")
OUTPUT_DIR.mkdir(exist_ok=True)
# Configure what fields you want in the JSON
def make_log_entry(role: str, text: str, language: str = None, meta: dict = None):
return {
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"role": role,
"text": text,
"language": language or "und",
"meta": meta or {},
}
# In-memory conversation log (append-only)
conversation_log = []
# Simple agent with tools (example)
class LocalTestAgent(Agent):
def __init__(self):
super().__init__(instructions="You are a helpful multilingual assistant. Reply in the user's language.")
@function_tool
async def check_order_status(self, order_id: str) -> str:
# Replace this with a real API call
return f"Order {order_id} is out for delivery."
@function_tool
async def cancel_order(self, order_id: str) -> str:
# Replace this with a real API call
return f"Order {order_id} has been canceled."
# Create server
server = AgentServer()
# Prewarm VAD to reduce cold start (optional)
def prewarm(proc):
proc.userdata["vad"] = silero.VAD.load()
server.setup_fnc = prewarm
@server.rtc_session(agent_name="local-console-agent")
async def entrypoint(ctx: JobContext):
# Choose STT/LLM/TTS config. Replace models or providers as desired.
stt = deepgram.STT(model="deepgram/nova-3-general", language="multi") # multilingual
llm = inference.LLM(model="openai/gpt-4o") # replace if you have a different LLM
tts = inference.TTS(model="cartesia/sonic-3") # replace if you want other TTS
session = AgentSession(
stt=stt,
llm=llm,
tts=tts,
vad=ctx.proc.userdata.get("vad"),
)
# Log incoming transcripts (user)
@session.on("user_input_transcribed")
async def on_user_transcript(trans):
if not hasattr(trans, "transcript"):
return
# Add user turn when final
if getattr(trans, "is_final", True):
entry = make_log_entry("user", trans.transcript, getattr(trans, "language", None), {
"confidence": getattr(trans, "confidence", None),
})
conversation_log.append(entry)
print(f"\n[USER] {trans.transcript}")
# Log assistant responses and tool calls
@session.on("conversation_item_added")
async def on_conversation_item(item):
# item may be assistant text, tool_result etc. Adjust based on item.shape
role = getattr(item, "role", "assistant")
text = getattr(item, "text", "") or getattr(item, "content", "")
entry = make_log_entry(role, text, getattr(item, "language", None), {
"type": getattr(item, "type", None),
"meta": getattr(item, "meta", None),
})
conversation_log.append(entry)
print(f"\n[ASSISTANT] {text}")
# Save JSON whenever a new entry is added (optional)
def save_json_snapshot():
fname = OUTPUT_DIR / f"conversation-{datetime.datetime.utcnow().strftime('%Y%m%d-%H%M%S')}.json"
with open(fname, "w", encoding="utf-8") as f:
json.dump(conversation_log, f, indent=2, ensure_ascii=False)
print(f"Saved snapshot to {fname}")
# Add a shutdown callback to persist final log
ctx.add_shutdown_callback(lambda: save_json_snapshot())
# Start session with the agent
await session.start(room=ctx.room, agent=LocalTestAgent())
# Optionally greet
await session.generate_reply(instructions="Greet the user briefly and ask how you can help.")
# Keep the session alive while connected to console
await ctx.connect()
if __name__ == "__main__":
# Run the agent server in console mode:
# This exposes microphone/speaker I/O in your terminal environment.
# Use: python agent_console.py console --record
cli.run_app(server)
How to run (local test)
1. Start:
◦ python agent_console.py console --record
◦ This opens console mode, captures mic input, plays TTS to speakers, and the LiveKit console UI displays transcripts.
1. Interact:
◦ Speak into your mic. LiveKit handles VAD & STT and the agent responds.
◦ You can press Ctrl+T in console mode to toggle between voice and text input.
1. Stop:
◦ Ctrl+C to stop; the script’s shutdown callback will save the final JSON snapshot into console-recordings.
Customizing the JSON format
• The function make_log_entry controls what fields go into each log row. Add:
◦ turn_id, session_id, audio_file (path), tool_calls (list), stt_confidence, latencies, etc.
• If you want a single file that updates in place, change save_json_snapshot to write to a fixed filename (e.g., conversation.json) after each new item.
Where to plug in tools (order/cancel)
• Implement the actual API calls inside the tool methods (check_order_status, cancel_order).
• When tools run, their outputs appear in the conversation stream and will be logged by the conversation_item_added handler.
Switching to cloud-only STT/LLM/TTS
• Replace deepgram.STT / inference.LLM / inference.TTS with the cloud providers and set keys in .env.
• For higher accuracy in Indian languages and code-switching, Deepgram’s multilingual models and Cartesia/Rime TTS are good options — configure language options as needed.
Next steps I can help with
• Provide a version that uses local STT/TTS if you want zero cloud usage.
• Add VAD tuning or push-to-talk to improve capture quality.
• Show exact code to wire OpenAI function-calling -> tool invocation -> structured logging.
• Provide a JSON schema example you want (e.g., include tool call entries, audio file references, confidence scores).
If you want, I can now:
• Send a variant that uses local STT/TTS (no API keys).
• Or send a version with exact Deepgram/OpenAI API calls wired in.
Which would you prefer?
Sources: Build a multilingual voice agent that automatically switches languages | Li... | Build Your First AI Voice Agent in Python: Complete Tutorial | LiveKit | Server startup modes | LiveKit Documentation | Build a multilingual voice agent that automatically switches languages | Li... | Agent session | LiveKit Documentation