选项
首页首页 Skill 数据科学与机器学习 azure-ai-contentunderstanding-py

azure-ai-contentunderstanding-py

microsoft/skills microsoft/skills

使用 Azure AI 内容理解 Python SDK 从文档、图像、音频和视频中提取语义内容。

...展开全部
0
更新时间 2026-09-16

Azure AI 内容理解 SDK(Python 版)

一种多模态 AI 服务,可从文档、视频、音频和图像文件中提取语义内容,用于 RAG 和自动化工作流。

安装

pip install azure-ai-contentunderstanding

环境变量

CONTENTUNDERSTANDING_ENDPOINT=https://.cognitiveservices.azure.com/  # 所有身份验证方法均需此变量
AZURE_TOKEN_CREDENTIALS=prod # 仅当在生产环境中使用 DefaultAzureCredential 时才需要此变量

身份验证与生命周期

🔑 以下每个代码示例均遵循两条规则:

  1. 优先使用DefaultAzureCredential它既可在本地(Azure CLI / VS Code / Developer CLI)使用,也可在 Azure 中(托管身份、工作负载身份)使用,且无需修改代码。请避免使用连接字符串、账户/API 密钥——它们会绕过 Entra 的审计和轮换机制。
    • 本地开发:DefaultAzureCredential可直接使用。
    • 生产环境:将AZURE_TOKEN_CREDENTIALS设置为prod(或AZURE_TOKEN_CREDENTIALS= ),以将凭据链限制为符合生产环境安全标准的凭据。
  2. 将每个客户端封装在上下文管理器中,以确保 HTTP 传输、套接字和令牌缓存能以确定性方式释放:
    • 同步模式:使用 `(...) as client:`:
    • 异步:async with(...) as client: 以及 async with DefaultAzureCredential() as credential:(来自azure.identity.aio

代码片段可能会简化此配置,但生产环境中的代码应始终遵循这两条规则。

import os
from azure.ai.contentunderstanding import ContentUnderstandingClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
# 本地开发:使用 DefaultAzureCredential。生产环境:设置 AZURE_TOKEN_CREDENTIALS=prod 或 AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# 或者在生产环境中直接使用特定凭据:
# 参见 https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
with ContentUnderstandingClient(endpoint=endpoint, credential=credential) as client:
    analyzers = list(client.list_analyzers())

核心工作流

内容理解操作是异步的长期运行操作:

  1. 开始分析— 使用begin_analyze()启动分析操作(返回一个轮询器)
  2. 轮询结果— 持续轮询直至分析完成(SDK 通过.result() 处理此过程)
  3. 处理结果— 从 `AnalyzeResult.contents` 中提取结构化结果

预构建分析器

分析器 内容类型 用途
预构建文档搜索 文档 为 RAG 应用程序提取 Markdown 内容
预构建图像搜索 图片 从图片中提取内容
预构建音频搜索 音频 带时间戳的音频转录
预构建视频搜索 视频 提取帧、文字记录、摘要
预构建发票 文档 提取发票字段

分析文档

import os
from azure.ai.contentunderstanding import ContentUnderstandingClient
from azure.ai.contentunderstanding.models import AnalyzeInput
from azure.identity import DefaultAzureCredential

endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
with ContentUnderstandingClient(
    endpoint=endpoint,
    credential=DefaultAzureCredential()
) as client:
    # 根据 URL 分析文档
    poller = client.begin_analyze(
        analyzer_id="prebuilt-documentSearch",
        inputs=[AnalyzeInput(url="https://example.com/document.pdf")]
    )

    result = poller.result()

    # 获取 Markdown 内容(contents 是一个列表)
    content = result.contents[0]
    print(content.markdown)

访问文档内容详情

from azure.ai.contentunderstanding.models import MediaContentKind, DocumentContent

content = result.contents[0]
if content.kind == MediaContentKind.DOCUMENT:
    document_content: DocumentContent = content  # 类型:忽略
    print(document_content.start_page_number)

分析图像

from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-imageSearch",
    inputs=[AnalyzeInput(url="https://example.com/image.jpg")]
)
result = poller.result()
content = result.contents[0]
print(content.markdown)

分析视频

from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-videoSearch",
    inputs=[AnalyzeInput(url="https://example.com/video.mp4")]
)

result = poller.result()

# 访问视频内容 (AudioVisualContent)
content = result.contents[0]

