OpenAI o-series max_completion_tokens Exhausted by Reasoning Tokens

from openai import OpenAI

client = OpenAI()


# ❌ BUG — same budget you'd use on gpt-4.1
resp_bad = client.chat.completions.create(
    model="o3",
    messages=[{"role": "user", "content": "Explain this stack trace: ..."}],
    max_completion_tokens=500,               # <-- 500 total, reasoning eats it all
)
print(repr(resp_bad.choices[0].message.content))
print(resp_bad.choices[0].finish_reason)      # "length"
print(resp_bad.usage.completion_tokens_details.reasoning_tokens)   # 500


# ✅ SIZED — reasoning budget + response budget
resp = client.chat.completions.create(
    model="o3",
    messages=[{"role": "user", "content": "Explain this stack trace: ..."}],
    reasoning_effort="medium",
    max_completion_tokens=20_000,             # 15K reasoning + 5K visible headroom
)
print(resp.choices[0].message.content)
print(f"Reasoning tokens: {resp.usage.completion_tokens_details.reasoning_tokens}")
print(f"Visible output:   {resp.usage.completion_tokens - resp.usage.completion_tokens_details.reasoning_tokens}")


# ✅ Adaptive budget — retry with larger cap if truncated
def robust_reasoning_call(model: str, messages: list, effort: str = "medium"):
    initial_budgets = {"low": 5_000, "medium": 20_000, "high": 60_000}
    budget = initial_budgets[effort]

    for attempt in range(3):
        resp = client.chat.completions.create(
            model=model,
            messages=messages,
            reasoning_effort=effort,
            max_completion_tokens=budget,
        )
        choice = resp.choices[0]
        content = choice.message.content or ""

        if choice.finish_reason != "length":
            return resp

        # Truncated — check if reasoning burned the budget
        reasoning_used = resp.usage.completion_tokens_details.reasoning_tokens
        if reasoning_used >= budget * 0.9:
            budget *= 2                        # double budget for retry
            continue
        return resp

    # Still truncated — either lower effort or accept truncation
    raise RuntimeError(f"Model kept exhausting budget even at {budget} tokens")


# ✅ Track cost per call — reasoning tokens are billed at output rate
def call_with_cost_tracking(prompt: str):
    resp = client.chat.completions.create(
        model="o3",
        messages=[{"role": "user", "content": prompt}],
        reasoning_effort="medium",
        max_completion_tokens=20_000,
    )

    u = resp.usage
    reasoning = u.completion_tokens_details.reasoning_tokens
    visible = u.completion_tokens - reasoning

    print(f"  input tokens:     {u.prompt_tokens}")
    print(f"  reasoning tokens: {reasoning}  (billed at output rate)")
    print(f"  visible tokens:   {visible}")
    print(f"  ratio (reasoning/visible): {reasoning / max(visible, 1):.1f}x")

    return resp.choices[0].message.content


# ✅ Responses API equivalent
resp_r = client.responses.create(
    model="o3",
    input="Explain this stack trace: ...",
    reasoning={"effort": "medium"},
    max_output_tokens=20_000,                 # NOTE: max_output_tokens, not max_completion_tokens
)

u = resp_r.usage
print(f"Reasoning: {u.output_tokens_details.reasoning_tokens}")
print(f"Visible:   {u.output_tokens - u.output_tokens_details.reasoning_tokens}")

from openai import OpenAI
from enum import Enum

client = OpenAI()


class TaskType(str, Enum):
    QUICK_LOOKUP = "quick_lookup"          # <200 tokens visible
    CODE_REVIEW  = "code_review"           # 500-2K tokens, moderate depth
    ARCHITECTURE = "architecture"          # deep analysis, 2K+ tokens
    SIMPLE_CHAT  = "simple_chat"           # no reasoning needed


EFFORT_MAP = {
    TaskType.QUICK_LOOKUP: ("o3", "low",    5_000),
    TaskType.CODE_REVIEW:  ("o3", "medium", 20_000),
    TaskType.ARCHITECTURE: ("o3", "high",   60_000),
    TaskType.SIMPLE_CHAT:  ("gpt-5.4", None, 2_000),   # no reasoning
}


def call(task_type: TaskType, prompt: str) -> str:
    model, effort, budget = EFFORT_MAP[task_type]

    kwargs = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_completion_tokens": budget,
    }
    if effort:
        kwargs["reasoning_effort"] = effort

    resp = client.chat.completions.create(**kwargs)
    return resp.choices[0].message.content


# Use it
answer = call(TaskType.QUICK_LOOKUP, "What is a callback URL?")
review = call(TaskType.CODE_REVIEW, f"Review this function: {code}")


# ✅ GPT-5 with minimal / none — very cheap reasoning modes
resp = client.responses.create(
    model="gpt-5.4",
    input="What time is it in Tokyo?",
    reasoning={"effort": "minimal"},           # ~100-800 reasoning tokens
    max_output_tokens=2_000,
)

