Azure AI 内容理解 Python SDK

azure-ai-contentunderstanding-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.60/5
使用5.6K

Azure AI Content Understanding Python SDK

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

安装

bash
pip install azure-ai-contentunderstanding

环境变量

bash
CONTENTUNDERSTANDING_ENDPOINT=https://<resource>.cognitiveservices.azure.com/

身份验证

python
import os
from azure.ai.contentunderstanding import ContentUnderstandingClient
from azure.identity import DefaultAzureCredential

endpoint = os.environ["CONTENTUNDERSTANDING_ENDPOINT"]
credential = DefaultAzureCredential()
client = ContentUnderstandingClient(endpoint=endpoint, credential=credential)

核心工作流

Content Understanding 的操作是异步长时运行操作:

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

预置分析器

| 分析器 | 内容类型 | 用途 |
|----------|--------------|---------|
| prebuilt-documentSearch | 文档 | 为 RAG 应用提取 markdown |
| prebuilt-imageSearch | 图像 | 从图像中提取内容 |
| prebuilt-audioSearch | 音频 | 带有时间戳的音频转录 |
| prebuilt-videoSearch | 视频 | 提取帧、转录文本和摘要 |
| prebuilt-invoice | 文档 | 提取发票字段 |

分析文档

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"]
client = ContentUnderstandingClient(
endpoint=endpoint,
credential=DefaultAzureCredential()
)

通过 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)

访问文档内容详情

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)

分析图像

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)

分析视频

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()

访问视频内容 (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 at {frame.time}: {frame.description}")

分析音频

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()

访问音频转录文本

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

自定义分析器

通过字段架构(field schemas)创建自定义分析器以进行专门的提取:

python
# 创建自定义分析器
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"}
                        }
                    }
                }
            }
        }
    }
)

使用自定义分析器

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"])

分析器管理

python
# 列出所有分析器
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")

异步客户端

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"]
credential = DefaultAzureCredential()

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 文档 | 页面、表格、插图、段落 |
| AudioVisualContent | 音频、视频文件 | 转录短语、时间戳、关键帧 |

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

模型导入

python
from azure.ai.contentunderstanding.models import (
    AnalyzeInput,
    AnalyzeResult,
    MediaContentKind,
    DocumentContent,
    AudioVisualContent,
)
客户端类型

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

最佳实践

1. 使用 begin_analyze 配合 AnalyzeInput —— 这是正确的方法签名
2. 通过 result.contents[0] 访问结果 —— 结果以列表形式返回
3. 针对常见场景使用预置分析器(文档/图像/音频/视频搜索)
4. 仅在需要领域特定字段提取时创建自定义分析器
5. 在高吞吐量场景下使用异步客户端,并配合 azure.identity.aio 凭据
6. 处理长时间运行的操作 —— 视频/音频分析可能需要数分钟
7. 尽可能使用 URL 来源以避免上传开销

适用场景

本技能适用于执行概览中所描述的工作流或操作。

局限性

  • 仅在任务明确符合上述范围时使用此技能。
  • 不要将输出结果视为环境特定验证、测试或专家评审的替代方案。
  • 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并寻求澄清。