Realtime Voice Translation Agent

%pip install -q "livekit-agents[openai,silero]>=1.3.0" python-dotenv

CEREBRAS_API_KEY=your-key-here
OPENAI_API_KEY=your-key-here
LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your-key-here
LIVEKIT_API_SECRET=your-key-here

import os
from dotenv import load_dotenv

load_dotenv()

required = ["OPENAI_API_KEY", "CEREBRAS_API_KEY", "LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"]
missing = [k for k in required if not os.getenv(k)]
if missing:
    raise RuntimeError(f"Missing API keys: {', '.join(missing)}. Add them to .env file.")

print("✅ API keys loaded")

TARGET_LANGUAGE = "Spanish"  # Change this to your desired target language

def get_translation_prompt(target_language: str) -> str:
    """Generate system prompt for translation."""
    return f"""You are a real-time translator. Your task is to translate spoken text to {target_language}.

Rules:
1. Translate the input text accurately to {target_language}
2. Preserve the tone and intent of the original message
3. Keep translations natural and conversational
4. If the input is already in {target_language}, repeat it with minor improvements if needed
5. Do NOT add explanations or commentary - just translate
6. Respond ONLY with the translated text"""

import logging
import os
from livekit.agents import (
    AutoSubscribe,
    JobContext,
    WorkerOptions,
    AgentSession,
)
from livekit.agents.voice import Agent as VoiceAgent
from livekit.plugins import openai, silero

# Suppress verbose logging
logging.getLogger("livekit").setLevel(logging.WARNING)
logging.getLogger("livekit.agents").setLevel(logging.WARNING)

async def entrypoint(ctx: JobContext):
    """LiveKit agent entrypoint."""
    # Connect to the room
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    
    print(f"🎤 Connected to room: {ctx.room.name}")
    
    # Create Cerebras LLM client (OpenAI-compatible)
    cerebras_llm = openai.LLM(
        model="llama-3.1-8b",
        base_url="https://api.cerebras.ai/v1",
        api_key=os.environ["CEREBRAS_API_KEY"],
        extra_headers={"X-Cerebras-3rd-Party-Integration": "realtime-translation"}
    )
    
    # Create voice agent with translation instructions
    agent = VoiceAgent(
        instructions=get_translation_prompt(TARGET_LANGUAGE),
    )
    
    # Create session with STT/LLM/TTS/VAD components
    session = AgentSession(
        vad=silero.VAD.load(),
        stt=openai.STT(model="whisper-1"),
        llm=cerebras_llm,
        tts=openai.TTS(model="tts-1", voice="alloy"),
    )
    
    # Start session with agent and room
    await session.start(agent=agent, room=ctx.room)
    
    # Greet the user
    await session.say(f"Translation agent ready. I will translate to {TARGET_LANGUAGE}. Please speak.")


print(f"✅ Agent defined. Target language: {TARGET_LANGUAGE}")

from livekit.agents import jupyter

# Run the agent inside the notebook
jupyter.run_app(
    WorkerOptions(entrypoint_fnc=entrypoint)
)