Install
openclaw skills install @thcjp/azure-transcription-tool-pro企业级Azure语音转写工具,支持实时流式转写、说话人分离、批量处理、自定义模型及多语言混合转写。
openclaw skills install @thcjp/azure-transcription-tool-proAzure语音转写专业版是企业级语音转写工具,在免费版批量转写基础上,提供实时流式转写、说话人分离、批量队列管理、自定义语音模型等高级能力。适用于企业会议系统、客服中心、专业领域语音文档化等高阶场景.
| 能力 | 免费版 | 专业版 |
|---|---|---|
| 批量转写 | 支持 | 支持 |
| 实时流式转写 | 不支持 | 支持 |
| 说话人分离 | 不支持 | 支持 |
| 批量队列管理 | 不支持 | 支持 |
| 自定义语音模型 | 不支持 | 支持 |
| 多语言混合 | 单语言 | 中英文混合 |
| 输出格式 | 纯文本 | SRT/VTT/JSON/纯文本 |
| 后处理 | 不支持 | 标点恢复/敏感词过滤 |
| 并发处理 | 单任务 | 多任务并发 |
| 参数名 | 类型 | 必填 | 说明 |
|---|---|---|---|
| input | string | 是 | Azure语音转写专业版处理的输入数据或指令 |
| options | object | 否 | 附加配置选项,如模式选择、格式偏好等 |
| callback_url | string | 否 | 异步处理完成后的回调通知URL |
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}")
处理: 解析实时流式转写的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回实时流式转写的响应数据,包含状态码、结果和日志.
input_params参数,支持创建/查询/导出操作# 启用说话人分离的批量转写
job = client.begin_transcription(
name="meeting-with-diarization",
locale="zh-CN",
content_urls=["https://<storage>/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)的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回说话人分离(Diarization)的响应数据,包含状态码、结果和日志.
input_params参数,支持创建/查询/导出操作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}"
处理: 解析批量转写队列管理的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回批量转写队列管理的响应数据,包含状态码、结果和日志. 能力覆盖范围:本skill的核心能力覆盖以下场景关键词:企业级语音转写工、支持实时流式转写、批量处理与自定义、语音转写专业版、面向企业团队与专、业用户的高级语音、转写工具、核心能力、实时流式语音转写、支持麦克风输入与、流式音频、自动识别不同说话、支持大规模音频文、件处理、自定义语音模型集、提升专业领域识别、准确率、多语言混合转写等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持.
input_params参数,支持创建/查询/导出操作企业会议系统实时生成字幕与会议纪要.
# 会议实时字幕系统
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://<storage>/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://<storage>/calls/call_001.wav", "name": "call_001"},
{"url": "https://<storage>/calls/call_002.wav", "name": "call_002"},
{"url": "https://<storage>/calls/call_003.wav", "name": "call_003"},
{"url": "https://<storage>/calls/call_004.wav", "name": "call_004"},
{"url": "https://<storage>/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://<storage>/videos/video_cn_01.wav", "name": "video_cn_01"},
{"url": "https://<storage>/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://<storage>/videos/video_en_01.wav", "name": "video_en_01"},
{"url": "https://<storage>/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')
以下场景Azure语音转写专业版不适合处理:
需要文本翻译、多语言转换、本地化处理时使用。不适用于非本工具能力范围的需求.
# 依赖说明
pip install azure-ai-transcription
# ...
# 配置环境变量
export TRANSCRIPTION_ENDPOINT="https://<resource>.cognitiveservices.azure.com"
export TRANSCRIPTION_KEY="配置值"
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)
manager = BatchTranscriptionManager(
endpoint=os.environ["TRANSCRIPTION_ENDPOINT"],
key=os.environ["TRANSCRIPTION_KEY"]
)
# ...
results = manager.batch_transcribe(
audio_files=[{"url": "https://<storage>/audio.wav", "name": "test"}],
locale="zh-CN",
diarization=True
)
# ...
manager.export_results(format='srt')
| 参数 | 说明 | 免费版 | 专业版 |
|---|---|---|---|
locale | 语言代码 | 支持 | 支持 |
diarization_enabled | 说话人分离 | 不支持 | 支持 |
max_speakers | 最大说话人数 | 不支持 | 可配置 |
custom_model | 自定义模型 | 不支持 | 支持 |
profanity_filter | 敏感词过滤 | 不支持 | 支持 |
punctuation | 标点恢复 | 不支持 | 支持 |
| 格式 | 用途 | 特点 |
|---|---|---|
| 纯文本 | 文档归档 | 最简格式 |
| SRT | 视频字幕 | 带序号与时间戳 |
| VTT | Web视频字幕 | HTML5标准 |
| JSON | 程序处理 | 含完整元数据 |
实时转写延迟受网络条件影响。建议使用稳定的网络连接,并适当增大音频缓冲区.
确保音频质量良好,说话人间隔清晰。设置合理的max_speakers参数,避免过多或过少.
专业版BatchTranscriptionManager自动记录失败任务,可通过retry_failed方法重试.
在Azure门户训练自定义模型后,在转写配置中指定custom_model参数即可.
完全兼容。专业版与免费版使用相同的TRANSCRIPTION_ENDPOINT和TRANSCRIPTION_KEY配置.
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| LLM API | API | 必需 | 由Agent内置LLM提供 |
| Python 3 | 运行时 | 必需 | python.org 下载安装 |
| azure-ai-transcription | Python SDK | 必需 | pip install azure-ai-transcription |
| Azure认知服务 | 云服务 | 必需 | Azure门户创建资源 |
| concurrent.futures | Python标准库 | 必需 | Python自带 |
TRANSCRIPTION_ENDPOINT:Azure认知服务端点URLTRANSCRIPTION_KEY:Azure认知服务订阅密钥| 错误场景 | 原因 | 处理方式 |
|---|---|---|
| 配置错误 | 参数缺失或格式错误 | 检查依赖说明中的配置要求 |
| 运行时错误 | 运行环境不满足 | 确认运行环境符合依赖说明 |
| 网络错误 | 连接超时或不可达 | 执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令,参考国内替代方案 |
| 风险类型 | 防范措施 |
|---|---|
| API密钥泄露 | 通过环境变量配置,禁止硬编码到代码或配置文件中 |
| 命令执行风险 | 仅执行白名单命令,避免拼接用户输入到命令行参数中 |
| 网络通信安全 | 使用HTTPS协议,验证SSL证书有效性 |
| 敏感数据暴露 | 输出结果中不包含密钥、令牌等敏感信息 |
使用前请确认已阅读依赖说明章节,确保运行环境满足安全要求。
| 操作场景 | 手动耗时 | 自动化耗时 | 效率提升 |
|---|---|---|---|
| 文件解析与提取 | 5-10分钟/个 | <5秒/个 | 60-120x |
| 批量文件处理(100个) | 8-16小时 | <5分钟 | 96-192x |
| API调用与响应解析 | 2-3分钟/次 | <1秒/次 | 120-180x |
| 多接口数据聚合 | 15-30分钟 | <10秒 | 90-180x |
| 命令执行与结果收集 | 3-5分钟/次 | <2秒/次 | 90-150x |
| 重复任务批量执行 | 因任务而异 | 线性缩减 | 5-50x |
| 错误排查与修复 | 10-30分钟 | <30秒 | 20-60x |
| 对比维度 | Azure语音转写专业版 | 传统手动方式 | 通用脚本工具 |
|---|---|---|---|
| 自动化程度 | 全流程自动 | 完全手动 | 部分自动 |
| 错误处理 | 内置错误恢复 | 依赖人工经验 | 基本try-catch |
| 可复用性 | 参数化配置 | 一次性脚本 | 模板化 |
| 安全合规 | 内置安全检查 | 无安全保障 | 无安全保障 |
| 适用场景 | 企业级语音转写工具,支持实时流式转写、说话人分离、批量处理与自定义模型。。Azu | 通用场景 | 通用场景 |