I need to use text model to get work and realtime model to talk to customers Can you help with idea how I can create agents for this

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

I need save token tell me how I can manage or build context

Thanks — I’m responding to your request about saving tokens and managing conversational context. I can walk you through practical strategies and show code examples tailored to your stack. Before I give code, a couple quick clarifying questions so I give the most useful, concrete guidance: • Which language do you want examples in: Python or TypeScript? • Are you already using LiveKit Agents and agent handoffs, or starting from scratch? • Do you want the text agent to keep most state off the LLM (preferred) or occasionally pass full conversation history? • Any particular types of “work” your text agent will do (DB lookups, bookings, payments, long documents)? While you answer, here are the most effective token-saving strategies you can adopt (short and actionable): 1) Store structured session state (use session.userdata) • Keep persistent facts (user name, account id, authentication status, recently-collected form fields) in

session.userdata

as a small dataclass or object. • Only surface needed fields into prompts when the LLM actually needs them. This avoids re-sending the same facts every turn. 2) Summarize and truncate chat history before sending to the LLM • Keep a short recent-window (e.g., last 5–10 user/assistant turns). • Maintain and update a concise summary of the whole conversation (one or two sentences) and include the summary + recent messages when context is required. • Use an automated summarizer (a cheap/fast model or a custom prompt) periodically to refresh the summary. 3) Pass context selectively between agents • Don’t auto-forward full chat_ctx on every handoff. Explicitly create a condensed context to pass to the next agent. • If handing off, pass: summarized conversation, relevant session.userdata fields, and only the last few messages. 4) Return structured results from background agents • Have your text/work agent return structured JSON like: ◦

{ status, data, reply_text, required_followups }

• The realtime/customer-facing agent uses

reply_text

for speech without needing the whole reasoning trace. 5) Edit or redact low-value messages before storing • Remove or compress repeated confirmations, long logs, or noisy transcripts before adding them to chat history. 6) Use role-based system prompts sparingly and keep them minimal • Long system prompts count toward tokens every time they are sent; store static instructions server-side and only pass necessary instructions. 7) Cache deterministic responses • For repeated queries (order status, plan pricing), cache responses and serve the cached reply instead of calling the LLM every time. 8) Control frequency of full-context calls • Only run full-context prompts for tasks that require deep reasoning. For simple intents use intent classifiers or rules which are cheaper. If you tell me your language (Python/TS) and whether you want examples for agent handoffs or for a single-agent flow with session.userdata, I’ll provide a concise example showing: •

session.userdata

dataclass/object usage, • a small ConversationManager that creates a truncated context, • sample function tool that returns structured results to the realtime agent. Which language and scenario would you like me to show? Sources: Building multi-agent architectures with LiveKit agents | LiveKit | Agents and handoffs | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation

I am looking for python code from scratch, yes I am fine with text agent use more tokens. Work is more about db and faq from the rag

Thanks — I’ll give you a concise, from-scratch Python implementation that matches what you asked for: • Realtime agent (low-latency voice) for customer interaction • Text agent for backend work (DB queries + RAG for FAQs) — this agent may use more tokens • Efficient context management using

session.userdata

so you don’t resend repeated facts • Clear handoff pattern between the two agents I’ll keep this compact and runnable-ish so you can adapt it quickly. Tell me if you want it wired to a real DB (Postgres/Supabase) or a vector store (Pinecone/Weaviate) and I’ll add that integration. Prereqs • Python 3.9+ • Install required packages (example): pip install "livekit-agents[mcp]" openai faiss-cpu python-dotenv .env (required environment variables) • LIVEKIT_API_KEY, LIVEKIT_API_SECRET, LIVEKIT_URL • OPENAI_API_KEY High-level approach 1. Store persistent, deterministic state in a small dataclass assigned to session.userdata (saves tokens). 2. RealtimeAgent handles speech I/O and short replies; when complex work is needed it calls a tool that returns a TextAgent (hand off). 3. TextAgent runs DB/RAG queries, returns a structured result and updates session.userdata with the answer. 4. RealtimeAgent speaks the prepared reply_text (no need to resend the whole reasoning trace). Complete minimal example (single file) • This example uses OpenAI embeddings + a tiny in-memory "FAQ" index to keep the example self-contained. • Replace the RAG index and DB calls with your real systems as needed.

# app.py
import os
import asyncio
from dataclasses import dataclass, field
from typing import Optional, List, Dict
import openai
import numpy as np

