项目名称：理性投资助手 — 语音识别与分析系统

报告日期：2024年8月18日

项目负责人：不想透露名字

项目概述：

我开发了一款名为“理性投资助手”的语音对话系统，旨在通过语音识别技术分析投资人的语音，判断其情绪状态是否理性，并据此提供投资建议。该系统特别适用于金融市场分析、投资顾问服务等领域，帮助投资者避免情绪化决策。

技术方案与实施步骤：

模型选择：

我们选择了基于深度学习的英伟达NIM平台和微软的phi-3-voice。这些模型能够准确识别语音中的关键词和情感倾向，为投资建议提供依据。

数据的构建：

数据构建包括收集投资相关的语音数据、标注情绪状态、进行特征提取和向量化处理。我们采用了深度学习技术来提高数据的表达能力和模型的泛化能力。

功能整合：

系统整合了语音识别、情感分析、投资建议生成等模块，实现了从语音输入到投资建议输出的全流程自动化。

实施步骤：

环境搭建：配置Python环境，

安装 .NET 8 (下载地址 [https://dotnet.microsoft.com/](https://dotnet.microsoft.com/))，

安装 Visual Studio Code，安装 Visual Studio Code 插件 （.NET Extension Pack 和 Python, Jupyter ），安装 Python 库 - Jupyter Notebook安装语音处理和机器学习相关库。

代码实现：实现语音信号预处理、模型训练、情感分析、投资建议生成等关键功能。

测试与调优：设计测试用例，进行系统测试，根据结果调优模型参数。

集成与部署：将语音识别、情感分析和建议生成模块集成，部署到Web或移动平台。

项目成果与展示：

代码实现：

部署所有所需库

<code># ! pip install gradio

# ! pip install openai-whisper==20231117

# ! pip install ffmpeg==1.4

# ! conda install ffmpeg -y

# ! pip install edge-tts

# ! pip install transformers

# ! pip install openai

from openai import OpenAI

使用openai 

client = OpenAI(

base_url = "https://integrate.api.nvidia.com/v1",

api_key = "api key"

)

使用phi-3模型输出陆蓉行为金融学讲义内容 

completion = client.chat.completions.create(

model="your_appropriate_model",code>

messages=[{"role":"user","content":"请解释陆蓉行为金融学讲义中投资者是否理性的内容"}],

temperature=0.4,

top_p=0.7,

max_tokens=2048,

stream=True

)

 下面这段Python代码的目的是使用edge-tts（一个基于Microsoft Edge的文本到语音转换工具）将文本转换为语音，并将生成的语音保存为名为"demo.mp3"的音频文件。

edge_command=f'edge-tts --text "{result}" --write-media ./content/demo.mp3'

下面使用edge_command，这是一个Python库，用于实现边缘检测。边缘检测是一种计算机视觉技术，用于检测图像中的边缘、线条和轮廓。在图像处理和计算机视觉中，边缘检测是一种基本操作，用于提取图像中的关键特征。

edge_command

下面这段Python代码的作用是执行一个名为`edge_command`的系统命令。`os.system()`函数是Python的os模块中的一个函数，用于执行一个外部命令并返回其状态码。这个状态码可以用来判断命令是否执行成功。

import os

os.system(edge_command)

 后面这段代码是在Python环境中安装pydub库的命令。pydub是一个用于音频处理的Python库，可以用来对音频文件进行剪切、连接、转换格式等操作。需要注意的是，安装pydub库需要先安装ffmpeg工具，因为pydub库依赖于ffmpeg来处理音频文件。安装ffmpeg的方法因操作系统而异，可以参考其官方文档进行安装。

! pip install pydub==0.25.1

这段代码比较复杂，具体展开说说：

 这段Python代码的目的是在Jupyter Notebook中获取实时的麦克风音频

import pyaudio

import numpy as np

chunk = 1024

sample_rate = 44100

p = pyaudio.PyAudio()

stream = p.open(format=pyaudio.paInt16,

channels=1,

rate=sample_rate,

input=True,

frames_per_buffer=chunk)

while True:

data = stream.read(chunk)

audio_data = np.frombuffer(data, dtype=np.int16)

在上面的代码中，我们使用PyAudio打开一个音频输入流，并设置了音频的格式（16位整数），声道数（单声道），采样率（44100 Hz）和缓冲区大小（1024）。

然后，我们使用一个无限循环来持续读取音频数据。每次读取数据后，我们将其转换为NumPy数组，并可以对音频数据进行处理。

下面这段没什么好说的

import whisper

select_model ="tiny" # ['tiny', 'base']

whisper_model = whisper.load_model(select_model)

#@title Edge TTS

def calculate_rate_string(input_value):

rate = (input_value - 1) * 100

sign = '+' if input_value >= 1 else '-'

return f"{sign}{abs(int(rate))}"

def make_chunks(input_text, language):

language="English"code>

if language == "English":

temp_list = input_text.strip().split(".")

filtered_list = [element.strip() + '.' for element in temp_list[:-1] if element.strip() and element.strip() != "'" and element.strip() != '"']

if temp_list[-1].strip():

filtered_list.append(temp_list[-1].strip())

return filtered_list

import re

import uuid

def tts_file_name(text):

if text.endswith("."):

text = text[:-1]

text = text.lower()

text = text.strip()

text = text.replace(" ","_")

truncated_text = text[:25] if len(text) > 25 else text if len(text) > 0 else "empty"

random_string = uuid.uuid4().hex[:8].upper()

file_name = f"./content/edge_tts_voice/{truncated_text}_{random_string}.mp3"

return file_name

from pydub import AudioSegment

import shutil

import os

def merge_audio_files(audio_paths, output_path):

# Initialize an empty AudioSegment

merged_audio = AudioSegment.silent(duration=0)

# Iterate through each audio file path

for audio_path in audio_paths:

# Load the audio file using Pydub

audio = AudioSegment.from_file(audio_path)

# Append the current audio file to the merged_audio

merged_audio += audio

# Export the merged audio to the specified output path

merged_audio.export(output_path, format="mp3")code>

def edge_free_tts(chunks_list,speed,voice_name,save_path):

# print(chunks_list)

if len(chunks_list)>1:

chunk_audio_list=[]

if os.path.exists("./content/edge_tts_voice"):

shutil.rmtree("./content/edge_tts_voice")

os.mkdir("./content/edge_tts_voice")

k=1

for i in chunks_list:

print(i)

edge_command=f'edge-tts --rate={calculate_rate_string(speed)}% --voice {voice_name} --text "{i}" --write-media ./content/edge_tts_voice/{k}.mp3'

print(edge_command)

var1=os.system(edge_command)

if var1==0:

pass

else:

print(f"Failed: {i}")

chunk_audio_list.append(f"./content/edge_tts_voice/{k}.mp3")

k+=1

# print(chunk_audio_list)

merge_audio_files(chunk_audio_list, save_path)

else:

edge_command=f'edge-tts --rate={calculate_rate_string(speed)}% --voice {voice_name} --text "{chunks_list[0]}" --write-media {save_path}'

print(edge_command)

var2=os.system(edge_command)

if var2==0:

pass

else:

print(f"Failed: {chunks_list[0]}")

return save_path

# text = "This is Microsoft Phi 3 mini 4k instruct Demo" Simply update the text variable with the text you want to convert to speech

text = 'This is Microsoft Phi 3 mini 4k instruct Demo' # @param {type: "string"}

Language = "English" # @param ['English']

# Gender of voice simply change from male to female and choose the voice you want to use

Gender = "Female"# @param ['Male', 'Female']

female_voice="en-US-AriaNeural"# @param["en-US-AriaNeural",'zh-CN-XiaoxiaoNeural','zh-CN-XiaoyiNeural']code>

speed = 1 # @param {type: "number"}

translate_text_flag = False

if len(text)>=600:

long_sentence = True

else:

long_sentence = False

# long_sentence = False # @param {type:"boolean"}

save_path = '' # @param {type: "string"}

if len(save_path)==0:

save_path=tts_file_name(text)

if Language == "English" :

if Gender=="Male":

voice_name="en-US-ChristopherNeural"code>

if Gender=="Female":

voice_name=female_voice

# voice_name="en-US-AriaNeural"code>

if translate_text_flag:

input_text=text

# input_text=translate_text(text, Language)

# print("Translateting")

else:

input_text=text

if long_sentence==True and translate_text_flag==True:

chunks_list=make_chunks(input_text,Language)

elif long_sentence==True and translate_text_flag==False:

chunks_list=make_chunks(input_text,"English")

else:

chunks_list=[input_text]

# print(chunks_list)

# edge_save_path=edge_free_tts(chunks_list,speed,voice_name,save_path)

# from IPython.display import clear_output

# clear_output()

# from IPython.display import Audio

# Audio(edge_save_path, autoplay=True)

from IPython.display import clear_output

from IPython.display import Audio

if not os.path.exists("./content/audio"):

os.mkdir("./content/audio")

import uuid

def random_audio_name_generate():

random_uuid = uuid.uuid4()

audio_extension = ".mp3"

random_audio_name = str(random_uuid)[:8] + audio_extension

return random_audio_name

def talk(input_text):

global translate_text_flag,Language,speed,voice_name

if len(input_text)>=600:

long_sentence = True

else:

long_sentence = False

if long_sentence==True and translate_text_flag==True:

chunks_list=make_chunks(input_text,Language)

elif long_sentence==True and translate_text_flag==False:

chunks_list=make_chunks(input_text,"English")

else:

chunks_list=[input_text]

save_path="./content/audio/"+random_audio_name_generate()code>

edge_save_path=edge_free_tts(chunks_list,speed,voice_name,save_path)

return edge_save_path

edge_save_path=talk(text)

Audio(edge_save_path, autoplay=True)

这段代码主要用于将文本转换为语音。

1. `calculate_rate_string(input_value)` 函数：根据输入的值计算语音速度，并返回一个字符串。

2. `make_chunks(input_text, language)` 函数：将文本分割成多个部分，以便在语音中正确断句。

3. `tts_file_name(text)` 函数：根据输入的文本生成一个唯一的音频文件名。

4. `merge_audio_files(audio_paths, output_path)` 函数：将多个音频文件合并为一个音频文件。

5. `edge_free_tts(chunks_list, speed, voice_name, save_path)` 函数：使用 Edge TTS 将文本转换为语音，并将生成的音频文件保存到指定的路径。

6. `talk(input_text)` 函数：将输入的文本转换为语音，并返回生成的音频文件的路径。

7. `edge_save_path=talk(text)` 调用 `talk()` 函数，将文本转换为语音。

8. `Audio(edge_save_path, autoplay=True)` 播放生成的音频文件。

综上所述，这段代码的主要用途是将文本转换为语音，并可以自定义语音的速度和音色。

def phi_demo(prompt):

client = OpenAI(

base_url = "https://integrate.api.nvidia.com/v1",

api_key = "API Key"

)

completion = client.chat.completions.create(

model="microsoft/phi-3-mini-128k-instruct",code>

messages=[{"role":"user","content":prompt}],

temperature=0.4,

top_p=0.7,

max_tokens=512,

stream=True

)

result = ""

for chunk in completion:

if chunk.choices[0].delta.content is not None:

result += chunk.choices[0].delta.content

return result

这段代码首先创建了一个名为client的OpenAI对象，用于与NVIDIA NIM API进行交互。需要提供API的base_url和api_key（处于隐私保护我并没有放出我的apikey）。

然后，使用client.chat.completions.create()方法创建一个对话生成请求。

#@title Run gradio app

def convert_to_text(audio_path):

result = whisper_model.transcribe(audio_path,word_timestamps=True,fp16=False,language='Chinese',task='translate')code>

with open('scan.txt', 'w') as file:

file.write(str(result))

return result["text"]

import gradio as gr

from IPython.display import Audio, display

def run_text_prompt(message, chat_history):

bot_message = phi_demo(message)

edge_save_path=talk(bot_message)

# print(edge_save_path)

display(Audio(edge_save_path, autoplay=True))

chat_history.append((message, bot_message))

return "", chat_history

def run_audio_prompt(audio, chat_history):

if audio is None:

return None, chat_history

print(audio)

message_transcription = convert_to_text(audio)

_, chat_history = run_text_prompt(message_transcription, chat_history)

return None, chat_history

with gr.Blocks() as demo:

chatbot = gr.Chatbot(label="Chat with Phi 3 mini 4k instruct")code>

msg = gr.Textbox(label="Ask anything")code>

msg.submit(run_text_prompt, [msg, chatbot], [msg, chatbot])

with gr.Row():

audio = gr.Audio(sources="microphone", type="filepath")code>

send_audio_button = gr.Button("Send Audio", interactive=True)

send_audio_button.click(run_audio_prompt, [audio, chatbot], [audio, chatbot])

demo.launch(share=True,debug=True)

基于上面的一切：我们只要再写亿点代码，即可使用Gradio库来创建一个聊天机器人。Gradio是一个用于创建交互式机器学习模型的库，可以让用户轻松地尝试和部署模型。

首先，定义了一个名为`convert_to_text`的函数，用于将音频文件转换为文本。这个函数使用了Whisper模型，Whisper是一个基于Transformer的语音识别模型，可以用来将音频文件转换为文本。

接下来，定义了一个名为`run_text_prompt`的函数，用于处理用户输入的文本。这个函数使用了Phi 3 mini 4k instruct模型，Phi是一个基于Transformer的聊天机器人模型，可以用来生成回复。

然后，定义了一个名为`run_audio_prompt`的函数，用于处理用户上传的音频文件。这个函数会调用`convert_to_text`函数将音频文件转换为文本，然后调用`run_text_prompt`函数处理文本。

最后，使用Gradio创建了一个聊天机器人界面，包括一个文本输入框、一个音频输入框和一个发送按钮。用户可以在文本输入框中输入问题，然后点击发送按钮将问题发送给聊天机器人。聊天机器人会根据用户输入的文本生成回复，并将回复显示在界面上。这样我们的理性投资决策RAG就编写完成了。

 应用场景展示：

系统在金融投资领域中，帮助投资者在情绪波动时做出更加理性的投资决策。

功能演示：

展示系统的主要功能，包括实时语音识别、情绪状态判断、投资建议生成等。

问题与解决方案：

问题分析：

在项目实施过程中，我们遇到了语音识别准确度不高、情绪分析的复杂性等问题。

解决措施：

通过增加训练数据、优化模型结构、引入先进的情感分析算法来提高识别准确度和分析复杂情绪的能力。

项目总结与展望：

项目评估：

项目在技术实现上取得了初步成功，但在语音识别的准确性和情绪分析的深度上还有提升空间。

未来方向：

未来计划进一步优化语音识别算法，增强情绪分析的准确性，并探索更多投资领域的应用。

附件与参考资料：

Microsoft-Phi-3-NvidiaNIMWorkshop/Phi-3-SK.ipynb at main · kinfey/Microsoft-Phi-3-NvidiaNIMWorkshop · GitHub

陆蓉行为金融学讲义