Azure-AI_Foundry

Lesson 3: Enterprise Patterns

Multi-Region Deployment

                    ┌──────────────┐
                    │ Azure Front  │
                    │    Door      │
                    └──────┬───────┘
                     ┌─────┴─────┐
              ┌──────┴──┐   ┌────┴─────┐
              │ East US  │   │ West EU  │
              │  Region  │   │  Region  │
              ├──────────┤   ├──────────┤
              │ APIM     │   │ APIM     │
              │ OpenAI   │   │ OpenAI   │
              │ AI Search│   │ AI Search│
              │ App Svc  │   │ App Svc  │
              └──────────┘   └──────────┘

Configuration Pattern

# Multi-region client with automatic failover
REGIONS = [
    {"endpoint": "https://openai-eastus.openai.azure.com/", "priority": 1},
    {"endpoint": "https://openai-westeu.openai.azure.com/", "priority": 2},
]

def get_client(region_index: int = 0) -> AzureOpenAI:
    return AzureOpenAI(
        azure_endpoint=REGIONS[region_index]["endpoint"],
        api_version="2024-08-01-preview",
    )

def call_with_failover(messages: list, max_retries: int = 3) -> str:
    for region in REGIONS:
        try:
            client = AzureOpenAI(
                azure_endpoint=region["endpoint"],
                api_version="2024-08-01-preview",
            )
            response = client.chat.completions.create(
                model="gpt-4o", messages=messages,
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"Region {region['endpoint']} failed: {e}")
            continue
    raise Exception("All regions exhausted")

Retry with Exponential Backoff + Jitter

import time
import random
from openai import RateLimitError, APITimeoutError, APIConnectionError

def call_with_retry(client, messages, max_retries=5):
    retryable_errors = (RateLimitError, APITimeoutError, APIConnectionError)

    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model="gpt-4o", messages=messages
            )
        except RateLimitError as e:
            # Use Retry-After header if available
            retry_after = int(e.response.headers.get("Retry-After", 0))
            if retry_after:
                time.sleep(retry_after)
                continue

            # Exponential backoff with jitter
            base_delay = min(2 ** attempt, 60)
            jitter = random.uniform(0, base_delay * 0.5)
            delay = base_delay + jitter
            print(f"Rate limited. Retrying in {delay:.1f}s (attempt {attempt + 1})")
            time.sleep(delay)

        except retryable_errors as e:
            delay = min(2 ** attempt, 30) + random.uniform(0, 1)
            print(f"Transient error: {e}. Retrying in {delay:.1f}s")
            time.sleep(delay)

    raise Exception(f"Failed after {max_retries} retries")

Tip: Always use jitter in retry delays. Without jitter, all clients retry at the same time, creating a thundering herd that prolongs the outage.

Token Management & Rate Limit Handling

import tiktoken

def count_tokens(messages: list, model: str = "gpt-4o") -> int:
    """Estimate token count before sending a request."""
    encoding = tiktoken.encoding_for_model(model)
    token_count = 0
    for msg in messages:
        token_count += 4  # message overhead
        token_count += len(encoding.encode(msg["content"]))
    token_count += 2  # reply priming
    return token_count

def truncate_to_budget(text: str, max_tokens: int, model: str = "gpt-4o") -> str:
    """Truncate text to fit within a token budget."""
    encoding = tiktoken.encoding_for_model(model)
    tokens = encoding.encode(text)
    if len(tokens) <= max_tokens:
        return text
    return encoding.decode(tokens[:max_tokens])

# Usage: ensure request fits within limits
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": truncate_to_budget(user_input, max_tokens=3000)},
]
prompt_tokens = count_tokens(messages)
max_completion = 128000 - prompt_tokens  # GPT-4o context window

Cost Allocation and Chargeback

# Track usage per department/project for chargeback
import logging

usage_logger = logging.getLogger("ai_usage")

def tracked_completion(client, messages, department: str, project: str):
    response = client.chat.completions.create(model="gpt-4o", messages=messages)

    usage_logger.info(
        "AI Usage",
        extra={
            "department": department,
            "project": project,
            "model": "gpt-4o",
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "estimated_cost_usd": (
                response.usage.prompt_tokens * 0.0025 / 1000
                + response.usage.completion_tokens * 0.01 / 1000
            ),
        },
    )
    return response

Cost Allocation via APIM

Use APIM policies to tag requests and aggregate usage in Azure Monitor:

<inbound>
  <set-header name="x-department" exists-action="override">
    <value>@(context.Request.Headers.GetValueOrDefault("x-department","unknown"))</value>
  </set-header>
  <azure-openai-emit-token-metric namespace="ai-costs">
    <dimension name="department" value="@(context.Request.Headers.GetValueOrDefault("x-department","unknown"))" />
    <dimension name="project" value="@(context.Request.Headers.GetValueOrDefault("x-project","unknown"))" />
  </azure-openai-emit-token-metric>
</inbound>

Enterprise Reference Architecture

┌────────────────────────────────────────────────────────────────┐
│                        Azure Front Door                        │
│                    (Global Load Balancing + WAF)                │
├───────────────────────────┬────────────────────────────────────┤
│        Region A           │           Region B                 │
├───────────────────────────┼────────────────────────────────────┤
│  ┌─────────────────────┐  │  ┌─────────────────────────────┐  │
│  │ APIM (AI Gateway)   │  │  │ APIM (AI Gateway)           │  │
│  │ - Rate limiting     │  │  │ - Rate limiting             │  │
│  │ - Semantic cache    │  │  │ - Semantic cache            │  │
│  │ - Token metrics     │  │  │ - Token metrics             │  │
│  └────────┬────────────┘  │  └────────┬────────────────────┘  │
│  ┌────────┴────────────┐  │  ┌────────┴────────────────────┐  │
│  │ App Service / ACA   │  │  │ App Service / ACA           │  │
│  │ (Application Layer) │  │  │ (Application Layer)         │  │
│  └────────┬────────────┘  │  └────────┬────────────────────┘  │
│  ┌────────┴────────────┐  │  ┌────────┴────────────────────┐  │
│  │ Azure OpenAI        │  │  │ Azure OpenAI                │  │
│  │ AI Search           │  │  │ AI Search                   │  │
│  │ Cosmos DB (GRS)     │  │  │ Cosmos DB (GRS)             │  │
│  │ Key Vault           │  │  │ Key Vault                   │  │
│  └─────────────────────┘  │  └─────────────────────────────┘  │
├───────────────────────────┴────────────────────────────────────┤
│  Shared: Azure Monitor │ Log Analytics │ Microsoft Entra ID   │
└────────────────────────────────────────────────────────────────┘

HA/DR Summary

Component HA Strategy RPO RTO
Azure OpenAI Multi-region via APIM 0 Seconds (failover)
AI Search Geo-replicated index Near 0 Minutes
Cosmos DB Multi-region write 0 Automatic
App Service Multi-region + Front Door 0 Seconds
Key Vault Built-in geo-replication 0 Automatic

Tip: Design for graceful degradation. If the AI service is unavailable, serve cached responses, show pre-computed content, or present a “try again” message — never leave users with an empty screen.

References