# Placeholder imports to match typical LiveKit Agents SDK structure
from livekit.agents import Agent, AgentSession, function_tool, ChatContext
from livekit.plugins import openai as lk_openai  # pseudonym for plugin objects
# Replace plugin imports with real SDK objects for your environment.

# ---------------- Session State ----------------
@dataclass
class SessionState:
    user_name: Optional[str] = None
    latest_query: Optional[str] = None
    latest_answer: Optional[str] = None
    # Add structured fields you frequently need (user_id, account_id, auth)
    meta: Dict = field(default_factory=dict)

# ---------------- Simple in-memory FAQ + RAG ----------------
FAQ_DATA = [
    {"id": "1", "question": "How do I reset my password?", "answer": "Visit the login page and click 'Forgot Password'."},
    {"id": "2", "question": "What is your refund policy?", "answer": "We offer refunds within 30 days of purchase."},
    {"id": "3", "question": "How long does shipping take?", "answer": "Shipping typically takes 3-5 business days."},
]

openai.api_key = os.getenv("OPENAI_API_KEY")

def embed_text(text: str) -> np.ndarray:
    resp = openai.Embedding.create(input=[text], model="text-embedding-3-small")
    emb = np.array(resp["data"][0]["embedding"], dtype="float32")
    return emb

# Precompute small embedding matrix and simple nearest neighbor search
FAQ_EMBEDDINGS = [embed_text(f"{f['question']} {f['answer']}") for f in FAQ_DATA]
FAQ_MATRIX = np.vstack(FAQ_EMBEDDINGS)

def faq_rag_lookup(question: str, k: int = 1) -> List[Dict]:
    q_emb = embed_text(question)
    # cosine similarity via dot (normalize to speed up)
    def normalize(v): return v / np.linalg.norm(v)
    qn = normalize(q_emb)
    matn = np.vstack([normalize(v) for v in FAQ_MATRIX])
    scores = matn.dot(qn)
    idxs = np.argsort(-scores)[:k]
    return [FAQ_DATA[i] for i in idxs]

# ---------------- Text Agent (backend heavy work) ----------------
class TextAgent(Agent):
    def __init__(self, chat_ctx: Optional[ChatContext] = None):
        super().__init__(instructions="You are a backend assistant that performs DB/RAG queries.", chat_ctx=chat_ctx)

    @function_tool()
    async def run_rag_and_db(self, context, question: str):
        """
        Perform RAG lookup and (simulated) DB query.
        Returns a structured dict: {status, data, reply_text}
        """
        # RAG FAQ lookup
        hits = faq_rag_lookup(question, k=1)
        faq_answer = hits[0]["answer"] if hits else None

        # Simulate DB query (replace with real DB call)
        db_result = None
        if "order" in question.lower():
            db_result = {"order_id": "12345", "status": "shipped", "eta": "tomorrow"}

        # Build reply text
        if db_result:
            reply = f"Order {db_result['order_id']} is {db_result['status']} and should arrive {db_result['eta']}."
        elif faq_answer:
            reply = faq_answer
        else:
            # fallback to LLM to craft helpful reply (text-model allowed to use more tokens)
            # Use openai.ChatCompletion or your preferred text model here.
            comp = openai.ChatCompletion.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "You are a helpful backend assistant."},
                    {"role": "user", "content": f"Answer succinctly: {question}"}
                ],
                max_tokens=300,
                temperature=0.0,
            )
            reply = comp["choices"][0]["message"]["content"].strip()

        # Save the structured result to session.userdata for the realtime agent
        state: SessionState = context.session.userdata
        state.latest_query = question
        state.latest_answer = reply

        return {"status": "ok", "data": {"faq_hit": bool(faq_answer), "db": db_result}, "reply_text": reply}

