Claude 3.7 소네 평가: 성능, 추론, 비용 최적화

import os
import json
from anthropic import Anthropic
import weave; weave.init("claude37_inference")

# Configuration
API_KEY = "your claude key" # Replace with your API key
MODEL = "claude-3-7-sonnet-20250219"
THINKING_BUDGET = 2000
ANSWER_TOKENS = 4000
MAX_TOKENS = ANSWER_TOKENS + THINKING_TOKENS
ENABLE_STREAMING = False  # Set to True to enable streaming


# Default prompt if none is provided
DEFAULT_PROMPT = """
Solve this puzzle: Three people check into a hotel. They pay $30 to the manager.
The manager finds out that the room only costs $25 so he gives $5 to the bellboy to return
to the three people. The bellboy, however, decides to keep $2 and gives $1 back to each person.
Now, each person paid $10 and got back $1, so they paid $9 each, totaling $27.
The bellboy kept $2, which makes $29. Where is the missing $1?
"""

# Initialize client
client = Anthropic(api_key=API_KEY)


@weave.op
def run_inference(prompt=None):
    """
    Run inference with the given prompt or default prompt if none provided.
    
    Args:
        prompt (str, optional): The prompt to send to Claude. Defaults to None.
    
    Returns:
        dict: The complete response object when using basic_inference
        or the final text response when using stream_inference
    """
    # Use the provided prompt or fall back to the default
    actual_prompt = prompt if prompt is not None else DEFAULT_PROMPT
    
    if ENABLE_STREAMING:
        return stream_inference(actual_prompt)
    else:
        return basic_inference(actual_prompt)

def basic_inference(prompt):
    """
    Basic non-streaming inference.
    
    Args:
        prompt (str): The prompt to send to Claude
    
    Returns:
        dict: The complete response object from the API
    """
    try:
        # Prepare request parameters
        request_params = {
            "model": MODEL,
            "max_tokens": MAX_TOKENS,
            "messages": [{"role": "user", "content": prompt}]
        }
        
        # Only include thinking if budget is greater than 0
        if THINKING_BUDGET > 0:
            request_params["thinking"] = {"type": "enabled", "budget_tokens": THINKING_BUDGET}
        
        response = client.messages.create(**request_params)
        
        # Print thinking blocks and answer
        for block in response.content:
            if block.type == "thinking":
                print("\n=== THINKING ===")
                print(block.thinking)
                print("===============")
            elif block.type == "text":
                print("\n=== ANSWER ===")
                print(block.text)
                print("==============")
        
        return response
    except Exception as e:
        print(f"Error: {e}")
        return None

def stream_inference(prompt):
    """
    Streaming inference.
    
    Args:
        prompt (str): The prompt to send to Claude
    
    Returns:
        str: The combined text response from the stream
    """
    try:
        # Prepare streaming request parameters
        stream_params = {
            "model": MODEL,
            "max_tokens": MAX_TOKENS,
            "messages": [{"role": "user", "content": prompt}]
        }
        
        # Only include thinking if budget is greater than 0
        if THINKING_BUDGET > 0:
            stream_params["thinking"] = {"type": "enabled", "budget_tokens": THINKING_BUDGET}
        
        full_text_response = ""
        
        with client.messages.stream(**stream_params) as stream:
            print("\n=== STREAMING RESPONSE ===")
            current_block_type = None
            
            for event in stream:
                if event.type == "content_block_start":
                    current_block_type = event.content_block.type
                    print(f"\n--- Starting {current_block_type} block ---")
                elif event.type == "content_block_delta":
                    if event.delta.type == "thinking_delta":
                        print(event.delta.thinking, end="", flush=True)
                    elif event.delta.type == "text_delta":
                        print(event.delta.text, end="", flush=True)
                        full_text_response += event.delta.text
                elif event.type == "content_block_stop":
                    print(f"\n--- End {current_block_type} block ---")
                elif event.type == "message_stop":
                    print("\n--- Message complete ---")
                    print("\n==========================")
        
        return full_text_response
    except Exception as e:
        print(f"Error in streaming: {e}")
        return None

if __name__ == "__main__":
    # Example 1: Using the default prompt
    default_response = run_inference()
    print("\nInference with default prompt complete.")
    
    # Example 2: Using a custom prompt
    custom_prompt = "Explain how quantum computing differs from classical computing in simple terms."
    custom_response = run_inference(custom_prompt)
    print("\nInference with custom prompt complete.")

