基于向量数据库实现音乐推荐

当听音乐的时候平台会推荐一些你喜欢的歌曲，如何更好的实现？用向量数据库的相似度搜索，可以基于语义而不是传统关键词进行更有“人情味儿”的推荐。

基于向量数据库进行歌曲推荐具有多方面的优势，主要体现如下：

1. 高效的数据检索与匹配

快速相似性查询：向量数据库采用高效的向量索引技术，能够迅速计算向量之间的相似性，从而快速找到与给定查询向量相似的歌曲数据。这种能力在处理大规模音乐库时尤为重要，可以大幅提升推荐系统的响应速度和效率。
降低查询时间：相比传统数据库，向量数据库通过优化存储结构和查询算法，能够显著降低查询时间，提升用户体验。
2. 精准的个性化推荐
捕捉用户喜好：通过将用户的行为（如听歌历史、点赞、分享等）和歌曲的属性（如旋律、节奏、歌词等）转换为向量表示，向量数据库能够更准确地捕捉用户的喜好特征。
提升推荐准确性：基于向量相似性的推荐算法能够为用户推荐与其兴趣更加匹配的歌曲，提高推荐的准确性和用户满意度。
3. 丰富的数据支持
支持复杂数据类型：向量数据库不仅支持传统的文本数据，还能够处理图像、音频等复杂数据类型。这使得歌曲推荐系统可以综合考虑歌曲的多个方面（如封面图、MV等）进行推荐。

下面以某云厂商向量数据库为基础，开始一步步Python代码来实现：

执行pip install requirements.txt初始化Python环境。

appbuilder_sdk==0.9.3

Flask==3.0.3

langchain==0.3.1

langchain_community==0.3.1

pymochow==1.3.1

qianfan==0.4.8.1

准备环境

准备环境可以参考VDB官方文档：https://cloud.baidu.com/doc/VDB/s/qlxqtaok5和https://cloud.baidu.com/doc/VDB/s/Nltgvlg7k的环境准备部分。

创建config.py文件

import os

from pymochow.auth.bce_credentials import BceCredentials

# 定义配置信息

account = 'root'

api_key = 'vdb$o*******a'

endpoint = 'http://120.*.*.31:5287'  #形如'http://127.0.0.1:5287'，数据库等访问地址

# 初始化BceCredentials对象

credentials = BceCredentials(account, api_key)

# 设置千帆AI平台的安全认证信息（AK/SK），通过环境变量

# 注意替换以下参数为您的Access Key和Secret Key

os.environ["QIANFAN_ACCESS_KEY"] = "ALTAKqk85Nx*********8BcS3"

os.environ["QIANFAN_SECRET_KEY"] = "3416b7f7271*********525c61757ae50"

创建数据库并初始化VDB数据库

文件1.creat_music_vdb.py代码如下：

import pymochow

from pymochow.configuration import Configuration

import config  # 导入配置文件

from pymochow.auth.bce_credentials import BceCredentials

# credentials（凭据）和 endpoint（终端点）信息。这些信息用于数据库。

config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)

# 这个客户端是用来与数据库服务进行交互的。

client = pymochow.MochowClient(config_obj)

try:

 db = client.create_database("music_vdb")   # 创建名称为 "music_vdb" 的数据库

except Exception as e:   # 捕获所有类型的异常

 print(f"Error: {e}")   # 打印异常信息

db_list = client.list_databases()  # 获取当前客户端连接的数据库列表

for db_name in db_list:

 print(db_name.database_name)  # 打印此连接下所有的数据库名称

client.close() # 关闭客户端

执行代码：python 1.creat_music_vdb.py，显示数据库music_vdb创建成功。

[root@iZb8bdvZ recommend-music]# python 1.creat_music_vdb.py 

Error: Database Already Exists

DocumentInsight

music_vdb

[root@iZb8bdvZ recommend-music]# 

创建表

文件2.creat_music_table.py代码如下：