# ---------------- Realtime Agent (voice I/O, token-sparing) ----------------
class RealtimeAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a friendly customer agent. Use short replies.")

    async def on_enter(self):
        # Greet based on session state
        state: SessionState = self.session.userdata
        if state.user_name:
            await self.session.generate_reply(instructions=f"Welcome back {state.user_name}, how can I help?")
        else:
            await self.session.generate_reply(instructions="Hello! What's your name?")

    @function_tool()
    async def handle_user_message(self, context, text: str):
        """
        Called when user asks something. If it looks like a backend task, handoff to TextAgent.
        Otherwise reply quickly from the realtime model directly.
        """
        state: SessionState = context.session.userdata

        # Simple heuristic: if contains keywords, delegate to backend
        backend_keywords = ["order", "status", "refund", "shipping", "reset password", "how do i"]
        if any(k in text.lower() for k in backend_keywords):
            # Acknowledge quickly, then handoff
            await self.session.generate_reply(instructions="Let me check that for you—one moment please.")

            # Create a small chat context for the text agent (condensed)
            chat_ctx = ChatContext()
            chat_ctx.add_message(role="user", content=f"User question: {text}")

            text_agent = TextAgent(chat_ctx=chat_ctx)

            # Execute the text agent's tool that performs RAG/DB work
            # Depending on the SDK, you might call text_agent.run(...) or use the tool directly.
            # Here we call the tool function via the agent's function API (pseudocode)
            result = await text_agent.run_rag_and_db(context=context, question=text)

            # Fetch result from session.userdata or returned result
            reply = result["reply_text"]
            return reply

        # Otherwise, short realtime reply using realtime model (cheap)
        # Use session.generate_reply or a direct LLM call configured for realtime.
        short_resp = "I can help with that. Could you give me a bit more detail?"
        return short_resp

# ---------------- Entrypoint (session setup) ----------------
async def start_session(room, connection):
    """
    This function represents starting an AgentSession for a user connection/room.
    Replace with real LiveKit session wiring for your environment.
    """
    # Create session and stash structured userdata
    session = AgentSession(
        # stt=..., llm=realtime_llm, tts=..., vad=..., config as needed
    )
    session.userdata = SessionState()

    # Start realtime agent
    rt_agent = RealtimeAgent()
    await session.start(agent=rt_agent, room=room)  # adapt to SDK call signatures

# ---------------- Run helper for local testing ----------------
if __name__ == "__main__":
    print("This file shows the core pattern. Integrate with your LiveKit runtime and SDK calls.")

Key points and token-saving guidance (summary) • Use session.userdata (SessionState dataclass) to store persistent facts (user name, last answer, IDs). These do not occupy LLM chat tokens. • Keep RealtimeAgent prompts short and only pass the minimal context to the text agent. Use a condensed ChatContext Sources: Tool definition and use | LiveKit Documentation | Agents and handoffs | LiveKit Documentation | Building multi-agent architectures with LiveKit agents | LiveKit | External data and RAG | LiveKit Documentation | OpenAI and LiveKit partner to turn Advanced Voice into an API

Can we stream? Will this have latency?

Thanks — I’m answering your questions about streaming and latency for the realtime + text-agent architecture (what parts can stream, where latency shows up, and how to mitigate it). Short answer • Yes — you can stream audio (and partial text) end-to-end so the customer hears responses incrementally. • The realtime agent offers low-latency speech I/O; the text agent work (DB/RAG/LLM) will be higher-latency when invoked. You can combine streaming and smart handoffs to keep perceived latency low. What can stream • Audio input: user speech is streamed to the realtime agent (STT or realtime speech model) with very low delay. • Partial speech or TTS: realtime models and modern TTS can stream audio output as it’s generated, so customers hear a response quickly rather than waiting for the entire LLM output. • Intermediate text: many LLMS support streaming token-by-token (or chunked) responses, which you can forward progressively to TTS or to the UI. Where latency arises • Network and media pipeline: audio capture → upload → model ingest → response stream. With good infrastructure this is tens to low hundreds of milliseconds. • Realtime model processing: typically low-latency (designed for real-time conversations). • Text-agent work: DB queries are usually fast; RAG retrieval and embedding lookups are fast if local/vector store is tuned. However, a text LLM run (especially a large reasoning run) can add 100s ms to multiple seconds depending on model, prompt size, and token budget. • End-to-end perceived latency: depends on when you wait for the text-agent result before speaking. If you wait for the entire text-agent output, the user may perceive a pause. Patterns to reduce perceived latency • Acknowledge immediately on realtime agent: have the realtime agent stream a brief acknowledgement (“One moment — I’m checking that now.”) while the text agent runs. This immediate streaming keeps the conversation natural. • Stream the text-agent output: if your text model supports streaming token-by-token, forward tokens as they arrive to TTS so speech begins earlier. • Incremental RAG: run fast retrieval first (FAQ match) and respond if high-confidence; only escalate to heavy LLM calls when needed. • Use summarization or short probing prompts: ask the text agent quick targeted questions to reduce compute vs. a full reasoning pass. • Cache and prefetch: cache frequent FAQ answers, precompute embeddings, and prewarm models; this drops lookup and inference time. • Use parallelism: run DB/RAG in parallel to LLM calls when possible, and synthesize partial replies from available info first. • Choose models by latency profile: use smaller/faster text models for quick lookups and reserve larger models for complex tasks. Implementation sketch (how to stream in your architecture) 1. RealtimeAgent receives audio stream → STT/ReatimeModel transcribes and partially responds (streaming). 2. If task needs backend: ◦ RealtimeAgent immediately streams an acknowledgement via TTS. ◦ Kick off TextAgent in background for DB/RAG/LLM. ◦ If TextAgent supports streaming, forward tokens to TTS as they arrive. ◦ When TextAgent returns structured result, finalize customer-facing reply (confirm or expand). 1. If TextAgent result arrives fast, you can skip the acknowledgement and speak full answer immediately; otherwise use the incremental path above. Trade-offs to consider • Early streaming (acknowledge + partial reply) improves UX but must be designed to avoid confusion when the backend later contradicts the partial reply. • Streaming TTS from tokens can sound less smooth than waiting for a polished full answer, so tune voice/pauses and consider using a dedicated streaming TTS optimized for low-latency. • Fully streaming large LLM outputs ties up compute longer; balance cost vs. user experience. Practical tips • Precompute FAQ embeddings and keep them in a low-latency vector store (or in-memory FAISS) to keep retrieval under ~50–100 ms. • Use a small, fast text model for intent classification and quick answers; fall back to a larger model only for complex reasoning. • If you use OpenAI Realtime or similar, use the model’s streaming API for token-level outputs and pair it with streaming TTS to reduce perceived delay. • Monitor metrics: measure STT latency, retrieval time, model latency, and TTS streaming time so you can optimize the slowest stages. If you want, I can: • Provide a small Python example showing streaming acknowledgement + background text-agent handoff with token streaming to TTS. • Suggest concrete model choices and expected latency ranges for each stage based on your hosting and vector store choices. Which would help most next: a code example for streaming handoffs, or model/store recommendations with estimated latencies?