import os
import json
from anthropic import Anthropic
import weave; weave.init("claude37_inference")


# Configuration
API_KEY = "your claude key" # Replace with your API key
MODEL = "claude-3-7-sonnet-20250219"
ENABLE_STREAMING = False  # Set to True to enable streaming output

# Initialize client
client = Anthropic(api_key=API_KEY)

# Tools definitions
TOOLS = [
    {
        "name": "weather",
        "description": "Get current weather information for a location.",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "The location to get weather for."}
            },
            "required": ["location"]
        }
    },
    {
        "name": "calculator",
        "description": "Perform mathematical calculations.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "The mathematical expression to evaluate."}
            },
            "required": ["expression"]
        }
    },
    {
        "name": "dictionary",
        "description": "Get definitions of words.",
        "input_schema": {
            "type": "object",
            "properties": {
                "word": {"type": "string", "description": "The word to look up."}
            },
            "required": ["word"]
        }
    }
]

# Mock data for tools
WEATHER_DATA = {
    "New York": {"temperature": 72, "condition": "Sunny", "humidity": "45%"},
    "London": {"temperature": 62, "condition": "Cloudy", "humidity": "78%"},
    "Tokyo": {"temperature": 80, "condition": "Partly cloudy", "humidity": "65%"},
    "Paris": {"temperature": 65, "condition": "Rainy", "humidity": "82%"},
    "Sydney": {"temperature": 85, "condition": "Clear", "humidity": "55%"},
}

DICTIONARY_DATA = {
    "ephemeral": {
        "definition": "Lasting for a very short time.",
        "part_of_speech": "adjective",
        "example": "Ephemeral snowflakes melted on her hand."
    },
    "ubiquitous": {
        "definition": "Present, appearing, or found everywhere.",
        "part_of_speech": "adjective",
        "example": "Smartphones have become ubiquitous in modern society."
    },
    "serendipity": {
        "definition": "The occurrence of events by chance in a happy or beneficial way.",
        "part_of_speech": "noun",
        "example": "The discovery was a perfect example of serendipity."
    },
    "algorithm": {
        "definition": "A process or set of rules to be followed in calculations or other problem-solving operations.",
        "part_of_speech": "noun",
        "example": "A search algorithm"
    }
}

def get_weather(location):
    """Mock weather data function."""
    return WEATHER_DATA.get(location, {"error": f"No weather data available for {location}"})

def calculate(expression):
    """Simple calculator function."""
    try:
        # Warning: eval can be dangerous in production code
        # Use a proper math expression parser in real applications
        result = eval(expression, {"__builtins__": {}})
        return {"result": result}
    except Exception as e:
        return {"error": f"Could not evaluate expression: {str(e)}"}

def get_definition(word):
    """Mock dictionary function."""
    return DICTIONARY_DATA.get(word.lower(), {"error": f"No definition found for '{word}'"})

def execute_tool(tool_name, tool_input):
    """Execute the appropriate tool based on the name and input."""
    if tool_name == "weather":
        return get_weather(tool_input["location"])
    elif tool_name == "calculator":
        return calculate(tool_input["expression"])
    elif tool_name == "dictionary":
        return get_definition(tool_input["word"])
    else:
        return {"error": "Unknown tool requested"}