import time

import pymochow  # 导入pymochow库，用于操作数据库

from pymochow.configuration import Configuration  # 用于配置客户端

import config  # 导入配置文件，包含身份验证和终端信息

# 导入pymochow模型相关的类和枚举类型

from pymochow.model.schema import Schema, Field, VectorIndex, HNSWParams

from pymochow.model.enum import FieldType, IndexType, MetricType, TableState

from pymochow.model.table import Partition

# 使用配置文件中的信息初始化客户端

config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)

client = pymochow.MochowClient(config_obj)

# 选择或创建数据库

db = client.database("music_vdb")

# 定义数据表的字段列表

fields = [

 # 定义 id 字段

 Field("id", FieldType.UINT64,   # 类型为 UINT64

 primary_key=True,     # 设为主键（primary_key=True）

 partition_key=True,   # 设为分区键（partition_key=True）

 auto_increment=False, # 不自增（auto_increment=False）

 not_null=True),

 # 定义 desc 字段，类型为 STRING

 Field("desc", FieldType.STRING),

 # 定义 music_name 字段，类型为 STRING

 Field("music_name", FieldType.STRING),

 # 定义 music_ url字段，类型为 STRING

 Field("music_url", FieldType.STRING),

 # 定义 metadata 字段，类型为 STRING

 Field("metadata", FieldType.STRING),

 # 定义 vector 字段，类型为 FLOAT_VECTOR，不能为空（not_null=True），维度为 384

 Field("vector", FieldType.FLOAT_VECTOR,

 not_null=True,

 dimension=384)

]

# 定义数据表的索引列表

indexes = [

 # 定义向量索引

 VectorIndex(

 index_name="vector_idx",  # 索引名称为 "vector_idx"

 field="vector",  # 索引的字段为 "vector"

 index_type=IndexType.HNSW,  # 索引类型为 HNSW

 metric_type=MetricType.L2,  # 度量类型为 L2

 params=HNSWParams(m=32, efconstruction=200)  # HNSW 索引的参数，m=32, efconstruction=200

 ),

]

# 尝试创建名为 "music_table" 的数据表

try:

 table = db.create_table(

 table_name="music_table",  # 数据表的名称为 "music_table"

 replication=1,  # 若您的vdb实例为免费版,这里的参数需要设置为1; 若vdb实例为标准版,这里的参数不能超过实例数据节点的数量

 partition=Partition(partition_num=1),  # 使用单个分区，分区数为1

 schema=Schema(fields=fields, indexes=indexes)  # 使用指定的字段和索引结构定义表的模式

 )

except Exception as e:  # 捕获所有类型的异常

 print(f"Error: {e}")  # 打印异常信息

# 轮询数据表状态，直到表状态为NORMAL，表示表已准备好

while True:

 time.sleep(2)  # 每次检查前暂停2秒，减少对服务器的压力

 table = db.describe_table("music_table")

 if table.state == TableState.NORMAL:  # 表状态为NORMAL，跳出循环

 break

# 打印数据表的详细信息

print("table: {}".format(table.to_dict()))

client.close()  # 关闭客户端连接

执行python 2.creat_music_table.py ，可以看到表创建成功并返回表的描述信息。

[root@iZb08bdvZ recommend-music]# python 2.creat_music_table.py 

table: {'database': 'music_vdb', 'table': 'music_table', 'description': '', 'replication': 1, 'partition': {'partitionType': <PartitionType.HASH: 'HASH'>, 'partitionNum': 1}, 'enableDynamicField': False, 'schema': {'fields': [{'fieldName': 'id', 'fieldType': 'UINT64', 'notNull': True, 'primaryKey': True, 'partitionKey': True}, {'fieldName': 'desc', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'music_name', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'music_url', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'metadata', 'fieldType': 'STRING', 'notNull': False}, {'fieldName': 'vector', 'fieldType': 'FLOAT_VECTOR', 'notNull': True, 'dimension': 384}], 'indexes': [{'indexName': 'vector_idx', 'indexType': <IndexType.HNSW: 'HNSW'>, 'field': 'vector', 'metricType': <MetricType.L2: 'L2'>, 'autoBuild': False, 'params': {'M': 32, 'efConstruction': 200}}]}, 'aliases': [], 'createTime': '2024-09-27 11:12:08', 'state': <TableState.NORMAL: 'NORMAL'>}

