Claude 4 Sonnet と Claude 4 Opus の始め方

if "ANTHROPIC_API_KEY" not in os.environ:
    os.environ["ANTHROPIC_API_KEY"] = "your api key"

!pip install anthropic weave 

import os 
# Optional: default API key fallback for demo (replace/remove for production!)
if "ANTHROPIC_API_KEY" not in os.environ:
    os.environ["ANTHROPIC_API_KEY"] = "your_claude_api_key"

if "WANDB_API_KEY" not in os.environ:
    os.environ["WANDB_API_KEY"] = "your_wandb_api_key"
import anthropic
import os
import weave; weave.init("claude_4")
# === CONFIGURABLE ARGS ===
streaming = True          # Set to True for streaming, False for classic response
enable_thinking = True    # Set to True to enable extended thinking

# ==== MODEL SELECTION ====
# Claude Opus 4 (highest reasoning, highest cost)
model = "claude-opus-4-20250514"
# Claude Sonnet 4 (fast, less expensive)
# model = "claude-sonnet-4-20250514"

client = anthropic.Anthropic()


@weave.op
def claude_inference(prompt, model, streaming=True, enable_thinking=True):
    """
    Run inference on given prompt with specified Claude model,
    printing thinking and response as they arrive,
    and returning the full final response text.

    Returns: text response as a string, or None on failure
    """
    kwargs = dict(
        model=model,
        max_tokens=2048 if enable_thinking else 512,
        messages=[{"role": "user", "content": prompt}],
    )

    if enable_thinking:
        kwargs["thinking"] = {"type": "enabled", "budget_tokens": 1024}

    response_text = ""

    if streaming:
        with client.messages.stream(**kwargs) as stream:
            for event in stream:
                if event.type == "content_block_start":
                    block_type = event.content_block.type
                    if block_type == "thinking":
                        print("\n[THINKING]: ", end="", flush=True)
                    elif block_type == "text":
                        print("\n[RESPONSE]: ", end="", flush=True)
                elif event.type == "content_block_delta":
                    d = event.delta
                    if getattr(d, "type", None) == "thinking_delta":
                        print(d.thinking, end="", flush=True)
                    elif getattr(d, "type", None) == "text_delta":
                        print(d.text, end="", flush=True)
                        response_text += d.text
                elif event.type == "content_block_stop":
                    print()  # Finish this block
    else:
        response = client.messages.create(**kwargs)
        for block in response.content:
            if block.type == "thinking" and block.thinking.strip():
                print("\n[THINKING]:", block.thinking.strip(), flush=True)
            elif block.type == "text" and block.text.strip():
                print("\n[RESPONSE]:", block.text.strip(), flush=True)
                response_text += block.text.strip()
    # return response_text if response_text else None
    return str(response_text)
# === USAGE EXAMPLE ===

prompt = (
    "If a train travels 60 miles per hour for 2.5 hours, how far does it go? "
    "Show your reasoning step by step."
)

final_answer = claude_inference(
    prompt,
    model=model,
    streaming=streaming,
    enable_thinking=enable_thinking
)

print("\n\n=== FINAL RETURNED RESPONSE ===\n", final_answer)

if enable_thinking:
    kwargs["thinking"] = {"type": "enabled", "budget_tokens": 1024}


import anthropic
import ast
import os
import weave; weave.init("claude_4")

REVENUE_PER_MONTH = [5000, 6000, 4800, 5300, 5500, 5100, 4950, 5400, 5200, 5100, 5300, 5000]

def handle_db_command(command):
    cmd = command.lower().strip()
    if cmd == "get average monthly revenue":
        avg = sum(REVENUE_PER_MONTH) / len(REVENUE_PER_MONTH)
        return f"{avg:.2f}"
    elif cmd == "get total yearly revenue":
        return str(sum(REVENUE_PER_MONTH))
    elif cmd == "get best month":
        month = REVENUE_PER_MONTH.index(max(REVENUE_PER_MONTH)) + 1
        val = max(REVENUE_PER_MONTH)
        return f"Month {month}, revenue: {val}"
    else:
        return "Error: unknown database command"

def safe_eval(expr):
    try:
        node = ast.parse(expr, mode='eval')
        for sub in ast.walk(node):
            if not isinstance(sub, (ast.Expression, ast.BinOp, ast.UnaryOp, ast.Num, ast.Load, ast.operator, ast.unaryop, ast.Constant)):
                raise ValueError("Unsafe")
        return eval(expr, {"__builtins__": {}})
    except Exception as e:
        return f"Error: {e}"