# 获取带时间戳的转录短语
for phrase in content.transcript_phrases:
    print(f"[{phrase.start_time} - {phrase.end_time}]: {phrase.text}")

# 获取关键帧(针对视频)
for frame in content.key_frames:
    print(f"时间点 {frame.time} 的帧:{frame.description}")

分析音频

from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-audioSearch",
    inputs=[AnalyzeInput(url="https://example.com/audio.mp3")]
)

result = poller.result()

# 获取音频转录文本
content = result.contents[0]
for phrase in content.transcript_phrases:
    print(f"[{phrase.start_time}] {phrase.text}")

自定义分析器

使用字段模式创建自定义分析器以实现专项提取:

# 创建自定义分析器
analyzer = client.create_analyzer(
    analyzer_id="my-invoice-analyzer",
    analyzer={
        "description": "自定义发票分析器",
        "base_analyzer_id": "prebuilt-documentSearch",
        "field_schema": {
            "fields": {
                "vendor_name": {"type": "string"},
                "invoice_total": {"type": "number"},
                "line_items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "description": {"type": "string"},
                            "amount": {"type": "number"}
                        }
                    }
                }
            }
        }
    }
)

# 使用自定义分析器
from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="my-invoice-analyzer",
    inputs=[AnalyzeInput(url="https://example.com/invoice.pdf")]
)

result = poller.result()

# 访问提取的字段
print(result.fields["vendor_name"])
print(result.fields["invoice_total"])

分析器管理

# 列出所有分析器
analyzers = client.list_analyzers()
for analyzer in analyzers:
    print(f"{analyzer.analyzer_id}: {analyzer.description}")

# 获取特定分析器
analyzer = client.get_analyzer("prebuilt-documentSearch")

# 删除自定义分析器
client.delete_analyzer("my-custom-analyzer")

异步客户端

import asyncio
import os
from azure.ai.contentunderstanding.aio import ContentUnderstandingClient
from azure.ai.contentunderstanding.models import AnalyzeInput
from azure.identity.aio import DefaultAzureCredential

async def analyze_document():
    endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
    async with DefaultAzureCredential() as credential:
        async with ContentUnderstandingClient(
            endpoint=endpoint,
            credential=credential
        ) as client:
            poller = await client.begin_analyze(
                analyzer_id="prebuilt-documentSearch",
                inputs=[AnalyzeInput(url="https://example.com/doc.pdf")]
            )
            result = await poller.result()
            content = result.contents[0]
            return content.markdown

asyncio.run(analyze_document())

内容类型

用于 提供
DocumentContent PDF、图片、Office文档 页面、表格、图、段落
视听内容 音频、视频文件 字幕短语、时间码、关键帧

两者均继承自MediaContent,该类提供基本信息和 Markdown 表示形式。

模型导入

from azure.ai.contentunderstanding.models import (
    AnalyzeInput,
    AnalyzeResult,
    MediaContentKind,
    DocumentContent,
    AudioVisualContent,
)

客户端类型

客户端 用途
ContentUnderstandingClient 适用于所有操作的同步客户端
ContentUnderstandingClient(aio) 适用于所有操作的异步客户端

最佳实践

  1. 请选择同步或异步模式并保持一致。请勿在同一调用路径中混合使用azure.ai.contentunderstanding同步客户端与azure.ai.contentunderstanding.aio异步客户端。每个模块应选择一种模式。
  2. 始终为客户端和异步凭据使用上下文管理器。将每个客户端封装为 `ContentUnderstandingClient(...) as client:(sync)` (同步模式)`ContentUnderstandingClient(...) as client:(async)`(异步模式)。 对于来自azure.identity.aio 的异步DefaultAzureCredential,也请使用async 模式并指定 credential:,以便对令牌和传输进行清理。
  3. 请配合AnalyzeInput使用begin_analyze—— 这是正确的方法签名
  4. 通过result.contents[0]访问结果——结果以列表形式返回
  5. 针对常见场景(文档/图像/音频/视频搜索)使用预构建的分析器
  6. 仅在需要进行特定领域字段提取时才创建自定义分析器
  7. 在需要高吞吐量的场景中,使用 azure.identity.aio凭据配合异步客户端
  8. 处理长时间运行的操作— 视频/音频分析可能需要数分钟
  9. 在可能的情况下使用 URL 源,以避免上传开销
在 GitHub 上查看
---
name: azure-ai-contentunderstanding-py
description: Extract semantic content from documents, images, audio, and video using Azure AI Content Understanding SDK for Python.
license: MIT
---