插入数据

插入3条音乐数据，每条包括歌曲描述、名称、url等信息，格式如下：{'desc':'这首歌曲融合了古典与现代的元素，以悠扬的钢琴旋律开场，逐渐引入电子节拍，营造出一种穿越时空的梦幻感。歌词深情而富有哲理，讲述着对过往的怀念与未来的憧憬，让人在旋律中感受到心灵的触动。','music_name':'《时空之吻》','music_url':'http://music.com/a.mp3'}。将歌曲desc信息向量化保存，供后续查询。

文件3.insert_data.py的代码如下：

from langchain_community.document_loaders import PDFPlumberLoader,Docx2txtLoader,UnstructuredExcelLoader

from langchain.text_splitter import RecursiveCharacterTextSplitter  # 用于文本分割

import qianfan  # 千帆AI平台SDK

import time  # 用于暂停执行，避免请求频率过高

import json

import pymochow

import config  # 导入配置文件

from pymochow.model.table import Row, AnnSearch, HNSWSearchParams  # 用于写入向量数据

from pymochow.configuration import Configuration

all_splits = [

{'desc':'这首歌曲融合了古典与现代的元素，以悠扬的钢琴旋律开场，逐渐引入电子节拍，营造出一种穿越时空的梦幻感。歌词深情而富有哲理，讲述着对过往的怀念与未来的憧憬，让人在旋律中感受到心灵的触动。','music_name':'《时空之吻》','music_url':'http://music.com/a.mp3'},

{'desc':'这是一首充满力量的摇滚乐曲，吉他riff强劲有力，鼓点密集而富有节奏感，主唱用其独特的嗓音嘶吼出对生活的热爱与不屈。歌曲传达了即使面对困难和挑战，也要勇往直前、永不放弃的精神，激励着每一个听众的心灵。','music_name':'《逆流而上》','music_url':'http://music.com/b.mp3'},

{'desc':'轻柔的吉他伴奏下，女歌手用温暖而富有磁性的嗓音缓缓唱出这首民谣歌曲。歌词简单却直击人心，讲述着关于爱情、梦想与成长的故事。旋律悠扬，仿佛能抚平人心中的浮躁与不安，让人在宁静中感受到生活的美好与希望。','music_name':'《岁月静好》','music_url':'http://music.com/c.mp3'}

]

print(all_splits[0])  # 打印分割后的第一个块内容

emb = qianfan.Embedding()  # 初始化嵌入模型对象

embeddings = []  # 用于存储每个文本块的嵌入向量

for chunk in all_splits:  # 遍历所有分割的文本块

 # 获取文本块的嵌入向量，使用默认模型Embedding-V1

 resp = emb.do(texts=[chunk['desc']])

 embeddings.append(resp['data'][0]['embedding'])  # 将嵌入向量添加到列表中

 time.sleep(0.3)  # 暂停0.3秒，避免请求过于频繁

print(embeddings[0])  # 打印第一个文本块的嵌入向量

# 初始化一个空列表，用于存储每行的数据对象

rows = []

# 遍历所有的数据块（all_splits），逐行写入向量化数据

