Azure OpenAI PTU capacity exceeded — throttling on reserved throughput

"""Size PTU from real request telemetry.

Feed this Application Insights or Log Analytics data:
  - p95 input tokens per request
  - p95 output tokens per request
  - target sustained RPS
"""
from dataclasses import dataclass

# Rough tokens-per-PTU-second for common models (2026-01, verify in Azure docs)
MODEL_INPUT_TPS_PER_PTU  = {"gpt-4o": 50,  "gpt-4o-mini": 250, "gpt-4.1": 40, "o3-mini": 30}
MODEL_OUTPUT_TPS_PER_PTU = {"gpt-4o": 13,  "gpt-4o-mini": 65,  "gpt-4.1": 10, "o3-mini": 8}

@dataclass
class TrafficShape:
    model: str
    p95_input_tokens: int
    p95_output_tokens: int
    target_rps: float
    burst_headroom: float = 1.4  # +40% for spikes

def size_ptu(t: TrafficShape) -> int:
    input_tps  = t.p95_input_tokens  * t.target_rps
    output_tps = t.p95_output_tokens * t.target_rps

    ptu_for_input  = input_tps  / MODEL_INPUT_TPS_PER_PTU[t.model]
    ptu_for_output = output_tps / MODEL_OUTPUT_TPS_PER_PTU[t.model]

    # PTU serves whichever side is heavier at a given moment
    base_ptu = max(ptu_for_input, ptu_for_output)
    sized = int(base_ptu * t.burst_headroom)

    # Round up to increments of 25 (Azure's granularity)
    return ((sized + 24) // 25) * 25

# Example: chat app doing 10 RPS with 1500-token prompts and 400-token responses
t = TrafficShape(model="gpt-4o", p95_input_tokens=1500, p95_output_tokens=400, target_rps=10.0)
print(f"Recommended PTU: {size_ptu(t)}")   # -> 100 PTU on gpt-4o

import os
from openai import AzureOpenAI

client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version="2024-10-21",
)

# PTU deployment, with header pointing to a Standard fallback
response = client.chat.completions.create(
    model="gpt4o-ptu",                     # PTU deployment
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=200,
    extra_headers={
        "x-ms-spillover-deployment-name": "gpt4o-standard",   # fallback
    },
)

# Response headers tell you which deployment served the request
served_by = response.model_extra.get("x-ms-deployment-name") or "unknown"
print(f"Served by: {served_by}")

# Provision a Standard deployment of the SAME model to receive spillover
az cognitiveservices account deployment create \
  --name my-aoai-resource \
  --resource-group my-rg \
  --deployment-name gpt4o-standard \
  --model-name gpt-4o \
  --model-version 2024-08-06 \
  --model-format OpenAI \
  --sku-capacity 100 \
  --sku-name Standard

# Verify PTU deployment carries the spillover default
az cognitiveservices account deployment show \
  --name my-aoai-resource -g my-rg \
  --deployment-name gpt4o-ptu \
  --query "properties.spilloverDeploymentName"

import time
import threading
from azure.identity import DefaultAzureCredential
from azure.monitor.query import MetricsQueryClient, MetricAggregationType

RESOURCE_ID = "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.CognitiveServices/accounts/{acct}"
credential = DefaultAzureCredential()
metrics_client = MetricsQueryClient(credential)

class PTUUtilizationTracker:
    """Polls PTU utilization and exposes a boolean 'saturating' flag."""

    def __init__(self, resource_id: str, poll_interval: int = 30, threshold: float = 85.0):
        self.resource_id = resource_id
        self.poll_interval = poll_interval
        self.threshold = threshold
        self._utilization = 0.0
        self._stop = threading.Event()

    @property
    def saturating(self) -> bool:
        return self._utilization >= self.threshold

    def start(self):
        threading.Thread(target=self._loop, daemon=True).start()

    def stop(self):
        self._stop.set()

    def _loop(self):
        while not self._stop.is_set():
            try:
                response = metrics_client.query_resource(
                    self.resource_id,
                    metric_names=["AzureOpenAIProvisionedManagedUtilization"],
                    granularity="PT1M",
                    aggregations=[MetricAggregationType.MAXIMUM],
                )
                for metric in response.metrics:
                    for ts in metric.timeseries:
                        latest = ts.data[-1].maximum if ts.data else 0
                        self._utilization = float(latest or 0)
            except Exception:
                pass  # Keep last known value on error
            self._stop.wait(self.poll_interval)

# Application code
tracker = PTUUtilizationTracker(RESOURCE_ID)
tracker.start()

def handle_request(is_urgent: bool):
    if tracker.saturating and not is_urgent:
        return {"error": "system_busy_retry_later", "utilization": tracker._utilization}
    # ... make the actual OpenAI call ...