Building Video Agents

from vision_agents.plugins import ultralytics

processor = ultralytics.YOLOPoseProcessor(
    model_path="yolo26n-pose.pt",
    conf_threshold=0.5,
    fps=30,
    enable_hand_tracking=True,
    enable_wrist_highlights=True,
)

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="AI Golf Coach"),
    instructions="Read @golf_coach.md",
    llm=gemini.Realtime(fps=3),
    processors=[processor],
)

from vision_agents.plugins import roboflow
import random

processor = roboflow.RoboflowLocalDetectionProcessor(
    classes=["person", "sports ball"],
    conf_threshold=0.5,
    fps=5,
)

agent = Agent(
    processors=[processor],
    ...
)

questions = [
    "What's happening on the field right now?",
    "Provide an update on the situation.",
    "What has just happened?",
]

@agent.events.subscribe
async def on_detection_completed(event: roboflow.DetectionCompletedEvent):
    """
    Trigger commentary when a sports ball is detected.
    """
    ball_detected = any(
        obj["label"] == "sports ball" 
        for obj in event.objects
    )
    
    if ball_detected:
        await agent.simple_response(random.choice(questions))

from vision_agents.plugins import moondream

vlm = moondream.CloudVLM(
    api_key="YOUR_API_KEY",
    model="moondream-2",
)

# Analyze a frame
response = await vlm.query(
    image=frame,
    prompt="Describe what you see in this image."
)

import random
from vision_agents.core import Agent, User
from vision_agents.plugins import getstream, openai, roboflow

class Debouncer:
    def __init__(self, interval: float):
        self.interval = interval
        self.last_call = 0
    
    def __call__(self) -> bool:
        import time
        now = time.time()
        if now - self.last_call >= self.interval:
            self.last_call = now
            return True
        return False

async def create_agent(**kwargs) -> Agent:
    llm = openai.Realtime()
    
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="AI Sports Commentator", id="agent"),
        instructions="You are a sports commentator. Provide exciting play-by-play.",
        processors=[
            roboflow.RoboflowLocalDetectionProcessor(
                classes=["person", "sports ball"],
                conf_threshold=0.5,
                fps=5,
            )
        ],
        llm=llm,
    )
    
    questions = [
        "Provide an update on the situation on the football field.",
        "What has just happened?",
        "What is happening on the field right now?",
    ]
    
    debouncer = Debouncer(8)  # Call LLM once every 8 seconds max
    
    @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))
    
    return agent

from vision_agents.core.processors.base_processor import VideoProcessorPublisher
from vision_agents.core.events import EventManager, PluginBaseEvent
from vision_agents.core.utils.video_track import QueuedVideoTrack
from dataclasses import dataclass
import av

@dataclass
class CustomDetectionEvent(PluginBaseEvent):
    type: str = "custom.detection"
    confidence: float = 0.0

class CustomProcessor(VideoProcessorPublisher):
    name = "custom_processor"
    
    def __init__(self, fps: int = 10):
        self.fps = fps
        self._video_track = QueuedVideoTrack()
        self.events = EventManager()
        self.events.register(CustomDetectionEvent)
    
    async def process_video(
        self,
        track: aiortc.VideoStreamTrack,
        participant_id: str | None,
        shared_forwarder: VideoForwarder | None = None,
    ) -> None:
        # Process incoming video frames
        async for frame in track:
            processed_frame = await self._process_frame(frame)
            await self._video_track.add_frame(processed_frame)
    
    async def _process_frame(self, frame: av.VideoFrame) -> av.VideoFrame:
        # Your custom processing logic
        return frame
    
    def publish_video_track(self) -> aiortc.VideoStreamTrack:
        return self._video_track
    
    async def stop_processing(self) -> None:
        pass
    
    async def close(self) -> None:
        pass