for index, chunk in enumerate(all_splits):

 # 初始化 metadata 字符串

 metadata = "{}"

 # 如果数据块的 metadata 不为空，则将其转换为 JSON 格式的字符串表示

 metadata = json.dumps("{'meta':'guo'}")

 # 创建一行数据对象（Row），包括 id、desc、music_url、music_name和向量（vector）

 row = Row(

 id=index,  # 行的唯一标识符为当前索引值

 desc=chunk['desc'],  # 文本内容为数据块的页面内容

 music_name= chunk['music_name'],

 music_url = chunk['music_url'],

 metadata=metadata,  # metadata 字段为数据块的元数据转换后的 JSON 字符串

 vector=embeddings[index]  # 向量字段为当前索引处的嵌入向量

 )

 # 将创建的行对象添加到行列表中

 rows.append(row)

# 打印第一个Row对象转换成的字典格式，以验证数据结构

print(rows[0].to_dict()["metadata"])

# 读取数据库配置文件，并且初始化连接

config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)

client = pymochow.MochowClient(config_obj)

# 选择或创建数据库

db = client.database("music_vdb")

try:

 table = db.describe_table("music_table")

 table.upsert(rows=rows) # 批量写入向量数据，一次最多支持写入1000条

 table.rebuild_index("vector_idx") # 创建向量索引，必要步骤

except Exception as e:  # 捕获所有类型的异常

 print(f"Error: {e}")  # 打印异常信息

执行该py文件，得到成功插入数据到信息：

[root@iZbp1bdvZ recommend-music]# python 3.insert_data.py 

{'desc': '这首歌曲融合了古典与现代的元素，以悠扬的钢琴旋律开场，逐渐引入电子节拍，营造出一种穿越时空的梦幻感。歌词深情而富有哲理，讲述着对过往的怀念与未来的憧憬，让人在旋律中感受到心灵的触动。', 'music_name': '《时空之吻》', 'music_url': 'http://music.com/a.mp3'}

[0.04450969398021698, -0.012606468982994556, 0.029245486482977867, 0.04335927963256836, 0.012867981567978859, -0.07021073251962662, 0.013255738653242588, 0.008808392100036144, -0.06885731220245361, -0.05070096626877785, 0.09620784968137741, 0.017057133838534355, 0.05768919363617897, -0.01192457415163517, -0.0893397182226181, 0.052833426743745804, -0.058397531509399414, 0.06879784911870956, 0.006234222557395697, -0.016128402203321457, -0.025326812639832497, -0.11005371063947678, -0.003927907440811396, -0.04725881665945053, 0.019934600219130516, -0.10630016028881073, -0.058785904198884964, 0.0589137077331543, 0.1025250107049942, -0.015004078857600689, -0.04456541687250137, 0.014082217589020729, 0.01440258789807558, -0.09728464484214783, 0.042103420943021774, 0.026821577921509743, 0.0392179861664772, -0.04989371821284294, -0.07824667543172836, -0.04206026718020439, -0.05403972789645195, 0.001970261335372925, 0.027647076174616814, -0.01116835605353117, 0.0396750345826149, -0.031651776283979416, -0.1068955659866333, 0.04741368442773819, -0.035898443311452866, 0.04879467561841011, -0.016227226704359055, -0.03340809792280197, -0.03595193848013878, -0.06449054926633835, 0.09188341349363327, 0.004208618775010109, 0.03133337199687958, -0.16060130298137665, 0.057378314435482025, 0.02127242647111416, -0.05610528588294983, -0.02468905597925186, 0.008001215755939484, -0.05551918223500252, 0.07316410541534424, -0.0034658778458833694, 0.011759597808122635, -0.02996266633272171, 0.03243081271648407, -0.18630355596542358, 0, 0, -0.010053005069494247, 0, 0, 0, 0]

"{'meta':'guo'}"

构建查询的API

文件名：RAG_API.py

import config  # 导入配置模块

from flask import Flask, request, jsonify,Response

import appbuilder

import os

import qianfan

import pymochow

from pymochow.configuration import Configuration

from pymochow.auth.bce_credentials import BceCredentials

from pymochow.model.table import AnnSearch, HNSWSearchParams

app = Flask(__name__)

# 读取数据库配置文件，并且初始化连接