Yes, I need to see the streaming +ack + handoff and token streaming to tts

Great — I’ll give you a focused, practical Python example that demonstrates: • Streaming acknowledgement (immediate voice feedback to the user) • Handoff from a low-latency realtime agent to a heavier text agent for DB/RAG work • Streaming token-by-token text output from the text agent into TTS so the user hears the answer as it’s generated I’ll keep it minimal and clearly annotated so you can adapt it. This example uses the LiveKit Agents patterns discussed earlier and OpenAI for embeddings / text LLMs; replace with your actual SDK bindings if names differ. Before running • Python 3.9+ • Install: pip install "livekit-agents[openai]" openai python-dotenv aiohttp • Create a .env with OPENAI_API_KEY (and LiveKit keys if you’ll run in LiveKit environment) Conceptual flow implemented 1. RealtimeAgent receives user audio → transcribes/understands it. 2. RealtimeAgent immediately streams an acknowledgement (TTS) so the user hears something quickly. 3. RealtimeAgent hands off the heavy work to TextAgent (DB/RAG). TextAgent runs and streams tokens as they are produced. 4. Each streaming token chunk is forwarded to TTS so speech begins before the full answer is finished. Key implementation notes • session.userdata holds small persistent state (saves tokens). • We stream tokens from the text LLM using an async generator and forward chunks to TTS as sentences or partial phrases for smoother speech. • The realtime agent never waits silently — it acknowledges first to reduce perceived latency. Compact runnable sketch (Adapt SDK class names and wiring for your environment; this is a working pattern rather than copy-paste turnkey code.)

# stream_handoff.py
import os
import asyncio
from dataclasses import dataclass
from typing import Optional, AsyncGenerator
import openai
from dotenv import load_dotenv

# Load keys
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# Placeholder imports — replace with your actual LiveKit Agents SDK imports
from livekit.agents import Agent, AgentSession, function_tool, ChatContext

# ---------- session userdata ----------
@dataclass
class SessionState:
    user_name: Optional[str] = None
    latest_query: Optional[str] = None
    latest_answer: Optional[str] = None