# Azure AI Content Understanding SDK for Python

Multimodal AI service that extracts semantic content from documents, video, audio, and image files for RAG and automated workflows.

## Installation

```bash
pip install azure-ai-contentunderstanding
```

## Environment Variables

```bash
CONTENTUNDERSTANDING_ENDPOINT=https://<resource>.cognitiveservices.azure.com/  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```

## Authentication & Lifecycle

> **🔑 Two rules apply to every code sample below:**
>
> 1. **Prefer `DefaultAzureCredential`.** It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
>    - Local dev: `DefaultAzureCredential` works as-is.
>    - Production: set `AZURE_TOKEN_CREDENTIALS=prod` (or `AZURE_TOKEN_CREDENTIALS=<specific_credential>`) to constrain the credential chain to production-safe credentials.
> 2. **Wrap every client in a context manager** so HTTP transports, sockets, and token caches are released deterministically:
>    - Sync: `with <Client>(...) as client:`
>    - Async: `async with <Client>(...) as client:` **and** `async with DefaultAzureCredential() as credential:` (from `azure.identity.aio`)
>
> Snippets may abbreviate this setup, but production code should always follow both rules.

```python
import os
from azure.ai.contentunderstanding import ContentUnderstandingClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
with ContentUnderstandingClient(endpoint=endpoint, credential=credential) as client:
    analyzers = list(client.list_analyzers())
```

## Core Workflow

Content Understanding operations are asynchronous long-running operations:

1. **Begin Analysis** — Start the analysis operation with `begin_analyze()` (returns a poller)
2. **Poll for Results** — Poll until analysis completes (SDK handles this with `.result()`)
3. **Process Results** — Extract structured results from `AnalyzeResult.contents`

## Prebuilt Analyzers

| Analyzer | Content Type | Purpose |
|----------|--------------|---------|
| `prebuilt-documentSearch` | Documents | Extract markdown for RAG applications |
| `prebuilt-imageSearch` | Images | Extract content from images |
| `prebuilt-audioSearch` | Audio | Transcribe audio with timing |
| `prebuilt-videoSearch` | Video | Extract frames, transcripts, summaries |
| `prebuilt-invoice` | Documents | Extract invoice fields |

## Analyze Document

```python
import os
from azure.ai.contentunderstanding import ContentUnderstandingClient
from azure.ai.contentunderstanding.models import AnalyzeInput
from azure.identity import DefaultAzureCredential

endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
with ContentUnderstandingClient(
    endpoint=endpoint,
    credential=DefaultAzureCredential()
) as client:
    # Analyze document from URL
    poller = client.begin_analyze(
        analyzer_id="prebuilt-documentSearch",
        inputs=[AnalyzeInput(url="https://example.com/document.pdf")]
    )

    result = poller.result()

    # Access markdown content (contents is a list)
    content = result.contents[0]
    print(content.markdown)
```

## Access Document Content Details

```python
from azure.ai.contentunderstanding.models import MediaContentKind, DocumentContent

content = result.contents[0]
if content.kind == MediaContentKind.DOCUMENT:
    document_content: DocumentContent = content  # type: ignore
    print(document_content.start_page_number)
```

## Analyze Image

```python
from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-imageSearch",
    inputs=[AnalyzeInput(url="https://example.com/image.jpg")]
)
result = poller.result()
content = result.contents[0]
print(content.markdown)
```

## Analyze Video

```python
from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-videoSearch",
    inputs=[AnalyzeInput(url="https://example.com/video.mp4")]
)

result = poller.result()

# Access video content (AudioVisualContent)
content = result.contents[0]

# Get transcript phrases with timing
for phrase in content.transcript_phrases:
    print(f"[{phrase.start_time} - {phrase.end_time}]: {phrase.text}")

# Get key frames (for video)
for frame in content.key_frames:
    print(f"Frame at {frame.time}: {frame.description}")
```

## Analyze Audio

```python
from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="prebuilt-audioSearch",
    inputs=[AnalyzeInput(url="https://example.com/audio.mp3")]
)

result = poller.result()

# Access audio transcript
content = result.contents[0]
for phrase in content.transcript_phrases:
    print(f"[{phrase.start_time}] {phrase.text}")
```

## Custom Analyzers

Create custom analyzers with field schemas for specialized extraction:

