推荐算法与RAG：数据挖掘的奇妙之旅
·
推荐算法和RAG的各种方法有异曲同工之妙，两者都是在海量的数据中查询（推荐）出和查询（目标）最相近最相关的一组数据。RAG主要用于大语言模型的效果提升，其方法和目标以及文档深度绑定，更注重文档的处理方法。推荐算法除了文本信息的处理外，更多了用户与用户、用户与物品、物品与物品之间的交互，再加上推荐算法中的文本信息一般是半结构化的，直接导致了其复杂多样的推荐算法。在这里，我将介绍几种简单的推荐算法并用python实现，让大家对这个领域有一个大致的了解。
在这里我使用的是MovieLens | GroupLens中的电影数据集进行试验（可以在该网站找到ml-32m数据集），下载后放到项目目录的data文件夹下。我们在这里实现了三个方法：根据文本的推荐、基于矩阵分解的推荐和基于图的推荐。
1. 数据预处理和展示
由于数据量比较庞大，这里我们先取其中的一部分进行展示：
def sample(src_dir = "data/ml-32m"):
# 保存目录
dst_dir = os.path.join(src_dir, "sample")
os.makedirs(dst_dir, exist_ok=True)
# 遍历目录下的所有 CSV 文件
for fname in os.listdir(src_dir):
if fname.endswith(".csv"):
src_path = os.path.join(src_dir, fname)
dst_path = os.path.join(dst_dir, fname)
# 读取 CSV
df = pd.read_csv(src_path)
# 取前 10 行
df_head = df.head(10)
# 保存到 sample 文件夹
df_head.to_csv(dst_path, index=False)
print(f"已保存: {dst_path}")为了后续算法的演示，在这里我过滤并且合并了其中的一些字段：
def zhengli(src_dir="data/ml-32m"):
need = ["movies.csv", "ratings.csv", "tags.csv"]
# 读取文件
movies = pd.read_csv(os.path.join(src_dir, need[0]))
ratings = pd.read_csv(os.path.join(src_dir, need[1]))
tags = pd.read_csv(os.path.join(src_dir, need[2]))
# 保存文件0：ratings中只保留 userId 和 movieId
save_file0 = ratings[["userId", "movieId","rating"]]
save_file0.to_csv("data/ratings.csv", index=False)
# 合并 movies 和 tags
merged = pd.merge(movies, tags[["movieId", "tag"]], on="movieId", how="left")
# 对 tag 聚合：同一电影的多个 tag 用逗号拼接
save_file1 = merged.groupby(["movieId", "title", "genres"], as_index=False)["tag"] \
.agg(lambda x: ",".join(sorted(set(filter(pd.notna, x)))))
save_file1.to_csv("data/movies.csv", index=False)
# 拆分 genres 字段并展开
genre_set = set()
for g in save_file1["genres"].dropna():
for item in g.split("|"):
genre_set.add(item)
# 保存到 txt 文件
with open("data/genres.txt", "w", encoding="utf-8") as f:
for genre in sorted(genre_set):
f.write(genre + "\n")
print(f"✅ 已保存所有类别到 data/genres.txt")
print("✅ 已保存 data/ratings.csv 和 data/movies.csv")2. 根据文本内容的匹配
（1）首先最简单也是最直观的一个方法就是通过文本描述的内容来进行推荐。这里的数据集由于缺少电影的评价和用户的个人信息，其实使用文本推荐的方法并不恰当。不过我们可以以此为例，在拥有互相的文本描述的数据集中就可以使用这个方法了。其中最常见的无非是TF-IDF算法，下面我们来计算他的得分并且持久化保存下来：
def get_tf_score(csv_path="data/movies.csv", top_n=20):
# 读取电影数据
df = pd.read_csv(csv_path)
# 拼接文本列
df["doc"] = df["title"].fillna("") + " " + df["genres"].fillna("") + " " + df["tag"].fillna("")
# 建立 TF-IDF 向量器
vectorizer = TfidfVectorizer(stop_words="english")
tfidf_matrix = vectorizer.fit_transform(df["doc"])
feature_names = vectorizer.get_feature_names_out()
# 转换为 numpy 数组
tfidf_array = tfidf_matrix.toarray()
# 保存每部电影 Top-N 高分词
top_words_list = []
top_scores_list = []
for row in tfidf_array:
top_idx = np.argsort(row)[::-1][:top_n]
top_words = feature_names[top_idx]
top_scores = row[top_idx]
top_words_list.append(",".join(top_words))
top_scores_list.append(",".join([f"{s:.4f}" for s in top_scores]))
# 构建结果 DataFrame
result_df = pd.DataFrame({
"movieId": df["movieId"],
"title": df["title"],
"top_words": top_words_list,
"top_scores": top_scores_list
})
# 保存到 CSV
result_df.to_csv("data/movies_tf.csv", index=False)
print("✅ 已保存 Top-20 TF-IDF 词到 data/movies_tf.csv")（2）之后我们创建一个名为my.txt的文本文件，将自己当作是目标用户来推荐合适的电影。这里我们采用余弦相似度来评分排序：
from sklearn.metrics.pairwise import cosine_similarity
def query_tfidf(csv_path="data/movies.csv", query:str=None, top_n=10):
if query is None or query.strip() == "":
return pd.DataFrame() # 返回空 DataFrame
# 读取电影数据
df = pd.read_csv(csv_path)
# 拼接文本列
df["doc"] = df["title"].fillna("") + " " + df["genres"].fillna("") + " " + df["tag"].fillna("")
# 建立 TF-IDF 向量器
vectorizer = TfidfVectorizer(stop_words="english")
tfidf_matrix = vectorizer.fit_transform(df["doc"])
query_vec = vectorizer.transform([query])
# 计算余弦相似度
cosine_sim = cosine_similarity(query_vec, tfidf_matrix).flatten()
# 获取最匹配的前 top_n 个电影索引
top_idx = cosine_sim.argsort()[::-1][:top_n]
# 返回结果
return df.iloc[top_idx][["movieId", "title", "genres"]].copy()（3）最后运行下面的代码，就可以看到推荐的电影列表了：
if __name__ == "__main__":
df = pd.read_csv("data/movies_tf.csv")
print(df[:5])
query = open("my.txt","r").read().strip()
movies_id = query_tfidf(csv_path="data/movies.csv", query=query, top_n=10)
print(movies_id)
pass2. 基于矩阵分解的推荐方法
基于文本的匹配存在着一些问题，比如如果两个电影有着同一个关键词，但是语义内容完全不同，该算法就没法将这两个电影区分开来。并且由于自然语言的丰富多样，简单地通过词组的重要性得分进行的推荐难以捕捉隐藏的联系。基于矩阵分解的推荐方法在一定程度上解决了这个问题。
（1）矩阵分解的思想在于，用户与物品之间的关系可以用各自的特征向量来表示，两者相乘则表示用户对于这个物品的评价。于是只要找到一个很好的矩阵分解方法，让用户和物品的特征向量在已有的数据集上拟合，那么即使用户对某个物品没有评分，也能预测出大致的分数。其中一种分解方法是svd矩阵分解：
def svd_recommend(data_path="data/ratings.csv", user_id=1, top_n=10, k_factors=20, save_path="data/ratings_svd.npy", reload=False):
# 读取评分数据
df = pd.read_csv(data_path)
# 创建用户-电影矩阵映射
user_ids = df["userId"].unique()
movie_ids = df["movieId"].unique()
user_map = {uid: idx for idx, uid in enumerate(user_ids)}
movie_map = {mid: idx for idx, mid in enumerate(movie_ids)}
inv_movie_map = {v: k for k, v in movie_map.items()}
if reload and os.path.exists(save_path):
# 直接加载预测评分矩阵
R_pred = np.load(save_path)
print(f"✅ 已从 {save_path} 加载预测评分矩阵")
else:
# 构建评分矩阵
R = np.zeros((len(user_ids), len(movie_ids)))
for _, row in df.iterrows():
u_idx = user_map[row["userId"]]
m_idx = movie_map[row["movieId"]]
R[u_idx, m_idx] = row["rating"]
# SVD 分解
U, S, Vt = np.linalg.svd(R, full_matrices=False)
S_k = np.diag(S[:k_factors])
U_k = U[:, :k_factors]
Vt_k = Vt[:k_factors, :]
# 预测评分矩阵
R_pred = np.dot(np.dot(U_k, S_k), Vt_k)
# 保存预测评分矩阵
np.save(save_path, R_pred)
print(f"✅ 已保存预测评分矩阵到 {save_path}")
# 找用户已评分电影
u_idx = user_map[user_id]
rated_idx = np.where(R_pred[u_idx] > 0)[0] # 可用原始 R 或 R_pred 判断已评分
unrated_idx = [i for i in range(R_pred.shape[1]) if i not in rated_idx]
# 获取预测评分并排序
pred_scores = [(inv_movie_map[i], R_pred[u_idx, i]) for i in unrated_idx]
top_recommend = sorted(pred_scores, key=lambda x: x[1], reverse=True)[:top_n]
# 展示结果
rated_movies = [inv_movie_map[i] for i in rated_idx]
print(f"🎬 用户 {user_id} 已评分电影ID：{rated_movies}")
print(f"🎯 推荐的 Top {top_n} 电影ID：{[m for m,_ in top_recommend]}")
return top_recommend（2）但是随着用户和物品的逐渐增多，矩阵一定是朝着非常稀疏而且巨大的方向发展，这时svd分解方法就不太适用了，针对稀疏矩阵的als分解方法可以很好地解决这个问题：
def als_recommend(
ratings_csv="data/ratings.csv",
user_id=1,
top_n=10,
factors=50,
iterations=20,
regularization=0.1,
model_path="data/ratings_als.pkl",
reload=False
):
"""
使用 ALS 对大规模稀疏评分矩阵进行推荐，并支持模型持久化
"""
# 读取评分数据
df = pd.read_csv(ratings_csv)
# 构建用户/电影映射（保证有序）
user_ids = sorted(df["userId"].unique())
movie_ids = sorted(df["movieId"].unique())
user_map = {uid: idx for idx, uid in enumerate(user_ids)}
movie_map = {mid: idx for idx, mid in enumerate(movie_ids)}
inv_movie_map = {v: k for k, v in movie_map.items()}
print(len(user_ids),len(movie_ids))
# 构建用户×电影稀疏矩阵
rows = df["userId"].map(user_map).values
cols = df["movieId"].map(movie_map).values
data = df["rating"].astype(float).values
R = coo_matrix((data, (rows, cols)), shape=(len(user_ids), len(movie_ids))).tocsr()
# 加载或训练模型
if reload and os.path.exists(model_path):
model, saved_user_map, saved_movie_map, saved_inv_movie_map = joblib.load(model_path)
print(f"✅ 已从 {model_path} 加载训练好的 ALS 模型")
user_map, movie_map, inv_movie_map = saved_user_map, saved_movie_map, saved_inv_movie_map
else:
model = implicit.als.AlternatingLeastSquares(
factors=factors,
regularization=regularization,
iterations=iterations,
calculate_training_loss=True
)
model.fit(R) # 注意这里 implicit 默认使用 item-user 矩阵
joblib.dump((model, user_map, movie_map, inv_movie_map), model_path)
print(f"✅ 已保存 ALS 模型到 {model_path}")
# 获取用户在矩阵中的行索引
if user_id not in user_map:
raise ValueError(f"User ID {user_id} 不在评分数据中")
u_idx = user_map[user_id]
# 获取用户已评分的电影
user_ratings = R[u_idx] # csr_matrix 单行
# print(user_ratings)
rated_movies = [inv_movie_map[i] for i in user_ratings.indices]
# print(f"🎬 用户 {user_id} 已评分电影ID：{rated_movies}")
# 推荐 Top-N 电影
ids, scores = model.recommend(
userid=u_idx,
user_items=R[u_idx], # ✅ csr_matrix 单行
N=top_n,
filter_already_liked_items=True
)
return ids.tolist()
（3）最后运行这个代码，就可以使用als推荐了：
# 示例调用
if __name__ == "__main__":
top10 = als_recommend(ratings_csv="data/ratings.csv", user_id=100948, top_n=10, factors=50, iterations=20, regularization=0.1,
model_path="data/ratings_als.pkl", reload=True)
print(top10)
3. 基于图的推荐方法
图结构似乎在绝大多数领域都有着强大的能力，推荐系统里也一样，无一例外的是，强大的能力往往伴随着极难的构造方法。在这里我也采用了类似的思想来进行推荐，只是极大地简化了图的结构的关系。
（1）确定图类和节点类型
首先我们需要定义出图结构的框架，我采用的是Pydantic包来进行类型检查和控制：
class NodeType(str, Enum):
"""节点类型枚举"""
PERSON = "person"
MOVIE = "movie"
class Relationship(str,Enum):
"""关系类型枚举"""
LIKE = "like"
DISLIKE = "dislike"
class TuNode(BaseModel):
type: NodeType = Field(..., description="节点的类型，只有persion和movie两类")
id: int = Field(..., description="节点id标识，其应该和type联合不唯一")
relationship: Optional[Dict[NodeType,Dict[int,str]]] = Field(
default=None,
description="节点之间的关系列表，包含关联的其他节点ID"
)
description: Optional[str] = Field(
None,
description="节点的详细描述信息"
)
metadata: Optional[Dict[str, Any]] = Field(
default_factory=dict,
description="节点的元数据，包含额外的属性和信息"
)
class Tu(BaseModel):
nodes:Optional[Dict[str,TuNode]] = Field(default_factory=dict,description="途中存储的所有nodes")
metadata:Optional[dict] = Field(default=None,description="元数据信息")
@field_validator("nodes")
def validate_nodes(cls, v):
"""验证节点字典"""
if not isinstance(v, dict):
if v:
raise ValueError('nodes must be a dictionary')
return v
def get_node(self,id:str):
return self.nodes.get(id,None)
def add_node(self,id:str,node:TuNode):
self.nodes[id]=node
return True
def del_node(self,id:str):
if self.nodes.get(id,False):
del self.nodes[id]
return True
return False
def update_node(self,id:str,node:TuNode):
self.nodes[id]=node
return True我们可以测试一下能否正常工作：
if __name__ == "__main__":
node1 = [{"type":"person","id":1,"relationship":{"person":{2:"like",3:"dislike"},"movie":{1:"like"}}},
{"type":"movie","id":1,"relationship":{"person":{1:"like"}}}]
t = Tu()
for n in node1:
node_id = f"{n.get("type","Unkown")}_{n.get("id","default")}"
t.add_node(id=node_id,node=TuNode(**n))
print(t.nodes)（2）构建图结构进行推荐
首先我们将图构造出来并持久化保存
def create_tu(ratings_csv="data/ratings.csv", save_path="data/tu.pkl") -> Tu:
# 读取评分数据
ratings = pd.read_csv(ratings_csv)
tu = Tu()
# 构建person和movie节点字典
person_dict: Dict[str, TuNode] = {}
movie_dict: Dict[str, TuNode] = {}
k = 0
for _, row in ratings.iterrows():
user_id = int(row["userId"])
movie_id = int(row["movieId"])
rating = float(row["rating"])
# 判断关系
if rating >= 4:
relation_type = Relationship.LIKE
elif rating <= 2:
relation_type = Relationship.DISLIKE
else:
continue # 评分3分不算关系
# person节点ID
p_id = f"person_{user_id}"
m_id = f"movie_{movie_id}"
# --- 构建用户节点 ---
if p_id not in person_dict:
person_dict[p_id] = TuNode(
type=NodeType.PERSON,
id=user_id,
relationship={NodeType.MOVIE:{}}
)
# 更新关系
person_dict[p_id].relationship[NodeType.MOVIE][movie_id] = relation_type.value
# --- 构建电影节点 ---
if m_id not in movie_dict:
movie_dict[m_id] = TuNode(
type=NodeType.MOVIE,
id=movie_id,
relationship={NodeType.PERSON:{}}
)
# 更新关系
movie_dict[m_id].relationship[NodeType.PERSON][user_id] = relation_type.value
k += 1
if k%100000 == 0:
print(f"已经处理了{k/10000:.1f}万条信息")
# 将节点添加到Tu对象
for pid, pnode in person_dict.items():
tu.add_node(pid, pnode)
for mid, mnode in movie_dict.items():
tu.add_node(mid, mnode)
# 持久化保存
joblib.dump(tu, save_path)
print(f"✅ 已构建并保存 Tu 对象到 {save_path}, 总节点数: {len(tu.nodes)}")
return tu可以通过下面的方法读取图：
def load_tu(tu_path="data/tu.pkl") -> Tu:
tu: Tu = joblib.load(tu_path)
print(f"✅ 已从 {tu_path} 加载 Tu 对象，总节点数: {len(tu.nodes)}")
return tu最后我采用了一个推荐策略来进行图的检索：
from collections import defaultdict, Counter
def search_tu(tu: Tu = None, person_id: int = 1, step: int = 1, top: int = 10) -> Tuple[List[int], List[int]]:
"""
基于 Tu 图结构查询推荐电影
返回：
top_like_movies: 喜欢最多的前 top 个电影ID
top_dislike_movies: 不喜欢最多的前 top 个电影ID
"""
if tu is None:
return [], []
# 获取用户节点
person_node = tu.get_node(f"person_{person_id}")
if not person_node:
return [], []
# 第一层：用户已评分电影
liked_movies = set()
disliked_movies = set()
for movie_id, relation in person_node.relationship.get("movie", {}).items():
if relation == "like":
liked_movies.add(movie_id)
elif relation == "dislike":
disliked_movies.add(movie_id)
# 收集邻居用户的电影评分
movie_counter_like = Counter()
movie_counter_dislike = Counter()
node_all = set()
# step=1 时，只考虑用户直接关系的电影
for movie_id in liked_movies.union(disliked_movies):
movie_node = tu.get_node(f"movie_{movie_id}")
if not movie_node:
continue
for uid, relation in movie_node.relationship.get("person", {}).items():
# 跳过原始用户
if uid == person_id:
continue
neighbor_node = tu.get_node(f"person_{uid}")
if not neighbor_node:
continue
for m_id, rel in neighbor_node.relationship.get("movie", {}).items():
node_all.add((m_id, rel))
if rel == "like":
movie_counter_like[m_id] += 1
elif rel == "dislike":
movie_counter_dislike[m_id] += 1
# 返回最多的前 top 个电影ID
top_like_movies = [m for m, _ in movie_counter_like.most_common(top)]
top_dislike_movies = [m for m, _ in movie_counter_dislike.most_common(top)]
return top_like_movies, top_dislike_movies
运行下面的代码就可以根据邻居来进行推荐：
# 示例调用
if __name__ == "__main__":
import joblib
tu = joblib.load("data/tu.pkl")
top_like, top_dislike = search_tu(tu=tu, person_id=1, step=1, top=10)
print("🎯 推荐的 Top 10 喜欢电影ID:", top_like)
print("⚠ 推荐的 Top 10 不喜欢电影ID:", top_dislike)4. 拓展
除此之外还有很多很多的推荐方法，如果你了解过RAG的相关内容，向量检索也是一个呼之欲出的方法。这个方法实现起来也比较方便，通过HuggingFace的包加载一个嵌入模型来获得嵌入向量，后续流程和RAG几乎是一摸一样的。另外还有基于强化学习的推荐策略，基于标签的推荐方法，基于RNN/ANN等神经网络的推荐，序列建模等。
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
14
0- 0
已为社区贡献6条内容
所有评论(0)