# ---------- Helper: stream OpenAI Chat tokens as async generator ----------
async def stream_chat_completion(prompt: str) -> AsyncGenerator[str, None]:
    """
    Uses OpenAI ChatCompletion streaming to yield text chunks.
    Replace this with your LLM's streaming API as needed.
    """
    # create streaming request
    loop = asyncio.get_event_loop()
    q = asyncio.Queue()

    def _on_data(data):
        # callback receiving streaming chunks — push text to queue
        if "choices" in data:
            for ch in data["choices"]:
                if ch.get("delta") and ch["delta"].get("content"):
                    loop.call_soon_threadsafe(q.put_nowait, ch["delta"]["content"])

    # Use openai.ChatCompletion.create with stream=True in a thread to avoid blocking
    def run_stream():
        resp = openai.ChatCompletion.create(
            model="gpt-4o-mini",
            messages=[{"role":"system","content":"You are a backend assistant."},
                      {"role":"user","content": prompt}],
            max_tokens=400,
            temperature=0.0,
            stream=True,
        )
        for event in resp:
            # each event is a chunk dict; push any content
            if "choices" in event:
                for ch in event["choices"]:
                    delta = ch.get("delta", {})
                    text = delta.get("content")
                    if text:
                        loop.call_soon_threadsafe(q.put_nowait, text)
        loop.call_soon_threadsafe(q.put_nowait, None)  # sentinel when finished

    # run streaming in executor to not block
    executor = asyncio.get_event_loop().run_in_executor
    asyncio.get_event_loop().run_in_executor(None, run_stream)

    while True:
        chunk = await q.get()
        if chunk is None:
            break
        yield chunk

# ---------- TextAgent: does RAG/DB work and streams answer ----------
class TextAgent(Agent):
    def __init__(self, chat_ctx: Optional[ChatContext] = None):
        super().__init__(instructions="Backend agent for DB/RAG", chat_ctx=chat_ctx)

    async def stream_answer_for_query(self, context, query: str) -> AsyncGenerator[str, None]:
        """
        Orchestrates retrieval + LLM streaming. Yields token chunks as they come.
        For demo: we directly stream an LLM completion on composed prompt.
        """
        # Example: run a fast RAG lookup (synchronous/fake here). Use your vector DB lookups.
        # rag_hit = rag_lookup(query)  # quick retrieval to include in prompt
        rag_hit = "Found FAQ: Resetting your password: Visit login -> 'Forgot Password'."

        prompt = (
            f"Question: {query}\n\n"
            f"Retrieved context: {rag_hit}\n\n"
            "Provide a concise helpful answer."
        )
        # stream LLM output token-by-token and yield each chunk
        async for chunk in stream_chat_completion(prompt):
            yield chunk

# ---------- RealtimeAgent: interacts with user, acknowledges, and hands off ----------
class RealtimeAgent(Agent):
    def __init__(self, session: AgentSession):
        super().__init__(instructions="Friendly voice agent")
        self.session = session

    async def on_enter(self):
        state: SessionState = self.session.userdata
        if state.user_name:
            await self.session.generate_reply(instructions=f"Welcome back {state.user_name}, how can I help?")
        else:
            await self.session.generate_reply(instructions="Hello! What's your name?")

    @function_tool()
    async def handle_user_text(self, context, user_text: str):
        """
        Called when user speech is transcribed to text.
        - Immediately stream acknowledgement TTS
        - Start TextAgent in background and stream its token output to TTS
        """
        state: SessionState = context.session.userdata
        state.latest_query = user_text

        # 1) Immediate acknowledgement (short, low-latency)
        # Use session TTS / generate_reply to speak quick ack
        await self.session.generate_reply(instructions="One moment — I'll check that for you.")

        # 2) Start backend text agent stream
        text_agent = TextAgent()
        # We pass current convo summary in chat context if wanted (keeps tokens small)
        chat_ctx = ChatContext()
        chat_ctx.add_message(role="user", content=f"User asked: {user_text}")

        # Kick off streaming generator from text agent
        # (If SDK supports streaming directly via tools, use that; here we call helper)
        async for token_chunk in text_agent.stream_answer_for_query(context, user_text):
            # Option A: buffer chunks until a sentence boundary then call TTS
            # Simple heuristic: forward every 200 chars or on punctuation
            # For demo, forward each chunk to TTS immediately in small batches
            # Replace session.send_audio_chunk(...) with your actual TTS streaming API
            await self.session.send_text_to_tts(token_chunk)

        # 3) After finished, update state.latest_answer (you could collect full text)
        # For simplicity, we assume `stream_answer_for_query` finalizes the spoken response.
        return "Answered via backend (streaming complete)."

# ---------- Mocked AgentSession methods you must implement per SDK ----------
# The real SDK will have methods to: start session, set userdata, generate_reply, send_text_to_tts
# Below is only illustrative of how you'd call them.

