https://livekit.io logo
Join Slack
Powered by
How can I integrate a RAG with uploading file?
# ask-ai
b
How can I integrate a RAG with uploading file?
t
Thanks for reaching out, Wesley. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m responding to your question about integrating a Retrieval‑Augmented Generation (RAG) system that accepts uploaded files. Below is a concise, practical plan you can implement with LiveKit Agents, plus the key code hooks and options to handle file uploads, indexing, and retrieval. High‑level flow 1. Frontend: user uploads a file and sends it to the agent using the LiveKit client SDK 
sendFile
(file is sent as a byte stream). 2. Agent: register a byte‑stream handler for the file topic to receive and reconstruct the file. 3. Extract: convert the received file into plain text (use a parser library appropriate for the file type). 4. Index: store the extracted text into a vector database or search index for retrieval. 5. RAG at query time: on relevant user turns, run a retrieval query, inject the retrieved context into the agent’s chat context, and let the LLM generate a response that uses the retrieved content. 6. UX: optionally stream progress/status updates to the user while processing and indexing. Concrete steps and implementation notes 1) Upload file from the frontend • Use the LiveKit client SDK’s 
sendFile
method to send the file as a byte stream to your agent. This ensures secure transfer and lets the agent receive chunks incrementally. 2) Receive the file in the agent • Register a byte stream handler in the agent for your chosen topic (e.g., 
"files"
). • The handler should read incoming byte chunks asynchronously and reconstruct the file in memory or write to temporary storage. 3) Convert file to text (extraction) • After reconstructing the file, convert it to text: ◦ PDFs → use a PDF parser (e.g., PyPDF2, pdfminer.six) or an OCR tool if it’s scanned. ◦ DOCX → use python-docx. ◦ Images → OCR (Tesseract or an OCR API). • The LiveKit docs don’t specify internal parsers for every file type, so use external libraries or services for extraction. 4) Index text into a vector store • Split the extracted text into chunks (e.g., 500–1000 tokens) with overlap. • Generate embeddings (your chosen embedding model) for each chunk. • Upsert those embeddings and chunk metadata into your vector DB (e.g., Pinecone, Milvus, Weaviate, or an in‑house store). 5) Trigger RAG lookups during conversation • Use an agent lifecycle hook such as 
on_user_turn_completed
to run retrieval when a user message arrives (or when you want to perform retrieval). • Perform a similarity search against your vector DB to fetch top relevant chunks. • Inject the retrieved text into the agent’s chat context as an assistant/system message or as additional context so the model can use it to generate an answer. Example pattern (pseudocode)
Copy code
async def on_user_turn_completed(self, turn_ctx, new_message):
    # 1) Do a vector search for new_message.text
    retrieved = await my_rag_lookup(new_message.text_content())

    # 2) Inject as context for generation
    turn_ctx.add_message(
        role="assistant",
        content=f"Retrieved context:\n{retrieved}"
    )

    # 3) Proceed with normal generation (agent will call LLM)
6) Real‑time feedback (optional) • While the file is being processed/indexed, use session generation/status updates to notify the user that processing is in progress. This improves perceived responsiveness. Practical considerations and tips • Chunking strategy and embedding latency impact retrieval relevance and response time; tune chunk size and overlap for your content. • Keep metadata (filename, upload time, uploader, doc type, page numbers) with indexed vectors to surface provenance in responses. • For large files, process and index asynchronously; inform the user when the document is ready for queries. • Sanitize and validate uploads (size limits, file types) before processing. Where to look for examples • There is an example repository demonstrating RAG integration with external data (LlamaIndex example). Use it as a reference for indexing and hooking retrieval into the agent flow. Next steps I can help with • If you want, tell me: ◦ Which file types you need to support (PDF, DOCX, images)? ◦ Which vector DB and embedding model you plan to use (or if you need recommendations)? ◦ Whether you want synchronous answers right after upload or asynchronous processing with a “document ready” notification? I’ll provide a step‑by‑step code example tailored to your stack. Sources: Vision | LiveKit Documentation | External data and RAG | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation
b
can you show me the python code?
t
Thanks — I’m responding to your request for a Python code example that shows how to integrate RAG with file uploads using LiveKit Agents. To give you a focused, runnable example, I need one quick clarifying detail: • Which vector store and embedding provider do you plan to use? (options: Pinecone, Milvus, Weaviate, or a simple local FAISS/local embeddings). If you don’t have a preference, I’ll use FAISS for local vector storage and OpenAI-style embeddings as a demonstrative default. If that’s fine, I’ll provide a concise example that covers: 1. Receiving a file byte stream in the agent (handler for 
files
topic). 2. Extracting text from a PDF or DOCX (I’ll include PDF extraction via 
pdfplumber
and a DOCX path with 
python-docx
). 3. Chunking and embedding text (example with OpenAI embeddings). 4. Indexing into FAISS. 5. Triggering retrieval in 
on_user_turn_completed
, injecting retrieved context, and generating the reply. Tell me which vector store you prefer, and I’ll produce the complete Python code.
b
FAISS
t
Thanks — I’ll show a concise, practical Python example that demonstrates these pieces together using FAISS: • receiving a file byte stream from the LiveKit agent (topic "files"), • extracting text from PDF and DOCX, • chunking text and creating embeddings (example uses OpenAI-style embeddings — replace with your embedding provider), • indexing into FAISS, • performing retrieval in 
on_user_turn_completed
, injecting retrieved context, and generating a reply. Before using this code, install required packages: • faiss-cpu, openai (or your embedding client), pdfplumber, python-docx, tiktoken (optional for token chunking). Example (single-file illustrative agent):
Copy code
import io
import asyncio
from typing import List

