专栏名称: Databri AI
创始人刘强出版过专著「推荐系统:算法、案例与大模型」、「构建企业级推荐系统」等。Databri AI聚焦金融、传统行业的数智化转型,提供咨询、培训、项目实施解决方案。过去3年服务过中国银联、中国移动、中盐、招商银行、广发银行等大客户。
目录
相关文章推荐
51好读  ›  专栏  ›  Databri AI

「大模型智能体」23| 定制一个个性化的语音助手

Databri AI  · 公众号  ·  · 2024-12-18 11:59

正文

请到「今天看啥」查看全文


你好,我是刘强。

本课程是关于大模型智能体的实战课程,包括原理、算法、应用场景、代码实战案例等,下表是本次课程的大纲。本课是第23节,代码实现一个个性化语音助手。。本课约4500字,阅读时长30min。

以下是本次课程的正文:




我们在前面课程中对智能体相关的架构、核心能力等进行了非常详细、完整的介绍,详细读者也大致了解了什么是智能体、智能体的核心是什么、智能体能解决哪些问题了。但是要自己去实现一个智能体还不是那么容易,因此,我们在接下来的几课会通过实战案例来帮助读者利用智能体技术做出4个实用的小工具,让读者通过代码实战更好地掌握大模型智能体技术。本课我们就讲解第一个实战案例:利用智能体实现一个个性化语音助手。

大家在看公众号文章的时候应该能注意到,微信很早就能听全文了,并且阅读的效果还不错,不是那么生硬。目前要实现这个功能其实不难,有非常多的开源工具和商业化的API可以帮忙做到。本课我们就基于我们前面课程学习的智能体技术,手把手教你做一个专属的个性化语音助手,然后你就可以将任何文字用自己的语音助手来帮你阅读了。

说明:从本课开始的实战案例的代码,我们已经开源到了github上了,读者可以通过https://github.com/liuq4360/llm_agent_abc下载代码进行针对性的学习。接下来的实战案例,我们会对代码实现流程、细节进行详细拆解,帮助读者更好地掌握相关知识。


23.1 整体业务流程

本课我们实现的个性化语音助手基于阶跃星辰的声音克隆API和大模型API来实现的,并且利用智能体技术将所有的流程串联起来,让智能体来自动调用工具实现整个业务流程,具体流程见下面图23-1。

图23-1:个性化语音助手实现流程

首先我们需要将待克隆的声音进行处理,提取10s内的音频(这个是阶跃星辰API的要求)及对应的文字,提取的音频上传到云端。然后利用克隆API对声音进行克隆,克隆好了声音后,我们就可以利用这个克隆的声音来朗读任何文本了。

针对任何一个长文本,我们需要将文本切片为1000字符以内的小片段(这个也是生成声音的API要求),针对每个片段我们利用声音合成API来生成该片段对应的声音。然后将多个声音片段合并起来,就得到了最终的长文本对应的音频文件了。

在下面2节,我们会对这个流程中最核心的工具模块和智能体模块进行详细介绍。让你能够熟悉智能体操作工具解决问题的详细思路和细节。

23.2 工具模块

我们将图23-1中的流程抽象成4个工具,具体在图中用黑体字进行了标记。下面我们分别讲解这4个工具的代码实现过程。在讲解之前先提一下,本小节会用到noisereduce、librosa、whisper等Python包(具体见下面包导入部分),读者需要自己去了解。特别是langchain的工具模块,这是下面代码中重要的组件,大家需要去阅读官方文献熟悉(见参考文献1)用法。

from langchain_core.tools import BaseTool # langchain的工具模块import noisereduce as nr  # 去噪import librosa  # 加载音频并获取前10simport soundfile as sfimport whisper  # 从声音中提取文字。import osimport requests from openai import OpenAI  # OpenAI模型的sdkimport refrom pydub import AudioSegment # 声音合并

  1. 抽取工具

抽取工具的主要目的是从待克隆的音频文件中抽取前10s对应的音频和文字(也可以是中间的任何10s以内的音频)。在抽取之前,我们可以对音频去除噪音,这样后面克隆的声音质量会更高。

去噪后,我们提取音频的前10s及对应的文字,然后将10s音频及对应的文字保存到本地。这里定义工具需要有name、description 2个重载的变量,name是工具的名字,description是对工具的说明。这2个要写的清楚一点,方便智能体能从名字和描述知道智能体的用途。我们在下面代码中描述中还对工具的输入、输出参数进行了详细说明。_run方法就是工具的具体逻辑,针对抽取工具就是将音频提取10s及生成对应的文字(后面工具类似,我们不会再详细解释,读者自行看代码)。

