azure-openai-llm

name	azure-openai-llm
description	Expert guidance for integrating Azure OpenAI services into Python applications. Use when implementing LLM-powered features, chat completions, embeddings, structured output extraction, rate limiting, fallback strategies, or Azure-specific authentication. Covers both sync and async patterns.

Azure OpenAI Integration Skill

Overview

Azure OpenAI provides access to OpenAI models (GPT-5.x, GPT-4.x, embeddings) with enterprise security, compliance, and regional availability. Uses the same OpenAI Python SDK with Azure-specific configuration.

When to Use This Skill

• Integrating GPT models for text generation or chat
• Implementing entity extraction or structured output
• Building multi-model fallback strategies
• Handling rate limits and API errors gracefully
• Using Azure authentication (API keys or Entra ID)
• Generating embeddings for semantic search

Configuration

Environment Variables

# .env (actual project configuration)
AZURE_OPENAI_ENDPOINT=https://aihubeastus26267492086.cognitiveservices.azure.com/
AZURE_OPENAI_API_KEY=your-api-key
AZURE_OPENAI_API_VERSION=2024-12-01-preview

# Model deployments (use YOUR deployment names)
AZURE_OPENAI_DEPLOYMENT_NAME=gpt-5.2-chat           # Primary chat model
AZURE_OPENAI_GPT5_DEPLOYMENT_NAME=gpt-5.1           # Fallback model
AZURE_OPENAI_CODEX_DEPLOYMENT_NAME=gpt-5.1-codex-max  # Code generation
AZURE_OPENAI_TEXTEMBEDDING_DEPLOYMENT_NAME=text-embedding-3-small

Client Initialization

import os
from openai import OpenAI, AsyncOpenAI

# Synchronous client
client = OpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
)

# Async client (preferred for high-throughput)
async_client = AsyncOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
)

Microsoft Entra ID Authentication (Recommended for Production)

from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(),
    "https://cognitiveservices.azure.com/.default"
)

client = OpenAI(
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
    api_key=token_provider,  # Token provider acts as API key
)

Chat Completions

Basic Usage

from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
)

response = client.chat.completions.create(
    model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Extract entities from this contract..."},
    ],
    temperature=0.1,  # Low for extraction tasks
    max_tokens=2000,
)

print(response.choices[0].message.content)

Async Chat Completion

from openai import AsyncOpenAI
import asyncio

async_client = AsyncOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
)

async def extract_entities(text: str) -> str:
    response = await async_client.chat.completions.create(
        model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
        messages=[
            {"role": "system", "content": "Extract key entities as JSON."},
            {"role": "user", "content": text},
        ],
        temperature=0,
        response_format={"type": "json_object"},
    )
    return response.choices[0].message.content

Structured Output with JSON Mode

import json
from pydantic import BaseModel

class ContractEntity(BaseModel):
    parties: list[str]
    effective_date: str | None
    expiration_date: str | None
    payment_amount: str | None
    key_risks: list[str]

async def extract_contract_entities(text: str) -> ContractEntity:
    response = await async_client.chat.completions.create(
        model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
        messages=[
            {
                "role": "system",
                "content": """Extract contract entities as JSON with this schema:
{
    "parties": ["Company A", "Company B"],
    "effective_date": "2025-01-15",
    "expiration_date": "2026-01-15",
    "payment_amount": "$50,000",
    "key_risks": ["Non-compete clause", "IP dispute"]
}"""
            },
            {"role": "user", "content": text},
        ],
        temperature=0,
        response_format={"type": "json_object"},
    )
    
    data = json.loads(response.choices[0].message.content)
    return ContractEntity(**data)

Responses API (New Stateful API)

Azure OpenAI's Responses API provides stateful conversations and tool calling:

from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
)

# Create a response
response = client.responses.create(
    model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
    input="Analyze this contract and extract key dates.",
)

print(response.output_text)

# Chain responses (maintains context)
second_response = client.responses.create(
    model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
    previous_response_id=response.id,
    input="Now find any payment terms.",
)

Multi-Model Fallback Strategy

import asyncio
from openai import AsyncOpenAI, RateLimitError, APITimeoutError
import logging

logger = logging.getLogger(__name__)

class AzureOpenAIClient:
    """Azure OpenAI client with automatic fallback."""
    
    def __init__(self):
        self.client = AsyncOpenAI(
            api_key=os.getenv("AZURE_OPENAI_API_KEY"),
            base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
            timeout=30.0,
        )
        self.primary_model = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME")  # gpt-5.2-chat
        self.fallback_model = os.getenv("AZURE_OPENAI_GPT5_DEPLOYMENT_NAME")  # gpt-5.1
    
    async def complete(
        self,
        messages: list[dict],
        temperature: float = 0.1,
        max_retries: int = 3,
    ) -> tuple[str, str]:
        """
        Complete with automatic fallback.
        Returns: (response_text, model_used)
        """
        # Try primary model
        try:
            response = await self.client.chat.completions.create(
                model=self.primary_model,
                messages=messages,
                temperature=temperature,
            )
            return response.choices[0].message.content, self.primary_model
            
        except (RateLimitError, APITimeoutError) as e:
            logger.warning(f"Primary model failed: {e}, trying fallback")
        
        # Try fallback model with retries
        for attempt in range(max_retries):
            try:
                response = await self.client.chat.completions.create(
                    model=self.fallback_model,
                    messages=messages,
                    temperature=temperature,
                )
                return response.choices[0].message.content, self.fallback_model
                
            except RateLimitError as e:
                wait_time = 2 ** attempt
                logger.warning(f"Fallback attempt {attempt+1} failed, waiting {wait_time}s")
                await asyncio.sleep(wait_time)
        
        raise Exception("All Azure OpenAI models exhausted")

