Azure Content Understanding

Content Understanding terms — defined before code

The problem of multi-modal content at scale

The four modalities — what Content Understanding processes

Choosing the right service for the job — the exam decision table

These three services (all in Part 5) are closely related and the exam will give you scenarios requiring you to choose between them. Here is the complete decision guide:

The analyzer pipeline — from content to structured output

What the output looks like — a document result example

{
  "status": "succeeded",
  "result": {
    "contents": [
      {
        "kind": "document",
        "markdown": "# Contract Agreement\n\nThis contract...",  // Full OCR text as markdown
        "fields": {
          "ContractValue": { "valueNumber": 150000.00, "confidence": 0.95 },
          "ExpiryDate":    { "valueDate": "2026-12-31",  "confidence": 0.98 },
          "PartyNames":    { "valueArray": ["Contoso Ltd", "Fabrikam Inc"], "confidence": 0.97 }
        },
        "summary": "This contract between Contoso Ltd and Fabrikam Inc...",
        "classificationLabel": "LegalContract",
        "tables": [                       // Extracted tables
          { "rowCount": 5, "columnCount": 3, "cells": [...] }
        ]
      }
    ]
  }
}

Creating an analyzer and submitting documents for analysis

Install: pip install requests azure-identity azure-ai-projects

import os
import time       # The 'time' module lets us pause execution (time.sleep) while polling
import requests   # The 'requests' library lets us make HTTP REST API calls
import json       # The 'json' module converts Python dicts to JSON strings and back

# ── Connection details ─────────────────────────────────────────────────
# These come from your Azure AI Foundry resource in Azure portal
# Navigate to: Azure AI Foundry → your project → Settings → API Keys
ENDPOINT = os.environ.get("CONTENTUNDERSTANDING_ENDPOINT")
# The endpoint format is: https://<your-resource>.cognitiveservices.azure.com/

API_KEY = os.environ.get("CONTENTUNDERSTANDING_KEY")
# Store API keys in environment variables — never hardcode them in your script

# API version — Content Understanding requires a specific API version in the URL
# Always check Microsoft docs for the latest stable version
API_VERSION = "2024-12-01-preview"

# HTTP headers — sent with every request to authenticate and specify content type
headers = {
    "Ocp-Apim-Subscription-Key": API_KEY,  # The standard Azure key header
    "Content-Type": "application/json"       # We are sending JSON in the request body
}


# ══════════════════════════════════════════════════════════════════════
# STEP 1 — Create a custom Analyzer
# An analyzer is a reusable configuration that defines:
#   - Which content modality (document, image, video, audio)
#   - Which AI operations to perform
#   - What fields to extract (the field schema)
# ══════════════════════════════════════════════════════════════════════

# A unique name for your analyzer (letters, numbers, hyphens only)
ANALYZER_NAME = "contract-analyzer-v1"

# The analyzer definition — a dictionary that becomes the JSON request body
analyzer_definition = {
    # description: a human-readable name for this analyzer (optional)
    "description": "Extracts key fields from legal contract documents",

    # scenario: tells the service what kind of content this analyzer handles
    # Options: "documentAnalysis" (docs/images), "videoAnalysis", "audioAnalysis"
    "scenario": "documentAnalysis",

    # fieldSchema: defines the specific fields you want extracted
    # The AI will look for these fields in every document submitted to this analyzer
    "fieldSchema": {
        "fields": {
            # Each key is the field name you want in the output
            # "type" tells the service what data type this field contains
            "ContractValue": {
                "type": "number",       # A numeric value (e.g. 150000.00)
                "description": "The total monetary value of the contract"
            },
            "ExpiryDate": {
                "type": "date",         # A date value (e.g. "2026-12-31")
                "description": "The date when the contract expires"
            },
            "PartyNames": {
                "type": "array",        # A list of values
                "items": {"type": "string"}, # Each item in the list is a string
                "description": "Names of all parties involved in the contract"
            },
            "GoverningLaw": {
                "type": "string",       # A text/string value
                "description": "The jurisdiction or country whose law governs the contract"
            }
        }
    }
}

