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）所需的指令和必要参数。 处理: 按照skill规范执行说话人分离（Diarization）操作,遵循单一意图原则。

• 执行说话人分离（Diarization）操作,处理输入数据并返回结果
• 验证执行结果,确认输出符合预期格式
• 异常时参考错误处理章节进行恢复
• 关键参数: 说话人分离（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':
                self._export_srt(result, output_dir)
            elif format == 'vtt':
                self._export_vtt(result, output_dir)
            elif format == 'json':
                self._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}nn")

    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}"

• 执行批量转写队列管理操作,处理输入数据并返回结果
• 验证执行结果,确认输出符合预期格式
• 异常时参考错误处理章节进行恢复
• 关键参数: 批量转写队列管理 选项

能力覆盖范围

本skill还覆盖以下能力场景: 企业级语音转写工、支持实时流式转写、批量处理与自定义、语音转写专业版、面向企业团队与专、业用户的高级语音、转写工具、核心能力、实时流式语音转写、支持麦克风输入与、流式音频、自动识别不同说话、支持大规模音频文、件处理、自定义语音模型集、提升专业领域识别、准确率、多语言混合转写。这些能力在上述核心功能中均有对应处理逻辑。

适用场景

场景一：企业会议实时字幕

企业会议系统实时生成字幕与会议纪要。

# 会议实时字幕系统
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.export_results(format='vtt', output_dir='./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

client = TranscriptionClient(
    endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
    credential=os.environ["TRANSCRIPTION_KEY"]
)

# 启动实时转写
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')

命令参数说明

1. -with-diarization: 命令参数,用于指定操作选项
2. -review-20260118: 命令参数,用于指定操作选项
3. -CN: 命令参数,用于指定操作选项
4. -US: 命令参数,用于指定操作选项

命令参数说明

• -US: 命令参数,用于指定操作选项
• -CN: 命令参数,用于指定操作选项

命令参数说明

• -CN: 命令参数,用于指定操作选项
• -US: 命令参数,用于指定操作选项

命令参数说明

• -CN: 命令参数,用于指定操作选项
• -US: 命令参数,用于指定操作选项

命令参数说明

• -US: 命令参数,用于指定操作选项
• -CN: 命令参数,用于指定操作选项

命令参数说明

• -US: 命令参数,用于指定操作选项
• -CN: 命令参数,用于指定操作选项

命令参数说明

• -CN: 命令参数,用于指定操作选项
• -US: 命令参数,用于指定操作选项

命令参数说明

• -US: 命令参数,用于指定操作选项
• -CN: 命令参数,用于指定操作选项

命令参数说明

• -US: 命令参数,用于指定操作选项
• -CN: 命令参数,用于指定操作选项

输入格式

输出格式

{
  "success": true,
  "data": {
    result: "相关说明",
    result: "相关说明",
    result: "相关说明",
    "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="your_api_key_here"

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

案例展示

转写参数配置

输出格式对比

常见问题

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

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

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

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

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

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

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

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

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

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

错误处理

已知限制

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