azure-transcription

Azure语音转写专业版

专业版增值服务

主要能力

1. 实时流式转写

import os
from azure.ai.transcription import TranscriptionClient
# ...
client = TranscriptionClient(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    credential=os.environ["TRANSCRIPTION_KEY"]
)
# ...
# 实时流式转写
stream = client.begin_stream_transcription(locale="zh-CN")
# ...
# 发送音频文件进行实时转写
stream.send_audio_file("realtime_audio.wav")
# ...
# 接收实时转写结果
for event in stream:
    if event.is_partial:
        print(f"[实时] {event.text}", end="\r")
    else:
        print(f"[完成] {event.text}")

2. 说话人分离（Diarization）

# 启用说话人分离的批量转写
job = client.begin_transcription(
    name="meeting-with-diarization",
    locale="zh-CN",
    content_urls=["https:///meeting.wav"],
    diarization_enabled=True,
    diarization_config={
        "max_speakers": 5,
        "enabled": True
    }
)
# ...
result = job.result()
# ...
# 输出带说话人标识的转写结果
for segment in result.segments:
    speaker = segment.speaker  # 说话人标识: Speaker1, Speaker2, ...
    print(f"[{speaker}] [{segment.start_time:.1f}s-{segment.end_time:.1f}s] {segment.text}")

• 异常时参考错误处理章节进行恢复
• 关键参数: 说话人分离（diarization） 选项

3. 批量转写队列管理

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from azure.ai.transcription import TranscriptionClient
# ...
class BatchTranscriptionManager:
    def __init__(self, endpoint, key, max_workers=5):
        self.client = TranscriptionClient(endpoint=endpoint, credential=key)
        self.max_workers = max_workers
        self.results = []
        self.failed = []
# ...
    def submit_transcription(self, audio_url, name, locale="zh-CN",
                              diarization=True, callback=None):
        """提交单个转写任务"""
        try:
            job = self.client.begin_transcription(
                name=name,
                locale=locale,
                content_urls=[audio_url],
                diarization_enabled=diarization
            )
            result = job.result()
# ...
            output = {
                'name': name,
                'status': result.status,
                'transcript': result.transcript,
                'segments': result.segments if hasattr(result, 'segments') else []
            }
            self.results.append(output)
# ...
            if callback:
                callback(output)
            return output
        except Exception as e:
            self.failed.append({'name': name, 'error': str(e)})
            return None
# ...
    def batch_transcribe(self, audio_files, locale="zh-CN", diarization=True):
        """批量转写"""
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            for audio in audio_files:
                future = executor.submit(
                    self.submit_transcription,
                    audio['url'], audio['name'], locale, diarization
                )
                futures.append(future)
# ...
            for future in as_completed(futures):
                future.result()
# ...
        return {'success': self.results, 'failed': self.failed}
# ...
    def export_results(self, format='srt', output_dir='./transcripts'):
        """导出转写结果"""
        os.makedirs(output_dir, exist_ok=True)
        for result in self.results:
            if format == 'srt':
_export_srt(result, output_dir)
            elif format == 'vtt':
_export_vtt(result, output_dir)
            elif format == 'json':
_export_json(result, output_dir)
# ...
    def _export_srt(self, result, output_dir):
        path = os.path.join(output_dir, f"{result['name']}.srt")
        with open(path, 'w', encoding='utf-8') as f:
            for i, seg in enumerate(result['segments'], 1):
                start = self._format_time(seg.start_time)
                end = self._format_time(seg.end_time)
                speaker = f"[{seg.speaker}] " if hasattr(seg, 'speaker') else ""
                f.write(f"{i}\n{start} --> {end}\n{speaker}{seg.text}\n\n")
