Football Commentator Example

import logging
import random

from dotenv import load_dotenv
from utils import Debouncer
from vision_agents.core import Agent, Runner, User
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import getstream, openai, roboflow

logger = logging.getLogger(__name__)

load_dotenv()


async def create_agent(**kwargs) -> Agent:
    llm = openai.Realtime()

    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="AI Sports Commentator", id="agent"),
        instructions="Read @instructions.md",
        processors=[
            roboflow.RoboflowLocalDetectionProcessor(
                classes=["person", "sports ball"],
                conf_threshold=0.5,
                fps=5,
            )
        ],
        llm=llm,
    )

    # Commentary prompts to rotate through
    questions = [
        "Provide an update on the situation on the football field.",
        "What has just happened?",
        "What is happening on the field right now?",
    ]

    # Debouncer: call LLM at most once every 8 seconds
    debouncer = Debouncer(8)

    @agent.events.subscribe
    async def on_detection_completed(event: roboflow.DetectionCompletedEvent):
        """
        Trigger commentary when Roboflow detects objects.
        
        Uses debouncer to avoid calling the LLM too frequently.
        """
        ball_detected = bool(
            [obj for obj in event.objects if obj["label"] == "sports ball"]
        )
        
        # Only comment when ball is detected and debouncer allows
        if ball_detected and debouncer:
            await agent.simple_response(random.choice(questions))

    return agent


async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    call = await agent.create_call(call_type, call_id)

    async with agent.join(call):
        await agent.finish()


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

processors=[
    roboflow.RoboflowLocalDetectionProcessor(
        classes=["person", "sports ball"],  # Only detect these objects
        conf_threshold=0.5,  # Confidence threshold (0-1)
        fps=5,  # Process 5 frames per second
    )
]

@agent.events.subscribe
async def on_detection_completed(event: roboflow.DetectionCompletedEvent):
    ball_detected = bool(
        [obj for obj in event.objects if obj["label"] == "sports ball"]
    )
    
    if ball_detected and debouncer:
        await agent.simple_response(random.choice(questions))

debouncer = Debouncer(8)  # Maximum one call every 8 seconds

if ball_detected and debouncer:  # debouncer returns True if enough time has passed
    await agent.simple_response(random.choice(questions))

class Debouncer:
    def __init__(self, interval: float):
        self.interval = interval
        self.last_time = 0.0
    
    def __bool__(self) -> bool:
        now = time.time()
        if now - self.last_time >= self.interval:
            self.last_time = now
            return True
        return False

You are an AI football commentator.
You will be shown a video of parts of a football match, and you need to analyze the events.

When asked what is going on right now, describe what you can see and the events taking place.
You may be asked with some regularity, don't sensationalize or add events - many could be 
mundane, normal things that occur in a football match!

It may make more sense to talk about ball possession, positioning, direction of play etc. 
depending on the context you're given.

Keep the replies very short, a few words max.
The colored boxes around the players and the ball are there to help you better detect objects.
Reply in English without using special symbols.

# OpenAI Realtime
from vision_agents.plugins import openai
llm = openai.Realtime()

# Gemini Live
from vision_agents.plugins import gemini
llm = gemini.Realtime()

ball_was_present = False

@agent.events.subscribe
async def on_detection_completed(event: roboflow.DetectionCompletedEvent):
    nonlocal ball_was_present
    
    ball_detected = bool(
        [obj for obj in event.objects if obj["label"] == "sports ball"]
    )
    
    # Trigger when ball comes back after being gone
    if ball_detected and not ball_was_present and debouncer:
        await agent.simple_response(
            "A play has just been made! Describe what happened, and the outcome"
        )
    
    ball_was_present = ball_detected