import faiss
import numpy as np
import pdfplumber
import docx

# Replace with your embeddings client. Example shows OpenAI-style interface.
import openai

# --- Simple text extraction helpers ---
def extract_text_from_pdf_bytes(b: bytes) -> str:
    with pdfplumber.open(io.BytesIO(b)) as pdf:
        pages = [p.extract_text() or "" for p in pdf.pages]
    return "\n".join(pages)

def extract_text_from_docx_bytes(b: bytes) -> str:
    doc = docx.Document(io.BytesIO(b))
    paras = [p.text for p in doc.paragraphs]
    return "\n".join(paras)

# --- Chunking helper ---
def chunk_text(text: str, max_chars: int = 2000, overlap: int = 200) -> List[str]:
    chunks = []
    start = 0
    length = len(text)
    while start < length:
        end = min(start + max_chars, length)
        chunks.append(text[start:end])
        start = end - overlap
        if start < 0:
            start = 0
    return chunks

# --- Embeddings helper (OpenAI-style) ---
def get_embeddings(texts: List[str]) -> List[List[float]]:
    # Batch calls if needed.
    # Replace 'text-embedding-3-small' with your model.
    resp = openai.Embedding.create(model="text-embedding-3-small", input=texts)
    return [d["embedding"] for d in resp["data"]]

# --- FAISS wrapper ---
class FaissIndex:
    def __init__(self, dim: int):
        self.dim = dim
        self.index = faiss.IndexFlatL2(dim)  # simple flat index
        self.metadatas = []  # keep metadata and original text

    def add(self, embeddings: List[List[float]], metadatas: List[dict]):
        arr = np.array(embeddings).astype("float32")
        self.index.add(arr)
        self.metadatas.extend(metadatas)

    def search(self, query_emb: List[float], top_k: int = 5):
        q = np.array([query_emb]).astype("float32")
        distances, indices = self.index.search(q, top_k)
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx == -1 or idx >= len(self.metadatas):
                continue
            results.append({"distance": float(dist), "metadata": self.metadatas[idx]})
        return results

# --- Simple in-memory document store and index manager ---
class DocumentManager:
    def __init__(self, embed_dim=1536):
        self.index = FaissIndex(embed_dim)

    async def add_document_from_bytes(self, filename: str, content_bytes: bytes):
        # Decide parser by file extension
        if filename.lower().endswith(".pdf"):
            text = extract_text_from_pdf_bytes(content_bytes)
        elif filename.lower().endswith(".docx"):
            text = extract_text_from_docx_bytes(content_bytes)
        else:
            # treat as plain text
            text = content_bytes.decode("utf-8", errors="ignore")

        chunks = chunk_text(text)
        embeddings = get_embeddings(chunks)
        metadatas = [{"filename": filename, "chunk": i, "text": chunks[i][:1000]} for i in range(len(chunks))]
        self.index.add(embeddings, metadatas)
        return {"status": "ok", "chunks_indexed": len(chunks)}

    async def query(self, query_text: str, top_k: int = 3):
        q_emb = get_embeddings([query_text])[0]
        return self.index.search(q_emb, top_k=top_k)