# ---------- Run / Integration ----------
async def demo_flow():
    # Create AgentSession (SDK-specific) and set userdata
    session = AgentSession(...)  # fill with your STT/Realtime/tts config
    session.userdata = SessionState()

    realtime_agent = RealtimeAgent(session=session)
    # Start agent session per SDK: session.start(agent=realtime_agent, room=...)
    await session.start(agent=realtime_agent)

    # Imagine we got a transcription "How do I reset my password?"
    user_text = "How do I reset my password?"
    # In real system, this would be triggered when STT delivers the text
    await realtime_agent.handle_user_text(context=session, user_text=user_text)

# Entry point
if __name__ == "__main__":
    asyncio.run(demo_flow())

Practical tips for using this pattern in production • Use a streaming-capable LLM (or the provider’s streaming API) so you can forward tokens progressively. • Forward tokens to a low-latency streaming TTS service rather than waiting for the entire text; many TTS services accept incremental text or SSML chunks. • For smoother speech, group tokens into sentence-level chunks before sending to TTS. A small buffer (e.g., 200–400 ms or punctuation detection) reduces choppy audio. • Acknowledge immediately in voice before heavy backend work to keep the conversation feeling responsive. • Precompute FAQ embeddings and keep RAG retrieval under ~50–100 ms by using an in-memory or optimized vector DB. • Monitor and tune: measure STT latency, LLM stream latency, and TTS latency to find bottlenecks. Would you like: • A more concrete end-to-end runnable example wired to a specific TTS streaming API (e.g., Amazon Polly/Neural, or your provider)? • A version that uses a vector DB (FAISS Sources: Google AI and LiveKit | LiveKit Documentation | LLM Output Replacement | LiveKit Documentation

raise RuntimeError("trying to generate speech from text without a TTS model")

Thanks — that RuntimeError means your session tried to convert text to audio but no TTS engine was configured for the AgentSession (or the agent-level call that generates speech wasn’t given a TTS). I’ll explain why it happens, how to fix it, and give a minimal Python example showing the correct places to configure TTS so token streaming can be sent to a TTS model. Why this error happens • The agent runtime tries to call a method that synthesizes speech (e.g., session.generate_reply, session.send_text_to_tts, or forwarding streaming tokens to the session TTS). • If the AgentSession (or the code path) does not have a TTS instance attached, the runtime raises: "trying to generate speech from text without a TTS model". • In short: you must provide a TTS model when you create the session or call a TTS-specific API. How to fix it (short) • Ensure you attach a TTS model when creating the AgentSession. • When streaming tokens, send them to the session's TTS/speech output API (the SDK method that expects a configured TTS). • If you want to control voice synthesis separately, instantiate and call your chosen TTS library/service directly (bypass session-level TTS) and stream its audio to the client. Minimal examples 1) Configure TTS on AgentSession (preferred when using LiveKit Agents) • Pass a TTS implementation to the session so session.generate_reply / built-in streaming to TTS works. Example (pseudo-code adapted to LiveKit Agents):

from livekit.plugins import openai, cartesia  # replace with real SDK imports

# create TTS model (provider-specific)
tts = cartesia.TTS(model="cartesia/sonic-3", voice="alloy")  # example provider

# create realtime session with STT and TTS configured
session = AgentSession(
    stt=openai.STT(model="deepgram/nova-3"),       # or your STT
    llm=openai.realtime.RealtimeModel(voice="alloy", modalities=["text"]),
    tts=tts,                                      # <<< important: attach TTS here
    vad=...,                                      # optional
)

# Now calls like session.generate_reply(...) or session.send_text_to_tts(...) will work
await session.start(agent=my_realtime_agent, room=room)

2) If you prefer to use an external TTS service directly (explicit streaming) • Instead of relying on AgentSession TTS, synthesize audio yourself and push audio frames to the LiveKit audio output or the session output API. Example sketch:

# 1) stream tokens from LLM
async for token in text_llm_stream(prompt):
    buffer += token
    if is_sentence_boundary(buffer):
        # 2) call external TTS (e.g., Amazon Polly, Google, or other) to synthesize 'buffer'
        audio_bytes = synthesize_text_to_audio(buffer)
        # 3) push audio to session output (SDK method; replace with your API)
        await session.output.push_audio(audio_bytes)
        buffer = ""
# after loop, flush remaining buffer

• With this approach, you must ensure your

session.output.push_audio

