"""
app.py — Interface Gradio pour le Voice Agent LiveKit
Démarre :
1. Le worker LiveKit agent (subprocess)
2. L'interface Gradio avec UI WebRTC
"""
import os
import sys
import json
import subprocess
import threading
import gradio as gr
import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pathlib import Path
from dotenv import load_dotenv
# Ajouter le dossier agent au path
sys.path.insert(0, str(Path(__file__).parent / "agent"))
load_dotenv()
# ──────────────────────────────────────────────
# Démarrage de l'agent en arrière-plan
# ──────────────────────────────────────────────
_agent_process: subprocess.Popen | None = None
def start_agent_worker():
"""Lance le worker LiveKit dans un sous-processus."""
global _agent_process
agent_path = Path(__file__).parent / "agent" / "voice_agent.py"
_agent_process = subprocess.Popen(
[sys.executable, str(agent_path), "start"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
)
print(f"[Agent] Worker démarré — PID {_agent_process.pid}")
for line in _agent_process.stdout:
print(f"[Agent] {line}", end="")
def launch_agent_thread():
t = threading.Thread(target=start_agent_worker, daemon=True)
t.start()
# ──────────────────────────────────────────────
# Génération de token LiveKit
# ──────────────────────────────────────────────
def get_token(room_name: str = "") -> str:
try:
from token_server import get_connection_details
details = get_connection_details(room_name=room_name or None)
return json.dumps(details, indent=2)
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
# ──────────────────────────────────────────────
# HTML de l'interface LiveKit WebRTC
# ──────────────────────────────────────────────
def build_livekit_html() -> str:
"""Génère le HTML en injectant URL + token depuis le .env."""
try:
from token_server import get_connection_details
details = get_connection_details()
injected_url = details["url"]
injected_token = details["token"]
auto_connect = "true"
except Exception:
injected_url = os.getenv("LIVEKIT_URL", "")
injected_token = ""
auto_connect = "false"
return LIVEKIT_CLIENT_HTML_TEMPLATE.replace("__LIVEKIT_URL__", injected_url) \
.replace("__LIVEKIT_TOKEN__", injected_token) \
.replace("__AUTO_CONNECT__", auto_connect)
LIVEKIT_CLIENT_HTML_TEMPLATE = """
Voice Agent
🎙️ Voice Agent
LiveKit
Déconnecté
"""
# ──────────────────────────────────────────────
# Interface Gradio
# ──────────────────────────────────────────────
def build_gradio_app() -> gr.Blocks:
with gr.Blocks(title="Voice Agent LiveKit") as demo:
gr.Markdown(
"""
# 🎙️ Voice Agent — LiveKit + Gradio
Assistant vocal propulsé par **LiveKit** et **Whisper**.
"""
)
with gr.Tabs():
# ── Onglet Agent ──────────────────────────────────────────────
with gr.TabItem("🔊 Agent Vocal"):
gr.HTML(
""
)
# ── Onglet Token ──────────────────────────────────────────────
with gr.TabItem("🔑 Générer un Token"):
gr.Markdown(
"Générez un token LiveKit depuis le serveur. "
"Copiez `url` et `token` dans l'interface ci-dessus."
)
room_input = gr.Textbox(
label="Nom de la room (optionnel)",
placeholder="voice-room-demo",
max_lines=1,
)
gen_btn = gr.Button("Générer le Token", variant="primary")
token_output = gr.Code(language="json", label="Détails de connexion")
gen_btn.click(fn=get_token, inputs=[room_input], outputs=[token_output])
# Pied de page
gr.Markdown(
""
"LiveKit Agents · OpenAI GPT-4o-mini · Silero VAD"
""
)
return demo
# ──────────────────────────────────────────────
# Main
# ──────────────────────────────────────────────
if __name__ == "__main__":
import argparse
import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
parser = argparse.ArgumentParser()
parser.add_argument("--no-agent", action="store_true", help="Ne pas lancer l'agent")
parser.add_argument("--port", type=int, default=7860)
args = parser.parse_args()
# Lancer le worker agent en arrière-plan
if not args.no_agent:
print("[Main] Démarrage du worker LiveKit Agent…")
launch_agent_thread()
# Construire l'app FastAPI et monter Gradio dessus
fastapi_app = FastAPI()
from fastapi.responses import Response
@fastapi_app.get("/voice-ui")
async def voice_ui():
html = build_livekit_html()
return Response(
content=html,
media_type="text/html",
headers={
"Permissions-Policy": "microphone=*, camera=*",
"Cross-Origin-Opener-Policy": "same-origin-allow-popups",
},
)
gradio_demo = build_gradio_app()
app = gr.mount_gradio_app(fastapi_app, gradio_demo, path="/")
print(f"[Main] App lancée sur http://localhost:{args.port}")
uvicorn.run(app, host="0.0.0.0", port=args.port)