# --- LiveKit Agent byte-stream handler & lifecycle hook (pseudocode integration) ---
# Replace the agent base classes and decorators with the actual LiveKit Agents SDK constructs.
class ExampleAgent:
    def __init__(self):
        # Set embedding dim to match your embedding model
        self.docs = DocumentManager(embed_dim=1536)

    async def handle_file_stream(self, filename: str, stream_async_iter):
        # stream_async_iter yields bytes chunks
        chunks = []
        async for b in stream_async_iter:
            chunks.append(b)
        file_bytes = b"".join(chunks)
        result = await self.docs.add_document_from_bytes(filename, file_bytes)
        # Optionally notify user the file was processed via session.generate_reply or similar.
        return result

    async def on_user_turn_completed(self, turn_ctx, new_message):
        # new_message.text_content() -> the user query
        user_q = new_message.text_content()
        # Run retrieval
        results = await self.docs.query(user_q, top_k=4)

        # Build a context string from retrieved results
        context_parts = []
        for r in results:
            md = r["metadata"]
            context_parts.append(f"From {md['filename']} (chunk {md['chunk']}):\n{md['text']}\n---\n")

        retrieved_context = "\n".join(context_parts) if context_parts else "No relevant docs found."

        # Inject the retrieved context into the turn context before generation
        turn_ctx.add_message(role="assistant", content=f"Retrieved context:\n{retrieved_context}")

        # The agent framework should now proceed to call the LLM to generate the reply
        # Optionally you can call session.generate_reply() or let the framework continue.

# --- Usage example (conceptual) ---
async def demo_flow():
    agent = ExampleAgent()

    # Simulate receiving a PDF upload (in practice this comes from LiveKit's byte stream)
    with open("example.pdf", "rb") as f:
        file_bytes = f.read()

    await agent.docs.add_document_from_bytes("example.pdf", file_bytes)

    # Simulate an incoming user message/turn
    class FakeMessage:
        def __init__(self, text): self._t = text
        def text_content(self): return self._t

    class FakeTurnCtx:
        def add_message(self, role, content):
            print(f"Injected to turn ctx ({role}):\n{content[:1000]}...\n")

    await agent.on_user_turn_completed(FakeTurnCtx(), FakeMessage("What does the document say about refund policy?"))

if __name__ == "__main__":
    asyncio.run(demo_flow())
Notes and next steps • Replace the embedding calls (
get_embeddings
) with your chosen provider if not using OpenAI. Ensure the embedding dimension passed to FAISS matches your model’s dimension. • For production: ◦ Persist FAISS index and metadata (save/load). ◦ Process large files asynchronously and notify users when indexing completes. ◦ Add file validation (max size, allowed types) and sanitize inputs. ◦ Use more advanced FAISS indexes (IVF, HNSW) for scale. ◦ Keep provenance (page numbers, offsets) in metadata to surface exact text snippets. • If you want, I can: ◦ Convert this pseudocode to match your exact LiveKit Agents SDK method names and event signatures. ◦ Add streaming progress updates to the user during upload/processing. ◦ Show persistence (save/load FAISS + metadata) and a production-ready indexing pipeline. Which would you like next?
b
how can I start an agent from scratch in python
t
You're asking how to start an agent from scratch in Python — great! Below is a concise, step‑by‑step guide plus a minimal working example you can use as a starting point. I’ll cover environment setup, project starter, dependencies, a minimal agent implementation, and how to run it locally or in dev mode. Before you begin • Use Python 3.10–3.13. • Install the uv tool (used for managing and running the starter project). 1) Initialize the project • Clone the Python agent starter repository: ◦ https://github.com/livekit-examples/agent-starter-python • That repo includes a production-ready layout (src/agent.py, pyproject.toml, Dockerfile, README). 2) Install dependencies • From the project root, install packages: ◦ uv pip install "livekit-agents[silero]" python-dotenv • The 
silero
plugin loads voice utilities (VAD) commonly used for turn detection. 3) Create your environment variables • Add a 
.env
file with your LiveKit credentials: ◦ LIVEKIT_URL=your_livekit_url ◦ LIVEKIT_API_KEY=your_api_key ◦ LIVEKIT_API_SECRET=your_api_secret 4) Download required model files • Download models used by plugins (VAD, noise cancellation, etc.): ◦ uv run agent.py download-files 5) Minimal agent code • Save this as 
agent.py
(or adapt starter repo). This is a minimal example that preloads VAD, defines an Agent, and runs via the provided CLI:
Copy code
import logging
from dotenv import load_dotenv
from livekit.agents import Agent, AgentSession, AgentServer, JobContext, cli
from livekit.plugins import silero