(or equivalent) is used to deliver audio; you will not use session.generate_reply. Troubleshooting checklist • Did you pass a TTS object into AgentSession at creation? If not — add it. • Are you calling a session method that expects a TTS but you only configured a text-only LLM? Provide the TTS or use an external TTS flow. • If you configured TTS but still see the error: ◦ Confirm the TTS object is non-null: print(session.tts) right before you generate speech. ◦ Check the SDK’s expected TTS type and constructor signature (model name, voice param). ◦ Ensure the TTS provider credentials are set in env vars and the TTS initialization succeeded. • If you want token streaming to TTS, confirm the SDK supports incremental TTS input; otherwise call your TTS API incrementally as you receive chunks. If you want, I can: • Provide a concrete runnable snippet using the exact LiveKit plugin names you’re using (OpenAI Realtime + Cartesia/another TTS). • Inspect the exact code where you call generate_reply / send_text_to_tts and suggest the minimal changes to attach TTS. Share: • The AgentSession creation code you’re using, or • The exact call stack / line that raises the RuntimeError and I’ll give the precise fix.

As I said I need to use realtime model for speaking

Thanks — I understand you want the realtime model to actually speak (not just do STT), and you hit the TTS error. Let’s fix that and show a concrete, minimal Python example that: • Uses a realtime-capable model for speech I/O (realtime model does speaking) • Streams an immediate acknowledgement, then hands off to a text agent • Streams token chunks from the text agent into speech with low perceived latency • Avoids the RuntimeError by correctly configuring a TTS path (either via the realtime model’s built-in TTS or by streaming text into an explicit TTS engine) Two viable approaches (pick one) • Option A — Realtime model synthesizes speech directly (easiest if your realtime provider supports streaming TTS output). You must configure the AgentSession with that realtime model that supports voice output. • Option B — Realtime model handles comprehension, but you stream text tokens into a separate TTS engine (recommended when you need full control over voice/streaming behavior). This is what I’ll demonstrate because it’s robust and avoids SDK-specific TTS pitfalls. Below is a focused Python example (adapt to your LiveKit SDK names). It shows: 1) RealtimeAgent that uses a realtime model for low-latency speech input and can synthesize quick acknowledgements via the realtime model TTS API (if available) or via the session’s TTS stub. 2) Background TextAgent that streams tokens using the provider streaming API. 3) A small streamer that forwards token chunks to a streaming TTS function (you can implement with any streaming TTS API), so audio plays as text is generated. 4) How to avoid the RuntimeError: make sure a TTS engine (either session.tts or your external TTS client) is present before sending text-to-audio. Important note: adjust class/method names to match the exact LiveKit Agents SDK you use — function names here are illustrative of the required wiring. Example (concise, adapt to your SDK):

import os
import asyncio
from dataclasses import dataclass
from typing import Optional, AsyncGenerator
import openai
from dotenv import load_dotenv

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# --- session state kept off LLM tokens ---
@dataclass
class SessionState:
    user_name: Optional[str] = None
    latest_query: Optional[str] = None
    latest_answer: Optional[str] = None

# --- Streaming chat completion using OpenAI streaming API (example) ---
async def stream_chat_completion(prompt: str) -> AsyncGenerator[str, None]:
    """
    Yields text chunks from the LLM as they arrive.
    Replace with your provider's streaming method if different.
    """
    loop = asyncio.get_event_loop()
    q = asyncio.Queue()

    def run_stream():
        resp = openai.ChatCompletion.create(
            model="gpt-4o-mini", messages=[{"role":"user","content":prompt}],
            stream=True, max_tokens=400, temperature=0.0
        )
        for event in resp:
            # event is chunk; push deltas with content
            for ch in event.get("choices", []):
                delta = ch.get("delta", {})
                text = delta.get("content")
                if text:
                    loop.call_soon_threadsafe(q.put_nowait, text)
        loop.call_soon_threadsafe(q.put_nowait, None)

    # run blocking stream in thread
    asyncio.get_event_loop().run_in_executor(None, run_stream)

    while True:
        chunk = await q.get()
        if chunk is None:
            break
        yield chunk

# --- Example external streaming TTS stub (replace with real TTS) ---
async def synthesize_and_play_incremental(text_chunk: str):
    """
    Replace this with your streaming TTS API call.
    The function should synthesize `text_chunk` to audio and play/send it to the user's audio output.
    """
    # Example placeholder: print to console to show streaming
    print("[TTS STREAM]:", text_chunk, flush=True)
    # In production: send audio frames to LiveKit (session.output.push_audio or equivalent)
    await asyncio.sleep(0.01)  # simulate tiny network/audio delay