# Build the URL for creating the analyzer
# Format: {endpoint}/contentunderstanding/analyzers/{name}?api-version={version}
create_url = (
    f"{ENDPOINT}contentunderstanding/analyzers/{ANALYZER_NAME}"
    f"?api-version={API_VERSION}"
)

# PUT request creates or updates the analyzer (idempotent — safe to run again)
# json.dumps() converts our Python dictionary into a JSON string for the HTTP body
response = requests.put(create_url, headers=headers, data=json.dumps(analyzer_definition))

# Check the response status code:
# 200 = updated existing, 201 = created new — both mean success
if response.status_code in (200, 201):
    print(f"✅ Analyzer '{ANALYZER_NAME}' created/updated successfully.")
else:
    # response.text contains the error message from the API
    print(f"❌ Failed to create analyzer: {response.status_code} — {response.text}")
    raise SystemExit(1)  # Stop the script — no point continuing without the analyzer


# ══════════════════════════════════════════════════════════════════════
# STEP 2 — Submit a document for analysis
# Same async pattern as Document Intelligence:  POST → get operation ID → poll
# ══════════════════════════════════════════════════════════════════════

# URL of the document to analyse — must be publicly accessible or in Azure Blob Storage
document_url = "https://your-storage.blob.core.windows.net/contracts/contract-001.pdf"

# Build the URL for submitting an analysis job to your analyzer
analyze_url = (
    f"{ENDPOINT}contentunderstanding/analyzers/{ANALYZER_NAME}:analyze"
    f"?api-version={API_VERSION}"
)

# The request body specifies what to analyse
# url: a publicly accessible URL to the document
analyze_body = {
    "url": document_url
}

# POST request submits the analysis job
# The service does NOT process it immediately — it queues it and returns an operation ID
response = requests.post(analyze_url, headers=headers, data=json.dumps(analyze_body))

# 202 Accepted means the job was queued successfully (not completed yet)
if response.status_code != 202:
    print(f"❌ Failed to submit document: {response.status_code} — {response.text}")
    raise SystemExit(1)

# The operation URL is in the "Operation-Location" response header
# This is the URL we will poll to check the status and get results
operation_url = response.headers["Operation-Location"]
print(f"✅ Analysis job submitted. Operation URL: {operation_url}")


# ══════════════════════════════════════════════════════════════════════
# STEP 3 — Poll until the operation completes
# Keep checking the operation URL every few seconds until status = "succeeded"
# ══════════════════════════════════════════════════════════════════════

print("⏳ Waiting for analysis to complete...")
result = None

# A while loop runs repeatedly until we tell it to stop
# We loop until the operation status is "succeeded" or "failed"
while True:
    # GET request to the operation URL — returns current status and (when done) results
    poll_response = requests.get(operation_url, headers=headers)
    poll_data = poll_response.json()  # Parse the JSON response into a Python dictionary

    status = poll_data.get("status")  # "running", "notStarted", "succeeded", or "failed"
    print(f"  Status: {status}")

    if status == "succeeded":
        # Analysis is complete — save the full result dictionary
        result = poll_data
        break  # Exit the while loop — we have our results
    elif status == "failed":
        # Something went wrong — print the error and stop
        print(f"❌ Analysis failed: {poll_data.get('error', {}).get('message', 'Unknown error')}")
        raise SystemExit(1)
    else:
        # Still processing — wait 3 seconds before checking again
        # time.sleep(3) pauses the script for 3 seconds without using CPU
        time.sleep(3)


# ══════════════════════════════════════════════════════════════════════
# STEP 4 — Read the results
# ══════════════════════════════════════════════════════════════════════

# result["result"]["contents"] is a list — one item per analysed content unit
# For a single document, it will have one item
contents = result.get("result", {}).get("contents", [])