load_dotenv()

logger = logging.getLogger("my-agent")
logger.setLevel(<http://logging.INFO|logging.INFO>)

server = AgentServer()

def prewarm(proc):
    # preload VAD to speed up connection handling
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

class MyAgent(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice assistant.")

    async def on_enter(self):
        # sends an initial greeting to the session participants
        self.session.generate_reply("Hello! How can I help you?")

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        vad=ctx.proc.userdata["vad"],
        stt="deepgram/nova-3-general",
        llm="openai/gpt-4.1-mini",
        tts="cartesia/sonic-2:6f84f4b8-58a2-430c-8c79-688dad597532"
    )
    await session.start(agent=MyAgent(), room=ctx.room)
    await ctx.connect()

if __name__ == "__main__":
    cli.run_app(server)
6) Run the agent locally • Development console mode: ◦ uv run agent.py console • Dev mode (connect to LiveKit Cloud for testing): ◦ uv run agent.py dev • Production start: ◦ uv run agent.py start 7) Deploying (optional) • Use the LiveKit CLI to scaffold and deploy: ◦ lk agent create • That will prepare Docker/config files and help deploy your agent. Practical tips • Adjust STT/LLM/TTS model strings to match your providers. • For production: persist session state as needed, add logging/metrics, validate uploaded files, and tune resource limits. • If you plan to add RAG or file uploads, I can integrate the FAISS example into this agent structure. Would you like: • A tailored agent template that includes the FAISS RAG flow (file upload + indexing + retrieval)? • Or help wiring specific STT/LLM/TTS providers you plan to use? Sources: Voice AI quickstart | LiveKit Documentation | Agents (Python) | livekit/livekit_composite | DeepWiki
b
I mean what is the lk command to create an agent and deploy it
t
You're asking for the exact LiveKit CLI commands to create and deploy an agent. Use these: • Create a new agent scaffold and deploy it (one-time setup):
Copy code
lk agent create
This creates Docker/config files (
Dockerfile
, 
livekit.toml
, etc.), packages your agent code, and deploys it to LiveKit Cloud. • Redeploy an updated agent:
Copy code
lk agent deploy
Notes: • Ensure your LiveKit credentials are configured (env vars or in 
livekit.toml
) before running the commands. • After 
lk agent create
, you can customize the generated 
Dockerfile
and 
livekit.toml
then use 
lk agent deploy
for subsequent updates. Would you like a quick checklist of the env vars or example 
livekit.toml
to use before running these commands? Sources: Deploy LiveKit Agents with GitHub Actions | LiveKit | Agent deployment overview | LiveKit Documentation | Voice AI quickstart | LiveKit Documentation
b
wesley@wesley:~/Documents/projects/popai/voice-agent-demo$ lk agent create Using default project [pb] Creating new agent deployment unable to determine agent language: expected package.json, requirements.txt, pyproject.toml, or lock files, please navigate to a directory containing an agent written in a supported language wesley@wesley:~/Documents/projects/popai/voice-agent-demo$
t
Thanks — I see the CLI error: it can’t detect the project language because your current folder is missing a recognized dependency/manifest file. I’ll help you fix that so 
lk agent create
can proceed. What happened • 
lk agent create
looks for one of these files to determine language and package your agent: ◦ For Python: 
pyproject.toml
or 
requirements.txt
◦ For Node: 
package.json
• Your directory doesn’t contain any of those files, so the CLI stops with the error you saw. How to fix it (quick options) • If you already have a Python agent, ensure you’re in the correct directory (the one that contains 
pyproject.toml
or 
requirements.txt
). Then re-run the command. • If you don’t have those files yet, create one now. The simplest approach for a Python agent: Option A — create a minimal 
pyproject.toml
(recommended) 1. In your project root, create 
pyproject.toml
with minimal content:
Copy code
[build-system]
   requires = ["setuptools", "wheel"]
   build-backend = "setuptools.build_meta"
1. Add your dependencies in 
pyproject.toml
or add a 
requirements.txt
if you prefer. Option B — create 
requirements.txt
1. Create a 
requirements.txt
file listing your runtime deps, for example:
Copy code
livekit-agents[silero]
   python-dotenv
   openai