# --- TextAgent that streams text tokens ---
class TextAgent:
    async def stream_answer(self, query: str) -> AsyncGenerator[str, None]:
        # e.g., optionally include fast RAG retrieval here
        rag_context = "Retrieved FAQ: Reset password -> Visit login > Forgot Password."
        prompt = f"Context: {rag_context}\nQuestion: {query}\nAnswer concisely:"
        async for token in stream_chat_completion(prompt):
            yield token

# --- RealtimeAgent orchestrates ack + handoff + streaming to TTS ---
class RealtimeAgent:
    def __init__(self, session):
        self.session = session  # your AgentSession, must include userdata and audio output hooks

    async def immediate_ack(self):
        """
        Speak an immediate acknowledgement to reduce perceived latency.
        Use session TTS if available, otherwise use external TTS.
        """
        ack_text = "One moment, I will check that for you."
        # Preferred: session-level TTS if configured
        if getattr(self.session, "tts", None):
            # SDK-specific method to speak text using session TTS
            await self.session.speak(ack_text)   # pseudocode: use your SDK method
        else:
            # fallback to external TTS streaming
            await synthesize_and_play_incremental(ack_text)

    async def handle_user_text(self, user_text: str):
        # 1) persist minimal state (saves tokens)
        state: SessionState = self.session.userdata
        state.latest_query = user_text

        # 2) speak immediate ack (low-latency)
        await self.immediate_ack()

        # 3) handoff to TextAgent and stream its tokens to TTS
        text_agent = TextAgent()
        buffer = ""
        sentence_endings = {".", "?", "!"}

        async for chunk in text_agent.stream_answer(user_text):
            buffer += chunk
            # optional: detect sentence boundaries for smoother TTS
            if any(p in buffer for p in sentence_endings) or len(buffer) > 200:
                # extract chunk up to last sentence boundary
                # simple heuristic: find last punctuation
                last_punc_idx = max((buffer.rfind(p) for p in sentence_endings))
                if last_punc_idx != -1:
                    to_speak = buffer[:last_punc_idx + 1]
                    buffer = buffer[last_punc_idx + 1:]
                else:
                    to_speak = buffer
                    buffer = ""
                # stream to TTS (incrementally)
                await synthesize_and_play_incremental(to_speak)

        # flush any remaining buffer
        if buffer.strip():
            await synthesize_and_play_incremental(buffer)

        # Optionally store final answer in userdata
        state.latest_answer = "stored locally or reconstructed if needed"

# ---------------- Example glue to run locally ----------------
class MockSession:
    def __init__(self):
        self.userdata = SessionState()
        self.tts = None  # set to a TTS object if you want session.speak to be available

    # Example placeholder method to demonstrate session.speak usage
    async def speak(self, text: str):
        # If you configure a TTS engine here, synthesize and output audio.
        # For demo: call the external synthesize stub
        await synthesize_and_play_incremental(text)

async def demo():
    session = MockSession()
    # If you want to use the session-level TTS, attach an object that exposes speak()
    # session.tts = YourTTSClient(...)
    agent = RealtimeAgent(session)
    await agent.handle_user_text("How do I reset my password?")

if __name__ == "__main__":
    asyncio.run(demo())

Troubleshooting checklist to avoid the RuntimeError • Make sure your AgentSession has a TTS configured if you call session.generate_reply or session.speak. Example: ◦ session = AgentSession(..., tts=YourTTSClient(...), ...) • If the SDK expects a specific TTS type, instantiate it using the SDK plugin (e.g., cartesia.TTS(...) or openai.TTS(...) depending on your SDK). • If you prefer to manage TTS yourself (Option B above), ensure you never call the SDK speech method without a configured TTS. Instead, call your external TTS and stream its audio frames to the user via the session output API. • Verify the exact error stack: it tells you which code path called the missing TTS — that helps you decide whether to attach a session TTS or to change to external TTS calls. Next steps I can provide • A concrete runnable example wired to a real streaming TTS provider you prefer (Amazon Polly Streaming / Google TTS streaming / provider X) and an example of sending audio frames to LiveKit. • An example using the realtime model’s built-in TTS if your realtime provider supports direct voice output (I’ll need the exact SDK/provider names you’re using). • Help update the exact line in your current code that triggers the RuntimeError if you paste that snippet or the session creation code. Which do you want next: a real provider example (pick one: Amazon Polly, Google Cloud TTS, Cartesia, or OpenAI Realtime TTS) — or paste your AgentSession creation code and I’ll point out the exact change to fix the RuntimeError?