for content in contents:
    # content["kind"] tells you the modality: "document", "image", "video", "audio"
    print(f"\n=== Content Kind: {content.get('kind')} ===")

    # ── Extracted Fields (from your field schema) ──
    fields = content.get("fields", {})
    if fields:
        print("\n📋 Extracted Fields:")
        for field_name, field_data in fields.items():
            # Each field has a value and a confidence score
            # The value key varies by type: valueNumber, valueDate, valueString, valueArray
            confidence = field_data.get("confidence", 0)
            # Look for any key starting with "value" to get the actual extracted value
            value = (field_data.get("valueNumber")
                  or field_data.get("valueDate")
                  or field_data.get("valueString")
                  or field_data.get("valueArray"))
            print(f"  {field_name}: {value}  (confidence: {confidence:.2f})")

    # ── Summary (if summarisation was enabled in the analyzer) ──
    summary = content.get("summary")
    if summary:
        print(f"\n📝 Summary:\n  {summary[:300]}...")   # Print first 300 characters

    # ── Classification Label (if classification was enabled) ──
    label = content.get("classificationLabel")
    if label:
        print(f"\n🏷️  Classification: {label}")

    # ── Tables (extracted table data) ──
    tables = content.get("tables", [])
    if tables:
        print(f"\n📊 Tables found: {len(tables)}")
        for i, table in enumerate(tables):
            print(f"  Table {i+1}: {table.get('rowCount')} rows × {table.get('columnCount')} cols")

    # ── Full OCR Text (as markdown) ──
    markdown = content.get("markdown")
    if markdown:
        # The full OCR text — headings, paragraphs, tables as markdown formatted text
        print(f"\n📄 Full OCR text ({len(markdown)} chars) — first 200:\n  {markdown[:200]}...")

Python — Submitting an Audio file for transcription and analysis

# ── Audio analyzer — transcription + entity extraction ──────────────────
# First: create an audio analyzer (same pattern as document analyzer above)
audio_analyzer_definition = {
    "description": "Transcribes call recordings and extracts customer entities",
    "scenario": "audioAnalysis",   # Must be "audioAnalysis" for audio files
    "fieldSchema": {
        "fields": {
            "CustomerName": {
                "type": "string",
                "description": "The name of the customer speaking in the recording"
            },
            "IssueType": {
                "type": "string",
                "description": "The type of issue the customer is reporting"
            },
            "Sentiment": {
                "type": "string",
                "description": "Overall customer sentiment: Positive, Neutral, or Negative"
            }
        }
    }
}

AUDIO_ANALYZER_NAME = "call-center-audio-analyzer"

# Create the audio analyzer (PUT request — same pattern as document analyzer)
audio_create_url = (
    f"{ENDPOINT}contentunderstanding/analyzers/{AUDIO_ANALYZER_NAME}"
    f"?api-version={API_VERSION}"
)
requests.put(audio_create_url, headers=headers, data=json.dumps(audio_analyzer_definition))

# Submit an audio file for analysis
audio_analyze_url = (
    f"{ENDPOINT}contentunderstanding/analyzers/{AUDIO_ANALYZER_NAME}:analyze"
    f"?api-version={API_VERSION}"
)

# The audio file URL — must be a WAV, MP3, or M4A file in Azure Blob Storage
audio_body = {
    "url": "https://your-storage.blob.core.windows.net/calls/call-20241201-001.wav"
}

response = requests.post(audio_analyze_url, headers=headers, data=json.dumps(audio_body))
# operation_url is in response.headers["Operation-Location"]
# Then poll the operation URL until status == "succeeded" (same loop as above)
# The result will contain:
#   content["transcript"]        → Full speech-to-text transcript with timestamps
#   content["fields"]["CustomerName"]  → Extracted customer name from the transcript
#   content["fields"]["IssueType"]     → Extracted issue type
#   content["fields"]["Sentiment"]     → Detected sentiment
print(f"Audio job submitted: {response.headers.get('Operation-Location', 'N/A')}")

Content Understanding — what the exam tests

Test your understanding