1. Save it in the project root. After adding one of those files, run:
Copy code
lk agent create
from the same directory. Extra tips • Confirm you’re in the exact folder that holds your agent entrypoint (e.g., 
agent.py
or 
src/agent.py
). The CLI packages the directory contents, so being in the right folder matters. • If you used the starter repo, run 
lk agent create
from that repo root where 
pyproject.toml
is present. • If you want, I can generate a minimal 
pyproject.toml
and example 
requirements.txt
tailored to your app. Tell me: ◦ Do you have an existing entrypoint file (e.g., 
agent.py
or 
src/agent.py
)? If so, what’s the path? ◦ Which dependencies do you need (STT/LLM/TTS plugins, RAG libs, FAISS, etc.)? Tell me which option you prefer and I’ll produce exact file contents you can paste into your project.
b
but i remember there was a command that would create a basic agent, it i would create the folders and everuything
t
Yes — the command you’re remembering is the CLI scaffold command. Use:
Copy code
lk agent init <agent_name> --lang python
What this does • Creates a new folder named 
<agent_name>
and scaffolds a complete Python agent project (source files like 
agent.py
, 
pyproject.toml
, 
Dockerfile
, 
.dockerignore
, README, etc.). • Gives you a ready starter you can run locally and then deploy. Quick next steps 1. Run the init command:
Copy code
lk agent init my-voice-agent --lang python
   cd my-voice-agent
1. Install dependencies and download model files (from the starter):
Copy code
uv pip install
   uv run agent.py download-files
1. Run in dev/console mode:
Copy code
uv run agent.py dev
   # or
   uv run agent.py console
1. Create/deploy to LiveKit Cloud:
Copy code
# one-time create (generates livekit.toml, Dockerfile and deploys)
   lk agent create --secrets-file .env

   # subsequent updates
   lk agent deploy
Troubleshooting tips • If 
lk
or 
uv
isn’t installed or authenticated, install the LiveKit CLI and authenticate (or run 
lk cloud auth
if your setup requires it). • Run 
lk agent init
from a directory where you want the new project created (it will make a new subfolder). • If you already have a project and prefer to scaffold inside it, you can copy files from the starter repo instead of 
init
. Would you like me to: • Generate the exact 
.env
and 
pyproject.toml
contents you can paste into your current folder? • Or walk through the `lk`/`uv` install and auth steps if you’re seeing CLI errors? Sources: Examples | livekit/livekit_composite | DeepWiki | Agents (Node.js) | livekit/livekit_composite | DeepWiki | Voice AI quickstart | LiveKit Documentation
b
Okay, this is my current agent.py code, please update to add the rag model
Copy code
import logging

from dotenv import load_dotenv
from livekit import rtc
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    cli,
    inference,
    room_io,
)
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

logger = logging.getLogger("agent")

load_dotenv(".env.local")


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are Jim the internal information technology support assistant for NorthBridge Consulting, a fictional consulting company. You help employees with technical issues, software access, device requests, and basic internal guidance.

General Behavior
- Communicate in a calm, friendly, and professional manner.
- Keep responses concise and natural.
- Ask only one question at a time.
- Prefer the simplest and safest action first.
- Never request sensitive information such as passwords.
- When identification is needed, ask only for the employee ID.
- Never mention internal systems, tools, or automation mechanisms.

Identity and Context

- The user is always an employee of NorthBridge Consulting.
- Request the employee ID only when required to perform actions or submit requests.

Company Approved Software Catalog

NorthBridge Consulting already provides and pays for the following software:

- CloudDocs, for document management
- MeetSpace, for video meetings
- InsightCRM, for client management
- SecureShare, for secure file exchange
- TimeTrack, for project hours and billing
- ExpenseFlow, for expense submissions
- PeopleHub, for human resources
- KnowledgeCore, for internal documentation
- TaskFlow, for internal task management
- DataBoard, for analytics and reporting

Software Access Rules

When an employee asks for access to software that exists in the approved catalog:

- Confirm assistance without mentioning approvals or back-end logic.
- Ask for the employee ID.
- After receiving the ID, confirm that the access request has been submitted.
- Inform the employee that updates will be sent by email.

Example behavior:

"I can help with that. Please provide your employee ID."

After receiving the ID:

"Thank you. Your access request has been submitted. You will receive updates by email soon."

When an employee asks for software not listed in the approved catalog:

- Inform the employee that the software is not currently available.
- Ask for a short description of the business need.
- After receiving the explanation, request the employee ID.
- Confirm that the request has been submitted for evaluation.
- Clarify that follow-up will occur by email.

