Observability for Google Gemini (Python) with Opik

1
import os
2
import getpass
3
4
if "GOOGLE_API_KEY" not in os.environ:
5
    os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Gemini API key: ")

1
from google import genai
2
from opik.integrations.genai import track_genai
3
4
os.environ["OPIK_PROJECT_NAME"] = "gemini-integration-demo"
5
6
client = genai.Client()
7
gemini_client = track_genai(client)
8
9
prompt = """
10
Write a short two sentence story about Opik.
11
"""
12
13
response = gemini_client.models.generate_content(
14
    model="gemini-2.0-flash-001", contents=prompt
15
)
16
print(response.text)

1
from google import genai
2
from opik.integrations.genai import track_genai
3
4
# Configure for VertexAI
5
PROJECT_ID = "your-project-id"
6
LOCATION = "us-central1"
7
8
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
9
vertexai_client = track_genai(client)
10
11
# Set project name for organization
12
os.environ["OPIK_PROJECT_NAME"] = "vertexai-integration-demo"
13
14
# Use the wrapped client
15
response = vertexai_client.models.generate_content(
16
    model="gemini-2.0-flash-001",
17
    contents="Write a short story about AI observability."
18
)
19
print(response.text)

1
from opik import track
2
3
@track
4
def generate_story(prompt):
5
    response = gemini_client.models.generate_content(
6
        model="gemini-2.0-flash-001", contents=prompt
7
    )
8
    return response.text
9
10
@track
11
def generate_topic():
12
    prompt = "Generate a topic for a story about Opik."
13
    response = gemini_client.models.generate_content(
14
        model="gemini-2.0-flash-001", contents=prompt
15
    )
16
    return response.text
17
18
@track
19
def generate_opik_story():
20
    topic = generate_topic()
21
    story = generate_story(topic)
22
    return story
23
24
# Execute the multi-step pipeline
25
generate_opik_story()

1
import os
2
import time
3
import opik
4
from opik import track, opik_context
5
from opik.integrations.genai import track_genai
6
import google.genai as genai
7
from google.genai.types import HttpOptions, GenerateVideosConfig
8
9
os.environ["OPIK_PROJECT_NAME"] = "genai-video-demo"
10
11
# Configure for VertexAI (required for Veo)
12
client = genai.Client(
13
    vertexai=True,
14
    http_options=HttpOptions(api_version="v1"),
15
)
16
genai_client = track_genai(client)
17
18
19
@track
20
def generate_video(
21
    prompt: str,
22
    number_of_videos: int = 1,
23
    duration_seconds: int = 4,
24
    resolution: str = "720p",
25
    generate_audio: bool = False,
26
) -> dict:
27
    """Generate a video using Google's Veo model."""
28
    # Create video
29
    operation = genai_client.models.generate_videos(
30
        model="veo-3.1-fast-generate-preview",
31
        prompt=prompt,
32
        config=GenerateVideosConfig(
33
            duration_seconds=duration_seconds,
34
            resolution=resolution,
35
            generate_audio=generate_audio,
36
            number_of_videos=number_of_videos,
37
        ),
38
    )
39
40
    # Wait for completion
41
    with opik.start_as_current_span(name="wait_for_completion") as span:
42
        while not operation.done:
43
            time.sleep(10)
44
            operation = genai_client.operations.get(operation)
45
            result = {"name": operation.name, "done": operation.done}
46
            opik_context.update_current_span(output=result)
47
48
    # Download all videos if generation succeeded
49
    if operation.response and operation.response.generated_videos:
50
        output_paths = []
51
        for i, generated_video in enumerate(operation.response.generated_videos):
52
            output_path = f"output_video_{i}.mp4"
53
            generated_video.video.save(output_path)
54
            output_paths.append(output_path)
55
        result["output_paths"] = output_paths
56
57
    return result
58
59
60
# Generate videos
61
generate_video("A golden retriever playing in the snow", number_of_videos=2)