drive_thru_ai_ordering.py

"""
# Environment setup and installation
uv init
uv venv && source .venv/bin/activate
uv add vision-agents 
uv add "vision-agents[getstream, gemini]"
"""

import logging

from dotenv import load_dotenv

from vision_agents.core.edge.types import User
from vision_agents.core.agents import Agent, AgentLauncher
from vision_agents.core import cli
from vision_agents.plugins import gemini, getstream

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)


async def create_agent(**kwargs) -> Agent:
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(
            name="Restaurant Ordering Agent"
        ),  # the user object for the agent (name, image etc)
        instructions="Read @drive_thru_ai_ordering_instructions.md",

        # Establish connection to the LLM
        llm=gemini.Realtime(
            #model = "gemini-live-2.5-flash-preview",
            model="gemini-2.5-flash-native-audio-preview-09-2025",
            config={
                "response_modalities": ["AUDIO"],
                "speech_config": {
                    "voice_config": {
                        "prebuilt_voice_config": {"voice_name": "Leda"}
                    }
                },
            },
        ),
        processors=[],  # processors can fetch extra data, check images/audio data or transform video
    )
    return agent


async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    # ensure the agent user is created
    await agent.create_user()
    # Create a call
    call = await agent.create_call(call_type, call_id)

    with await agent.join(call):
        await agent.llm.simple_response(text="Great the user and ask what they would like to order.")
        await agent.finish()  # run till the call ends


if __name__ == "__main__":
    cli(AgentLauncher(create_agent=create_agent, join_call=join_call))