# ✅ GPT-5 with effort=none — behaves like non-reasoning model
resp = client.responses.create(
    model="gpt-5.4",
    input="Translate: Hello",
    reasoning={"effort": "none"},              # no reasoning, latency-sensitive
    max_output_tokens=500,
)


# ✅ o4-mini — fixed internal budget, reasoning_effort is a no-op
# Do NOT pass reasoning_effort; use it as a normal reasoning model.
resp = client.chat.completions.create(
    model="o4-mini",
    messages=[{"role": "user", "content": "Explain event loops."}],
    max_completion_tokens=15_000,
    # reasoning_effort=...                     # ignored / warning
)


# ✅ Route via a classifier for high-volume production
def classify_and_call(user_input: str) -> str:
    """Cheap classifier picks the right depth."""
    classifier = client.chat.completions.create(
        model="gpt-5.4-nano",                  # cheap classifier
        messages=[{
            "role": "user",
            "content": (
                f"Classify complexity: 'quick_lookup' (fact/lookup), "
                f"'code_review' (analyze code), "
                f"'architecture' (multi-file design), "
                f"'simple_chat' (chit-chat). "
                f"Only output the label.\n\nInput: {user_input}"
            ),
        }],
        max_tokens=10,
    )
    task = TaskType(classifier.choices[0].message.content.strip())
    return call(task, user_input)

from openai import OpenAI

client = OpenAI()


class ReasoningExhausted(Exception):
    """Raised when the model burned budget on reasoning without producing output."""


def safe_reasoning_call(
    prompt: str,
    model: str = "o3",
    effort: str = "medium",
    initial_budget: int = 20_000,
    max_retries: int = 2,
) -> str:
    budget = initial_budget

    for attempt in range(max_retries + 1):
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            reasoning_effort=effort,
            max_completion_tokens=budget,
        )
        choice = resp.choices[0]
        content = choice.message.content or ""
        reasoning_used = resp.usage.completion_tokens_details.reasoning_tokens

        if choice.finish_reason == "stop":
            # Normal completion
            return content

        if choice.finish_reason == "length":
            if reasoning_used >= budget * 0.95:
                # Reasoning ate the budget
                if attempt < max_retries:
                    budget *= 2
                    continue
                raise ReasoningExhausted(
                    f"Model burned {reasoning_used}/{budget} tokens on reasoning "
                    f"across {attempt+1} attempts. Consider lower effort or shorter prompts."
                )
            # Content was written but truncated — return partial
            return content + "\n\n[TRUNCATED]"

        if choice.finish_reason == "content_filter":
            raise RuntimeError("Response blocked by content filter")

        return content


# ✅ Responses API — same pattern with different field names
from openai import BadRequestError

def safe_responses_call(prompt: str, effort: str = "medium", budget: int = 20_000):
    resp = client.responses.create(
        model="o3",
        input=prompt,
        reasoning={"effort": effort},
        max_output_tokens=budget,
    )

    if resp.status == "incomplete":
        reason = resp.incomplete_details.reason if resp.incomplete_details else None
        reasoning_tokens = resp.usage.output_tokens_details.reasoning_tokens
        if reason == "max_output_tokens" and reasoning_tokens >= budget * 0.95:
            raise ReasoningExhausted(
                f"Reasoning exhausted budget ({reasoning_tokens}/{budget})"
            )

    return resp.output_text


# ✅ Streaming — detect exhaustion at the terminal event
def stream_with_exhaustion_check(prompt: str):
    stream = client.responses.create(
        model="o3",
        input=prompt,
        reasoning={"effort": "medium"},
        max_output_tokens=20_000,
        stream=True,
    )

    text = []
    for event in stream:
        if event.type == "response.output_text.delta":
            chunk = getattr(event, "delta", "")
            text.append(chunk)
            print(chunk, end="", flush=True)

        elif event.type == "response.incomplete":
            reason = event.response.incomplete_details.reason
            reasoning = event.response.usage.output_tokens_details.reasoning_tokens
            if reason == "max_output_tokens" and not text:
                print(f"\n[EMPTY: {reasoning} reasoning tokens consumed]")
                raise ReasoningExhausted("stream ended empty due to reasoning exhaustion")

        elif event.type == "response.completed":
            return "".join(text)


# ✅ Log exhaustion for observability
import logging

def log_reasoning_metrics(resp):
    u = resp.usage
    reasoning = u.completion_tokens_details.reasoning_tokens
    visible = u.completion_tokens - reasoning
    logging.info(
        "reasoning_call model=%s reasoning_tokens=%d visible_tokens=%d ratio=%.2f",
        resp.model, reasoning, visible, reasoning / max(visible, 1),
    )
    # Alert if ratio > 20 (unusually reasoning-heavy)
    if reasoning / max(visible, 1) > 20:
        logging.warning("high reasoning/visible ratio on %s", resp.id)