config_obj = Configuration(credentials=config.credentials, endpoint=config.endpoint)

client = pymochow.MochowClient(config_obj)

# 选择或创建数据库

db = client.database("music_vdb")

try:

 table = db.table("music_table")

except Exception as e:  # 捕获所有类型的异常

 print(f"Error: {e}")  # 打印异常信息

@app.route('/chatwithVDB', endpoint='chatwithVDB', methods=['POST'])

def chat_with_VDB():

 # 接收用户输入的问题

 user_input = request.args.get('user_input')

 emb = qianfan.Embedding()  # 初始化嵌入模型对象

 resp = emb.do(texts=[user_input])

 anns = AnnSearch(vector_field="vector", vector_floats=resp['data'][0]['embedding'],

 params=HNSWSearchParams(ef=200, limit=2), # 设置HNSW算法参数和返回结果的限制数量

 # filter="music_name='《岁月静好》'" # 提供标量的过滤条件

 ) 

 res = table.search(anns=anns)

 # client.close()

 # 打印用户提出的问题和系统给出的回答

 print("\n\n> Response:")

 print(res.rows)  # 打印返回内容

 return jsonify({'message': res.rows}), 200

if __name__ == '__main__':

 app.run(debug=True)

启动该API：

# 服务工作进程数

SERVER_WORKER_AMOUNT=1

# 服务工作进程启动方式

SERVER_WORKER_CLASS=gevent

# 服务超时时间

GUNICORN_TIMEOUT=60

gunicorn   --bind "0.0.0.0:5000"   --workers ${SERVER_WORKER_AMOUNT}   --worker-class ${SERVER_WORKER_CLASS}   --timeout ${GUNICORN_TIMEOUT}   "RAG_API:app" 

Postman测试效果：

访问地址：http://114.55.*.35:5000/chatwithVDB?user_input=架子鼓

查询条件是“架子鼓”，返回语义最相似的是摇滚歌曲《逆流而上》，其次是《时空之吻》。可见其将“架子鼓”进行了向量化进行相似度搜索，与歌曲《逆流而上》描述中的“鼓点”匹配度最高"score": 1.0，故第一个被返回。

{

 "message": [

 {

 "distance": 1.4065943956375122,

 "row": {

 "desc": "这是一首充满力量的摇滚乐曲，吉他riff强劲有力，鼓点密集而富有节奏感，主唱用其独特的嗓音嘶吼出对生活的热爱与不屈。歌曲传达了即使面对困难和挑战，也要勇往直前、永不放弃的精神，激励着每一个听众的心灵。",

 "id": 1,

 "metadata": "\"{'meta':'guo'}\"",

 "music_name": "《逆流而上》",

 "music_url": "http://music.com/b.mp3"

 },

 "score": 1.0

 },

 {

 "distance": 1.5477650165557861,

 "row": {

 "desc": "这首歌曲融合了古典与现代的元素，以悠扬的钢琴旋律开场，逐渐引入电子节拍，营造出一种穿越时空的梦幻感。歌词深情而富有哲理，讲述着对过往的怀念与未来的憧憬，让人在旋律中感受到心灵的触动。",

 "id": 0,

 "metadata": "\"{'meta':'guo'}\"",

 "music_name": "《时空之吻》",

 "music_url": "http://music.com/a.mp3"

 },

 "score": 0.5

 }

 ]

}

至此我们根据搜索词的语义从向量数据库搜索相似歌曲的示例讲解完毕。综上所述，基于向量数据库进行歌曲推荐具有高效、精准、丰富、可扩展和自动化等优势。这些优势共同作用于提升用户体验和推荐系统的整体性能，使得歌曲推荐更加智能化和个性化。

进阶实现

前面提到：歌曲的属性（如旋律、节奏等）转换为向量表示，向量数据库能够更准确地捕捉用户的喜好特征。下面提供如何提取这些信息，提取后使用示例代码的向量转化代码完成文本向量化。

