Spaces:
Runtime error
Runtime error
| """ | |
| Bot runner for LiveKit transport. | |
| This script is called as a subprocess by the LiveKit server. | |
| It reads configuration from command line arguments and starts the bot with LiveKit transport. | |
| """ | |
| import argparse | |
| import asyncio | |
| import json | |
| import os | |
| import sys | |
| from loguru import logger | |
| # Add parent directory (project root) to Python path for imports | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from bot_config import BotConfig, TransportType | |
| def parse_args(): | |
| """Parse command line arguments.""" | |
| parser = argparse.ArgumentParser(description="Pipecat LiveKit bot runner") | |
| parser.add_argument("--session-id", required=True, help="Session ID for the bot") | |
| parser.add_argument("--room-name", required=True, help="LiveKit room name") | |
| parser.add_argument("--livekit-url", required=True, help="LiveKit server URL") | |
| parser.add_argument("--token", required=True, help="LiveKit access token") | |
| # Bot data can be provided as JSON string or file | |
| bot_data_group = parser.add_mutually_exclusive_group(required=True) | |
| bot_data_group.add_argument("--bot-data", help="JSON bot configuration data") | |
| bot_data_group.add_argument( | |
| "--bot-config-file", help="Path to JSON bot configuration file" | |
| ) | |
| parser.add_argument("--bot-name", default="bot", help="Bot participant identity") | |
| parser.add_argument( | |
| "--disable-rtvi", | |
| action="store_true", | |
| help="Disable RTVI processor and observer", | |
| ) | |
| return parser.parse_args() | |
| async def main(): | |
| """Main entry point for the LiveKit bot runner.""" | |
| try: | |
| args = parse_args() | |
| # Parse and validate bot data | |
| try: | |
| if args.bot_data: | |
| raw_bot_data = json.loads(args.bot_data) | |
| elif args.bot_config_file: | |
| with open(args.bot_config_file, "r") as f: | |
| raw_bot_data = json.load(f) | |
| logger.info(f"Loaded bot config from file: {args.bot_config_file}") | |
| else: | |
| logger.error("No bot data provided") | |
| sys.exit(1) | |
| except FileNotFoundError as e: | |
| logger.error(f"Bot config file not found: {e}") | |
| sys.exit(1) | |
| except json.JSONDecodeError as e: | |
| logger.error(f"Failed to parse bot data JSON: {e}") | |
| sys.exit(1) | |
| except Exception as e: | |
| logger.error(f"Failed to load bot data: {e}") | |
| sys.exit(1) | |
| # Create bot config for LiveKit | |
| config = BotConfig( | |
| transport_type=TransportType.LIVEKIT, | |
| session_id=args.session_id, | |
| livekit_url=args.livekit_url, | |
| livekit_token=args.token, | |
| livekit_room_name=args.room_name, | |
| participant_identity=args.bot_name, | |
| bot_data=raw_bot_data, | |
| rtvi_enabled=not args.disable_rtvi, | |
| ) | |
| logger.info(f"Starting LiveKit bot: {config.session_id}") | |
| logger.info(f"Room: {args.room_name}") | |
| logger.info(f"LiveKit URL: {config.livekit_url}") | |
| logger.info(f"Bot identity: {config.participant_identity}") | |
| # Import and run the bot | |
| from bot import run_livekit_pipeline | |
| await run_livekit_pipeline(config) | |
| except KeyboardInterrupt: | |
| logger.info("Bot runner interrupted by user") | |
| if __name__ == "__main__": | |
| # Configure logging | |
| logger.remove() | |
| logger.add( | |
| sys.stdout, | |
| format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>livekit-bot</cyan> | <level>{message}</level>", | |
| level="INFO", | |
| ) | |
| asyncio.run(main()) | |