# Updated code for HF Spaces compatibility import os import json import logging from dotenv import load_dotenv # Set up proper cache directories BEFORE importing other modules os.environ['HF_HOME'] = '/app/.cache' os.environ['TRANSFORMERS_CACHE'] = '/app/.cache' os.environ['HF_DATASETS_CACHE'] = '/app/.cache' from livekit import agents from livekit.agents import WorkerOptions, AutoSubscribe, AgentSession, Agent, RoomInputOptions, BackgroundAudioPlayer, AudioConfig, BuiltinAudioClip from livekit.plugins import ( openai, cartesia, deepgram, noise_cancellation, silero, hume, baseten ) from livekit.plugins import groq from livekit.plugins import elevenlabs, sarvam # Import turn detector with error handling try: from livekit.plugins.turn_detector.multilingual import MultilingualModel TURN_DETECTION_AVAILABLE = True except Exception as e: print(f"Turn detection model not available: {e}") TURN_DETECTION_AVAILABLE = False load_dotenv() # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[logging.StreamHandler()] ) def safe_save_transcript(content, filename): """Safely save transcript with proper error handling""" try: # Try to save in /app/transcripts directory first safe_path = f"/app/transcripts/{filename}" with open(safe_path, 'w') as f: json.dump(content, f, indent=2) print(f"Transcript saved to {safe_path}") return True except Exception as e: print(f"Failed to save transcript to transcripts directory: {e}") try: # Fallback to /tmp directory fallback_path = f"/tmp/{filename}" with open(fallback_path, 'w') as f: json.dump(content, f, indent=2) print(f"Transcript saved to fallback location: {fallback_path}") return True except Exception as e2: print(f"Failed to save transcript to fallback location: {e2}") return False async def entrypoint(ctx: agents.JobContext): """Main entrypoint with session cleanup""" async def write_transcript(): current_date = "12-12-12" filename = f"transcript_{ctx.room.name}_{current_date}.json" try: transcript_data = session.history.to_dict() safe_save_transcript(transcript_data, filename) except Exception as e: logging.error(f"Failed to save transcript: {e}") print(f"Transcript for {ctx.room.name} saved to {filename}") ctx.add_shutdown_callback(write_transcript) await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) prompt = "" if ctx.room.remote_participants.values(): for participant in ctx.room.remote_participants.values(): if participant.metadata: try: metadata = json.loads(participant.metadata) logging.info(f"Room metadata: {metadata}") prompt = metadata.get("prompt", "interview call") except json.JSONDecodeError: prompt = "interview call" continue # Set up turn detection with fallback turn_detection = None if TURN_DETECTION_AVAILABLE: try: turn_detection = MultilingualModel() print("Turn detection loaded successfully") except Exception as e: print(f"Failed to load turn detection: {e}") turn_detection = None # Create agent with error handling try: agent = Agent( instructions=( f"""You are Aish, speaking on behalf of AgentReach Systems. Your mission is to communicate the call purpose provided in `{prompt}` in a natural, friendly, and professional tone—as if you were a helpful colleague. Do not read verbatim; adapt to the flow of the conversation. Guidelines: • Open with a brief, warm greeting (use your own phrasing). • Optionally confirm identity if needed. • State the call purpose using the injected `{prompt}`, paraphrasing for a conversational feel. • Offer further assistance or next steps in your own words. • Close politely with a brief sign-off. • If it goes to voicemail, leave a concise message that includes your name, the purpose, and instructions for a callback. Never mention “virtual assistant” or recite a full script—keep it human, succinct, and on-point. """ ), vad=silero.VAD.load(), stt=groq.STT( model="whisper-large-v3", language="en", ), llm=groq.LLM( model="llama3-8b-8192" ), tts=deepgram.TTS( model="aura-2-andromeda-en", ), turn_detection=turn_detection, # This might be None ) print("Agent created successfully") except Exception as e: logging.error(f"Failed to create full agent: {e}") # Fallback agent without turn detection and with basic TTS try: agent = Agent( instructions=( f"""You are Aish, speaking on behalf of AgentReach Systems. Your mission is to communicate the call purpose provided in `{prompt}` in a natural, friendly, and professional tone—as if you were a helpful colleague. Do not read verbatim; adapt to the flow of the conversation. Guidelines: • Open with a brief, warm greeting (use your own phrasing). • Optionally confirm identity if needed. • State the call purpose using the injected `{prompt}`, paraphrasing for a conversational feel. • Offer further assistance or next steps in your own words. • Close politely with a brief sign-off. • If it goes to voicemail, leave a concise message that includes your name, the purpose, and instructions for a callback. Never mention “virtual assistant” or recite a full script—keep it human, succinct, and on-point. """ ), vad=silero.VAD.load(), stt=groq.STT( model="whisper-large-v3", language="en", ), llm=groq.LLM( model="llama3-8b-8192" ), tts=cartesia.TTS(), # Fallback TTS # No turn detection in fallback ) print("Fallback agent created successfully") except Exception as e2: logging.error(f"Failed to create fallback agent: {e2}") raise e2 # Initialize background audio with error handling try: background_audio = BackgroundAudioPlayer( ambient_sound=AudioConfig(BuiltinAudioClip.OFFICE_AMBIENCE, volume=0.8), thinking_sound=[ AudioConfig(BuiltinAudioClip.KEYBOARD_TYPING, volume=0.6), AudioConfig(BuiltinAudioClip.KEYBOARD_TYPING2, volume=0.6), ], ) print("Background audio initialized") except Exception as e: logging.error(f"Failed to initialize background audio: {e}") background_audio = None session = AgentSession() try: await session.start( room=ctx.room, agent=agent, room_input_options=RoomInputOptions( noise_cancellation=noise_cancellation.BVCTelephony(), ), ) print("Session started successfully") # Start background audio if available if background_audio: try: await background_audio.start(room=ctx.room, agent_session=session) print("Background audio started") except Exception as e: logging.error(f"Failed to start background audio: {e}") await session.generate_reply( instructions="Greet the user and offer your assistance." ) except Exception as e: logging.error(f"Error in session: {e}") # Continue running even with errors pass opts = WorkerOptions( entrypoint_fnc=entrypoint, initialize_process_timeout=120.0, shutdown_process_timeout=120.0, ) if __name__ == "__main__": agents.cli.run_app(opts)