Rate Limit Handling

from openai import RateLimitError, APITimeoutError, APIConnectionError
import asyncio

async def call_with_retry(
    client: AsyncOpenAI,
    model: str,
    messages: list[dict],
    max_retries: int = 5,
) -> str:
    """Call Azure OpenAI with exponential backoff."""
    
    for attempt in range(max_retries):
        try:
            response = await client.chat.completions.create(
                model=model,
                messages=messages,
            )
            return response.choices[0].message.content
            
        except RateLimitError as e:
            # Check for retry-after header
            retry_after = getattr(e, 'retry_after', None) or (2 ** attempt)
            print(f"Rate limited. Waiting {retry_after}s...")
            await asyncio.sleep(retry_after)
            
        except APITimeoutError:
            wait = 2 ** attempt
            print(f"Timeout. Retry {attempt+1}/{max_retries} in {wait}s")
            await asyncio.sleep(wait)
            
        except APIConnectionError as e:
            print(f"Connection error: {e}")
            raise
    
    raise Exception(f"Max retries ({max_retries}) exceeded")

Embeddings

async def generate_embeddings(texts: list[str]) -> list[list[float]]:
    """Generate embeddings using Azure OpenAI."""
    embedding_model = os.getenv("AZURE_OPENAI_TEXTEMBEDDING_DEPLOYMENT_NAME")  # text-embedding-3-small
    
    response = await async_client.embeddings.create(
        model=embedding_model,
        input=texts,
    )
    
    return [item.embedding for item in response.data]

# Batch processing (max 2048 items per request)
async def batch_embed(texts: list[str], batch_size: int = 100) -> list[list[float]]:
    """Batch embedding generation."""
    all_embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        embeddings = await generate_embeddings(batch)
        all_embeddings.extend(embeddings)
    
    return all_embeddings

Streaming Responses

async def stream_response(prompt: str):
    """Stream response for real-time output."""
    
    stream = await async_client.chat.completions.create(
        model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

Error Handling Best Practices

from openai import (
    RateLimitError,
    APITimeoutError,
    APIConnectionError,
    BadRequestError,
    AuthenticationError,
)

async def safe_completion(messages: list[dict]) -> str | None:
    """Comprehensive error handling."""
    
    try:
        response = await async_client.chat.completions.create(
            model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
            messages=messages,
        )
        return response.choices[0].message.content
        
    except AuthenticationError:
        # API key invalid or expired
        raise  # Don't retry
        
    except BadRequestError as e:
        # Invalid request (content filter, too many tokens, etc.)
        logger.error(f"Bad request: {e}")
        raise  # Don't retry
        
    except RateLimitError:
        # 429 - Too many requests
        # Retry with backoff
        raise
        
    except APITimeoutError:
        # Request timed out
        # Retry with backoff
        raise
        
    except APIConnectionError:
        # Network issue
        raise

Integration with Temporal Activities

from temporalio import activity
from openai import AsyncOpenAI, RateLimitError

@activity.defn(name="extract_entities")
async def extract_entities(text: str) -> dict:
    """Activity that calls Azure OpenAI with proper error handling."""
    
    client = AsyncOpenAI(
        api_key=os.getenv("AZURE_OPENAI_API_KEY"),
        base_url=f"{os.getenv('AZURE_OPENAI_ENDPOINT')}openai/v1/",
    )
    
    activity.heartbeat("Calling Azure OpenAI...")
    
    try:
        response = await client.chat.completions.create(
            model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),  # gpt-5.2-chat
            messages=[
                {"role": "system", "content": "Extract entities as JSON."},
                {"role": "user", "content": text},
            ],
            response_format={"type": "json_object"},
        )
        return {
            "entities": response.choices[0].message.content,
            "model": os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
            "tokens": response.usage.total_tokens,
        }
        
    except RateLimitError:
        # Let Temporal retry the activity
        activity.heartbeat("Rate limited, Temporal will retry")
        raise

Available Models (Azure OpenAI - Jan 2026)

Model	Use Case	Notes
gpt-5.2	Highest accuracy, complex reasoning	Latest flagship
gpt-5.1	Fast, cost-effective	Good fallback
gpt-5.1-codex	Code generation	Specialized
text-embedding-3-large	High-quality embeddings	3072 dimensions
text-embedding-3-small	Fast embeddings	1536 dimensions

Installation

pip install openai>=1.60.0 azure-identity

References

• Azure OpenAI Documentation
• OpenAI Python SDK
• Azure OpenAI Responses API
• Rate Limits