calculator_tool = {
    "name": "calculator",
    "description": "Safely evaluates arithmetic expressions (e.g. '150 * 50')",
    "input_schema": {
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "Math expression"}
        },
        "required": ["expression"]
    }
}
database_tool = {
    "name": "database",
    "description": (
        "A company revenue database. Supported commands (use as the `command` field):\n"
        "- 'get average monthly revenue' (Returns yearly average)\n"
        "- 'get total yearly revenue'\n"
        "- 'get best month' (Returns best month and its value)\n"
        "Do NOT use SQL. Only use one of the 3 supported commands."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "command": {"type": "string", "description": "Plain database command"}
        },
        "required": ["command"]
    }
}

PROMPT = (
    "What's the total revenue if we sold 150 units of product A at $50 each?\n"
    "Also, compare this to our average monthly revenue in our database.\n\n"
    "Instructions for the database tool:\n"
    "- Only use one of these commands for the 'command' field:\n"
    "    * get average monthly revenue\n"
    "    * get total yearly revenue\n"
    "    * get best month\n"
    "Do not use SQL. Only use one of these simple command phrases."
)

client = anthropic.Anthropic()

def execute_tools(blocks):
    """
    Find all tool calls in blocks.
    Return tool_result dicts to send as messages back to Claude.
    """
    tool_results = []
    for b in blocks:
        if b.type == "tool_use":
            if b.name == "calculator":
                val = safe_eval(b.input['expression'])
                out = str(val)
            elif b.name == "database":
                out = handle_db_command(b.input['command'])
            else:
                out = "Error: unknown tool"
            tool_results.append({"type": "tool_result", "tool_use_id": b.id, "content": out})
    return tool_results

# === ONE MODEL CALL (one assistant reply, may be tool call or final text) ===
@weave.op()
def claude_message_op(msg_history, interleaved=False):
    """
    Runs a single Anthropic API call (one assistant step).
    Returns list of blocks (tool calls or text/thinking blocks).
    """
    extra_headers = (
        {"anthropic-beta": "interleaved-thinking-2025-05-14"} if interleaved else None
    )
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2512,
        tools=[calculator_tool, database_tool],
        thinking={"type": "enabled", "budget_tokens": 2000},
        extra_headers=extra_headers,
        messages=msg_history,
    )
    return response.content

# === MAIN CHAIN ===
@weave.op()
def run_claude_tools_full_chain(
    prompt_text=PROMPT, interleaved=False, max_loops=5):
    """
    Main loop: calls single-step Claude; if tool call, executes it and calls Claude again etc.
    Each step is a tracked @weave.op for graph lineage.
    Returns the full set of blocks and tool call/results for inspection in Weave.
    """
    all_steps = []
    msg_history = [{"role": "user", "content": prompt_text}]
    for loop in range(max_loops):
        assistant_blocks = claude_message_op(msg_history, interleaved=interleaved)
        all_steps.append({"assistant_blocks": assistant_blocks, "msg_history": list(msg_history)})
        tool_results = execute_tools(assistant_blocks)
        if not tool_results:
            break
        msg_history.append({"role": "assistant", "content": assistant_blocks})
        msg_history.append({"role": "user", "content": tool_results})
        all_steps[-1]["tool_results"] = tool_results
    return all_steps

def print_blocks(blocks):
    for block in blocks:
        if block.type == "thinking":
            print(f"[THINKING BLOCK]: {block.thinking}\n{'-'*60}")
        elif block.type == "tool_use":
            print(f"[TOOL USE BLOCK]: {block.name} with input {block.input} (id={block.id}) {'-'*40}")
        elif block.type == "text":
            print(f"[RESPONSE BLOCK]:\n{block.text}\n{'-'*60}")



# interleaved = True 

steps = run_claude_tools_full_chain(
    prompt_text=PROMPT,
    interleaved=True,
    max_loops=5
)

print("\n====== Full Conversation Step Trace ======")
for i, step in enumerate(steps):
    print(f"\n----- Assistant Step {i+1} -----")
    print_blocks(step["assistant_blocks"])
    if "tool_results" in step:
        print(f"\nTOOL RESULTS: {step['tool_results']}")

# Optionally, the "final" response blocks (which is the last 'assistant_blocks' returned)
final_blocks = steps[-1]["assistant_blocks"]
print("\n--- Final Assistant Response ---\n")
print_blocks(final_blocks)


# interleaved = False 

steps = run_claude_tools_full_chain(
    prompt_text=PROMPT,
    interleaved=False,
    max_loops=5
)

print("\n====== Full Conversation Step Trace ======")
for i, step in enumerate(steps):
    print(f"\n----- Assistant Step {i+1} -----")
    print_blocks(step["assistant_blocks"])
    if "tool_results" in step:
        print(f"\nTOOL RESULTS: {step['tool_results']}")

# Optionally, the "final" response blocks (which is the last 'assistant_blocks' returned)
final_blocks = steps[-1]["assistant_blocks"]
print("\n--- Final Assistant Response ---\n")
print_blocks(final_blocks)