Azure OpenAI content_filter — filtered by policy, no completion returned

from openai import AzureOpenAI, BadRequestError
from typing import Any

client = AzureOpenAI(
    api_key="...", azure_endpoint="...", api_version="2024-10-21",
)

def call_with_filter_handling(messages: list) -> dict:
    try:
        response = client.chat.completions.create(
            model="gpt4o-prod",
            messages=messages,
            max_tokens=500,
        )
    except BadRequestError as e:
        # Prompt-side filter — request never ran
        body = e.body or {}
        inner = body.get("innererror", {})
        cf = inner.get("content_filter_result", {})
        triggered = [k for k, v in cf.items() if v.get("filtered") or v.get("detected")]
        return {
            "status": "prompt_filtered",
            "triggered_categories": triggered,
            "user_message": (
                "Your request could not be processed because it contains content "
                f"related to: {', '.join(triggered)}. Please rephrase and try again."
            ),
        }

    choice = response.choices[0]
    if choice.finish_reason == "content_filter":
        # Response-side filter — model generated something we cannot show
        cf = getattr(choice, "content_filter_results", {}) or {}
        triggered = [k for k, v in cf.items() if v.get("filtered")]
        severity = {k: v.get("severity") for k, v in cf.items() if v.get("filtered")}
        return {
            "status": "response_filtered",
            "triggered_categories": triggered,
            "severity": severity,
            "user_message": "The generated response was blocked. Try rephrasing your question.",
        }

    return {"status": "ok", "content": choice.message.content}

# 1) Create the custom filter in Azure AI Foundry:
#    Portal -> Azure AI Foundry -> Your project -> Content Filters -> New
#
#    For a medical Q&A app, typical config:
#      hate:      block at medium
#      sexual:    block at high     (allow medium for reproductive-health topics)
#      violence:  block at high     (allow medium for injury descriptions)
#      self_harm: block at medium
#      jailbreak: block on detection
#      protected_material_text: annotate only (do not block)
#
# 2) Attach the filter to your deployment via CLI

az cognitiveservices account deployment update \
  --name my-aoai-resource \
  --resource-group my-rg \
  --deployment-name gpt4o-medical \
  --set properties.raiPolicyName="medical-qa-filter-v1"

# 3) Verify the attachment
az cognitiveservices account deployment show \
  --name my-aoai-resource -g my-rg \
  --deployment-name gpt4o-medical \
  --query "properties.raiPolicyName"

from openai import AzureOpenAI

client = AzureOpenAI(api_key="...", azure_endpoint="...", api_version="2024-10-21")

def stream_with_filter_awareness(messages: list):
    """Yields (event_type, payload) tuples: ('token', str) or ('filter', dict) or ('done', None)."""
    stream = client.chat.completions.create(
        model="gpt4o-prod",
        messages=messages,
        stream=True,
        max_tokens=500,
    )

    partial = []
    for chunk in stream:
        if not chunk.choices:
            continue
        choice = chunk.choices[0]

        # Filter fired mid-stream
        if choice.finish_reason == "content_filter":
            cf = getattr(choice, "content_filter_results", {}) or {}
            triggered = [k for k, v in cf.items() if v.get("filtered")]
            yield ("filter", {
                "triggered": triggered,
                "partial_length": len("".join(partial)),
                "message": "Response filtered mid-generation — content discarded.",
            })
            return

        if choice.delta and choice.delta.content:
            partial.append(choice.delta.content)
            yield ("token", choice.delta.content)

    yield ("done", None)

# UI-side usage
for event, payload in stream_with_filter_awareness([{"role": "user", "content": "..."}]):
    if event == "token":
        print(payload, end="", flush=True)
    elif event == "filter":
        print(f"\n\n[Filtered: {payload['triggered']}]")
    elif event == "done":
        print("\n[Complete]")