class ExtractAudioTool(BaseTool):    name: str = "extract_audio"    description: str = ("Extracts the first 10 seconds of an audio file, Input should be a JSON with keys "                        "audio_path, output_audio_file and output_text_file. The return of this tools should be a "                        "JSON with keys output_audio_file and output_text_file, the output_audio_file is the "                        "extracted audio file path, the output_text_file is the extracted audio's text file path.")

def _run(self, inputs: dict) -> dict: audio_path = inputs['audio_path'] output_audio_file = inputs['output_audio_file'] output_text_file = inputs['output_text_file'] whisper_model = "large" # 1. 提取音频的前10秒 print("Extracting first 10 seconds of audio...") audio = AudioSegment.from_file(audio_path) first_10_seconds = audio[:10000] # 单位为毫秒 temp_audio_file = "temp_10s.wav" first_10_seconds.export(temp_audio_file, format="wav")

# 2. 去掉背景声音 print("Reducing background noise...") y, sr_rate = librosa.load(temp_audio_file, sr=None) # 加载音频 noise_profile = y[:sr_rate] reduced_noise_audio = nr.reduce_noise(y=y, sr=sr_rate, y_noise=noise_profile) sf.write(output_audio_file, reduced_noise_audio, sr_rate)

# 删除临时文件 os.remove(temp_audio_file)

# 3. 转录音频为文字 print("Transcribing audio with Whisper...") model = whisper.load_model(whisper_model) result = model.transcribe(output_audio_file, language="zh") voice_text = result["text"]

# 保存文字到文件 with open(output_text_file, "w", encoding="utf-8") as f: f.write(voice_text)

print("Text successfully extracted and saved.")

outputs = { "output_audio_file": output_audio_file, "output_text_file": output_text_file }
return outputs

2. 文件上传工具
抽取10s的音频后,我们需要利用API将10s音频文件上传到阶跃星辰的云端(见参考文献2)方便后面克隆。上传文件后会返回给我们一个文件id,这个id会用于进行声音克隆。上传工具的代码实现如下:

class UploadAudioTool(BaseTool):    name: str = "upload_audio"    description: str = ("Uploads audio(output_audio_file) to cloud storage and returns "                        "the file id:voice_file_id.")

def _run(self, output_audio_file: str) -> str: client = OpenAI(api_key=STEP_KEY, base_url="https://api.stepfun.com/v1")

response = client.files.create( file=open(output_audio_file, "rb"), purpose="storage" )

# 获取文件 ID 和文件名 voice_file_id = response.id # filename = response.filename # print(f"File ID: {file_id}, Filename: {filename}") return voice_file_id

这里说明一下,上面的STEP_KEY是阶跃星辰申请的接口秘钥,他们通过秘钥来进行计费。读者要自己实现的话,需要在他们的官网中自己申请并充值。


3. 克隆工具

克隆工具是本课的核心。通过将上传的10s音频的文件id和10s音频对应的声音文字提供给克隆工具,克隆工具调用阶跃星辰的API(见参考文献3)来进行声音克隆,克隆完成后,也会返回一个克隆声音的id,这个id会用来进行声音生成。克隆的具体代码逻辑如下:

class CloneVoiceTool(BaseTool):    name: str = "clone_voice"    description: str = (        "Clones the voice from an uploaded audio file id and voice text,"        "Input should be a JSON with keys voice_file_id and output_text_file."        "return the cloned voice id:cloned_voice_id.")

def _run(self, inputs: dict) -> str: voice_file_id = inputs['voice_file_id'] output_text_file = inputs['output_text_file']

# 读取文本文件内容 with open(output_text_file, "r", encoding="utf-8") as file: voice_text = file.read()

url = "https://api.stepfun.com/v1/audio/voices" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {STEP_KEY}" # 替换为实际的 API Key } data = { "file_id": voice_file_id, "model": "step-tts-mini", "text": voice_text }

response = requests.post(url, headers=headers, json=data)

# 打印响应结果 if response.status_code == 200: print("成功:", response.json()) # {'id': 'voice-tone-DCV8jkPKFM', 'object': 'audio.voice', 'duplicated': True} cloned_voice_id = response.json()['id'] return cloned_voice_id else: print("失败:", response.status_code, response.text) return ""


4. 声音合成工具







请到「今天看啥」查看全文