o3-mini vs. DeepSeek-R1: API 설정, 성능 테스트 및 모델 평가

pip install weave openai datasets

import os
from openai import OpenAI
# Option to hardcode API key (leave empty to use env variable)
OPENAI_API_KEY = ""
# Use hardcoded key if provided, else fetch from environment
api_key = OPENAI_API_KEY if OPENAI_API_KEY else os.getenv("OPENAI_API_KEY")

if not api_key:
    raise ValueError("No API key provided. Set it manually or through the environment variable.")

O3_MODEL = "o3-mini"  # Replace with actual o3 model ID
# Initialize OpenAI client
openai_client = OpenAI(api_key=api_key)

# Math problem
problem = "Find the derivative of f(x) = 3x^2 + 5x - 7."
# Run inference
try:
    response = openai_client.chat.completions.create(
        model=O3_MODEL,
        messages=[{"role": "user", "content": problem}]
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"Failed to get OpenAI response: {e}")

import os
import asyncio
import requests
from datasets import load_dataset
from openai import OpenAI
import json
import weave; weave.init("aime_evaluation")

# API Keys and Clients
openai_client = OpenAI(api_key="your api key")
deepseek_client = OpenAI(
    api_key="your-deepseek-key",
    base_url="https://api.deepseek.com"
)

# Model constants
O3_MINI = "o3-mini"  
JUDGE_MODEL = "gpt-4o-2024-08-06"  # For evaluation

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

def run_openai_inference(prompt: str) -> str:
    """Run inference using OpenAI API"""
    try:
        response = openai_client.chat.completions.create(
            model=O3_MINI,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Failed to get OpenAI response: {e}")
        return None

class O3MiniModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        print("running inference")
        return run_openai_inference(f"{system_message}{text}")

class DeepseekR1(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        try:
            response = deepseek_client.chat.completions.create(
                model="deepseek-reasoner",
                messages=[
                    {"role": "user", "content": system_message + text}
                ]
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"Failed to get Deepseek response: {e}")
            return None

@weave.op
async def gpt4_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 }}
       ```
    """
    
    try:
        response = openai_client.chat.completions.create(
            model=JUDGE_MODEL,
            messages=[{"role": "user", "content": query}]
        )
        response_text = response.choices[0].message.content
        
        json_start = response_text.index("```json") + 7
        json_end = response_text.index("```", json_start)
        correctness = json.loads(response_text[json_start:json_end].strip()).get("correctness", False)
        
        return {"correctness": correctness, "reasoning": response_text}
    except Exception as e:
        print(f"Scoring failed: {e}")
        return {"correctness": False, "reasoning": str(e)}

def load_ds():
    print("Loading AIME dataset...")
    dataset = load_dataset("Maxwell-Jia/AIME_2024")["train"]
    return [{"text": row["Problem"], "label": row["Answer"]} for row in dataset]

async def run_evaluations():
    print("Loading dataset...")
    dataset = load_ds()
    print("Initializing models...")

    models = {
        "o3-mini": O3MiniModel(),
        "deepseek-r1": DeepseekR1()
    }

    print("Preparing dataset for evaluation...")
    dataset_prepared = [{"text": row["text"], "label": row["label"]} for row in dataset]

    print("Running evaluations...")
    scorers = [gpt4_scorer]
    for model_name, model in models.items():
        print(f"\nEvaluating {model_name}...")
        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}")

if __name__ == "__main__":
    asyncio.run(run_evaluations())