1import time
2from pathlib import Path
3
4from kern.agent import Agent
5from kern.media import Video
6from kern.models.google import Gemini
7from kern.utils.log import logger
8
9model = Gemini(id="gemini-2.0-flash-exp")
10agent = Agent(
11 model=model,
12 markdown=True,
13)
14
15# Please download a sample video file to test this Agent
16# Run: `wget https://storage.googleapis.com/generativeai-downloads/images/GreatRedSpot.mp4` to download a sample video
17video_path = Path(__file__).parent.joinpath("samplevideo.mp4")
18video_file = None
19remote_file_name = f"files/{video_path.stem.lower().replace('_', '')}"
20try:
21 video_file = model.get_client().files.get(name=remote_file_name)
22except Exception as e:
23 logger.info(f"Error getting file {video_path.stem}: {e}")
24 pass
25
26# Upload the video file if it doesn't exist
27if not video_file:
28 try:
29 logger.info(f"Uploading video: {video_path}")
30 video_file = model.get_client().files.upload(
31 file=video_path,
32 config=dict(name=video_path.stem, display_name=video_path.stem),
33 )
34
35 # Check whether the file is ready to be used.
36 while video_file.state.name == "PROCESSING":
37 time.sleep(2)
38 video_file = model.get_client().files.get(name=video_file.name)
39
40 logger.info(f"Uploaded video: {video_file}")
41 except Exception as e:
42 logger.error(f"Error uploading video: {e}")
43
44if __name__ == "__main__":
45 agent.print_response(
46 "Tell me about this video",
47 videos=[Video(content=video_file)],
48 stream=True,
49 )
