Azure AI 内容理解 Python SDK
Azure AI Content Understanding Python SDK
多模态 AI 服务,可从文档、视频、音频和图像文件中提取语义内容,用于 RAG 和自动化工作流。
安装
pip install azure-ai-contentunderstanding环境变量
CONTENTUNDERSTANDING_ENDPOINT=https://<resource>.cognitiveservices.azure.com/身份验证
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 | 文档 | 提取发票字段 |
分析文档
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)访问文档内容详情
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)
分析图像
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 at {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}")自定义分析器
通过字段架构(field schemas)创建自定义分析器以进行专门的提取:
# 创建自定义分析器
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"])分析器管理
# 列出所有分析器
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"]
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 表示形式。
模型导入
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 来源以避免上传开销
适用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务明确符合上述范围时使用此技能。
- 不要将输出结果视为环境特定验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并寻求澄清。