你是不是也遇到过这样的问题想用Python做一个电商推荐系统网上搜到的教程要么只讲协同过滤算法要么只教Django框架要么单独讲Scrapy爬虫但就是没人告诉你如何把这些技术栈真正串联起来构建一个完整的、可运行的推荐系统更让人困惑的是当你想引入“大数据”或“大模型”这些听起来很酷的概念时发现资料要么过于理论化要么工程实现复杂到让人望而却步。Hadoop、Spark这些词经常和推荐系统一起出现但它们在你的项目里到底扮演什么角色是必须的吗这篇文章要解决的核心问题就是如何用Python技术栈从零开始构建一个结构清晰、可扩展的电商推荐系统原型并理清何时需要引入Hadoop/Spark这类大数据组件以及“大模型”在当前语境下的真实含义。我将带你完成一个完整的项目实践涵盖数据采集Scrapy、后端服务Django、核心算法协同过滤、以及数据存储与处理的可扩展思路。你会发现一个能跑起来的推荐系统其核心在于清晰的数据流设计和适度的技术选型而不是盲目堆砌“高大上”的技术名词。读完本文你将能理解电商推荐系统的基本架构和数据流。使用Scrapy爬取模拟商品数据并结构化存储。使用Django搭建具备用户、商品、评分行为管理功能的后端API。实现基于用户的协同过滤算法并提供实时推荐接口。掌握如何设计系统以便在数据量增大时平滑地将核心计算如相似度矩阵迁移到Spark等大数据平台进行计算而Django服务只负责结果服务和实时逻辑。澄清“大模型”在推荐系统中的常见误解与应用边界。1. 为什么“Python电商推荐系统”值得你投入时间推荐系统早已不是大厂的专利它已成为提升任何带有内容或商品展示的应用用户粘性和转化率的关键技术。对于Python开发者而言这是一个绝佳的、能将数据爬取、Web开发、算法应用、数据工程等多个技能点串联起来的综合项目。这个项目的真正价值在于“贯通”。很多教程止步于算法理论或单一框架使用导致学习者无法形成系统观。我们将构建的不仅仅是一个算法Demo而是一个具备完整数据流水线的微系统数据从哪来- 用Scrapy爬虫模拟避免法律风险用公开数据集或模拟数据。数据怎么存、怎么管- 用Django的ORM和Admin管理商品、用户、行为数据。算法怎么集成和提供服务- 在Django中实现协同过滤算法并通过API暴露推荐结果。何时考虑“大数据”技术- 当用户-物品矩阵巨大内存计算如Pandas成为瓶颈时我们将设计一个方案将耗时的离线计算如全局用户相似度交给SparkDjango只使用计算结果或处理实时轻量计算。关于“大模型”在当前2023-2024年的推荐系统领域它通常指用于特征提取如用BERT提取商品描述文本特征或排序阶段的深度模型而不是直接替代协同过滤等召回算法。对于入门和中级项目基于协同过滤的召回基于逻辑回归/深度模型的精排是更实际的架构。本文重点在召回层协同过滤的实现但会指出引入“大模型”进行特征增强的可能位置。2. 核心概念与项目架构设计在动手写代码前必须理清几个核心概念和它们在我们系统里的角色。协同过滤算法核心思想是“物以类聚人以群分”。基于用户的协同过滤找到与目标用户兴趣相似的其他用户将这些用户喜欢而目标用户未接触过的物品推荐给他。关键在于计算用户之间的相似度如余弦相似度、皮尔逊相关系数。基于物品的协同过滤找到与目标用户历史喜欢物品相似的其他物品进行推荐。关键在于计算物品之间的相似度。矩阵分解一种更高级的方法将庞大的用户-物品评分矩阵分解为低维的用户隐向量和物品隐向量矩阵用内积预测评分。它属于协同过滤的范畴能更好地处理稀疏矩阵。我们的系统架构 我们将采用基于用户的协同过滤作为核心算法因为它更直观且便于解释“找到相似用户”这一逻辑。整个系统的数据流如下[Scrapy爬虫] - (爬取/模拟商品数据) - [MySQL/PostgreSQL数据库] | v [Django后台] - (管理、展示、记录用户行为) - [用户访问前端] | v [协同过滤算法模块] | v [推荐结果API] - [返回给前端]技术栈选型说明Django作为Web框架和业务逻辑的核心。其强大的ORM、Admin后台、路由和视图系统能快速搭建数据管理和API服务。Scrapy用于初始数据构建。在实际项目中数据可能来自业务数据库、日志文件等。这里用Scrapy模拟从电商网站抓取商品信息的过程重点是学习如何构建结构化的数据管道。协同过滤算法使用Python的pandas和numpy在内存中实现。这对于万级用户、十万级物品以内的数据量是可行的。Hadoop/Spark它们不是本项目初始必需的。它们是“扩展选项”。当用户行为数据达到千万、亿级用户-物品矩阵无法放入单机内存时就需要用Spark的分布式计算能力来离线计算用户相似度矩阵然后将计算结果例如每个用户的Top-N相似用户列表存储到数据库或缓存中供Django API查询。本文会讲解这种“离线计算在线服务”的架构思想。大模型如前所述是“增强选项”。例如可以用开源的Sentence-BERT模型为商品标题和描述生成向量作为物品的补充特征融入相似度计算。这属于特征工程层面不影响主流程。3. 环境准备与项目初始化我们使用Python 3.8版本进行开发。建议使用虚拟环境如venv或conda隔离项目依赖。3.1 创建项目目录与虚拟环境# 创建项目目录 mkdir ecommerce_recommender cd ecommerce_recommender # 创建并激活虚拟环境 (以venv为例) python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 安装核心依赖创建requirements.txt文件内容如下Django4.2.0 djangorestframework3.14.0 pandas2.0.3 numpy1.24.3 scrapy2.9.0 scikit-learn1.3.0 # 用于计算余弦相似度 mysqlclient2.2.0 # 如果使用MySQL否则使用psycopg2-binary for PostgreSQL然后安装pip install -r requirements.txt3.3 初始化Django项目和应用# 创建Django项目 django-admin startproject recommender_system . # 创建核心应用 python manage.py startapp core # 创建用于推荐算法的应用 python manage.py startapp rec_engine项目结构预览ecommerce_recommender/ ├── venv/ ├── recommender_system/ # Django项目配置目录 │ ├── __init__.py │ ├── settings.py │ ├── urls.py │ └── wsgi.py ├── core/ # 核心应用用户、商品、行为模型 │ ├── migrations/ │ ├── __init__.py │ ├── admin.py │ ├── apps.py │ ├── models.py │ ├── tests.py │ └── views.py ├── rec_engine/ # 推荐引擎应用算法实现与API │ ├── __init__.py │ ├── apps.py │ ├── algorithms.py # 协同过滤算法实现 │ ├── services.py # 推荐服务封装 │ └── views.py ├── scrapy_project/ # Scrapy爬虫项目后续创建 ├── requirements.txt └── manage.py3.4 配置数据库在recommender_system/settings.py中配置数据库这里以MySQL为例# settings.py DATABASES { default: { ENGINE: django.db.backends.mysql, NAME: recommender_db, # 数据库名 USER: your_username, PASSWORD: your_password, HOST: localhost, PORT: 3306, } } # 将应用添加到INSTALLED_APPS INSTALLED_APPS [ django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, rest_framework, # DRF core, # 核心应用 rec_engine, # 推荐引擎应用 ]运行迁移创建基础表python manage.py migrate4. 构建数据模型用户、商品与行为推荐系统的基石是数据。我们在core/models.py中定义三个核心模型。4.1 定义模型# core/models.py from django.db import models from django.contrib.auth.models import AbstractUser # 扩展Django自带的用户模型便于后续添加更多字段 class User(AbstractUser): # 可以添加额外字段如年龄、性别用于混合推荐 # age models.IntegerField(nullTrue, blankTrue) # gender models.CharField(max_length10, choices((M,Male),(F,Female)), nullTrue, blankTrue) class Meta: db_table core_user class Product(models.Model): 商品模型 # 基础信息 title models.CharField(max_length200, verbose_name商品标题) description models.TextField(verbose_name商品描述, blankTrue) price models.DecimalField(max_digits10, decimal_places2, verbose_name价格) category models.CharField(max_length100, verbose_name类别, db_indexTrue) # 加索引便于过滤 image_url models.URLField(verbose_name图片链接, blankTrue) # 用于协同过滤的隐式/显式特征可后续用模型生成 # feature_vector models.JSONField(nullTrue, blankTrue) # 例如从大模型提取的向量 created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table core_product verbose_name 商品 verbose_name_plural 商品 def __str__(self): return self.title class UserBehavior(models.Model): 用户行为记录评分/点击/购买等 BEHAVIOR_TYPES ( (view, 浏览), (cart, 加入购物车), (buy, 购买), (rate, 评分), ) user models.ForeignKey(User, on_deletemodels.CASCADE, related_namebehaviors, verbose_name用户) product models.ForeignKey(Product, on_deletemodels.CASCADE, related_namebehaviors, verbose_name商品) behavior_type models.CharField(max_length10, choicesBEHAVIOR_TYPES, verbose_name行为类型) # 对于评分行为value存储分数如1-5对于点击/购买可以存储1或次数 value models.FloatField(default1.0, verbose_name行为值) timestamp models.DateTimeField(auto_now_addTrue, verbose_name发生时间) class Meta: db_table core_user_behavior verbose_name 用户行为 verbose_name_plural 用户行为 # 一个用户对同一个商品同一种行为只记录最后一次或可累计这里用唯一约束示例 unique_together [user, product, behavior_type] def __str__(self): return f{self.user.username} - {self.product.title} - {self.behavior_type}4.2 注册模型到Admin并迁移在core/admin.py中注册模型方便后台管理# core/admin.py from django.contrib import admin from .models import Product, UserBehavior admin.register(Product) class ProductAdmin(admin.ModelAdmin): list_display (id, title, category, price) search_fields (title, category) list_filter (category,) admin.register(UserBehavior) class UserBehaviorAdmin(admin.ModelAdmin): list_display (user, product, behavior_type, value, timestamp) list_filter (behavior_type, timestamp) search_fields (user__username, product__title)运行命令创建数据表python manage.py makemigrations core python manage.py migrate core5. 使用Scrapy爬取模拟商品数据我们不会爬取真实电商网站避免法律风险而是创建一个Scrapy项目来生成模拟数据并学习如何将数据导入Django数据库。5.1 创建Scrapy项目与爬虫# 在项目根目录下 scrapy startproject scrapy_project cd scrapy_project scrapy genspider product_spider example.com5.2 编写模拟数据爬虫修改scrapy_project/spiders/product_spider.py# scrapy_project/spiders/product_spider.py import scrapy import random from faker import Faker # 用于生成假数据 class ProductSpider(scrapy.Spider): name product_spider allowed_domains [example.com] # 仅作占位实际不访问 start_urls [http://example.com/] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fake Faker() self.categories [电子产品, 图书, 服装, 家居, 食品, 美妆] def parse(self, response): 生成模拟商品数据 # 假设我们需要生成1000个商品 for i in range(1000): product { title: self.fake.catch_phrase(), description: self.fake.text(max_nb_chars200), price: round(random.uniform(10, 2000), 2), category: random.choice(self.categories), image_url: self.fake.image_url(), external_id: fprod_{i:06d}, # 模拟外部ID } yield product5.3 编写Item Pipeline导入Django数据库关键步骤在Scrapy的Pipeline中调用Django的ORM来保存数据。首先需要在Scrapy设置中激活Django环境。 修改scrapy_project/pipelines.py# scrapy_project/pipelines.py import sys import os # 将Django项目根目录添加到Python路径并设置Django环境 sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ..)) os.environ.setdefault(DJANGO_SETTINGS_MODULE, recommender_system.settings) import django django.setup() from core.models import Product class DjangoProductPipeline: def process_item(self, item, spider): # 避免重复导入根据external_id或title判断 product, created Product.objects.update_or_create( titleitem[title], defaults{ description: item.get(description, ), price: item[price], category: item[category], image_url: item.get(image_url, ), } ) spider.logger.info(f{Created if created else Updated} product: {product.title}) return item在scrapy_project/settings.py中启用Pipeline并降低日志级别# scrapy_project/settings.py ITEM_PIPELINES { scrapy_project.pipelines.DjangoProductPipeline: 300, } LOG_LEVEL INFO5.4 运行爬虫并导入数据# 在scrapy_project目录下 scrapy crawl product_spider -o products.json运行后检查Django Admin后台 (http://127.0.0.1:8000/admin/core/product/)应该能看到爬取的1000个商品。6. 实现协同过滤推荐算法这是系统的核心。我们在rec_engine/algorithms.py中实现基于用户的协同过滤。6.1 算法原理与步骤构建用户-物品评分矩阵行是用户列是商品值是用户对商品的行为评分如浏览1购买5评分评分值。计算用户相似度使用余弦相似度或皮尔逊相关系数计算用户向量之间的相似度。寻找最近邻为目标用户找出最相似的K个用户。生成推荐聚合最近邻用户喜欢的物品剔除目标用户已接触过的按预估兴趣度排序。6.2 算法实现代码# rec_engine/algorithms.py import numpy as np import pandas as pd from sklearn.metrics.pairwise import cosine_similarity from django.db.models import Count, Avg from core.models import UserBehavior, Product, User class UserBasedCF: 基于用户的协同过滤推荐 def __init__(self): self.user_sim_matrix None # 用户相似度矩阵 self.user_item_matrix None # 用户-物品评分矩阵 self.users None self.products None def prepare_data(self): 从数据库准备数据构建用户-物品矩阵 # 获取所有用户行为计算平均评分或行为权重 behaviors UserBehavior.objects.filter(behavior_typerate).values(user_id, product_id, value) # 如果没有评分数据可以用其他行为如购买模拟这里假设有评分 if not behaviors: # 示例如果没有评分用浏览和购买行为生成模拟评分 behaviors UserBehavior.objects.all().values(user_id, product_id, behavior_type) # 行为权重映射 weight_map {view: 1, cart: 2, buy: 5} behaviors [ {user_id: b[user_id], product_id: b[product_id], value: weight_map.get(b[behavior_type], 1)} for b in behaviors ] df pd.DataFrame(list(behaviors)) if df.empty: return False # 创建用户-物品评分矩阵 (稀疏矩阵未评分为0或NaN) self.user_item_matrix df.pivot_table( indexuser_id, columnsproduct_id, valuesvalue, aggfuncmean, # 同一用户对同一商品多次行为取平均 fill_value0 ) self.users self.user_item_matrix.index.tolist() self.products self.user_item_matrix.columns.tolist() return True def calculate_similarity(self): 计算用户之间的余弦相似度 if self.user_item_matrix is None: if not self.prepare_data(): raise ValueError(No data available to calculate similarity.) # 计算余弦相似度 # 注意这里计算的是稠密矩阵的相似度用户/物品很多时会内存爆炸。这是单机版的局限。 self.user_sim_matrix cosine_similarity(self.user_item_matrix.values) # 将相似度矩阵转换为DataFrame方便索引 self.user_sim_df pd.DataFrame( self.user_sim_matrix, indexself.users, columnsself.users ) def recommend(self, user_id, top_k10, neighbor_k20): 为目标用户推荐商品 Args: user_id: 目标用户ID top_k: 返回推荐商品的数量 neighbor_k: 使用的最近邻用户数量 Returns: list: 推荐商品ID列表 if self.user_sim_df is None: self.calculate_similarity() if user_id not in self.users: # 新用户冷启动问题返回热门商品或基于类别的推荐 return self._cold_start_recommendation(top_k) # 1. 获取目标用户的相似用户排除自己 sim_users self.user_sim_df[user_id].sort_values(ascendingFalse).iloc[1:neighbor_k1] # 2. 获取这些相似用户有过行为但目标用户没有的商品 target_user_items set(self.user_item_matrix.loc[user_id][self.user_item_matrix.loc[user_id] 0].index) # 计算候选商品的推荐分数 item_scores {} for sim_user_id, similarity in sim_users.items(): # 相似用户评分过的商品 sim_user_items self.user_item_matrix.loc[sim_user_id] for item_id in sim_user_items[sim_user_items 0].index: if item_id not in target_user_items: # 累加 (相似度 * 评分) item_scores[item_id] item_scores.get(item_id, 0) similarity * sim_user_items[item_id] # 3. 按分数排序返回top_k recommended_items sorted(item_scores.items(), keylambda x: x[1], reverseTrue)[:top_k] return [item[0] for item in recommended_items] def _cold_start_recommendation(self, top_k10): 冷启动处理返回最热门的商品 from django.db.models import Count hot_products UserBehavior.objects.values(product_id).annotate( countCount(id) ).order_by(-count)[:top_k] return [item[product_id] for item in hot_products]6.3 封装推荐服务在rec_engine/services.py中封装一个更易用的服务层可以加入缓存如Redis来存储用户相似度矩阵避免每次请求都重新计算。# rec_engine/services.py from django.core.cache import cache from .algorithms import UserBasedCF import pickle import hashlib class RecommendationService: CACHE_KEY_PREFIX rec_sim_matrix_ def __init__(self): self.cf_engine UserBasedCF() def get_user_recommendations(self, user_id, top_k10, force_updateFalse): 获取用户推荐列表 Args: force_update: 强制重新计算相似度矩阵例如每天离线更新一次 cache_key self.CACHE_KEY_PREFIX hashlib.md5(str(user_id).encode()).hexdigest()[:8] # 如果强制更新或缓存不存在则重新计算 if force_update or cache.get(cache_key) is None: print(fCalculating similarity matrix for recommendations...) self.cf_engine.calculate_similarity() # 将相似度矩阵和用户-物品矩阵序列化缓存示例实际可能只缓存结果 # 注意大矩阵缓存可能效率低这里缓存推荐结果更合适 cache.set(cache_key, True, timeout86400) # 缓存24小时 else: # 确保引擎已加载数据 if self.cf_engine.user_sim_df is None: self.cf_engine.prepare_data() # 这里可以只从缓存加载矩阵但为简化我们假设缓存只是标志 # 实际生产环境相似度矩阵应存储在数据库或分布式缓存中 pass # 获取推荐结果 recommendations self.cf_engine.recommend(user_id, top_ktop_k) return recommendations def update_similarity_matrix(self): 离线任务更新所有用户的相似度矩阵存入缓存或数据库 self.cf_engine.calculate_similarity() # 这里可以将self.cf_engine.user_sim_df存储到数据库或Redis # 例如存储每个用户的Top-N相似用户列表 # 为简单起见我们只设置一个缓存标志 cache.set(similarity_matrix_updated, True, timeoutNone) print(Similarity matrix updated.)7. 创建Django API提供推荐服务现在我们通过Django REST Framework (DRF) 创建API让前端或其他服务能获取推荐结果。7.1 创建序列化器在rec_engine目录下创建serializers.py# rec_engine/serializers.py from rest_framework import serializers from core.models import Product class ProductRecommendationSerializer(serializers.ModelSerializer): 用于推荐结果的产品序列化器 class Meta: model Product fields [id, title, price, category, image_url]7.2 创建API视图修改rec_engine/views.py# rec_engine/views.py from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from django.shortcuts import get_object_or_404 from core.models import Product, User from .services import RecommendationService from .serializers import ProductRecommendationSerializer class UserRecommendationAPIView(APIView): 获取用户的个性化推荐商品列表 GET /api/recommendations/{user_id}?top_k10 def get(self, request, user_id, formatNone): # 验证用户是否存在 user get_object_or_404(User, iduser_id) # 获取请求参数 top_k request.query_params.get(top_k, 10) try: top_k int(top_k) if top_k 0 or top_k 100: top_k 10 except ValueError: top_k 10 # 获取推荐服务实例 rec_service RecommendationService() # 获取推荐的商品ID列表 recommended_product_ids rec_service.get_user_recommendations(user_id, top_ktop_k) # 查询商品详细信息 recommended_products Product.objects.filter(id__inrecommended_product_ids) # 保持推荐顺序注意filter的in查询不保证顺序需要手动排序 id_to_product {p.id: p for p in recommended_products} ordered_products [id_to_product[pid] for pid in recommended_product_ids if pid in id_to_product] # 序列化返回 serializer ProductRecommendationSerializer(ordered_products, manyTrue) return Response({ user_id: user_id, username: user.username, recommendations: serializer.data, count: len(serializer.data) }) class ColdStartRecommendationAPIView(APIView): 冷启动推荐返回热门商品或随机商品 GET /api/recommendations/hot?top_k10 def get(self, request, formatNone): from django.db.models import Count top_k request.query_params.get(top_k, 10) try: top_k int(top_k) except ValueError: top_k 10 # 按行为次数排序获取热门商品 hot_products Product.objects.annotate( behavior_countCount(behaviors) ).order_by(-behavior_count)[:top_k] serializer ProductRecommendationSerializer(hot_products, manyTrue) return Response({ type: hot, recommendations: serializer.data })7.3 配置URL路由在rec_engine目录下创建urls.py# rec_engine/urls.py from django.urls import path from .views import UserRecommendationAPIView, ColdStartRecommendationAPIView urlpatterns [ path(recommendations/int:user_id/, UserRecommendationAPIView.as_view(), nameuser_recommendations), path(recommendations/hot/, ColdStartRecommendationAPIView.as_view(), namehot_recommendations), ]在项目根urls.py中引入# recommender_system/urls.py from django.contrib import admin from django.urls import path, include urlpatterns [ path(admin/, admin.site.urls), path(api/, include(rec_engine.urls)), ]8. 生成模拟用户行为数据并测试API算法和API准备好了但数据库里还没有用户行为数据。我们需要创建一个管理命令来生成模拟行为数据。8.1 创建Django管理命令在core/management/commands目录下创建generate_behavior.py# core/management/commands/generate_behavior.py import random from django.core.management.base import BaseCommand from django.contrib.auth import get_user_model from core.models import Product, UserBehavior User get_user_model() class Command(BaseCommand): help Generate simulated user behavior data for testing recommendations def add_arguments(self, parser): parser.add_argument( --num_users, typeint, default50, helpNumber of users to generate behaviors for ) parser.add_argument( --behaviors_per_user, typeint, default30, helpAverage number of behaviors per user ) def handle(self, *args, **options): num_users options[num_users] behaviors_per_user options[behaviors_per_user] users list(User.objects.all()[:num_users]) or [User.objects.create(usernameftest_user_{i}) for i in range(num_users)] products list(Product.objects.all()) if not products: self.stdout.write(self.style.ERROR(No products found. Please run scrapy spider first.)) return behavior_types [view, cart, buy, rate] behavior_weights {view: 0.6, cart: 0.2, buy: 0.1, rate: 0.1} # 行为概率分布 behaviors_created 0 for user in users: # 每个用户随机选择一些商品进行行为 num_behaviors random.randint(int(behaviors_per_user*0.5), int(behaviors_per_user*1.5)) selected_products random.sample(products, min(num_behaviors, len(products))) for product in selected_products: # 根据权重随机选择行为类型 b_type random.choices(behavior_types, weights[behavior_weights[t] for t in behavior_types])[0] if b_type rate: value random.randint(1, 5) # 1-5星评分 else: value 1.0 # 创建或更新行为记录 UserBehavior.objects.update_or_create( useruser, productproduct, behavior_typeb_type, defaults{value: value} ) behaviors_created 1 self.stdout.write(self.style.SUCCESS(fSuccessfully created/updated {behaviors_created} behavior records for {len(users)} users.))8.2 运行命令生成数据并测试# 首先确保有用户Django自带admin用户或创建一些测试用户 python manage.py createsuperuser # 按提示创建管理员用户 # 生成模拟行为数据 python manage.py generate_behavior --num_users30 --behaviors_per_user208.3 启动服务并测试API# 启动Django开发服务器 python manage.py runserver打开浏览器或使用curl、Postman测试API获取用户推荐GET http://127.0.0.1:8000/api/recommendations/1/?top_k5(假设用户ID 1存在)获取热门推荐GET http://127.0.0.1:8000/api/recommendations/hot/?top_k5你应该能收到JSON格式的推荐商品列表。9. 性能优化与扩展当数据量变大时我们上面实现的是单机内存版协同过滤。当用户和商品数量达到十万、百万级时user_item_matrix会变得巨大无法放入内存计算相似度矩阵O(n^2)的复杂度也无法接受。这时就需要引入大数据技术。9.1 架构升级离线计算 在线服务这是工业界常见做法离线计算层使用Spark在Hadoop YARN或独立集群上处理全量历史行为日志计算用户相似度矩阵。这个过程可能每天或每小时运行一次。结果存储将计算结果每个用户的Top-N相似用户列表或用户-物品评分预测矩阵存储到高效的查询系统中如Redis缓存、HBase或MySQL。在线服务层我们的Django服务基本不变。当收到推荐请求时从存储中查询该用户的相似用户或预计算的推荐结果快速返回。实时行为可以用于轻量级的实时更新如实时加权。9.2 示例Spark离线计算用户相似度假设我们已将用户行为日志存储在HDFS上格式user_id, product_id, rating, timestamp以下是一个PySpark作业的伪代码思路# 这是一个概念性示例非可执行完整代码 from pyspark.sql import SparkSession from pyspark.mllib.recommendation import ALS, Rating from pyspark.sql.functions import col import pandas as pd # 初始化Spark spark SparkSession.builder.appName(UserSimilarityCF).getOrCreate() # 1. 从HDFS加载行为数据 behavior_df spark.read.parquet(hdfs://path/to/user_behavior/) # 2. 转换为Rating对象 (用户ID, 商品ID, 评分) ratings_rdd behavior_df.rdd.map(lambda row: Rating(row[user_id], row[product_id], row[rating])) # 3. 使用ALS交替最小二乘法一种矩阵分解方法训练模型 # ALS比直接计算全量用户相似度更高效能处理稀疏矩阵 rank 10 # 隐向量维度 iterations 10 model ALS.train(ratings_rdd, rank, iterations) # 4. 获取所有用户的隐向量userFeatures user_features model.userFeatures() # RDD of (user_id, array_of_features) # 5. 计算用户相似度例如计算目标用户与所有用户的余弦相似度 def cosine_similarity(vec1, vec2): import numpy as np return float(np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))) # 广播目标用户向量计算与所有用户的相似度 target_user_id 123 target_vector user_features.lookup(target_user_id)[0] target_vector_bc spark.sparkContext.broadcast(target_vector) similarities_rdd user_features.map( lambda (uid, vec): (uid, cosine_similarity(vec, target_vector_bc.value)) ).filter(lambda (uid, sim): uid ! target_user_id).sortBy(lambda x: x[1], ascendingFalse) # 6. 取Top-K相似用户存入数据库如MySQL供Django查询 top_similar_users similarities_rdd.take(100) # 将top_similar_users写入MySQL表 user_similarity然后Django的RecommendationService可以从user_similarity表查询目标用户的相似用户再聚合推荐商品避免了内存中的大规模矩阵运算。9.3 引入“大模型”进行特征增强“大模型”如BERT可以用于提升推荐系统的效果应用点在Product模型中我们预留了feature_vector字段。你可以使用Sentence-BERT为商品标题和描述生成文本向量。流程离线处理所有商品描述通过BERT模型生成向量存入feature_vector。在计算商品相似度用于基于物品的协同过滤或用户兴趣向量时可以将文本向量作为特征之一与行为向量融合。注意这属于特征工程能提升精度但会增加系统复杂度。对于入门项目基于行为的协同过滤已足够。10. 常见问题与排查思路问题现象可能原因排查方式解决方案运行scrapy crawl时报DJANGO_SETTINGS_MODULE错误Django环境未正确配置在Scrapy项目中检查pipelines.py中sys.path添加和os.environ.setdefault确保路径正确或在Scrapy项目的settings.py中配置DJANGO_SETTINGS_MODULE推荐API返回空列表或商品数少于top_k1. 用户行为数据太少或太稀疏2. 目标用户是新用户冷启动3. 算法中的neighbor_k参数太小1. 检查数据库UserBehavior表记录数2. 检查目标用户ID是否存在及是否有行为3. 打印算法中间变量相似用户数、候选商品数1. 增加模拟行为数据量 (generate_behavior)2. 实现并触发冷启动推荐逻辑3. 调整neighbor_k参数或引入物品相似度作为补充计算相似度矩阵时内存溢出用户或商品数量过多矩阵太大监控程序内存使用打印矩阵形状self.user_item_matrix.shape1. 过滤掉行为过少的用户和商品2. 采用稀疏矩阵存储 (scipy.sparse)3. 迁移到Spark进行分布式计算Django Admin中看不到core应用下的模型未在admin.py中注册模型或注册错误检查core/admin.py文件确认使用了admin.register或admin.site.register正确注册模型并确保应用在settings.INSTALLED_APPS中API请求返回404URL配置错误或视图未正确导入检查recommender_system/urls.py和rec_engine/urls.py的路径确保URL模式拼写正确视图类已导入生成模拟行为数据时外键错误用户或商品不存在检查generate_behavior命令中获取用户和商品的查询逻辑确保数据库中存在用户和商品记录或修改命令先创建测试数据11. 最佳实践与项目进阶方向11.1 工程化最佳实践配置分离将数据库连接、缓存设置、Spark集群地址等敏感信息放入环境变量或settings/local.py不要硬编码。日志记录在关键步骤如算法计算、API调用添加日志便于监控和调试。异常处理在API视图和算法模块中做好异常捕获返回友好的错误信息避免服务崩溃。缓存策略使用Redis等缓存推荐结果、用户相似度列表。对于新行为可设置较短的缓存过期时间以实现准实时更新。代码复用将算法模块设计为可插拔的便于未来替换为基于物品的协同过滤、矩阵分解ALS或深度学习模型。11.2 推荐系统进阶方向多策略融合混合推荐结合协同过滤、基于内容的推荐利用商品类别、文本特征和热门推荐。加权打分不同策略产生候选集后用一个统一的排序模型如逻辑回归、GBDT、深度神经网络进行精排。实时推荐使用Kafka、Flink等流处理框架实时处理用户点击、搜索行为实时更新用户兴趣向量实现秒级推荐更新。探索与利用引入Bandit算法如UCB、Thompson Sampling在推荐已知兴趣商品利用和探索用户可能的新兴趣探索之间取得平衡。评估体系离线评估划分训练集/测试集计算准确率、召回率、覆盖率、新颖度。A/B测试在线对比不同推荐策略的实际业务指标点击率、转化率、停留时长。11.3 部署与监控容器化使用Docker将Django应用、Redis、MySQL等服务容器化用Docker Compose编排便于部署。异步任务将耗时的相似度矩阵计算任务放入Celery等异步队列由后台Worker执行不阻塞Web请求。监控告警对API响应时间、推荐列表覆盖率、缓存命中率等关键指标进行监控。通过这个项目你不仅实现了一个可运行的推荐系统原型更重要的是掌握了从数据采集、存储、算法实现、服务封装到性能扩展的完整思维链条。当数据量增长时你也清楚地知道该从哪个环节计算、存储、缓存入手进行优化和架构升级。这才是应对“Hadoop”、“Spark”、“大模型”这些技术名词的正确姿势——先理解核心问题再用合适的工具去解决它。 Python构建电商推荐系统：从协同过滤到大数据架构实战
你是不是也遇到过这样的问题想用Python做一个电商推荐系统网上搜到的教程要么只讲协同过滤算法要么只教Django框架要么单独讲Scrapy爬虫但就是没人告诉你如何把这些技术栈真正串联起来构建一个完整的、可运行的推荐系统更让人困惑的是当你想引入“大数据”或“大模型”这些听起来很酷的概念时发现资料要么过于理论化要么工程实现复杂到让人望而却步。Hadoop、Spark这些词经常和推荐系统一起出现但它们在你的项目里到底扮演什么角色是必须的吗这篇文章要解决的核心问题就是如何用Python技术栈从零开始构建一个结构清晰、可扩展的电商推荐系统原型并理清何时需要引入Hadoop/Spark这类大数据组件以及“大模型”在当前语境下的真实含义。我将带你完成一个完整的项目实践涵盖数据采集Scrapy、后端服务Django、核心算法协同过滤、以及数据存储与处理的可扩展思路。你会发现一个能跑起来的推荐系统其核心在于清晰的数据流设计和适度的技术选型而不是盲目堆砌“高大上”的技术名词。读完本文你将能理解电商推荐系统的基本架构和数据流。使用Scrapy爬取模拟商品数据并结构化存储。使用Django搭建具备用户、商品、评分行为管理功能的后端API。实现基于用户的协同过滤算法并提供实时推荐接口。掌握如何设计系统以便在数据量增大时平滑地将核心计算如相似度矩阵迁移到Spark等大数据平台进行计算而Django服务只负责结果服务和实时逻辑。澄清“大模型”在推荐系统中的常见误解与应用边界。1. 为什么“Python电商推荐系统”值得你投入时间推荐系统早已不是大厂的专利它已成为提升任何带有内容或商品展示的应用用户粘性和转化率的关键技术。对于Python开发者而言这是一个绝佳的、能将数据爬取、Web开发、算法应用、数据工程等多个技能点串联起来的综合项目。这个项目的真正价值在于“贯通”。很多教程止步于算法理论或单一框架使用导致学习者无法形成系统观。我们将构建的不仅仅是一个算法Demo而是一个具备完整数据流水线的微系统数据从哪来- 用Scrapy爬虫模拟避免法律风险用公开数据集或模拟数据。数据怎么存、怎么管- 用Django的ORM和Admin管理商品、用户、行为数据。算法怎么集成和提供服务- 在Django中实现协同过滤算法并通过API暴露推荐结果。何时考虑“大数据”技术- 当用户-物品矩阵巨大内存计算如Pandas成为瓶颈时我们将设计一个方案将耗时的离线计算如全局用户相似度交给SparkDjango只使用计算结果或处理实时轻量计算。关于“大模型”在当前2023-2024年的推荐系统领域它通常指用于特征提取如用BERT提取商品描述文本特征或排序阶段的深度模型而不是直接替代协同过滤等召回算法。对于入门和中级项目基于协同过滤的召回基于逻辑回归/深度模型的精排是更实际的架构。本文重点在召回层协同过滤的实现但会指出引入“大模型”进行特征增强的可能位置。2. 核心概念与项目架构设计在动手写代码前必须理清几个核心概念和它们在我们系统里的角色。协同过滤算法核心思想是“物以类聚人以群分”。基于用户的协同过滤找到与目标用户兴趣相似的其他用户将这些用户喜欢而目标用户未接触过的物品推荐给他。关键在于计算用户之间的相似度如余弦相似度、皮尔逊相关系数。基于物品的协同过滤找到与目标用户历史喜欢物品相似的其他物品进行推荐。关键在于计算物品之间的相似度。矩阵分解一种更高级的方法将庞大的用户-物品评分矩阵分解为低维的用户隐向量和物品隐向量矩阵用内积预测评分。它属于协同过滤的范畴能更好地处理稀疏矩阵。我们的系统架构 我们将采用基于用户的协同过滤作为核心算法因为它更直观且便于解释“找到相似用户”这一逻辑。整个系统的数据流如下[Scrapy爬虫] - (爬取/模拟商品数据) - [MySQL/PostgreSQL数据库] | v [Django后台] - (管理、展示、记录用户行为) - [用户访问前端] | v [协同过滤算法模块] | v [推荐结果API] - [返回给前端]技术栈选型说明Django作为Web框架和业务逻辑的核心。其强大的ORM、Admin后台、路由和视图系统能快速搭建数据管理和API服务。Scrapy用于初始数据构建。在实际项目中数据可能来自业务数据库、日志文件等。这里用Scrapy模拟从电商网站抓取商品信息的过程重点是学习如何构建结构化的数据管道。协同过滤算法使用Python的pandas和numpy在内存中实现。这对于万级用户、十万级物品以内的数据量是可行的。Hadoop/Spark它们不是本项目初始必需的。它们是“扩展选项”。当用户行为数据达到千万、亿级用户-物品矩阵无法放入单机内存时就需要用Spark的分布式计算能力来离线计算用户相似度矩阵然后将计算结果例如每个用户的Top-N相似用户列表存储到数据库或缓存中供Django API查询。本文会讲解这种“离线计算在线服务”的架构思想。大模型如前所述是“增强选项”。例如可以用开源的Sentence-BERT模型为商品标题和描述生成向量作为物品的补充特征融入相似度计算。这属于特征工程层面不影响主流程。3. 环境准备与项目初始化我们使用Python 3.8版本进行开发。建议使用虚拟环境如venv或conda隔离项目依赖。3.1 创建项目目录与虚拟环境# 创建项目目录 mkdir ecommerce_recommender cd ecommerce_recommender # 创建并激活虚拟环境 (以venv为例) python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 安装核心依赖创建requirements.txt文件内容如下Django4.2.0 djangorestframework3.14.0 pandas2.0.3 numpy1.24.3 scrapy2.9.0 scikit-learn1.3.0 # 用于计算余弦相似度 mysqlclient2.2.0 # 如果使用MySQL否则使用psycopg2-binary for PostgreSQL然后安装pip install -r requirements.txt3.3 初始化Django项目和应用# 创建Django项目 django-admin startproject recommender_system . # 创建核心应用 python manage.py startapp core # 创建用于推荐算法的应用 python manage.py startapp rec_engine项目结构预览ecommerce_recommender/ ├── venv/ ├── recommender_system/ # Django项目配置目录 │ ├── __init__.py │ ├── settings.py │ ├── urls.py │ └── wsgi.py ├── core/ # 核心应用用户、商品、行为模型 │ ├── migrations/ │ ├── __init__.py │ ├── admin.py │ ├── apps.py │ ├── models.py │ ├── tests.py │ └── views.py ├── rec_engine/ # 推荐引擎应用算法实现与API │ ├── __init__.py │ ├── apps.py │ ├── algorithms.py # 协同过滤算法实现 │ ├── services.py # 推荐服务封装 │ └── views.py ├── scrapy_project/ # Scrapy爬虫项目后续创建 ├── requirements.txt └── manage.py3.4 配置数据库在recommender_system/settings.py中配置数据库这里以MySQL为例# settings.py DATABASES { default: { ENGINE: django.db.backends.mysql, NAME: recommender_db, # 数据库名 USER: your_username, PASSWORD: your_password, HOST: localhost, PORT: 3306, } } # 将应用添加到INSTALLED_APPS INSTALLED_APPS [ django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, rest_framework, # DRF core, # 核心应用 rec_engine, # 推荐引擎应用 ]运行迁移创建基础表python manage.py migrate4. 构建数据模型用户、商品与行为推荐系统的基石是数据。我们在core/models.py中定义三个核心模型。4.1 定义模型# core/models.py from django.db import models from django.contrib.auth.models import AbstractUser # 扩展Django自带的用户模型便于后续添加更多字段 class User(AbstractUser): # 可以添加额外字段如年龄、性别用于混合推荐 # age models.IntegerField(nullTrue, blankTrue) # gender models.CharField(max_length10, choices((M,Male),(F,Female)), nullTrue, blankTrue) class Meta: db_table core_user class Product(models.Model): 商品模型 # 基础信息 title models.CharField(max_length200, verbose_name商品标题) description models.TextField(verbose_name商品描述, blankTrue) price models.DecimalField(max_digits10, decimal_places2, verbose_name价格) category models.CharField(max_length100, verbose_name类别, db_indexTrue) # 加索引便于过滤 image_url models.URLField(verbose_name图片链接, blankTrue) # 用于协同过滤的隐式/显式特征可后续用模型生成 # feature_vector models.JSONField(nullTrue, blankTrue) # 例如从大模型提取的向量 created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table core_product verbose_name 商品 verbose_name_plural 商品 def __str__(self): return self.title class UserBehavior(models.Model): 用户行为记录评分/点击/购买等 BEHAVIOR_TYPES ( (view, 浏览), (cart, 加入购物车), (buy, 购买), (rate, 评分), ) user models.ForeignKey(User, on_deletemodels.CASCADE, related_namebehaviors, verbose_name用户) product models.ForeignKey(Product, on_deletemodels.CASCADE, related_namebehaviors, verbose_name商品) behavior_type models.CharField(max_length10, choicesBEHAVIOR_TYPES, verbose_name行为类型) # 对于评分行为value存储分数如1-5对于点击/购买可以存储1或次数 value models.FloatField(default1.0, verbose_name行为值) timestamp models.DateTimeField(auto_now_addTrue, verbose_name发生时间) class Meta: db_table core_user_behavior verbose_name 用户行为 verbose_name_plural 用户行为 # 一个用户对同一个商品同一种行为只记录最后一次或可累计这里用唯一约束示例 unique_together [user, product, behavior_type] def __str__(self): return f{self.user.username} - {self.product.title} - {self.behavior_type}4.2 注册模型到Admin并迁移在core/admin.py中注册模型方便后台管理# core/admin.py from django.contrib import admin from .models import Product, UserBehavior admin.register(Product) class ProductAdmin(admin.ModelAdmin): list_display (id, title, category, price) search_fields (title, category) list_filter (category,) admin.register(UserBehavior) class UserBehaviorAdmin(admin.ModelAdmin): list_display (user, product, behavior_type, value, timestamp) list_filter (behavior_type, timestamp) search_fields (user__username, product__title)运行命令创建数据表python manage.py makemigrations core python manage.py migrate core5. 使用Scrapy爬取模拟商品数据我们不会爬取真实电商网站避免法律风险而是创建一个Scrapy项目来生成模拟数据并学习如何将数据导入Django数据库。5.1 创建Scrapy项目与爬虫# 在项目根目录下 scrapy startproject scrapy_project cd scrapy_project scrapy genspider product_spider example.com5.2 编写模拟数据爬虫修改scrapy_project/spiders/product_spider.py# scrapy_project/spiders/product_spider.py import scrapy import random from faker import Faker # 用于生成假数据 class ProductSpider(scrapy.Spider): name product_spider allowed_domains [example.com] # 仅作占位实际不访问 start_urls [http://example.com/] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.fake Faker() self.categories [电子产品, 图书, 服装, 家居, 食品, 美妆] def parse(self, response): 生成模拟商品数据 # 假设我们需要生成1000个商品 for i in range(1000): product { title: self.fake.catch_phrase(), description: self.fake.text(max_nb_chars200), price: round(random.uniform(10, 2000), 2), category: random.choice(self.categories), image_url: self.fake.image_url(), external_id: fprod_{i:06d}, # 模拟外部ID } yield product5.3 编写Item Pipeline导入Django数据库关键步骤在Scrapy的Pipeline中调用Django的ORM来保存数据。首先需要在Scrapy设置中激活Django环境。 修改scrapy_project/pipelines.py# scrapy_project/pipelines.py import sys import os # 将Django项目根目录添加到Python路径并设置Django环境 sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ..)) os.environ.setdefault(DJANGO_SETTINGS_MODULE, recommender_system.settings) import django django.setup() from core.models import Product class DjangoProductPipeline: def process_item(self, item, spider): # 避免重复导入根据external_id或title判断 product, created Product.objects.update_or_create( titleitem[title], defaults{ description: item.get(description, ), price: item[price], category: item[category], image_url: item.get(image_url, ), } ) spider.logger.info(f{Created if created else Updated} product: {product.title}) return item在scrapy_project/settings.py中启用Pipeline并降低日志级别# scrapy_project/settings.py ITEM_PIPELINES { scrapy_project.pipelines.DjangoProductPipeline: 300, } LOG_LEVEL INFO5.4 运行爬虫并导入数据# 在scrapy_project目录下 scrapy crawl product_spider -o products.json运行后检查Django Admin后台 (http://127.0.0.1:8000/admin/core/product/)应该能看到爬取的1000个商品。6. 实现协同过滤推荐算法这是系统的核心。我们在rec_engine/algorithms.py中实现基于用户的协同过滤。6.1 算法原理与步骤构建用户-物品评分矩阵行是用户列是商品值是用户对商品的行为评分如浏览1购买5评分评分值。计算用户相似度使用余弦相似度或皮尔逊相关系数计算用户向量之间的相似度。寻找最近邻为目标用户找出最相似的K个用户。生成推荐聚合最近邻用户喜欢的物品剔除目标用户已接触过的按预估兴趣度排序。6.2 算法实现代码# rec_engine/algorithms.py import numpy as np import pandas as pd from sklearn.metrics.pairwise import cosine_similarity from django.db.models import Count, Avg from core.models import UserBehavior, Product, User class UserBasedCF: 基于用户的协同过滤推荐 def __init__(self): self.user_sim_matrix None # 用户相似度矩阵 self.user_item_matrix None # 用户-物品评分矩阵 self.users None self.products None def prepare_data(self): 从数据库准备数据构建用户-物品矩阵 # 获取所有用户行为计算平均评分或行为权重 behaviors UserBehavior.objects.filter(behavior_typerate).values(user_id, product_id, value) # 如果没有评分数据可以用其他行为如购买模拟这里假设有评分 if not behaviors: # 示例如果没有评分用浏览和购买行为生成模拟评分 behaviors UserBehavior.objects.all().values(user_id, product_id, behavior_type) # 行为权重映射 weight_map {view: 1, cart: 2, buy: 5} behaviors [ {user_id: b[user_id], product_id: b[product_id], value: weight_map.get(b[behavior_type], 1)} for b in behaviors ] df pd.DataFrame(list(behaviors)) if df.empty: return False # 创建用户-物品评分矩阵 (稀疏矩阵未评分为0或NaN) self.user_item_matrix df.pivot_table( indexuser_id, columnsproduct_id, valuesvalue, aggfuncmean, # 同一用户对同一商品多次行为取平均 fill_value0 ) self.users self.user_item_matrix.index.tolist() self.products self.user_item_matrix.columns.tolist() return True def calculate_similarity(self): 计算用户之间的余弦相似度 if self.user_item_matrix is None: if not self.prepare_data(): raise ValueError(No data available to calculate similarity.) # 计算余弦相似度 # 注意这里计算的是稠密矩阵的相似度用户/物品很多时会内存爆炸。这是单机版的局限。 self.user_sim_matrix cosine_similarity(self.user_item_matrix.values) # 将相似度矩阵转换为DataFrame方便索引 self.user_sim_df pd.DataFrame( self.user_sim_matrix, indexself.users, columnsself.users ) def recommend(self, user_id, top_k10, neighbor_k20): 为目标用户推荐商品 Args: user_id: 目标用户ID top_k: 返回推荐商品的数量 neighbor_k: 使用的最近邻用户数量 Returns: list: 推荐商品ID列表 if self.user_sim_df is None: self.calculate_similarity() if user_id not in self.users: # 新用户冷启动问题返回热门商品或基于类别的推荐 return self._cold_start_recommendation(top_k) # 1. 获取目标用户的相似用户排除自己 sim_users self.user_sim_df[user_id].sort_values(ascendingFalse).iloc[1:neighbor_k1] # 2. 获取这些相似用户有过行为但目标用户没有的商品 target_user_items set(self.user_item_matrix.loc[user_id][self.user_item_matrix.loc[user_id] 0].index) # 计算候选商品的推荐分数 item_scores {} for sim_user_id, similarity in sim_users.items(): # 相似用户评分过的商品 sim_user_items self.user_item_matrix.loc[sim_user_id] for item_id in sim_user_items[sim_user_items 0].index: if item_id not in target_user_items: # 累加 (相似度 * 评分) item_scores[item_id] item_scores.get(item_id, 0) similarity * sim_user_items[item_id] # 3. 按分数排序返回top_k recommended_items sorted(item_scores.items(), keylambda x: x[1], reverseTrue)[:top_k] return [item[0] for item in recommended_items] def _cold_start_recommendation(self, top_k10): 冷启动处理返回最热门的商品 from django.db.models import Count hot_products UserBehavior.objects.values(product_id).annotate( countCount(id) ).order_by(-count)[:top_k] return [item[product_id] for item in hot_products]6.3 封装推荐服务在rec_engine/services.py中封装一个更易用的服务层可以加入缓存如Redis来存储用户相似度矩阵避免每次请求都重新计算。# rec_engine/services.py from django.core.cache import cache from .algorithms import UserBasedCF import pickle import hashlib class RecommendationService: CACHE_KEY_PREFIX rec_sim_matrix_ def __init__(self): self.cf_engine UserBasedCF() def get_user_recommendations(self, user_id, top_k10, force_updateFalse): 获取用户推荐列表 Args: force_update: 强制重新计算相似度矩阵例如每天离线更新一次 cache_key self.CACHE_KEY_PREFIX hashlib.md5(str(user_id).encode()).hexdigest()[:8] # 如果强制更新或缓存不存在则重新计算 if force_update or cache.get(cache_key) is None: print(fCalculating similarity matrix for recommendations...) self.cf_engine.calculate_similarity() # 将相似度矩阵和用户-物品矩阵序列化缓存示例实际可能只缓存结果 # 注意大矩阵缓存可能效率低这里缓存推荐结果更合适 cache.set(cache_key, True, timeout86400) # 缓存24小时 else: # 确保引擎已加载数据 if self.cf_engine.user_sim_df is None: self.cf_engine.prepare_data() # 这里可以只从缓存加载矩阵但为简化我们假设缓存只是标志 # 实际生产环境相似度矩阵应存储在数据库或分布式缓存中 pass # 获取推荐结果 recommendations self.cf_engine.recommend(user_id, top_ktop_k) return recommendations def update_similarity_matrix(self): 离线任务更新所有用户的相似度矩阵存入缓存或数据库 self.cf_engine.calculate_similarity() # 这里可以将self.cf_engine.user_sim_df存储到数据库或Redis # 例如存储每个用户的Top-N相似用户列表 # 为简单起见我们只设置一个缓存标志 cache.set(similarity_matrix_updated, True, timeoutNone) print(Similarity matrix updated.)7. 创建Django API提供推荐服务现在我们通过Django REST Framework (DRF) 创建API让前端或其他服务能获取推荐结果。7.1 创建序列化器在rec_engine目录下创建serializers.py# rec_engine/serializers.py from rest_framework import serializers from core.models import Product class ProductRecommendationSerializer(serializers.ModelSerializer): 用于推荐结果的产品序列化器 class Meta: model Product fields [id, title, price, category, image_url]7.2 创建API视图修改rec_engine/views.py# rec_engine/views.py from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from django.shortcuts import get_object_or_404 from core.models import Product, User from .services import RecommendationService from .serializers import ProductRecommendationSerializer class UserRecommendationAPIView(APIView): 获取用户的个性化推荐商品列表 GET /api/recommendations/{user_id}?top_k10 def get(self, request, user_id, formatNone): # 验证用户是否存在 user get_object_or_404(User, iduser_id) # 获取请求参数 top_k request.query_params.get(top_k, 10) try: top_k int(top_k) if top_k 0 or top_k 100: top_k 10 except ValueError: top_k 10 # 获取推荐服务实例 rec_service RecommendationService() # 获取推荐的商品ID列表 recommended_product_ids rec_service.get_user_recommendations(user_id, top_ktop_k) # 查询商品详细信息 recommended_products Product.objects.filter(id__inrecommended_product_ids) # 保持推荐顺序注意filter的in查询不保证顺序需要手动排序 id_to_product {p.id: p for p in recommended_products} ordered_products [id_to_product[pid] for pid in recommended_product_ids if pid in id_to_product] # 序列化返回 serializer ProductRecommendationSerializer(ordered_products, manyTrue) return Response({ user_id: user_id, username: user.username, recommendations: serializer.data, count: len(serializer.data) }) class ColdStartRecommendationAPIView(APIView): 冷启动推荐返回热门商品或随机商品 GET /api/recommendations/hot?top_k10 def get(self, request, formatNone): from django.db.models import Count top_k request.query_params.get(top_k, 10) try: top_k int(top_k) except ValueError: top_k 10 # 按行为次数排序获取热门商品 hot_products Product.objects.annotate( behavior_countCount(behaviors) ).order_by(-behavior_count)[:top_k] serializer ProductRecommendationSerializer(hot_products, manyTrue) return Response({ type: hot, recommendations: serializer.data })7.3 配置URL路由在rec_engine目录下创建urls.py# rec_engine/urls.py from django.urls import path from .views import UserRecommendationAPIView, ColdStartRecommendationAPIView urlpatterns [ path(recommendations/int:user_id/, UserRecommendationAPIView.as_view(), nameuser_recommendations), path(recommendations/hot/, ColdStartRecommendationAPIView.as_view(), namehot_recommendations), ]在项目根urls.py中引入# recommender_system/urls.py from django.contrib import admin from django.urls import path, include urlpatterns [ path(admin/, admin.site.urls), path(api/, include(rec_engine.urls)), ]8. 生成模拟用户行为数据并测试API算法和API准备好了但数据库里还没有用户行为数据。我们需要创建一个管理命令来生成模拟行为数据。8.1 创建Django管理命令在core/management/commands目录下创建generate_behavior.py# core/management/commands/generate_behavior.py import random from django.core.management.base import BaseCommand from django.contrib.auth import get_user_model from core.models import Product, UserBehavior User get_user_model() class Command(BaseCommand): help Generate simulated user behavior data for testing recommendations def add_arguments(self, parser): parser.add_argument( --num_users, typeint, default50, helpNumber of users to generate behaviors for ) parser.add_argument( --behaviors_per_user, typeint, default30, helpAverage number of behaviors per user ) def handle(self, *args, **options): num_users options[num_users] behaviors_per_user options[behaviors_per_user] users list(User.objects.all()[:num_users]) or [User.objects.create(usernameftest_user_{i}) for i in range(num_users)] products list(Product.objects.all()) if not products: self.stdout.write(self.style.ERROR(No products found. Please run scrapy spider first.)) return behavior_types [view, cart, buy, rate] behavior_weights {view: 0.6, cart: 0.2, buy: 0.1, rate: 0.1} # 行为概率分布 behaviors_created 0 for user in users: # 每个用户随机选择一些商品进行行为 num_behaviors random.randint(int(behaviors_per_user*0.5), int(behaviors_per_user*1.5)) selected_products random.sample(products, min(num_behaviors, len(products))) for product in selected_products: # 根据权重随机选择行为类型 b_type random.choices(behavior_types, weights[behavior_weights[t] for t in behavior_types])[0] if b_type rate: value random.randint(1, 5) # 1-5星评分 else: value 1.0 # 创建或更新行为记录 UserBehavior.objects.update_or_create( useruser, productproduct, behavior_typeb_type, defaults{value: value} ) behaviors_created 1 self.stdout.write(self.style.SUCCESS(fSuccessfully created/updated {behaviors_created} behavior records for {len(users)} users.))8.2 运行命令生成数据并测试# 首先确保有用户Django自带admin用户或创建一些测试用户 python manage.py createsuperuser # 按提示创建管理员用户 # 生成模拟行为数据 python manage.py generate_behavior --num_users30 --behaviors_per_user208.3 启动服务并测试API# 启动Django开发服务器 python manage.py runserver打开浏览器或使用curl、Postman测试API获取用户推荐GET http://127.0.0.1:8000/api/recommendations/1/?top_k5(假设用户ID 1存在)获取热门推荐GET http://127.0.0.1:8000/api/recommendations/hot/?top_k5你应该能收到JSON格式的推荐商品列表。9. 性能优化与扩展当数据量变大时我们上面实现的是单机内存版协同过滤。当用户和商品数量达到十万、百万级时user_item_matrix会变得巨大无法放入内存计算相似度矩阵O(n^2)的复杂度也无法接受。这时就需要引入大数据技术。9.1 架构升级离线计算 在线服务这是工业界常见做法离线计算层使用Spark在Hadoop YARN或独立集群上处理全量历史行为日志计算用户相似度矩阵。这个过程可能每天或每小时运行一次。结果存储将计算结果每个用户的Top-N相似用户列表或用户-物品评分预测矩阵存储到高效的查询系统中如Redis缓存、HBase或MySQL。在线服务层我们的Django服务基本不变。当收到推荐请求时从存储中查询该用户的相似用户或预计算的推荐结果快速返回。实时行为可以用于轻量级的实时更新如实时加权。9.2 示例Spark离线计算用户相似度假设我们已将用户行为日志存储在HDFS上格式user_id, product_id, rating, timestamp以下是一个PySpark作业的伪代码思路# 这是一个概念性示例非可执行完整代码 from pyspark.sql import SparkSession from pyspark.mllib.recommendation import ALS, Rating from pyspark.sql.functions import col import pandas as pd # 初始化Spark spark SparkSession.builder.appName(UserSimilarityCF).getOrCreate() # 1. 从HDFS加载行为数据 behavior_df spark.read.parquet(hdfs://path/to/user_behavior/) # 2. 转换为Rating对象 (用户ID, 商品ID, 评分) ratings_rdd behavior_df.rdd.map(lambda row: Rating(row[user_id], row[product_id], row[rating])) # 3. 使用ALS交替最小二乘法一种矩阵分解方法训练模型 # ALS比直接计算全量用户相似度更高效能处理稀疏矩阵 rank 10 # 隐向量维度 iterations 10 model ALS.train(ratings_rdd, rank, iterations) # 4. 获取所有用户的隐向量userFeatures user_features model.userFeatures() # RDD of (user_id, array_of_features) # 5. 计算用户相似度例如计算目标用户与所有用户的余弦相似度 def cosine_similarity(vec1, vec2): import numpy as np return float(np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))) # 广播目标用户向量计算与所有用户的相似度 target_user_id 123 target_vector user_features.lookup(target_user_id)[0] target_vector_bc spark.sparkContext.broadcast(target_vector) similarities_rdd user_features.map( lambda (uid, vec): (uid, cosine_similarity(vec, target_vector_bc.value)) ).filter(lambda (uid, sim): uid ! target_user_id).sortBy(lambda x: x[1], ascendingFalse) # 6. 取Top-K相似用户存入数据库如MySQL供Django查询 top_similar_users similarities_rdd.take(100) # 将top_similar_users写入MySQL表 user_similarity然后Django的RecommendationService可以从user_similarity表查询目标用户的相似用户再聚合推荐商品避免了内存中的大规模矩阵运算。9.3 引入“大模型”进行特征增强“大模型”如BERT可以用于提升推荐系统的效果应用点在Product模型中我们预留了feature_vector字段。你可以使用Sentence-BERT为商品标题和描述生成文本向量。流程离线处理所有商品描述通过BERT模型生成向量存入feature_vector。在计算商品相似度用于基于物品的协同过滤或用户兴趣向量时可以将文本向量作为特征之一与行为向量融合。注意这属于特征工程能提升精度但会增加系统复杂度。对于入门项目基于行为的协同过滤已足够。10. 常见问题与排查思路问题现象可能原因排查方式解决方案运行scrapy crawl时报DJANGO_SETTINGS_MODULE错误Django环境未正确配置在Scrapy项目中检查pipelines.py中sys.path添加和os.environ.setdefault确保路径正确或在Scrapy项目的settings.py中配置DJANGO_SETTINGS_MODULE推荐API返回空列表或商品数少于top_k1. 用户行为数据太少或太稀疏2. 目标用户是新用户冷启动3. 算法中的neighbor_k参数太小1. 检查数据库UserBehavior表记录数2. 检查目标用户ID是否存在及是否有行为3. 打印算法中间变量相似用户数、候选商品数1. 增加模拟行为数据量 (generate_behavior)2. 实现并触发冷启动推荐逻辑3. 调整neighbor_k参数或引入物品相似度作为补充计算相似度矩阵时内存溢出用户或商品数量过多矩阵太大监控程序内存使用打印矩阵形状self.user_item_matrix.shape1. 过滤掉行为过少的用户和商品2. 采用稀疏矩阵存储 (scipy.sparse)3. 迁移到Spark进行分布式计算Django Admin中看不到core应用下的模型未在admin.py中注册模型或注册错误检查core/admin.py文件确认使用了admin.register或admin.site.register正确注册模型并确保应用在settings.INSTALLED_APPS中API请求返回404URL配置错误或视图未正确导入检查recommender_system/urls.py和rec_engine/urls.py的路径确保URL模式拼写正确视图类已导入生成模拟行为数据时外键错误用户或商品不存在检查generate_behavior命令中获取用户和商品的查询逻辑确保数据库中存在用户和商品记录或修改命令先创建测试数据11. 最佳实践与项目进阶方向11.1 工程化最佳实践配置分离将数据库连接、缓存设置、Spark集群地址等敏感信息放入环境变量或settings/local.py不要硬编码。日志记录在关键步骤如算法计算、API调用添加日志便于监控和调试。异常处理在API视图和算法模块中做好异常捕获返回友好的错误信息避免服务崩溃。缓存策略使用Redis等缓存推荐结果、用户相似度列表。对于新行为可设置较短的缓存过期时间以实现准实时更新。代码复用将算法模块设计为可插拔的便于未来替换为基于物品的协同过滤、矩阵分解ALS或深度学习模型。11.2 推荐系统进阶方向多策略融合混合推荐结合协同过滤、基于内容的推荐利用商品类别、文本特征和热门推荐。加权打分不同策略产生候选集后用一个统一的排序模型如逻辑回归、GBDT、深度神经网络进行精排。实时推荐使用Kafka、Flink等流处理框架实时处理用户点击、搜索行为实时更新用户兴趣向量实现秒级推荐更新。探索与利用引入Bandit算法如UCB、Thompson Sampling在推荐已知兴趣商品利用和探索用户可能的新兴趣探索之间取得平衡。评估体系离线评估划分训练集/测试集计算准确率、召回率、覆盖率、新颖度。A/B测试在线对比不同推荐策略的实际业务指标点击率、转化率、停留时长。11.3 部署与监控容器化使用Docker将Django应用、Redis、MySQL等服务容器化用Docker Compose编排便于部署。异步任务将耗时的相似度矩阵计算任务放入Celery等异步队列由后台Worker执行不阻塞Web请求。监控告警对API响应时间、推荐列表覆盖率、缓存命中率等关键指标进行监控。通过这个项目你不仅实现了一个可运行的推荐系统原型更重要的是掌握了从数据采集、存储、算法实现、服务封装到性能扩展的完整思维链条。当数据量增长时你也清楚地知道该从哪个环节计算、存储、缓存入手进行优化和架构升级。这才是应对“Hadoop”、“Spark”、“大模型”这些技术名词的正确姿势——先理解核心问题再用合适的工具去解决它。