gemini_live_clone.py

import asyncio
from uuid import uuid4

from vision_agents.core.edge.types import User
from vision_agents.core import agents
from vision_agents.plugins import openai, getstream

async def start_agent() -> None:
    # create an agent to run with Stream's edge, openAI realtime llm
    agent = agents.Agent(
        edge=getstream.Edge(),  # low latency edge. clients for React, iOS, Android, RN, Flutter etc.
        agent_user=User(name="My happy AI friend", id="agent"),  # the user object for the agent (name, image etc)
        #instructions="You're a voice AI assistant. Keep responses short and conversational. Don't use special characters or formatting. Be friendly and helpful. Respond only in English",
        instructions="Read @agent_instructions.md",
        processors=[],  # processors can fetch extra data, check images/audio data or transform video
        # realtime version (speech-to-speech, vad/tts/stt not needed)
        llm=openai.Realtime()
        # Alternative: llm with separate tts & stt (requires conversations API)
        # llm=openai.LLM(model="gpt-4o-mini"),
        # tts=cartesia.TTS(),
        # stt=deepgram.STT(),
    )
    await agent.create_user()

    # Create a call
    call = agent.edge.client.video.call("default", str(uuid4()))

    # Open the demo UI
    await agent.edge.open_demo(call)

    # Have the agent join the call/room
    with await agent.join(call):
        print("🎙️  Agent is now active and listening...")
        print("Press Ctrl+C to stop the agent")
        
        # Keep the agent running indefinitely until interrupted
        try:
            await asyncio.Event().wait()  # Wait forever until interrupted
        except KeyboardInterrupt:
            print("\n👋 Shutting down agent...")
        finally:
            await agent.finish()


if __name__ == "__main__":
    asyncio.run(start_agent())