@weave.op
def multi_tool_example(prompt):
    """
    Example showing Claude choosing between multiple tools.
    
    Args:
        prompt (str): The user's input prompt
        
    Returns:
        dict: The final response object from the API
    """
    # Initial request with the user's prompt
    print(f"\n=== QUESTION: {prompt} ===")
    response = client.messages.create(
        model=MODEL,
        max_tokens=4000,
        thinking={"type": "enabled", "budget_tokens": 2000},
        tools=TOOLS,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # Display thinking and tool selection
    thinking_blocks = [b for b in response.content if b.type == "thinking"]
    for block in thinking_blocks:
        print("\n🧠 THINKING:")
        print(block.thinking[:300] + "..." if len(block.thinking) > 300 else block.thinking)
    
    # Process tool use if needed
    conversation = [{"role": "user", "content": prompt}]
    
    # We might need multiple tool calls, so loop until we get a final answer
    while response.stop_reason == "tool_use":
        tool_block = next((b for b in response.content if b.type == "tool_use"), None)
        
        if tool_block:
            # Show which tool was selected
            print(f"\n🔧 SELECTED TOOL: {tool_block.name}")
            print(f"Tool input: {tool_block.input}")
            
            # Execute the appropriate tool
            tool_result = execute_tool(tool_block.name, tool_block.input)
            print(f"Tool result: {json.dumps(tool_result, indent=2)}")
            
            # Save assistant's response (thinking + tool use)
            assistant_blocks = thinking_blocks + [tool_block]
            conversation.append({"role": "assistant", "content": assistant_blocks})
            
            # Add tool result to conversation
            conversation.append({
                "role": "user", 
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_block.id,
                    "content": json.dumps(tool_result)
                }]
            })
            
            # Get next response
            response = client.messages.create(
                model=MODEL,
                max_tokens=4000,
                thinking={"type": "enabled", "budget_tokens": 2000},
                tools=TOOLS,
                messages=conversation
            )
            
            # Update thinking blocks for next iteration
            thinking_blocks = [b for b in response.content if b.type == "thinking"]
            for block in thinking_blocks:
                print("\n🧠 ADDITIONAL THINKING:")
                print(block.thinking[:300] + "..." if len(block.thinking) > 300 else block.thinking)
    
    # Print final answer
    print("\n✓ FINAL ANSWER:")
    final_text = ""
    for block in response.content:
        if block.type == "text":
            print(block.text)
            final_text += block.text
    
    # Collect all tools used throughout the conversation
    tools_used = []
    for msg in conversation:
        if msg["role"] == "assistant" and isinstance(msg["content"], list):
            for block in msg["content"]:
                if hasattr(block, "type") and block.type == "tool_use":
                    tools_used.append({
                        "name": block.name,
                        "input": block.input,
                        "id": block.id
                    })
    
    # Return the complete response object for further processing
    return {
        "response_object": response,
        "final_text": final_text,
        "conversation_history": conversation,
        "tools_used": tools_used
    }

if __name__ == "__main__":
    # Example 1: Weather question
    result1 = multi_tool_example("What's the current weather in Tokyo?")
    
    # Example 2: Math calculation
    result2 = multi_tool_example("Calculate the square root of 144 plus 25")
    
    # Example 3: Dictionary definition
    result3 = multi_tool_example("Define the word 'algorithm' for me")
    
    # Example 4: Complex question requiring multiple tools
    result4 = multi_tool_example("If it's 62°F in London, what's that in Celsius?")
    
    # You can now use the returned results for further processing
    print("\n=== Example of accessing returned data ===")
    print(f"Example 4 final text: {result4['final_text']}")
    
    # Print tools used in the complex query example
    print("\n=== Tools used in Example 4 ===")
    if result4['tools_used']:
        for i, tool in enumerate(result4['tools_used'], 1):
            print(f"Tool {i}: {tool['name']} (ID: {tool['id']})")
            print(f"Input: {json.dumps(tool['input'], indent=2)}")
    else:
        print("No tools were used in this example.")

import os
import asyncio
import json
from datasets import load_dataset
from anthropic import Anthropic
from litellm import completion  # Using litellm instead of Azure OpenAI
import weave
weave.init("aime_evaluation")  # Initialize Weave

# Initialize Anthropic client
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "your key")
client = Anthropic(api_key=ANTHROPIC_API_KEY)

# Set OpenAI API key for litellm
os.environ["OPENAI_API_KEY"] = "your key"  # Replace with your actual OpenAI API key

# Constants
MODEL_NAME = "claude-3-7-sonnet-20250219"
MAX_ANSWER_TOKENS = 8000  # Maximum tokens for the answer

# Define different reasoning setups to test with much larger thinking budgets
REASONING_CONFIGS = [
    {
        "name": "standard",
        "thinking": None,  # Standard Claude (no extended thinking)
        "max_tokens": MAX_ANSWER_TOKENS
    },
    {
        "name": "thinking_4k",
        "thinking": {"type": "enabled", "budget_tokens": 4000},
        "max_tokens": MAX_ANSWER_TOKENS + 4000
    },
    {
        "name": "thinking_8k",
        "thinking": {"type": "enabled", "budget_tokens": 8000},
        "max_tokens": MAX_ANSWER_TOKENS + 8000
    },
    {
        "name": "thinking_16k",
        "thinking": {"type": "enabled", "budget_tokens": 16000},
        "max_tokens": MAX_ANSWER_TOKENS + 16000
    },
    {
        "name": "thinking_24k",
        "thinking": {"type": "enabled", "budget_tokens": 24000},
        "max_tokens": MAX_ANSWER_TOKENS + 24000
    }
]

