AI-103 CERTIFICATION BOOTCAMP

# Use the virtual environment from Session 5
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install openai azure-identity requests
az login

# From Session 5: your resource's /openai/v1/ route (my variable name)
export FOUNDRY_OPENAI_BASE_URL="https://<resource-name>.openai.azure.com/openai/v1/"
# My own variable name, not Microsoft's: the DEPLOYMENT name of your image model
export IMAGE_MODEL_DEPLOYMENT="<your-image-deployment-name>"

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

# Keyless: a Microsoft Entra token provider instead of an API key
token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)
client = OpenAI(base_url=os.environ["FOUNDRY_OPENAI_BASE_URL"], api_key=token_provider)

result = client.images.generate(
    model=os.environ["IMAGE_MODEL_DEPLOYMENT"],  # deployment name, not catalog name
    prompt="Studio photo of a blue enamel water bottle on a plain white table, "
           "soft daylight from the left, minimal style",
    n=1,
    size="1024x1024",
    quality="medium",
    output_format="png",
)

# GPT-image models always return base64 image data
image_bytes = base64.b64decode(result.data[0].b64_json)
with open("generated_1.png", "wb") as image_file:
    image_file.write(image_bytes)
print("Saved generated_1.png")

Saved generated_1.png

source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install --upgrade openai azure-identity
az login

export FOUNDRY_OPENAI_BASE_URL="https://<resource-name>.openai.azure.com/openai/v1/"
# My own variable name, not Microsoft's: the DEPLOYMENT name of your sora-2 model
export VIDEO_MODEL_DEPLOYMENT="<your-video-deployment-name>"

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

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)
client = OpenAI(base_url=os.environ["FOUNDRY_OPENAI_BASE_URL"], api_key=token_provider)

# The ID printed by make_video.py. The original job must be COMPLETED.
video = client.videos.remix(
    video_id="<original-video-id>",
    prompt="Change the light to warm late-afternoon sun. Keep everything else the same.",
)
print(f"Remix started. ID: {video.id}")

while video.status not in ["completed", "failed", "cancelled"]:
    time.sleep(20)
    video = client.videos.retrieve(video.id)
    print(f"Status: {video.status}")

if video.status == "completed":
    content = client.videos.download_content(video.id, variant="video")
    content.write_to_file("remix.mp4")
    print("Saved remix.mp4")
else:
    print(f"Ended with status {video.status}: {video.error}")

Remix started. ID: video_<new-id>
Status: in_progress
...
Status: completed
Saved remix.mp4

source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install azure-ai-contentsafety azure-ai-documentintelligence azure-identity requests
az login

# Both point at your Foundry resource's Foundry Tools endpoint.
# The variable names are the ones Microsoft's samples use.
export CONTENT_SAFETY_ENDPOINT="https://<resource-name>.cognitiveservices.azure.com/"
export DOCUMENTINTELLIGENCE_ENDPOINT="https://<resource-name>.cognitiveservices.azure.com/"

import os
import sys
from azure.ai.contentsafety import ContentSafetyClient
from azure.ai.contentsafety.models import AnalyzeImageOptions, ImageData
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential

client = ContentSafetyClient(os.environ["CONTENT_SAFETY_ENDPOINT"], DefaultAzureCredential())
BLOCK_AT = 4  # my policy for this app: block Medium (4) and High (6)

with open(sys.argv[1], "rb") as file:
    request = AnalyzeImageOptions(image=ImageData(content=file.read()))

try:
    response = client.analyze_image(request)
except HttpResponseError as e:
    print("Analyze image failed.")
    if e.error:
        print(f"Error code: {e.error.code}")
        print(f"Error message: {e.error.message}")
    raise

worst = 0
for item in response.categories_analysis:
    print(f"{item.category} severity: {item.severity}")
    worst = max(worst, item.severity or 0)

print("DECISION:", "BLOCK" if worst >= BLOCK_AT else "ALLOW")

<category> severity: 0      (one line each for Hate, SelfHarm, Sexual, Violence)
...
DECISION: ALLOW

{
  "text": {
    "format": {
      "type": "json_schema",
      "name": "CaseExtraction",
      "strict": true,
      "schema": {
        "type": "object",
        "properties": {
          "people": {"type": "array", "items": {"type": "string"}},
          "organisations": {"type": "array", "items": {"type": "string"}},
          "topics": {"type": "array", "items": {"type": "string"}},
          "request_type": {"type": "string", "enum": ["refund", "delivery", "complaint", "other"]},
          "summary": {"type": "string"},
          "due_date": {"type": ["string", "null"]}
        },
        "required": ["people", "organisations", "topics", "request_type", "summary", "due_date"],
        "additionalProperties": false
      }
    }
  }
}

import os
from typing import Literal, Optional
from openai import OpenAI
from pydantic import BaseModel
from azure.identity import DefaultAzureCredential, get_bearer_token_provider

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)
openai = OpenAI(base_url=os.environ["FOUNDRY_OPENAI_BASE_URL"], api_key=token_provider)

# The schema, as a Pydantic model. The SDK turns it into a strict json_schema.
class CaseExtraction(BaseModel):
    people: list[str]
    organisations: list[str]
    topics: list[str]
    request_type: Literal["refund", "delivery", "complaint", "other"]
    summary: str
    due_date: Optional[str]  # may be null: Learn's way to emulate an optional field

text = open("sample.txt", encoding="utf-8").read()

response = openai.responses.parse(
    model=os.environ["FOUNDRY_MODEL_NAME"],  # a deployment that supports structured outputs
    input="Extract only what the text states. The summary is one sentence under 30 words. "
          f"Text: {text}",
    text_format=CaseExtraction,
)
print(response.output_parsed.model_dump_json(indent=2))

{
  "people": ["<names the text states>"],
  "organisations": ["<organisations the text states>"],
  "topics": ["<topic>", "<topic>"],
  "request_type": "delivery",
  "summary": "<one sentence>",
  "due_date": null
}

import os
from azure.ai.contentsafety import ContentSafetyClient
from azure.ai.contentsafety.models import AnalyzeTextOptions
from azure.ai.textanalytics import TextAnalyticsClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
language = TextAnalyticsClient(os.environ["AZURE_LANGUAGE_ENDPOINT"], credential=credential)
safety = ContentSafetyClient(os.environ["CONTENT_SAFETY_ENDPOINT"], credential)
text = open("complaint.txt", encoding="utf-8").read()

# 1. Sentiment, with opinion mining (targets and their assessments)
doc = language.analyze_sentiment([text], show_opinion_mining=True)[0]
print("Document sentiment:", doc.sentiment)
for sentence in doc.sentences:
    for opinion in sentence.mined_opinions:
        words = ", ".join(a.text for a in opinion.assessments)
        print(f"  {opinion.target.sentiment} target '{opinion.target.text}': {words}")

# 2. PII detection. Print categories only, never the values.
pii = language.recognize_pii_entities([text], language="en")[0]
for entity in pii.entities:
    print(f"  PII {entity.category} ({entity.confidence_score:.2f})")
with open("redacted.txt", "w", encoding="utf-8") as f:
    f.write(pii.redacted_text)

# 3. Harmful content in the text
result = safety.analyze_text(AnalyzeTextOptions(text=text))
for item in result.categories_analysis:
    print(f"  Safety {item.category}: {item.severity}")

Document sentiment: <positive, negative, neutral or mixed>
  <sentiment> target '<target>': <assessment>
  <sentiment> target '<target>': <assessment>
  PII <category> (<score>)
  PII <category> (<score>)
  Safety <category>: 0      (one line for each of the four categories)