Example behavior:

"That software is not currently available. Could you briefly describe why you need it?"

After receiving the justification:

"Thank you for the explanation. Please provide your employee ID so I can submit the request."

After receiving the ID:

"Your request has been submitted for review. You will receive updates by email."

Login and Access Issues

- Trigger when employees report sign-in problems or access errors.
- Provide reassurance and simple checks.
- Never request passwords.
- Escalate when needed.

Troubleshooting Behavior

- Trigger when employees report systems not working, slowness, or errors.
- Provide one simple step at a time.
- Confirm results before continuing.

Hardware and Device Requests

- Handle requests for new devices or hardware problems.
- Offer ticket creation or contact with responsible staff.

Ticketing and Escalation

- When issues require investigation or approval, submit a support request.
- Clearly state that updates will be provided by email.
- Do not expose internal identifiers or workflows.

Tone and Interaction Safety

- Maintain a supportive and professional tone.
- Avoid technical jargon unless clearly necessary.
- Never blame the employee.
- Keep conversations clear and efficient.
""",
        )

    # To add tools, use the @function_tool decorator.
    # Here's an example that adds a simple weather tool.
    # You also have to add `from livekit.agents import function_tool, RunContext` to the top of this file
    # @function_tool
    # async def lookup_weather(self, context: RunContext, location: str):
    #     """Use this tool to look up current weather information in the given location.
    #
    #     If the location is not supported by the weather service, the tool will indicate this. You must tell the user the location's weather is unavailable.
    #
    #     Args:
    #         location: The location to look up weather information for (e.g. city name)
    #     """
    #
    #     <http://logger.info|logger.info>(f"Looking up weather for {location}")
    #
    #     return "sunny with a temperature of 70 degrees."


server = AgentServer()


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()


server.setup_fnc = prewarm


@server.rtc_session()
async def my_agent(ctx: JobContext):
    # Logging setup
    # Add any other context you want in all log entries here
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }

    # Set up a voice AI pipeline using OpenAI, Cartesia, Deepgram, and the LiveKit turn detector
    session = AgentSession(
        # Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand
        # See all available models at <https://docs.livekit.io/agents/models/stt/>
        stt=inference.STT(model="deepgram/nova-3", language="multi"),
        # A Large Language Model (LLM) is your agent's brain, processing user input and generating a response
        # See all available models at <https://docs.livekit.io/agents/models/llm/>
        llm=inference.LLM(model="openai/gpt-4.1-mini"),
        # Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear
        # See all available models as well as voice selections at <https://docs.livekit.io/agents/models/tts/>
        tts=inference.TTS(
            model="cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"
        ),
        # VAD and turn detection are used to determine when the user is speaking and when the agent should respond
        # See more at <https://docs.livekit.io/agents/build/turns>
        turn_detection=MultilingualModel(),
        vad=ctx.proc.userdata["vad"],
        # allow the LLM to generate a response while waiting for the end of turn
        # See more at <https://docs.livekit.io/agents/build/audio/#preemptive-generation>
        preemptive_generation=True,
    )

    # To use a realtime model instead of a voice pipeline, use the following session setup instead.
    # (Note: This is for the OpenAI Realtime API. For other providers, see <https://docs.livekit.io/agents/models/realtime/))>
    # 1. Install livekit-agents[openai]
    # 2. Set OPENAI_API_KEY in .env.local
    # 3. Add `from livekit.plugins import openai` to the top of this file
    # 4. Use the following session setup instead of the version above
    # session = AgentSession(
    #     llm=openai.realtime.RealtimeModel(voice="marin")
    # )

    # # Add a virtual avatar to the session, if desired
    # # For other providers, see <https://docs.livekit.io/agents/models/avatar/>
    # avatar = hedra.AvatarSession(
    #   avatar_id="...",  # See <https://docs.livekit.io/agents/models/avatar/plugins/hedra>
    # )
    # # Start the avatar and wait for it to join
    # await avatar.start(session, room=ctx.room)

    # Start the session, which initializes the voice pipeline and warms up the models
    await session.start(
        agent=Assistant(),
        room=ctx.room,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=lambda params: (
                    noise_cancellation.BVCTelephony()
                    if params.participant.kind
                    == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
                    else noise_cancellation.BVC()
                ),
            ),
        ),
    )

    # Join the room and connect to the user
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(server)
t
I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!
2 Views
Open in Slack
PreviousNext
Powered by