# ...
    def _format_time(self, seconds):
        h = int(seconds // 3600)
        m = int((seconds % 3600) // 60)
        s = int(seconds % 60)
        ms = int((seconds % 1) * 1000)
        return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
```- 验证执行结果,确认输出符合预期格式
- 异常时参考错误处理章节进行恢复
- 关键参数: `批量转写队列管理` 选项
## 适用范围
### 场景一：企业会议实时字幕
企业会议系统实时生成字幕与会议纪要.
```python
# 会议实时字幕系统
meeting_manager = BatchTranscriptionManager(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    key=os.environ["TRANSCRIPTION_KEY"]
)
# ...
# 实时流式转写会议音频
stream = client.begin_stream_transcription(locale="zh-CN")
# ...
print("=== 会议实时字幕 ===")
for event in stream:
    if not event.is_partial:
        timestamp = event.timestamp
        print(f"[{timestamp}] {event.text}")
# ...
# 会后批量处理录音（带说话人分离）
meeting_manager.submit_transcription(
    audio_url="https:///meetings/full_meeting.wav",
    name="quarterly-review-20260118",
    locale="zh-CN",
    diarization=True
)

场景二：客服通话批量转写

客服中心批量转写通话录音，用于质检与分析.

# 批量转写客服通话
manager = BatchTranscriptionManager(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    key=os.environ["TRANSCRIPTION_KEY"],
    max_workers=5
)
# ...
# 定义批量任务
call_recordings = [
    {"url": "https:///calls/call_001.wav", "name": "call_001"},
    {"url": "https:///calls/call_002.wav", "name": "call_002"},
    {"url": "https:///calls/call_003.wav", "name": "call_003"},
    {"url": "https:///calls/call_004.wav", "name": "call_004"},
    {"url": "https:///calls/call_005.wav", "name": "call_005"},
]
# ...
# 批量转写（启用说话人分离）
results = manager.batch_transcribe(call_recordings, locale="zh-CN", diarization=True)
# ...
# 导出为SRT格式
manager.export_results(format='srt', output_dir='./call_transcripts')
# ...
print(f"成功: {len(results['success'])}, 失败: {len(results['failed'])}")

场景三：多语言视频字幕批量生成

视频平台批量生成多语言字幕.

# 多语言字幕生成
video_manager = BatchTranscriptionManager(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    key=os.environ["TRANSCRIPTION_KEY"],
    max_workers=3
)
# ...
# 中文视频生成中文字幕
chinese_videos = [
    {"url": "https:///videos/video_cn_01.wav", "name": "video_cn_01"},
    {"url": "https:///videos/video_cn_02.wav", "name": "video_cn_02"},
]
video_manager.batch_transcribe(chinese_videos, locale="zh-CN", diarization=False)
video_manager.export_results(format='vtt', output_dir='./subtitles/zh-CN')
# ...
# 英文视频生成英文字幕
english_videos = [
    {"url": "https:///videos/video_en_01.wav", "name": "video_en_01"},
    {"url": "https:///videos/video_en_02.wav", "name": "video_en_02"},
]
video_manager.batch_transcribe(english_videos, locale="en-US", diarization=False)
video_manager./subtitles/en-US')

使用方法

1. 环境准备

# 依赖说明
pip install azure-ai-transcription
# ...
# 配置环境变量
export TRANSCRIPTION_ENDPOINT="https://.cognitiveservices.azure.com"
export TRANSCRIPTION_KEY="API_KEY"

2. 实时转写

import os
from azure.ai.transcription import TranscriptionClient
# ...
# 启动实时转写
stream = client.begin_stream_transcription(locale="zh-CN")
stream.send_audio_file("audio.wav")
# ...
for event in stream:
    print(event.text)

3. 批量转写

manager = BatchTranscriptionManager(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    key=os.environ["TRANSCRIPTION_KEY"]
)
# ...
results = manager.batch_transcribe(
    audio_files=[{"url": "https:///audio.wav", "name": "test"}],
    locale="zh-CN",
    diarization=True
)
# ...
manager.export_results(format='srt')

请求格式

结果格式

{
  "success": true,
  "data": {
    "result": "处理结果",
    "status": "success",
    "metadata": {
    "metadata": {
      "template_used": "reviewer",
      "word_count": 0,
      "style": "专业"
    }
  },
  "error": null
}

输出模板参考: assets/output.json

异常管理

安装与配置

运行环境

• Agent平台: 支持SKILL.md的任意AI Agent（Claude Code / Cursor / Codex / Gemini CLI等）
• 操作系统: Windows / macOS / Linux
• Python版本: 3.8及以上

第三方依赖

API Key 配置

• TRANSCRIPTION_ENDPOINT：Azure认知服务端点URL
• TRANSCRIPTION_KEY：Azure认知服务订阅密钥
• 与免费版使用相同的认证配置，完全兼容

可用性分类

• 分类: MD+EXEC（纯Markdown指令，核心功能需要exec命令行执行能力）
• 说明: 基于Markdown的AI Skill，通过自然语言指令驱动Agent执行专业语音转写任务。支持实时流式转写、说话人分离、批量队列等企业级功能，通过Python SDK调用Azure AI服务。与免费版完全兼容，可直接复用免费版的认证配置与基础转写流程. API Key配置方式:

export API_KEY="${API_KEY:?请设置环境变量}"

配置后需重启会话或开启新终端生效。API Key应妥善保管,避免泄露到版本控制系统.

案例展示

转写参数配置

输出格式对比

问答总汇

Q1：实时转写有延迟怎么办？

实时转写延迟受网络条件影响。建议使用稳定的网络连接，并适当增大音频缓冲区.

Q2：说话人分离准确率如何提升？

确保音频质量良好，说话人间隔清晰。设置合理的max_speakers参数，避免过多或过少.

Q3：批量转写如何处理失败任务？

专业版BatchTranscriptionManager自动记录失败任务，可通过retry_failed方法重试.

Q4：自定义语音模型如何接入？

在Azure门户训练自定义模型后，在转写配置中指定custom_model参数即可.

Q5：与免费版的API配置是否兼容？

完全兼容。专业版与免费版使用相同的TRANSCRIPTION_ENDPOINT和TRANSCRIPTION_KEY配置.

错误恢复方案

注意事项

• 需要API Key，无Key环境无法使用
• 依赖云服务，需要网络连接

问题处理指引

安全保证

创新特色

功能介绍

• 自动化执行: 企业级语音转写工具，支持实时流式转写、说话人分离、批量处理与自定义模型。Azure语音转写专业版 —— 面向企业团队与专
• 文件处理: 支持多种文件格式的读取、解析和写入操作
• API集成: 通过标准化接口调用外部服务并处理响应
• 命令执行: 在安全沙箱中执行系统命令并收集结果

帮助文档

Q1: Azure语音转写专业版支持哪些输入格式？

A1: 企业级语音转写工具，支持实时流式转写、说话人分离、批量处理与自定义模型。Azure语音转写专业版 —— 面向企业团队与专业用户的高级语音转写工具。核心能力: -。支持文本指令和结构化参数输入，具体格式参考使用流程章节。

Q2: 需要配置API Key吗？

A2: 是的，部分功能需要配置对应平台的API Key。请在依赖说明章节查看具体要求，并通过环境变量安全配置。

Q3: 命令行执行失败怎么办？

A3: 检查命令参数是否正确，确认运行环境支持exec能力。如遇权限问题，请参照错误处理章节排查。

性能数据

特色分析

Azure语音转写专业版通用排查步骤

1. 检查输入参数: 确认所有必填参数已提供且格式正确
2. 查看日志输出: 定位具体错误行和异常类型
3. 验证环境配置: 确认依赖库版本和运行环境满足要求
4. 逐步调试: 缩小问题范围,隔离故障模块