emb = qianfan.Embedding()  # 初始化嵌入模型对象

embeddings = []  # 用于存储每个文本块的嵌入向量

for chunk in all_splits:  # 遍历所有分割的文本块

 # 获取文本块的嵌入向量，使用默认模型Embedding-V1

 resp = emb.do(texts=[chunk['desc']])

 embeddings.append(resp['data'][0]['embedding'])  # 将嵌入向量添加到列表中

音乐特征提取

音乐特征提取是将音乐信号转换为一系列数值特征的过程，这些特征能够代表音乐的某些属性，如旋律、节奏、和声等。以下是音乐特征提取的一些常见方法和步骤：

1. 时域特征提取
   振幅: 音频信号的强度或响度。
   零交叉率: 信号波形穿越零点的速率，与音频的纹理有关。
   能量: 一定时间窗口内音频信号的能量总和。
   均方根能量: 能量的平方根，常用于估计音频的响度。
2. 频域特征提取
   傅里叶变换: 将时域信号转换为频域信号，揭示音频的频率成分。
   频谱质心: 音频频谱的“重心”，反映音频的亮度。
   频谱展宽: 音频频谱的分散程度，与音频的丰富性有关。
   频谱衰减: 音频频谱的下降速度，与音频的平滑度有关。
3. 梅尔频谱特征提取
   梅尔频率倒谱系数(MFCC): 基于人类听觉感知的频率尺度，广泛用于语音和音乐识别。
   梅尔频谱: 类似于频谱，但使用梅尔尺度对频率进行量化，更接近人耳的感知。
4. 节奏特征提取
   节拍: 音乐的基本节奏单位，通常以BPM（每分钟节拍数）表示。
   节奏模式: 音乐中的重复节奏模式，如打击乐的节奏。
5. 和声特征提取
   和弦: 音乐中的和声结构，如大三和弦、小三和弦等。
   调性: 音乐的主音和调式，如C大调、A小调等。

工具和库

Librosa: 一个强大的Python库，提供了许多音频分析和音乐信息检索功能。
Essentia: 一个开源的音频分析和音频基础研究库，适用于音乐信息检索。

代码示例

import librosa
import numpy as np

# 加载音频文件
audio_path = 'path_to_your_audio_file.wav'
y, sr = librosa.load(audio_path)

# 提取时域特征
zero_crossings = librosa.zero_crossings(y).sum()
rms = librosa.feature.rms(y=y).mean()

# 提取频域特征
stft = librosa.stft(y)
spectral_centroids = librosa.feature.spectral_centroid(S=np.abs(stft)).mean()
spectral_rolloff = librosa.feature.spectral_rolloff(S=np.abs(stft)).mean()

# 提取梅尔频谱特征
mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
mfcc_mean = mfccs.mean(axis=1)

# 提取节奏特征
tempo, beats = librosa.beat.beat_track(y=y, sr=sr)
beat_times = librosa.frames_to_time(beats, sr=sr)

# 提取和声特征
# 和弦识别可能需要额外的处理，例如音高跟踪和音高类概率矩阵
chords = librosa.chroma_cqt(y=y, sr=sr)
chord_mean = chords.mean(axis=1)

# 打印提取的特征
print(f'Zero Crossing Rate: {zero_crossings}')
print(f'RMS Energy: {rms}')
print(f'Spectral Centroid: {spectral_centroids}')
print(f'Spectral Rolloff: {spectral_rolloff}')
print(f'MFCC Mean: {mfcc_mean}')
print(f'Tempo: {tempo} BPM')
print(f'Chord Mean: {chord_mean}')

在这个示例中，我们首先加载了一个音频文件，然后提取了时域特征（零交叉率和均方根能量）、频域特征（频谱质心和频谱下降点）、梅尔频谱特征（MFCCs的均值）、节奏特征（节拍和节拍时间）、和声特征。这些特征可以用于音乐分类、推荐系统或其他音乐信息检索任务。