EcomGPT-7B会员系统开发Django REST框架实战最近在做一个电商项目客户提了个需求想要一个能根据用户历史行为自动推荐商品的会员系统。这听起来挺酷的但实现起来要考虑的东西不少用户数据怎么管、推荐算法怎么接、API接口怎么设计才安全。我琢磨了一下决定用Django REST框架来搭后台再配上专门为电商场景优化的EcomGPT-7B模型来做个性化推荐。这套组合拳打下来效果还真不错今天就跟大家分享一下具体的实现思路和代码。1. 项目准备与环境搭建1.1 技术栈选择先说说我为什么选这些技术Django REST FrameworkPython里做REST API的老牌选手了文档全、社区活跃权限控制、序列化这些功能都封装得很好能省不少事。EcomGPT-7B阿里开源的电商领域大模型在商品理解、评论分析这些电商任务上表现挺不错的关键是专门为电商场景优化过比用通用模型效果要好。PostgreSQL关系型数据库会员系统的数据关系比较复杂用这个比较合适。Redis缓存用户会话和推荐结果提升响应速度。1.2 环境安装咱们从零开始先把需要的包都装上# 创建项目目录 mkdir ecomgpt-membership cd ecomgpt-membership # 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install django djangorestframework pip install django-cors-headers pip install psycopg2-binary redis pip install transformers torch pip install python-dotenv1.3 项目初始化用Django的命令创建项目和应用# 创建Django项目 django-admin startproject membership_backend . # 创建核心应用 python manage.py startapp membership python manage.py startapp recommendations项目结构大概长这样ecomgpt-membership/ ├── membership_backend/ # 项目配置 ├── membership/ # 会员管理应用 ├── recommendations/ # 推荐系统应用 ├── venv/ # 虚拟环境 ├── .env # 环境变量 ├── manage.py └── requirements.txt2. 数据库模型设计会员系统的核心是数据模型设计好了后面开发会轻松很多。2.1 会员相关模型在membership/models.py里定义用户和会员等级from django.contrib.auth.models import AbstractUser from django.db import models class User(AbstractUser): 扩展Django默认用户模型 phone models.CharField(max_length20, uniqueTrue, nullTrue, blankTrue) avatar models.URLField(max_length500, blankTrue) registration_date models.DateTimeField(auto_now_addTrue) last_login_ip models.GenericIPAddressField(nullTrue, blankTrue) # 会员相关字段 membership_level models.ForeignKey(MembershipLevel, on_deletemodels.SET_NULL, nullTrue) points models.IntegerField(default0) total_spent models.DecimalField(max_digits10, decimal_places2, default0) class Meta: db_table users def __str__(self): return f{self.username} ({self.email}) class MembershipLevel(models.Model): 会员等级配置 LEVEL_CHOICES [ (bronze, 青铜会员), (silver, 白银会员), (gold, 黄金会员), (platinum, 铂金会员), (diamond, 钻石会员), ] name models.CharField(max_length50, choicesLEVEL_CHOICES, uniqueTrue) display_name models.CharField(max_length50) min_points models.IntegerField(default0) discount_rate models.DecimalField(max_digits5, decimal_places2, default0) # 折扣率 monthly_free_shipping models.BooleanField(defaultFalse) priority_support models.BooleanField(defaultFalse) created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table membership_levels ordering [min_points] def __str__(self): return self.display_name2.2 用户行为模型为了做个性化推荐我们需要记录用户的各种行为class UserBehavior(models.Model): 用户行为记录 BEHAVIOR_TYPES [ (view, 浏览商品), (click, 点击商品), (add_to_cart, 加入购物车), (purchase, 购买商品), (search, 搜索关键词), (review, 发表评价), ] user models.ForeignKey(User, on_deletemodels.CASCADE, related_namebehaviors) behavior_type models.CharField(max_length20, choicesBEHAVIOR_TYPES) product_id models.CharField(max_length100, blankTrue) # 商品ID product_category models.CharField(max_length100, blankTrue) # 商品类目 search_query models.CharField(max_length200, blankTrue) # 搜索词 rating models.IntegerField(nullTrue, blankTrue) # 评分 duration models.IntegerField(default0) # 停留时长秒 # 行为发生时的上下文 device_type models.CharField(max_length50, blankTrue) browser models.CharField(max_length50, blankTrue) ip_address models.GenericIPAddressField() created_at models.DateTimeField(auto_now_addTrue) class Meta: db_table user_behaviors indexes [ models.Index(fields[user, created_at]), models.Index(fields[behavior_type, created_at]), ] def __str__(self): return f{self.user.username} - {self.behavior_type} at {self.created_at}2.3 推荐相关模型在recommendations/models.py里from django.db import models from membership.models import User class Recommendation(models.Model): 推荐结果存储 user models.ForeignKey(User, on_deletemodels.CASCADE, related_namerecommendations) recommended_products models.JSONField() # 推荐商品列表 recommendation_type models.CharField(max_length50) # 推荐类型personalized, trending, etc. model_version models.CharField(max_length50, defaultecomgpt-7b-v1) # 推荐效果追踪 impressions models.IntegerField(default0) # 展示次数 clicks models.IntegerField(default0) # 点击次数 conversions models.IntegerField(default0) # 转化次数 created_at models.DateTimeField(auto_now_addTrue) expires_at models.DateTimeField() # 推荐结果过期时间 class Meta: db_table recommendations ordering [-created_at] def click_through_rate(self): 计算点击率 if self.impressions 0: return 0 return (self.clicks / self.impressions) * 100 def conversion_rate(self): 计算转化率 if self.clicks 0: return 0 return (self.conversions / self.clicks) * 1003. EcomGPT-7B集成与推荐引擎3.1 模型加载与初始化先创建一个推荐服务类负责加载EcomGPT模型# recommendations/services/recommendation_service.py import json import redis from datetime import datetime, timedelta from typing import List, Dict, Any import torch from transformers import AutoTokenizer, AutoModelForCausalLM class EcomGPTRecommendationService: EcomGPT-7B推荐服务 def __init__(self, model_pathiic/nlp_ecomgpt_multilingual-7B-ecom): self.model_path model_path self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.tokenizer None self.model None self.redis_client redis.Redis(hostlocalhost, port6379, db0) # 缓存配置 self.cache_ttl 3600 # 1小时 def load_model(self): 加载EcomGPT模型 if self.model is None: print(正在加载EcomGPT-7B模型...) self.tokenizer AutoTokenizer.from_pretrained( self.model_path, trust_remote_codeTrue ) self.model AutoModelForCausalLM.from_pretrained( self.model_path, torch_dtypetorch.float16 if torch.cuda.is_available() else torch.float32, device_mapauto if torch.cuda.is_available() else None, trust_remote_codeTrue ) if not torch.cuda.is_available(): self.model self.model.to(self.device) print(模型加载完成) def generate_prompt(self, user_data: Dict[str, Any], behavior_history: List[Dict]) - str: 生成推荐提示词 # 提取用户兴趣标签 interests self._extract_user_interests(behavior_history) prompt_template Below is an instruction that describes a task. Write a response that appropriately completes the request. ### Instruction: 作为电商推荐系统请为以下用户推荐10个最相关的商品 用户信息 - 用户ID: {user_id} - 会员等级: {membership_level} - 历史兴趣: {interests} - 最近浏览: {recent_views} - 购买历史: {purchase_history} 请根据用户的兴趣和浏览历史推荐10个最可能感兴趣的商品。 每个推荐请包含商品ID、商品名称、推荐理由、预估点击率。 ### Response: # 填充模板 prompt prompt_template.format( user_iduser_data.get(id), membership_leveluser_data.get(membership_level, 普通会员), interests, .join(interests[:5]), recent_viewsself._format_recent_views(behavior_history), purchase_historyself._format_purchase_history(behavior_history) ) return prompt def _extract_user_interests(self, behavior_history: List[Dict]) - List[str]: 从行为历史中提取用户兴趣 interests [] category_counts {} for behavior in behavior_history: if behavior[product_category]: category behavior[product_category] category_counts[category] category_counts.get(category, 0) 1 # 按出现频率排序 sorted_categories sorted(category_counts.items(), keylambda x: x[1], reverseTrue) interests [category for category, _ in sorted_categories[:10]] return interests def get_recommendations(self, user_id: int, force_refresh: bool False) - List[Dict]: 获取用户推荐 cache_key frecommendations:user:{user_id} # 检查缓存 if not force_refresh: cached self.redis_client.get(cache_key) if cached: return json.loads(cached) # 加载模型懒加载 self.load_model() # 获取用户数据和行为历史 user_data self._get_user_data(user_id) behavior_history self._get_user_behavior_history(user_id) # 生成提示词 prompt self.generate_prompt(user_data, behavior_history) # 调用模型生成推荐 recommendations self._generate_with_model(prompt) # 缓存结果 self.redis_client.setex( cache_key, self.cache_ttl, json.dumps(recommendations) ) return recommendations def _generate_with_model(self, prompt: str) - List[Dict]: 使用EcomGPT生成推荐 inputs self.tokenizer(prompt, return_tensorspt).to(self.device) with torch.no_grad(): outputs self.model.generate( **inputs, max_new_tokens500, temperature0.7, do_sampleTrue, top_p0.9, pad_token_idself.tokenizer.eos_token_id ) response self.tokenizer.decode(outputs[0], skip_special_tokensTrue) # 解析模型响应 recommendations self._parse_model_response(response) return recommendations def _parse_model_response(self, response: str) - List[Dict]: 解析模型返回的推荐结果 # 这里简化处理实际项目中需要更复杂的解析逻辑 recommendations [] # 示例解析逻辑 lines response.split(\n) for line in lines: if 商品ID: in line and 商品名称: in line: parts line.split(|) if len(parts) 4: rec { product_id: parts[0].replace(商品ID:, ).strip(), product_name: parts[1].replace(商品名称:, ).strip(), reason: parts[2].replace(推荐理由:, ).strip(), estimated_ctr: float(parts[3].replace(预估点击率:, ).replace(%, ).strip()) } recommendations.append(rec) return recommendations[:10] # 最多返回10个推荐3.2 实时推荐API创建推荐视图提供实时推荐接口# recommendations/views.py from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from rest_framework.permissions import IsAuthenticated from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_page from .services.recommendation_service import EcomGPTRecommendationService from .models import Recommendation class PersonalizedRecommendationView(APIView): 个性化推荐接口 permission_classes [IsAuthenticated] def __init__(self): super().__init__() self.recommendation_service EcomGPTRecommendationService() method_decorator(cache_page(60 * 5)) # 缓存5分钟 def get(self, request): 获取个性化推荐 user request.user force_refresh request.GET.get(refresh, false).lower() true try: # 获取推荐结果 recommendations self.recommendation_service.get_recommendations( user.id, force_refreshforce_refresh ) # 记录推荐结果 recommendation_record Recommendation.objects.create( useruser, recommended_productsrecommendations, recommendation_typepersonalized, expires_atdatetime.now() timedelta(hours1) ) # 返回格式化结果 response_data { user_id: user.id, recommendation_id: recommendation_record.id, generated_at: recommendation_record.created_at.isoformat(), expires_at: recommendation_record.expires_at.isoformat(), recommendations: recommendations, metadata: { model_version: ecomgpt-7b-v1, cache_hit: not force_refresh, total_recommendations: len(recommendations) } } return Response(response_data, statusstatus.HTTP_200_OK) except Exception as e: # 降级策略返回热门商品 fallback_recommendations self._get_fallback_recommendations() return Response({ user_id: user.id, recommendations: fallback_recommendations, metadata: { model_version: fallback, error: str(e), note: 使用降级推荐策略 } }, statusstatus.HTTP_200_OK) def _get_fallback_recommendations(self) - List[Dict]: 降级策略返回热门商品 # 这里可以查询数据库中的热门商品 return [ { product_id: fallback_001, product_name: 热门商品A, reason: 当前热门选择, estimated_ctr: 0.05 }, # ... 更多降级推荐 ]4. REST API设计与权限控制4.1 用户认证与权限用Django REST Framework的权限系统# membership/permissions.py from rest_framework import permissions class IsOwnerOrAdmin(permissions.BasePermission): 只允许用户自己或管理员访问 def has_object_permission(self, request, view, obj): # 管理员有所有权限 if request.user.is_staff: return True # 用户只能访问自己的数据 return obj.user request.user class MembershipLevelPermission(permissions.BasePermission): 基于会员等级的权限控制 def has_permission(self, request, view): # 公开接口不需要权限 if view.action in [list, retrieve]: return True # 需要特定会员等级才能访问 required_level getattr(view, required_membership_level, bronze) user_level request.user.membership_level.name if request.user.membership_level else bronze # 会员等级映射到权重 level_weights { bronze: 1, silver: 2, gold: 3, platinum: 4, diamond: 5 } user_weight level_weights.get(user_level, 1) required_weight level_weights.get(required_level, 1) return user_weight required_weight4.2 用户管理API# membership/views.py from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, IsAdminUser from django.contrib.auth import authenticate from django.utils import timezone from .models import User, MembershipLevel, UserBehavior from .serializers import UserSerializer, MembershipLevelSerializer, UserBehaviorSerializer from .permissions import IsOwnerOrAdmin, MembershipLevelPermission class UserViewSet(viewsets.ModelViewSet): 用户管理API queryset User.objects.all() serializer_class UserSerializer permission_classes [IsAuthenticated, IsOwnerOrAdmin] def get_permissions(self): 根据动作调整权限 if self.action create: return [] # 注册不需要认证 elif self.action in [update, partial_update, destroy]: return [IsAuthenticated(), IsOwnerOrAdmin()] return super().get_permissions() action(detailFalse, methods[post]) def register(self, request): 用户注册 serializer self.get_serializer(datarequest.data) if serializer.is_valid(): user serializer.save() # 设置默认会员等级 default_level MembershipLevel.objects.filter(namebronze).first() if default_level: user.membership_level default_level user.save() # 返回token实际项目中应该用JWT return Response({ user: UserSerializer(user).data, message: 注册成功 }, statusstatus.HTTP_201_CREATED) return Response(serializer.errors, statusstatus.HTTP_400_BAD_REQUEST) action(detailFalse, methods[post]) def login(self, request): 用户登录 username request.data.get(username) password request.data.get(password) user authenticate(usernameusername, passwordpassword) if user is not None: # 更新最后登录信息 user.last_login timezone.now() user.last_login_ip self._get_client_ip(request) user.save() return Response({ user: UserSerializer(user).data, message: 登录成功 }) return Response( {error: 用户名或密码错误}, statusstatus.HTTP_401_UNAUTHORIZED ) action(detailTrue, methods[get]) def behaviors(self, request, pkNone): 获取用户行为记录 user self.get_object() behaviors UserBehavior.objects.filter(useruser).order_by(-created_at)[:100] serializer UserBehaviorSerializer(behaviors, manyTrue) return Response(serializer.data) def _get_client_ip(self, request): 获取客户端IP x_forwarded_for request.META.get(HTTP_X_FORWARDED_FOR) if x_forwarded_for: ip x_forwarded_for.split(,)[0] else: ip request.META.get(REMOTE_ADDR) return ip4.3 序列化器定义# membership/serializers.py from rest_framework import serializers from django.contrib.auth.hashers import make_password from .models import User, MembershipLevel, UserBehavior class UserSerializer(serializers.ModelSerializer): 用户序列化器 password serializers.CharField(write_onlyTrue, requiredFalse) membership_level_display serializers.CharField(sourcemembership_level.display_name, read_onlyTrue) class Meta: model User fields [ id, username, email, phone, avatar, membership_level, membership_level_display, points, total_spent, registration_date, password, first_name, last_name ] read_only_fields [registration_date, points, total_spent] def create(self, validated_data): 创建用户时加密密码 if password in validated_data: validated_data[password] make_password(validated_data[password]) return super().create(validated_data) def update(self, instance, validated_data): 更新用户时处理密码 if password in validated_data: validated_data[password] make_password(validated_data[password]) return super().update(instance, validated_data) class MembershipLevelSerializer(serializers.ModelSerializer): 会员等级序列化器 class Meta: model MembershipLevel fields __all__ class UserBehaviorSerializer(serializers.ModelSerializer): 用户行为序列化器 behavior_type_display serializers.CharField(sourceget_behavior_type_display, read_onlyTrue) class Meta: model UserBehavior fields __all__ read_only_fields [created_at]5. API安全与限流5.1 安全中间件配置在settings.py中配置安全相关设置# membership_backend/settings.py # 安全设置 SECURE_SSL_REDIRECT False # 生产环境应该设为True SECURE_HSTS_SECONDS 31536000 # 1年HSTS SECURE_HSTS_INCLUDE_SUBDOMAINS True SECURE_HSTS_PRELOAD True SECURE_CONTENT_TYPE_NOSNIFF True X_FRAME_OPTIONS DENY # CORS设置 CORS_ALLOWED_ORIGINS [ http://localhost:3000, http://127.0.0.1:3000, ] # REST Framework设置 REST_FRAMEWORK { DEFAULT_AUTHENTICATION_CLASSES: [ rest_framework.authentication.SessionAuthentication, rest_framework.authentication.BasicAuthentication, ], DEFAULT_PERMISSION_CLASSES: [ rest_framework.permissions.IsAuthenticated, ], DEFAULT_THROTTLE_CLASSES: [ rest_framework.throttling.AnonRateThrottle, rest_framework.throttling.UserRateThrottle, ], DEFAULT_THROTTLE_RATES: { anon: 100/day, # 匿名用户每天100次 user: 1000/day, # 认证用户每天1000次 recommendation: 60/minute, # 推荐接口每分钟60次 }, DEFAULT_RENDERER_CLASSES: [ rest_framework.renderers.JSONRenderer, ], DEFAULT_PARSER_CLASSES: [ rest_framework.parsers.JSONParser, rest_framework.parsers.FormParser, rest_framework.parsers.MultiPartParser, ], }5.2 自定义限流类# membership/throttles.py from rest_framework.throttling import SimpleRateThrottle class RecommendationThrottle(SimpleRateThrottle): 推荐接口限流 scope recommendation def get_cache_key(self, request, view): if request.user.is_authenticated: ident request.user.pk else: ident self.get_ident(request) return self.cache_format % { scope: self.scope, ident: ident } class MembershipLevelThrottle(SimpleRateThrottle): 基于会员等级的限流 scope membership_level def get_cache_key(self, request, view): if request.user.is_authenticated: # 根据会员等级设置不同的限流 level request.user.membership_level.name if request.user.membership_level else bronze ident f{request.user.pk}_{level} else: ident self.get_ident(request) return self.cache_format % { scope: self.scope, ident: ident } def get_rate(self): 根据会员等级返回不同的速率限制 if not hasattr(self, user): return 60/minute level_rates { bronze: 60/minute, silver: 120/minute, gold: 300/minute, platinum: 600/minute, diamond: 1200/minute, } user_level self.user.membership_level.name if self.user.membership_level else bronze return level_rates.get(user_level, 60/minute)5.3 API版本控制# membership_backend/urls.py from django.urls import path, include from rest_framework.routers import DefaultRouter from membership.views import UserViewSet, MembershipLevelViewSet from recommendations.views import PersonalizedRecommendationView router DefaultRouter() router.register(rusers, UserViewSet) router.register(rmembership-levels, MembershipLevelViewSet) urlpatterns [ path(api/v1/, include([ path(, include(router.urls)), path(recommendations/, PersonalizedRecommendationView.as_view(), namerecommendations), path(auth/, include(rest_framework.urls)), ])), # 健康检查 path(health/, lambda request: JsonResponse({status: healthy})), # API文档 path(api/docs/, include_docs_urls(title会员系统API文档)), ]6. 部署与优化建议6.1 生产环境配置创建生产环境配置文件# membership_backend/settings_production.py from .settings import * DEBUG False ALLOWED_HOSTS [yourdomain.com, api.yourdomain.com] # 数据库配置 DATABASES { default: { ENGINE: django.db.backends.postgresql, NAME: os.getenv(DB_NAME), USER: os.getenv(DB_USER), PASSWORD: os.getenv(DB_PASSWORD), HOST: os.getenv(DB_HOST, localhost), PORT: os.getenv(DB_PORT, 5432), } } # Redis缓存 CACHES { default: { BACKEND: django_redis.cache.RedisCache, LOCATION: fredis://{os.getenv(REDIS_HOST, localhost)}:{os.getenv(REDIS_PORT, 6379)}/1, OPTIONS: { CLIENT_CLASS: django_redis.client.DefaultClient, PASSWORD: os.getenv(REDIS_PASSWORD, None), } } } # 安全设置 SECURE_SSL_REDIRECT True SESSION_COOKIE_SECURE True CSRF_COOKIE_SECURE True SECURE_HSTS_SECONDS 31536000 # 静态文件 STATIC_ROOT /var/www/static/ MEDIA_ROOT /var/www/media/ # 日志配置 LOGGING { version: 1, disable_existing_loggers: False, handlers: { file: { level: ERROR, class: logging.FileHandler, filename: /var/log/django/error.log, }, recommendation_file: { level: INFO, class: logging.FileHandler, filename: /var/log/django/recommendation.log, }, }, loggers: { django: { handlers: [file], level: ERROR, propagate: True, }, recommendations: { handlers: [recommendation_file], level: INFO, propagate: False, }, }, }6.2 Docker部署配置# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ postgresql-client \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制项目文件 COPY . . # 收集静态文件 RUN python manage.py collectstatic --noinput # 创建非root用户 RUN useradd -m -u 1000 django USER django # 启动命令 CMD [gunicorn, membership_backend.wsgi:application, --bind, 0.0.0.0:8000]# docker-compose.yml version: 3.8 services: db: image: postgres:13 environment: POSTGRES_DB: membership_db POSTGRES_USER: membership_user POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - postgres_data:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U membership_user] interval: 10s timeout: 5s retries: 5 redis: image: redis:7-alpine command: redis-server --requirepass ${REDIS_PASSWORD} volumes: - redis_data:/data healthcheck: test: [CMD, redis-cli, ping] interval: 10s timeout: 5s retries: 5 web: build: . command: sh -c python manage.py migrate python manage.py runserver 0.0.0.0:8000 volumes: - .:/app ports: - 8000:8000 environment: DB_HOST: db DB_NAME: membership_db DB_USER: membership_user DB_PASSWORD: ${DB_PASSWORD} REDIS_HOST: redis REDIS_PASSWORD: ${REDIS_PASSWORD} depends_on: db: condition: service_healthy redis: condition: service_healthy volumes: postgres_data: redis_data:6.3 性能优化建议数据库优化为常用查询字段添加索引使用select_related和prefetch_related减少查询次数定期清理过期数据缓存策略用户推荐结果缓存1小时热门商品列表缓存30分钟会员等级信息缓存24小时异步任务使用Celery处理推荐模型推理用户行为分析异步进行邮件通知等非实时任务异步处理监控与告警使用Prometheus监控API性能设置推荐准确率告警监控模型推理延迟7. 总结这套基于Django REST框架和EcomGPT-7B的会员系统在实际项目中跑起来效果还不错。Django REST Framework的权限系统和序列化功能确实省了不少事EcomGPT-7B在电商推荐上的表现也比通用模型要好。部署的时候要注意安全配置特别是生产环境下的HTTPS、CORS这些设置。性能方面合理使用缓存和异步任务能明显提升用户体验。如果数据量特别大可以考虑把推荐服务单独部署用消息队列和主服务解耦。模型版本更新的时候要做好A/B测试确保新版本效果确实有提升。这套方案比较适合中小型电商项目如果规模再大可能就需要考虑更复杂的微服务架构了。不过对于大多数场景这个方案应该够用了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。