Evaluating the new Gemini 2.5 Pro Experimental model

pip install google-generativeai weave datasets pillow requests litellm google-genai

from google import genai
from google.genai import types
from PIL import Image
import requests
from io import BytesIO
import weave

weave.init("gemini-vision-demo")

# Download the image
url = "https://d1yhils6iwh5l5.cloudfront.net/charts/resized/124357/large/cotd.png"
response = requests.get(url)
image = Image.open(BytesIO(response.content))

# Gemini client
client = genai.Client(api_key="")

@weave.op()
def analyze_chart(prompt: str, img: Image.Image) -> str:
    res = client.models.generate_content(
        model="gemini-2.5-pro-exp-03-25",
        contents=[prompt, img]
    )
    return res.text

# Call and print
result = analyze_chart("What's shown in this chart?", image)
print(result)


from google import genai
from google.genai import types
import weave

weave.init("gemini-function-call")

# Define the mock function
def set_light_values(brightness: int, color_temp: str) -> dict[str, int | str]:
    print("simulate function to set light values")
    return {
        "brightness": brightness,
        "colorTemperature": color_temp
    }

tool_list = [set_light_values]
tool_map = {fn.__name__: fn for fn in tool_list}

@weave.op()
def run_light_prompt(prompt: str):
    config = {
        'tools': tool_list,
        'automatic_function_calling': {'disable': True},
        'tool_config': {
            'function_calling_config': {
                'mode': 'any'
            }
        }
    }

    client = genai.Client(api_key="")
    chat = client.chats.create(model='gemini-2.5-pro-exp-03-25', config=config)
    response = chat.send_message(prompt)

    raw_calls = []
    results = []

    if response.function_calls:
        for fn_call in response.function_calls:
            raw_calls.append({
                "function": fn_call.name,
                "args": fn_call.args
            })
            fn = tool_map.get(fn_call.name)
            if fn:
                result = fn(**fn_call.args)
                results.append({
                    "function": fn_call.name,
                    "args": fn_call.args,
                    "result": result
                })

    return {
        "text": response.text,
        "raw_function_calls": raw_calls,
        "executed_results": results
    }

# Run it
result = run_light_prompt("Make the room cozy for reading.")
print(result)



from google import genai
from google.genai import types
import weave

weave.init("gemini-code-exec")

# Parse response helper
def parse_code_execution_response(response):
    code = None
    output = None
    explanation = None

    for part in response.candidates[0].content.parts:
        if part.executable_code:
            code = part.executable_code.code
        if part.code_execution_result:
            output = part.code_execution_result.output
        if part.text:
            explanation = part.text

    return {
        "code": code,
        "execution_output": output,
        "explanation": explanation
    }

# Inference op
@weave.op()
def run_code_inference(prompt: str):
    client = genai.Client(api_key="")

    chat = client.chats.create(
        model='gemini-2.5-pro-exp-03-25',
        config=types.GenerateContentConfig(
            tools=[types.Tool(code_execution=types.ToolCodeExecution)]
        )
    )

    response = chat.send_message(prompt)
    return parse_code_execution_response(response)

# Run it
result = run_code_inference("Please run Python code to calculate the sum of [5, 10, 15, 20].")
print(result)

import os
import asyncio
import json
from datasets import load_dataset
from litellm import completion
import google.generativeai as genai
import weave
weave.init("aime_evaluation")  # Initialize Weave

# Set API keys
OPENAI_API_KEY = "your_openai_api_key"  # Replace with your actual OpenAI API key
GOOGLE_API_KEY = "your_google_ai_studio_api_key"  # Replace with your actual Google API key

# Configure API clients
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
genai.configure(api_key=GOOGLE_API_KEY)

# Define Gemini model configurations
GEMINI_CONFIGS = [
    {
        "name": "gemini-2.5-pro-exp-03-25",
        "model_name": "gemini-2.5-pro-exp-03-25"
    },
    {
        "name": "gemini-2.0-flash-thinking-exp-01-21",
        "model_name": "gemini-2.0-flash-thinking-exp-01-21" 
    }
]

# Consistent system message for all models
system_message = "Solve the following problem. put your final answer within \\boxed{}: "