```python
# Create custom analyzer
analyzer = client.create_analyzer(
    analyzer_id="my-invoice-analyzer",
    analyzer={
        "description": "Custom invoice analyzer",
        "base_analyzer_id": "prebuilt-documentSearch",
        "field_schema": {
            "fields": {
                "vendor_name": {"type": "string"},
                "invoice_total": {"type": "number"},
                "line_items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "description": {"type": "string"},
                            "amount": {"type": "number"}
                        }
                    }
                }
            }
        }
    }
)

# Use custom analyzer
from azure.ai.contentunderstanding.models import AnalyzeInput

poller = client.begin_analyze(
    analyzer_id="my-invoice-analyzer",
    inputs=[AnalyzeInput(url="https://example.com/invoice.pdf")]
)

result = poller.result()

# Access extracted fields
print(result.fields["vendor_name"])
print(result.fields["invoice_total"])
```

## Analyzer Management

```python
# List all analyzers
analyzers = client.list_analyzers()
for analyzer in analyzers:
    print(f"{analyzer.analyzer_id}: {analyzer.description}")

# Get specific analyzer
analyzer = client.get_analyzer("prebuilt-documentSearch")

# Delete custom analyzer
client.delete_analyzer("my-custom-analyzer")
```

## Async Client

```python
import asyncio
import os
from azure.ai.contentunderstanding.aio import ContentUnderstandingClient
from azure.ai.contentunderstanding.models import AnalyzeInput
from azure.identity.aio import DefaultAzureCredential

async def analyze_document():
    endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
    async with DefaultAzureCredential() as credential:
        async with ContentUnderstandingClient(
            endpoint=endpoint,
            credential=credential
        ) as client:
            poller = await client.begin_analyze(
                analyzer_id="prebuilt-documentSearch",
                inputs=[AnalyzeInput(url="https://example.com/doc.pdf")]
            )
            result = await poller.result()
            content = result.contents[0]
            return content.markdown

asyncio.run(analyze_document())
```

## Content Types

| Class | For | Provides |
|-------|-----|----------|
| `DocumentContent` | PDF, images, Office docs | Pages, tables, figures, paragraphs |
| `AudioVisualContent` | Audio, video files | Transcript phrases, timing, key frames |

Both derive from `MediaContent` which provides basic info and markdown representation.

## Model Imports

```python
from azure.ai.contentunderstanding.models import (
    AnalyzeInput,
    AnalyzeResult,
    MediaContentKind,
    DocumentContent,
    AudioVisualContent,
)
```

## Client Types

| Client | Purpose |
|--------|---------|
| `ContentUnderstandingClient` | Sync client for all operations |
| `ContentUnderstandingClient` (aio) | Async client for all operations |

## Best Practices

1. **Pick sync OR async and stay consistent.** Do not mix `azure.ai.contentunderstanding` sync clients with `azure.ai.contentunderstanding.aio` async clients in the same call path. Choose one mode per module.
2. **Always use context managers for clients and async credentials.** Wrap every client in `with ContentUnderstandingClient(...) as client:` (sync) or `async with ContentUnderstandingClient(...) as client:` (async). For async `DefaultAzureCredential` from `azure.identity.aio`, also use `async with credential:` so tokens and transports are cleaned up.
3. **Use `begin_analyze` with `AnalyzeInput`** — this is the correct method signature
4. **Access results via `result.contents[0]`** — results are returned as a list
5. **Use prebuilt analyzers** for common scenarios (document/image/audio/video search)
6. **Create custom analyzers** only for domain-specific field extraction
7. **Use async client** for high-throughput scenarios with `azure.identity.aio` credentials
8. **Handle long-running operations** — video/audio analysis can take minutes
9. **Use URL sources** when possible to avoid upload overhead

所有文件

1 个文件

安装 azure-ai-contentunderstanding-py

下载技能文件并将其解压到 .claude/skills/ 目录下。

下载ZIP

克隆仓库并复制技能文件到您的项目中。

git clone https://github.com/microsoft/skills/tree/main/.github/plugins/azure-sdk-python/skills/azure-ai-contentunderstanding-py # Copy SKILL.md to your .claude/skills/ directory

复制 复制
快速设置: 将技能文件夹复制到 .claude/skills/ Claude 会自动检测并使用该技能

相关技能

web-search
更新时间 2026-06-29
webapp-testing
更新时间 2026-06-29
lark-base
更新时间 2026-07-05
agentmail
更新时间 2026-06-29
OR