Building an AI Meeting Assistant with Gemini Vision API and VideoSDK: Analyzing Screen Content in Real-time

1git clone https://github.com/videosdk-community/videosdk-gemini-vision-agent.git
2cd videosdk-gemini-vision-agent
3
4# Server environment setup
5python -m venv .venv
6pip install -r requirements.txt
7cp .env.example .env
8
9# Add your API keys to .env
10# OPENAI_API_KEY=your_openai_key_here
11# GEMINI_API_KEY=your_gemini_api_key
12
13# Client setup
14cd client
15cp .env.example .env
16# Add VITE_VIDEOSDK_TOKEN=your_videosdk_auth_token_here
17

1class AIAgent:
2    def __init__(self, meeting_id: str, authToken: str, name: str):
3        # Initialize components
4        self.loop = asyncio.get_event_loop()
5        self.audio_track = CustomAudioStreamTrack(
6            loop=self.loop,
7            handle_interruption=True
8        )
9        
10        # Configure the meeting
11        self.meeting_config = MeetingConfig(
12            name=name,
13            meeting_id=meeting_id,
14            token=authToken,
15            mic_enabled=True,
16            webcam_enabled=False,
17            custom_microphone_audio_track=self.audio_track,
18        )
19        
20        # Initialize Gemini vision model
21        if gemini_api_key:
22            genai.configure(api_key=gemini_api_key)
23            self.vision_model = genai.GenerativeModel('gemini-1.5-flash')
24        
25        # Initialize OpenAI for transcription and conversation
26        self.intelligence = OpenAIIntelligence(
27            loop=self.loop,
28            api_key=openai_api_key,
29            input_audio_transcription=InputAudioTranscription(model="whisper-1"),
30            tools=[screen_tool],
31            audio_track=self.audio_track,
32            handle_function_call=self.handle_function_call,
33        )
34        
35        # Storage for the latest screen frame
36        self.frame_queue = deque(maxlen=1)
37        self.latest_frame = None
38

1async def handle_function_call(self, function_call):
2    """Process function calls from OpenAI, particularly for screen analysis."""
3    if function_call.name == "analyze_screen":
4        if not self.latest_frame:
5            return "No screen content available"
6        
7        # Convert frame to image
8        image_data = self.latest_frame.to_ndarray()
9        image = Image.fromarray(image_data)
10        
11        try:
12            # Request analysis from Gemini
13            response = await self.loop.run_in_executor(
14                None,
15                lambda: self.vision_model.generate_content([
16                    "Analyze this screen to help user. Focus on relevant UI elements, text, code, and context.",
17                    image
18                ])
19            )
20            return response.text
21        except Exception as e:
22            return f"Analysis error: {str(e)}"
23    return "Unknown command"
24

1async def add_screenshare_listener(self, stream: Stream, peer_name: str):
2    """Store the latest frame from a screen share stream."""
3    print("Participant screenshare enabled", peer_name)
4    while True:
5        try:                
6            frame = await stream.track.recv()
7            self.latest_frame = frame  # Update latest frame
8        except Exception as e:
9            traceback.print_exc()
10            print("Screenshare processing error:", e)
11            break
12

1async def add_audio_listener(self, stream: Stream, peer_name: str):
2    """Process audio from a participant and send it to OpenAI for transcription."""
3    while True:
4        try:
5            # Get audio frame
6            frame = await stream.track.recv()      
7            audio_data = frame.to_ndarray()[0]
8            
9            # Process audio (convert format, resample)
10            audio_data_float = (
11                audio_data.astype(np.float32) / np.iinfo(np.int16).max
12            )
13            audio_mono = librosa.to_mono(audio_data_float.T)
14            audio_resampled = librosa.resample(
15                audio_mono, orig_sr=48000, target_sr=16000
16            )
17            
18            # Convert to PCM and send to OpenAI
19            pcm_frame = (
20                (audio_resampled * np.iinfo(np.int16).max)
21                .astype(np.int16)
22                .tobytes()
23            )
24            
25            await self.intelligence.send_audio_data(pcm_frame)
26        except Exception as e:
27            print("Audio processing error:", e)
28            break
29

1def on_participant_joined(self, participant: Participant):
2    """Setup listeners for participant audio and screen shares."""
3    peer_name = participant.display_name
4    print("Participant joined:", peer_name)
5    
6    # Set instructions for the AI assistant
7    intelligence_instructions = """
8    You are an AI meeting assistant. Follow these rules:
9    1. Use analyze_screen tool when user asks about:
10    - Visible UI elements
11    - On-screen content
12    - Application help
13    - Workflow guidance
14    2. Keep responses under 2 sentences
15    3. Always acknowledge requests first
16    """
17
18    # Update OpenAI with instructions
19    asyncio.create_task(
20        self.intelligence.update_session_instructions(intelligence_instructions)
21    )
22
23    # Setup stream handlers
24    participant.add_event_listener(
25        ParticipantHandler(
26            participant_id=participant.id,
27            on_stream_enabled=on_stream_enabled,
28            on_stream_disabled=on_stream_disabled
29        )
30    )
31

1class OpenAIIntelligence:
2    """Manages real-time communication with OpenAI's streaming API."""
3    
4    async def connect(self):
5        """Establish WebSocket connection to OpenAI's real-time API."""
6        url = f"wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
7        
8        self.ws = await self._http_session.ws_connect(
9            url=url,
10            headers={
11                "Authorization": f"Bearer {self.api_key}",
12                "OpenAI-Beta": "realtime=v1",
13            },
14        )
15        
16        # Start message handling and update session
17        self.receive_message_task = self.loop.create_task(
18            self.receive_message_handler()
19        )
20        await self.update_session(self.session_update_params)
21

1class CustomAudioStreamTrack(CustomAudioTrack):
2    """Custom audio track for streaming AI-generated speech."""
3    
4    async def add_new_bytes(self, audio_data_stream: Iterator[bytes]):
5        """Add new audio data to be processed and played."""
6        await self._process_audio_task_queue.put(audio_data_stream)
7    
8    async def recv(self) -> AudioFrame:
9        """Provide the next audio frame for streaming."""
10        # Complex audio buffering and timing logic...
11        if len(self.frame_buffer) > 0:
12            frame = self.frame_buffer.pop(0)
13        else:
14            # Create silent frame if buffer is empty
15            frame = AudioFrame(format="s16", layout="mono", samples=self.samples)
16            for p in frame.planes:
17                p.update(bytes(p.buffer_size))
18        
19        # Set frame timing properties
20        frame.pts = pts
21        frame.time_base = time_base
22        frame.sample_rate = self.sample_rate
23        return frame
24

1# Start the server (from project root)
2uvicorn app:app
3
4# Start the client (from /client folder)
5npm run dev
6