# Function to perform inference using litellm for the scorer
def run_inference_openai(prompt, model_id="gpt-4o-2024-08-06"):
    try:
        response = completion(
            model=model_id,
            temperature=0.0,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        # Extract content from litellm response
        if response and hasattr(response, 'choices') and len(response.choices) > 0:
            content = response.choices[0].message.content
            return content
        else:
            print("No content found in response")
            return None
            
    except Exception as e:
        print(f"Failed to get response: {e}")
        return None

# Gemini inference function
async def gemini_inference(prompt: str, config: dict) -> str:
    """
    Generate solution using Gemini with specified config.
    """
    try:
        model = genai.GenerativeModel(config["model_name"])
        
        # Enhanced prompt with step-by-step instruction
        enhanced_prompt = f"{system_message} {prompt}\n"
        
        print(f"\n--- Starting inference with {config['name']} ---")
        
        # Generate content
        response = model.generate_content(
            enhanced_prompt,

        )
        
        # Extract text from response
        if response.text:
            text_content = response.text
            print(f"Successfully generated content with {config['name']}")
            return text_content
        else:
            return "No response generated"
                
    except Exception as e:
        print(f"Error in inference with {config['name']}: {e}")
        return f"Error: {str(e)}"

# Define model classes for each Gemini configuration
class Gemini25ProExpModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await gemini_inference(text, GEMINI_CONFIGS[0])

class Gemini20FlashModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await gemini_inference(text, GEMINI_CONFIGS[1])

# Score function using GPT-4o via litellm
@weave.op
async def gpt4o_scorer(label: str, model_output: str) -> dict:
    """Score the model's output by comparing it with the ground truth."""
    query = f"""
    YOU ARE A LLM JUDGE DETERMINING IF THE FOLLOWING MODEL GENERATED ANSWER IS THE SAME AS THE CORRECT ANSWER 
    I WILL GIVE YOU THE LAST 100 CHARS OF THE MODEL'S REASONING PATH, WHICH WILL CONTAIN THE FINAL ANSWER -> 

    Model's Answer (last 100 chars): {str(model_output)[-100:]}
    Correct Answer: {label}
    
    Your task:
    1. State the model's predicted answer (answer only).
    2. State the ground truth (answer only).
    3. Determine if the model's final answer is correct (ignore formatting differences, etc.). RESPOND with the predicted and ground truth answer, followed with a JSON object containing the correctness encapsulated within the following delimiters:
       ```json
       {{ "correctness": true/false }}
       ```
    """
    
    # Perform inference using litellm
    response = run_inference_openai(query, "gpt-4o-2024-08-06")
    
    if response is None:
        return {"correctness": False, "reasoning": "Inference failed."}
    
    try:
        # Extract correctness JSON object from the response
        json_start = response.index("```json") + 7
        json_end = response.index("```", json_start)
        correctness = json.loads(response[json_start:json_end].strip()).get("correctness", False)
    except (ValueError, IndexError):
        correctness = False

    return {"correctness": correctness, "reasoning": response}


def load_ds():
    print("Loading AIME dataset...")
    all_problems = []
    
    # Load AIME2025-I
    aime_i = load_dataset("opencompass/AIME2025", "AIME2025-I")
    # Check what splits are available
    splits_i = list(aime_i.keys())
    print(f"Available splits in AIME2025-I: {splits_i}")
    
    # Use the first available split
    first_split_i = splits_i[0]
    problems_i = [{"text": row["question"], "label": row["answer"]} for row in aime_i[first_split_i]]
    print(f"Loaded {len(problems_i)} problems from AIME2025-I")
    all_problems.extend(problems_i)
    
    # Load AIME2025-II
    aime_ii = load_dataset("opencompass/AIME2025", "AIME2025-II")
    splits_ii = list(aime_ii.keys())
    print(f"Available splits in AIME2025-II: {splits_ii}")
    
    # Use the first available split
    first_split_ii = splits_ii[0]
    problems_ii = [{"text": row["question"], "label": row["answer"]} for row in aime_ii[first_split_ii]]
    print(f"Loaded {len(problems_ii)} problems from AIME2025-II")
    all_problems.extend(problems_ii)
    
    print(f"Total problems loaded: {len(all_problems)}")
    return all_problems



# Run evaluations for each model
async def run_evaluations():
    print("Loading dataset...")
    dataset = load_ds()
    print(f"Loaded {len(dataset)} problems")
    
    print("Initializing models...")
    models = {
        # "gemini-2.5-pro-exp": Gemini25ProExpModel(),
        "gemini-2.0-flash": Gemini20FlashModel(),
    }

    print("Preparing dataset for evaluation...")
    # You can uncomment this to use a subset for testing
    # test_size = 5  # Number of problems to evaluate per model
    # dataset_prepared = dataset[:test_size]
    dataset_prepared = dataset

    print("Running evaluations...")
    scorers = [gpt4o_scorer]
    
    for model_name, model in models.items():
        print(f"\n\n=== EVALUATING {model_name.upper()} ===")
        evaluation = weave.Evaluation(
            dataset=dataset_prepared,
            scorers=scorers,
            name=f"{model_name} Evaluation"
        )
        results = await evaluation.evaluate(model)
        print(f"Results for {model_name}: {results}")
        
        # Calculate and print accuracy
        if hasattr(results, 'scores') and 'gpt4o_scorer' in results.scores:
            correct = sum(1 for score in results.scores['gpt4o_scorer'] if score['correctness'])
            accuracy = correct / len(dataset_prepared) if dataset_prepared else 0
            print(f"{model_name} accuracy: {accuracy:.2%} ({correct}/{len(dataset_prepared)})")

if __name__ == "__main__":
    # Set your API keys here or in environment variables
    if not os.environ.get("OPENAI_API_KEY"):
        os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY  # Required for litellm
    
    # Run evaluations
    asyncio.run(run_evaluations())