Azure-AI_Foundry

Lesson 2: Batch Processing & Real-Time Streaming

Global Batch API

The Global Batch API processes large volumes of requests asynchronously at 50% lower cost compared to real-time endpoints, with a 24-hour SLA for completion.

When to Use Batch

Criteria	Batch	Real-Time
Latency tolerance	Hours acceptable	Milliseconds needed
Volume	1,000+ requests	Individual requests
Cost sensitivity	High — need savings	Performance > cost
Use cases	Document processing, bulk classification, dataset generation	Chat, search, live apps

Submitting a Batch Job

from openai import AzureOpenAI
import json

client = AzureOpenAI(
    azure_endpoint="https://my-openai.openai.azure.com/",
    api_version="2024-08-01-preview",
)

# Step 1: Prepare the JSONL input file
requests = [
    {
        "custom_id": f"doc-{i}",
        "method": "POST",
        "url": "/chat/completions",
        "body": {
            "model": "gpt-4o",
            "messages": [
                {"role": "system", "content": "Summarize the document in 2 sentences."},
                {"role": "user", "content": doc_text},
            ],
            "max_tokens": 200,
        },
    }
    for i, doc_text in enumerate(documents)
]

with open("batch_input.jsonl", "w") as f:
    for req in requests:
        f.write(json.dumps(req) + "\n")

# Step 2: Upload the input file
batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")

# Step 3: Create the batch job
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/chat/completions",
    completion_window="24h",
    metadata={"project": "document-processing", "owner": "data-team"},
)
print(f"Batch job created: {batch_job.id}")

# Step 4: Poll for completion
import time

while True:
    status = client.batches.retrieve(batch_job.id)
    print(f"Status: {status.status} | Completed: {status.request_counts.completed}/{status.request_counts.total}")
    if status.status in ("completed", "failed", "expired"):
        break
    time.sleep(60)

# Step 5: Download results
if status.status == "completed":
    result_file = client.files.content(status.output_file_id)
    results = [json.loads(line) for line in result_file.text.strip().split("\n")]
    for r in results:
        print(f"{r['custom_id']}: {r['response']['body']['choices'][0]['message']['content']}")

Real-Time Streaming with Server-Sent Events (SSE)

Streaming delivers tokens as they’re generated, reducing perceived latency for chat applications.

Python Streaming

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain distributed systems in detail."},
    ],
    stream=True,
)

full_response = ""
for chunk in stream:
    if chunk.choices[0].delta.content:
        token = chunk.choices[0].delta.content
        print(token, end="", flush=True)
        full_response += token

FastAPI Streaming Endpoint

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(request: ChatRequest):
    async def generate():
        stream = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": request.message}],
            stream=True,
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

JavaScript Client for SSE

async function streamChat(message, onToken) {
  const response = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ message }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    const text = decoder.decode(value);
    const lines = text.split("\n").filter((l) => l.startsWith("data: "));

    for (const line of lines) {
      const data = line.slice(6);
      if (data === "[DONE]") return;
      const parsed = JSON.parse(data);
      onToken(parsed.token);
    }
  }
}

// Usage
streamChat("Explain AI", (token) => {
  document.getElementById("output").textContent += token;
});

Async Patterns for Production

import asyncio
from openai import AsyncAzureOpenAI

async_client = AsyncAzureOpenAI(
    azure_endpoint="https://my-openai.openai.azure.com/",
    api_version="2024-08-01-preview",
)

async def process_document(doc: str) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"Summarize: {doc}"}],
    )
    return response.choices[0].message.content

async def process_batch_async(documents: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_process(doc):
        async with semaphore:
            return await process_document(doc)

    results = await asyncio.gather(*[limited_process(d) for d in documents])
    return results

Tip: Use the Global Batch API for non-urgent workloads — it’s 50% cheaper than real-time. For interactive apps, use streaming to improve perceived performance. For moderate concurrency (10-100 docs), use async Python with semaphores to stay within rate limits.