Fastwhisper + Pyannote 实现 ASR + 说话者识别

from pyannote.core import Segment
def get_text_with_timestamp(transcribe_res):timestamp_texts = []for item in transcribe_res:start = item.startend = item.endtext = item.text.strip()timestamp_texts.append((Segment(start, end), text))return timestamp_texts

def add_speaker_info_to_text(timestamp_texts, ann):spk_text = []for seg, text in timestamp_texts:spk = ann.crop(seg).argmax()spk_text.append((seg, spk, text))return spk_text

def merge_cache(text_cache):sentence = ''.join([item[-1] for item in text_cache])spk = text_cache[0][1]start = round(text_cache[0][0].start, 1)end = round(text_cache[-1][0].end, 1)return Segment(start, end), spk, sentence

PUNC_SENT_END = [',', '.', '?', '!', "，", "。", "？", "！"]

def merge_sentence(spk_text):merged_spk_text = []pre_spk = Nonetext_cache = []for seg, spk, text in spk_text:if spk != pre_spk and pre_spk is not None and len(text_cache) > 0:merged_spk_text.append(merge_cache(text_cache))text_cache = [(seg, spk, text)]pre_spk = spk
elif text and len(text) > 0 and text[-1] in PUNC_SENT_END:text_cache.append((seg, spk, text))merged_spk_text.append(merge_cache(text_cache))text_cache = []pre_spk = spkelse:text_cache.append((seg, spk, text))pre_spk = spkif len(text_cache) > 0:merged_spk_text.append(merge_cache(text_cache))return merged_spk_text

def diarize_text(transcribe_res, diarization_result):timestamp_texts = get_text_with_timestamp(transcribe_res)spk_text = add_speaker_info_to_text(timestamp_texts, diarization_result)res_processed = merge_sentence(spk_text)return res_processed

def write_to_txt(spk_sent, file):with open(file, 'w') as fp:for seg, spk, sentence in spk_sent:line = f'{seg.start:.2f} {seg.end:.2f} {spk} {sentence}\n'fp.write(line)



import torchimport whisperimport numpy as npfrom pydub import AudioSegmentfrom loguru import loggerfrom faster_whisper import WhisperModelfrom pyannote.audio import Pipelinefrom pyannote.audio import Audio
from common.error import ErrorCode
model_path = config["asr"]["faster-whisper-large-v3"]
# 测试音频： https://isv-data.oss-cn-hangzhou.aliyuncs.com/ics/MaaS/ASR/test_audio/asr_speaker_demo.wavaudio = "./test/asr/data/asr_speaker_demo.wav"asr_model = WhisperModel(model_path, device="cuda", compute_type="float16")spk_rec_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization-3.1", use_auth_token="your huggingface token")spk_rec_pipeline.to(torch.device("cuda"))
asr_result, info = asr_model.transcribe(audio, language="zh", beam_size=5)diarization_result = spk_rec_pipeline(audio)
final_result = diarize_text(asr_result, diarization_result)for segment, spk, sent in final_result:print("[%.2fs -> %.2fs] %s %s" % (segment.start, segment.end, sent, spk))

[0.00s -> 9.80s] 非常高兴能够和几位的话一起来讨论互联网企业如何决胜全球化新高地这个话题 SPEAKER_01[9.80s -> 20.20s] 然后第二块其实是游戏平台所谓游戏平台它主要是简单来说就是一个商店家社区的这样一个模式而这么多年 SPEAKER_03[20.20s -> 35.70s] 我们随着整个业务的拓张会发现跟阿里云有非常紧密的联系因为刚开始伟光在介绍的时候也讲阿里云也是阿里巴巴的云所以这个过程中一会儿也可以稍微展开跟大家讲一下我们跟云是怎么一路走来的 SPEAKER_04[35.70s -> 62.40s] 其实的确的话就对我们互联网公司来说如果不能够问当地的人口的话我想我们可能整个的就失去了后边所有的动力不知道你们各位怎么看就是我们最大的这个问题是不是效率优先Yes or No然后如果是讲一个最关键的你们是怎么来克服这一些挑战的 SPEAKER_01[62.40s -> 90.50s] 因为其实我们最近一直在做海外业务所以说我们碰到了一些问题可以一起分享出来给大家其实一起探讨一下其实海外我们还是这个观点说是无论你准备工作做得有多充分无论你有学习能力有多强你一个中国企业的负责人其实在出海的时候它整体还是一个强势的是做的过程 SPEAKER_03[90.50s -> 101.60s] 后来推到德国或者推到新加坡 印尼 越南等等这些地方每一个地方走过去都面临的一个问题是建站的效率怎么样能够快速地把这个站点建起来 SPEAKER_04[101.60s -> 122.90s] 一方面我们当初刚好从2014年刚好开始要出去的时候国内就是三个北上广深但在海外要同时开服北美 美东 美西 欧洲 日本我还记得那个时候我们在海外如何去建立这种IDC的康碳建设基础设施建设云服务的部署那都是一个全新的挑战 SPEAKER_02

spk_rec_pipeline = Pipeline.from_pretrained("./data/models/speaker-diarization-3.1/config.yaml")