# 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(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

# Claude inference function with streaming
async def claude_inference(prompt: str, config: dict) -> str:
    """
    Generate solution using Claude with specified reasoning config and streaming.
    """
    try:
        kwargs = {
            "model": MODEL_NAME,
            "max_tokens": config["max_tokens"],
            "messages": [
                {"role": "user", "content": system_message + prompt}
            ]
        }
        
        # Add thinking parameters if specified
        if config["thinking"]:
            kwargs["thinking"] = config["thinking"]
        
        print(f"\n--- Starting inference with {config['name']} config ---")
        print(f"Max tokens: {config['max_tokens']}")
        if config["thinking"]:
            print(f"Thinking budget: {config['thinking']['budget_tokens']} tokens")
        
        thinking_content = ""
        text_content = ""
        
        # Use streaming
        with client.messages.stream(**kwargs) as stream:
            for event in stream:
                if event.type == "content_block_start":
                    if event.content_block.type == "thinking":
                        print(f"\n[Starting thinking block...]")
                    elif event.content_block.type == "text":
                        print(f"\n[Starting response...]")
                    elif event.content_block.type == "redacted_thinking":
                        print(f"\n[Redacted thinking block]")
                
                elif event.type == "content_block_delta":
                    if event.delta.type == "thinking_delta":
                        # Just print dots to show progress without flooding console
                        print(".", end="", flush=True)
                        thinking_content += event.delta.thinking
                    elif event.delta.type == "text_delta":
                        print(event.delta.text, end="", flush=True)
                        text_content += event.delta.text
                
                elif event.type == "content_block_stop":
                    if hasattr(event, 'content_block') and event.content_block.type == "thinking":
                        print(f"\n[Thinking block complete - {len(thinking_content)} chars]")
                    elif hasattr(event, 'content_block') and event.content_block.type == "text":
                        print(f"\n[Response complete - {len(text_content)} chars]")
                    else:
                        print(f"\n[Block complete]")
        
        print(f"\n--- Inference complete ---")
        
        # If no text was generated, return an error message
        if not text_content:
            return "No response generated"
            
        return text_content
                
    except Exception as e:
        print(f"Error in inference: {e}")
        return f"Error: {str(e)}"

# Define model classes for each reasoning configuration
class ClaudeStandardModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await claude_inference(text, REASONING_CONFIGS[0])

class ClaudeThinking4kModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await claude_inference(text, REASONING_CONFIGS[1])

class ClaudeThinking8kModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await claude_inference(text, REASONING_CONFIGS[2])

class ClaudeThinking16kModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await claude_inference(text, REASONING_CONFIGS[3])

class ClaudeThinking24kModel(weave.Model):
    @weave.op
    async def predict(self, text: str) -> str:
        return await claude_inference(text, REASONING_CONFIGS[4])


# 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(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}

# Load and preprocess dataset
def load_ds():
    print("Loading AIME dataset...")
    try:
        dataset = load_dataset("Maxwell-Jia/AIME_2024")["train"]  # No test set here
        return [{"text": row["Problem"], "label": row["Answer"]} for row in dataset]
    except Exception as e:
        print(f"Error loading AIME dataset: {e}")
        return []

# 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 = {
        "standard": ClaudeStandardModel(),
        "thinking_4k": ClaudeThinking4kModel(),
        "thinking_8k": ClaudeThinking8kModel(),
        "thinking_16k": ClaudeThinking16kModel(),
        "thinking_24k": ClaudeThinking24kModel(),
    
    }

    print("Preparing dataset for evaluation...")
    # Take a subset for faster testing - adjust as needed
    # test_size = 5  # Number of problems to evaluate per model
    dataset_prepared = [{"text": row["text"], "label": row["label"]} for row in dataset]

    print("Running evaluations...")
    scorers = [gpt4o_scorer]
    results = {}
    
    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"
        )
        await evaluation.evaluate(model)


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