Claude extended_thinking — budget exceeded, no final answer returned

"""Size max_tokens for extended thinking, safely."""
import anthropic

client = anthropic.Anthropic()

def call_with_thinking(prompt: str,
                      task_complexity: str = "medium",
                      expected_answer_tokens: int = 2000):
    """
    task_complexity: 'simple' | 'medium' | 'complex' | 'research'
    Determines the reasoning budget.
    """
    budgets = {
        "simple":   2_000,   # yes/no with light reasoning
        "medium":   6_000,   # code review, single-hop analysis
        "complex": 16_000,   # multi-hop planning, math
        "research": 32_000,  # deep reasoning tasks (Opus recommended)
    }
    budget = budgets[task_complexity]
    safety_margin = 500                              # room for token accounting slop
    max_tokens = budget + expected_answer_tokens + safety_margin

    # Never let budget exceed 70% of max_tokens
    if budget / max_tokens > 0.70:
        max_tokens = int(budget / 0.70) + safety_margin

    return client.messages.create(
        model="claude-opus-4-7",
        max_tokens=max_tokens,
        thinking={"type": "enabled", "budget_tokens": budget},
        messages=[{"role": "user", "content": prompt}],
    )

response = call_with_thinking(
    "Prove that there are infinitely many prime numbers, from scratch.",
    task_complexity="complex",
    expected_answer_tokens=3000,
)

# Extract the final text (skip thinking blocks)
text = "".join(b.text for b in response.content if b.type == "text")
reasoning = "".join(b.thinking for b in response.content if b.type == "thinking")
print(f"Reasoning: {len(reasoning)} chars; Answer: {len(text)} chars")
print(f"stop_reason: {response.stop_reason}")

import anthropic
from anthropic.types import Message

client = anthropic.Anthropic()

MAX_ESCALATIONS = 2  # bail after two retries to keep cost bounded

def call_with_escalation(prompt: str,
                         initial_budget: int = 6_000,
                         initial_answer_headroom: int = 2_000) -> Message:
    budget = initial_budget
    headroom = initial_answer_headroom

    for attempt in range(MAX_ESCALATIONS + 1):
        response = client.messages.create(
            model="claude-opus-4-7",
            max_tokens=budget + headroom + 500,
            thinking={"type": "enabled", "budget_tokens": budget},
            messages=[{"role": "user", "content": prompt}],
        )

        if response.stop_reason != "max_tokens":
            return response

        # Grew reasoning by 50%, answer headroom by 2x
        budget = int(budget * 1.5)
        headroom = int(headroom * 2)
        print(f"[attempt {attempt}] truncated — retrying with "
              f"budget={budget}, headroom={headroom}")

    # Gave up — return the truncated response but flag it
    return response

resp = call_with_escalation("Design a distributed rate limiter for 100M req/day.",
                            initial_budget=8000, initial_answer_headroom=3000)
print("stop_reason:", resp.stop_reason)

import anthropic

client = anthropic.Anthropic()

def stream_with_thinking_events(prompt: str):
    """Yields {"type": "thinking"|"text"|"stop", ...} events."""
    with client.messages.stream(
        model="claude-opus-4-7",
        max_tokens=12_000,
        thinking={"type": "enabled", "budget_tokens": 8_000},
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        current_block_type = None

        for event in stream:
            if event.type == "content_block_start":
                current_block_type = event.content_block.type
                if current_block_type == "thinking":
                    yield {"type": "thinking_start"}
                elif current_block_type == "text":
                    yield {"type": "text_start"}

            elif event.type == "content_block_delta":
                delta = event.delta
                if delta.type == "thinking_delta":
                    yield {"type": "thinking", "delta": delta.thinking}
                elif delta.type == "text_delta":
                    yield {"type": "text", "delta": delta.text}

            elif event.type == "message_stop":
                yield {"type": "stop", "stop_reason": stream.current_message_snapshot.stop_reason}

# UI-side rendering
for event in stream_with_thinking_events("Solve: what is 17 * 43 without using multiplication."):
    if event["type"] == "thinking_start":
        print("[thinking...]", end="", flush=True)
    elif event["type"] == "thinking":
        # For internal debug or dedicated UI panel; do NOT show in main output
        pass
    elif event["type"] == "text_start":
        print("\n[answer:]\n", end="")
    elif event["type"] == "text":
        print(event["delta"], end="", flush=True)
    elif event["type"] == "stop":
        print(f"\n[stop_reason: {event['stop_reason']}]")