一、環境安裝準備

   #git拉取 bert-base-chinese 文件#創建 虛擬運行環境python -m venv myicrplatenv#刷新source myicrplatenv/bin/activate#python Django 集成nacospip install nacos-sdk-python#安裝 Djangopip3 install Django==5.1#安裝 pymysql settings.py 里面需要 # 強制用pymysql替代默認的MySQLdb pymysql.install_as_MySQLdb()pip install pymysql# 安裝mongopip install djongo pymongopip install transformerspip install torch#安裝Daphne: pip install daphne#項目通過 通過 daphne啟動daphne icrplat.asgi:application

二、構建項目及app模塊

#創建app模塊
python3 manage.py startapp cs

icrplat├── README.md
├── cs
│?? ├── __init__.py
│?? ├── __pycache__
│?? ├── admin.py
│?? ├── apps.py
│?? ├── migrations
│?? ├── models.py
│?? ├── tests.py
│?? └── views.py
├── icrplat
│?? ├── __init__.py
│?? ├── __pycache__
│?? ├── asgi.py
│?? ├── common
│?? ├── settings.py
│?? ├── urls.py
│?? └── wsgi.py
├── manage.py
├── myicrplatenv
│?? ├── bin
│?? ├── include
│?? ├── lib
│?? ├── pyvenv.cfg
│?? └── share
├── nacos-data
│?? └── snapshot
└── templates

三、準備數據mongo

 db.products.insertMany([{'name': '手機', 'description': '最新款智能手機，支持5G網絡，高清攝像頭'},{'name': '無線耳機', 'description': '降噪無線耳機，藍牙連接，長續航'},{'name': '智能手表', 'description': '健康監測，運動記錄，支持通知提醒'},{'name': '平板電腦', 'description': '輕薄便攜，高性能處理器，適合辦公和娛樂'},{'name': '筆記本電腦', 'description': '高性能筆記本，適合游戲和設計工作'},{'name': '相機', 'description': '專業級相機，支持4K視頻拍攝'},{'name': '耳機', 'description': '高保真音質，舒適佩戴'},{'name': '充電寶', 'description': '大容量充電寶，支持快充'},{'name': '手機殼', 'description': '防摔手機殼，支持多種機型'},{'name': '路由器', 'description': '高速無線路由器，支持千兆網絡'},])

四、相關代碼

#################################settings.py#######################################from nacos import NacosClientfrom icrplat.common.config.nacos.NacosConfigWatcher import nacos_config_watcherimport pymysql# 強制用pymysql替代默認的MySQLdb
pymysql.install_as_MySQLdb()INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','cs'  #模塊app 注入
]# Nacos 配置
NACOS_SERVER = "xx"  # Nacos 服務器地址 ip 換成Nacos對應的IP地址
NACOS_NAMESPACE = "dev"  # Nacos 命名空間
NACOS_GROUP = "MICRO_GROUP"  # Nacos 分組
NACOS_DATA_ID = "aics_config_dev"  # Nacos 配置 ID# 初始化 Nacos 客戶端
nacos_client = NacosClient(NACOS_SERVER, namespace=NACOS_NAMESPACE)# 從 Nacos 獲取 config配置
nacos_config = nacos_client.get_config(NACOS_DATA_ID, NACOS_GROUP)
# 將 JSON 字符串轉換為字典
nacos_config = eval(nacos_config)
print(f"nacos_config: {nacos_config}")#解析 mysql 初始配置
mysql_config = nacos_config.get("mysql", {})
print(f"mysql_config: {mysql_config}")#解析 mongodb 初始配置
mongo_config = nacos_config.get("mongodb",{})
print(f"mongo_config: {mongo_config}")# 配置 MySQL 數據庫
DATABASES = {'default': {'ENGINE': 'django.db.backends.mysql','NAME': mysql_config.get("DATABASE_NAME"),'USER': mysql_config.get("DATABASE_USER"),'PASSWORD': mysql_config.get("DATABASE_PASSWORD"),'HOST': mysql_config.get("DATABASE_HOST"),'PORT': mysql_config.get("DATABASE_PORT", "8081"),'OPTIONS': {'charset': 'utf8mb4',  # 支持更廣泛的字符集},},'mongodb': {  # MongoDB配置'ENGINE': 'djongo','NAME': mongo_config.get('DB_NAME'),'ENFORCE_SCHEMA': False,'CLIENT': {'host': mongo_config.get('DB_HOST'),'port': int(mongo_config.get('DB_PORT')),'username': mongo_config.get('DB_USERNAME'),'password': mongo_config.get('DB_PASSWORD'),'authSource': mongo_config.get('AUTH_DB_SOURCE'),}}
}
#使用多數據庫，需要在 settings.py 中設置數據庫路由。
DATABASE_ROUTERS = ['icrpPlat.dbconfig.routers.DatabaseRouter']print(f"更新前 Databases: {DATABASES}")def update_nacos_config(config):"""動態更新 MySQL 配置"""print("收到 Nacos 配置更新通知")print(f"收到更新后 config: {config}")# 使用 json.loads 解析 JSON 字符串  contentconfig_dict = json.loads(config['content'])#獲取更新后的 mysql 配置mysql_config = config_dict.get("mysql", {})print(f"更新后 Databases: {mysql_config}")nacos_config_watcher.update_mysql_config(mysql_config)#獲取更新后的 elasticsearch 配置mongo_config = config_dict.get("mongodb",{})print(f"更新后 mongodb: {mongo_config}")nacos_config_watcher.update_mongodb_config(mongo_config)# 監聽 Nacos 配置變化
nacos_client.add_config_watcher(NACOS_DATA_ID, NACOS_GROUP, update_nacos_config)#########################nacos_config_watcher.py 方法##############################import threadingfrom django.conf import settingsclass NacosConfigWatcher:def __init__(self):# 創建一個鎖對象 可以確保在同一時間只有一個線程能夠訪問某個共享資源，從而避免多線程環境下的數據競爭問題。self.lock = threading.Lock()"""更新myslq 配置with 語句主要用于上下文管理，通常用于處理資源的管理和釋放。它的核心作用是確保在代碼塊執行完畢后，資源能夠被正確地關閉或清理，避免資源泄漏。常見的場景包括文件操作、數據庫連接、線程鎖等。"""def update_mysql_config(self, mysql_config):with self.lock:settings.DATABASES['default'].update({'NAME': mysql_config.get("DATABASE_NAME"),'USER': mysql_config.get("DATABASE_USER"),'PASSWORD': mysql_config.get("DATABASE_PASSWORD"),'HOST': mysql_config.get("DATABASE_HOST"),'PORT': mysql_config.get("DATABASE_PORT", "3306"),})def update_mongodb_config(self, mongo_config):with self.lock:settings.DATABASES['mongodb'].update({'host': mongo_config.get('DB_HOST'),'port': int(mongo_config.get('DB_PORT')),'username': mongo_config.get('DB_USERNAME'),'password': mongo_config.get('DB_PASSWORD'),'authSource': mongo_config.get('AUTH_DB_SOURCE'),})#使用模塊級別的單例 可以有多種實現單例的形式
nacos_config_watcher = NacosConfigWatcher()

##################################urls.yml####################################
urlpatterns = [path('admin/', admin.site.urls),path('cs/handleUserRequest',cs_views.handleUserRequest, name='handleUserRequest'),path('cs/getProducts',cs_views.getProducts,name='get_product'),path('cs/getdata',cs_views.getdata,name='get_data')
]

####################################views.py################################
from django.shortcuts import renderimport json
import logging
import torch
from torch.nn.functional import cosine_similarity
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exemptfrom icrplat.common.enum.ResponeCodeEnum import ResponseCodeEnum
from icrplat.common.exception.BusinessException import BusinessException
from icrplat.common.utils.CommonResult import CommonResult
from cs.models import Productsfrom functools import lru_cachefrom icrplat.common.utils.TransformerUtils import TransformerUtils# import os
# os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"# Create your views here.
@csrf_exempt  # 如果需要關閉CSRF驗證，可以加上這個裝飾器
def handleUserRequest(request):try:if request.method == 'POST':json_data = request.bodydata = json.loads(json_data)# 假設你需要處理的數據結構如下message = data.get('message')logging.info(f"data:{data}")else:raise BusinessException(ResponseCodeEnum.METHOD_ERROR.status_code, ResponseCodeEnum.METHOD_ERROR.status_message)return JsonResponse(CommonResult.success_data(None), json_dumps_params={'ensure_ascii': False})except BusinessException as e:return JsonResponse(CommonResult.error(e.code, e.message), json_dumps_params={'ensure_ascii': False})def getProducts(request):try:if request.method == 'GET':products_mongo = Products.objects.using('mongodb').all()else:raise BusinessException(ResponseCodeEnum.METHOD_ERROR.status_code,ResponseCodeEnum.METHOD_ERROR.status_message)# 將查詢結果轉化為字典或列表，確保可以被序列化為 JSON# products_data = [{'name': product.name, 'description': product.description} for product in products_mongo]# 使用列表推導式調用 to_dict 方法，將查詢結果轉化為字典#注意點： 需要通過實例來調用，而不是通過類直接調用products_data = [product.to_dict() for product in products_mongo]return JsonResponse(CommonResult.success_data(products_data), json_dumps_params={'ensure_ascii': False})except BusinessException as e:return JsonResponse(CommonResult.error(e.code, e.message), json_dumps_params={'ensure_ascii': False})@csrf_exempt
def getdata(request):try:if request.method == 'GET':description = request.GET.get("description")if not description:raise BusinessException(ResponseCodeEnum.PARAM_ERROR.status_code, "描述不能為空")products_mongo = Products.objects.using('mongodb').all()products_data = [product.to_dict() for product in products_mongo]# 生成商品嵌入向量product_embeddings = {product['name']: get_cached_bert_embeddings(product['description']) for product in products_data}# 生成用戶查詢的嵌入向量user_embeddings = get_cached_bert_embeddings(description)# 獲取 計算相似度后的結果 商品product_embeddings 向量 用戶user_embeddings 向量top_products = TransformerUtils.similarities(product_embeddings,user_embeddings)if not top_products:return JsonResponse({"message": "未找到相關商品"}, json_dumps_params={'ensure_ascii': False})result_with_description = {}for product_name, similarity in top_products.items():product_description = next((product['description'] for product in products_data if product['name'] == product_name),"暫無描述")result_with_description[product_name] = {"相似度": similarity,"描述": product_description}return JsonResponse(CommonResult.success_data(result_with_description), json_dumps_params={'ensure_ascii': False})else:raise BusinessException(ResponseCodeEnum.METHOD_ERROR.status_code,ResponseCodeEnum.METHOD_ERROR.status_message)except BusinessException as e:return JsonResponse(CommonResult.error(e.code, e.message), json_dumps_params={'ensure_ascii': False})"""在代碼中使用 get_cached_bert_embeddings 替代 get_bert_embeddings，以減少重復計算。
"""
@lru_cache(maxsize=1000)
def get_cached_bert_embeddings(text):return TransformerUtils.get_bert_embeddings(text)

#################################TransformerUtils.py##########################import loggingfrom transformers import BertTokenizer, BertModel
import torch
from torch.nn.functional import cosine_similarityimport osfrom transformers.utils.hub import TRANSFORMERS_CACHEclass TransformerUtils:"""這個函數的主要功能是通過 BERT 模型生成文本的嵌入向量。如果在加載模型或生成嵌入的過程中遇到任何問題，函數會返回一個默認的零向量，確保程序能夠繼續運行，而不會因為異常而中斷。"""@staticmethoddef get_bert_embeddings(texts):# 輸出 BERT 模型的默認緩存路徑。通常，預訓練模型和分詞器會從緩存中加載，以減少重復下載的時間。logging.info(f"Default cache path: {TRANSFORMERS_CACHE}")""" 如果輸入的 texts 為空或者不是字符串類型，函數會返回一個長度為 768 的零向量。BERT 模型的嵌入向量通常是 768 維的，所以這里返回一個零向量作為默認值。 0.0返回一個長度為 768 的列表，所有元素均為浮點數 0.0，例如：[0.0, 0.0, ..., 0.0]  # 共768個元素  """if not texts or not isinstance(texts, str):return [0.0] * 768# 指定模型路徑path = "/Users/jiajiamao/soft/python/space/bert-base-chinese"logging.info(f"模型路徑指定 path:{path}")try:# 檢查路徑是否存在if not os.path.exists(path):logging.info(f"模型路徑 不存在!")# 這里指定了 BERT 模型的路徑，并檢查該路徑是否存在。如果路徑不存在，會拋出一個 FileNotFoundError 異常raise FileNotFoundError(f"Model path {path} does not exist.")"""加載分詞器:使用 BertTokenizer.from_pretrained 方法加載分詞器。輸入?："高性能"  轉變成 [高,性,能]分詞器的作用是將輸入文本轉換為模型能夠理解的 token 序列。詞向量轉換(tokens)  # 轉為數字矩陣"""tokenizer = BertTokenizer.from_pretrained(path)logging.info(f"********************加載分詞器:Tokenizer loaded successfully")"""加載模型:使用 BertModel.from_pretrained 方法加載 BERT 模型。加載成功后，model 變量將包含預訓練的 BERT 模型。"""model = BertModel.from_pretrained(path)logging.info(f"********************加載模型:BertModel loaded successfully")# 確保加載成功if tokenizer is None:raise ValueError("Tokenizer failed to load.")if model is None:raise ValueError("Model failed to load.")"""# 原理實現步驟：# 1. 添加特殊標記 → [CLS] 我 在 學習 BERT 模型 [SEP]# 2. 分詞 → ['降', '噪', '無', '線', '耳', '機', '，', '藍', '牙', '連', '接', '，', '長', '續', '航']# 3. 轉換為ID → [ 101, 7360, 1692, 3187, 5296, 5455, 3322, 8024, 5905, 4280, 6825, 2970,#          8024, 7270, 5330, 5661,  102] 方便輸入模型"""inputs = tokenizer(texts, return_tensors='pt', max_length=512, truncation=True, padding=True)logging.info(f"********************tokenizer 分詞:{tokenizer.tokenize(texts)}")logging.info(f"********************input:{inputs}")"""with torch.no_grad(): 表示在推理過程中不計算梯度，以提高效率。把分好詞的句子變成一系列向量，每個詞對應一個向量。每句話里的每個詞都會得到一個最終的向量表示模型被調用時它會自動執行自注意力機制的計算。BERT由多個Transformer編碼層堆疊而成，每一層都包含一個自注意力模塊:# 每個詞生成3個向量：Q（查詢向量）：用來詢問其他詞和它的關系。K（鑰匙向量）：用來衡量其他詞和它的相關性。V（值向量）：用來表示這個詞的實際內容。# 計算"高"對每個詞的關注度：attention_score = Q_高 ? [K_高, K_性, K_能]  # → 得到 [0.8, 0.1, 0.1]（更關注自己）用“高”的查詢向量（Q_高）去分別與“高”、“性”、“能”的鑰匙向量（K_高, K_性, K_能）做點積運算。點積的結果就是注意力分數，比如這里得到 [0.8, 0.1, 0.1]。這表示“高”更關注自己（0.8），而對“性”和“能”的關注度較低。# 加權求和：用注意力分數對“高”、“性”、“能”的值向量（V_高, V_性, V_能）進行加權求和。比如，新的“高”的表示 = 0.8 * V_高 + 0.1 * V_性 + 0.1 * V_能。這個過程的意義在于，通過計算每個詞對其他詞的關注度，讓模型能夠更好地理解上下文關系。比如“高”這個詞在這里更關注自己，所以它會更多地保留自己的信息，而稍微融入一點“性”和“能”的信息。"""with torch.no_grad():outputs = model(**inputs)"""1.從BERT的輸出中提取最后一層的隱藏狀態（last_hidden_state）。2.對所有詞的向量做平均，得到整個句子的向量表示（mean(dim=1)）。3.去掉多余的維度（squeeze()）。4.把Tensor轉換成NumPy數組（.numpy()）。5.最后把數組轉換成列表（.tolist()）"""embeddings = outputs.last_hidden_state.mean(dim=1).squeeze().numpy().tolist()logging.info(f"********************原內容Text: {texts}")logging.info(f"********************BERT生成一個768維的向量: {embeddings[:10]}")  # 只打印前10維，方便檢查return embeddingsexcept Exception as e:print(f"Error in get_bert_embeddings: {e}")return [0.0] * 768  # 出現異常時返回默認值"""計算相似度通過分析用戶的興趣（用戶向量）和商品的特點（商品向量），計算它們之間的匹配度。然后根據匹配度排序，篩選出最相關的前 5 個商品。最后，只保留那些匹配度足夠高（大于閾值）的商品，確保推薦的內容對用戶來說是有意義的。@:param product_embeddings 商品向量 @:param user_embeddings 用戶向量"""@staticmethoddef similarities(product_embeddings,user_embeddings):#將用戶向量轉換為張量形式user_embeds_tensor = torch.tensor(user_embeddings, dtype=torch.float32).unsqueeze(0)# 計算相似度similarities = {}for product_name, embedding in product_embeddings.items():#將商品的向量轉換為張量形式，并調整其維度以匹配用戶向量的計算product_embeds_tensor = torch.tensor(embedding, dtype=torch.float32).unsqueeze(0)#計算用戶向量與商品向量之間的余弦相似度，值范圍為 [-1, 1]，值越接近 1 表示相似度越高。similarity = cosine_similarity(user_embeds_tensor, product_embeds_tensor).item()#將結果存儲在一個字典中，鍵為商品名稱，值為相似度值similarities[product_name] = similarityprint(f"Product: {product_name}, Similarity: {similarity}")# 根據相似度排序（從高到低）sorted_products = sorted(similarities.items(), key=lambda item: item[1], reverse=True)print(f"Sorted Products: {sorted_products}")# 設置相似度閾值，并返回相似度最高的前5個商品（設置閾值并篩選商品）threshold = 0.6top_products = {k: v for k, v in sorted_products[:5] if v > threshold}return top_products

###############################ResponseCodeEnum###############################from enum import Enum"""返回通用響應類
"""
class ResponseCodeEnum(Enum):###################################公共響應#############################SUCCESS = (200, "操作成功！")PARAMS_ERROR = (400, "參數解析失敗,請核對參數！")UNAUTHORIZED = (401, "未認證（簽名錯誤）")FORBIDDEN = (402, "請求錯誤")  # 返回失敗業務公共codeMEDIA_TYPE_ERROR = (403, "不支持的媒體異常,請核對contentType！")URL_REQ_NULL = (404, "請求路徑不存在")METHOD_ERROR = (405, "不支持當前請求方法,請核對請求方法！")TIMEOUT_EXPIRE_ERROR = (406, "token登錄過期！")TOKEN_ILLEGAL = (407, "非法token！")INTERNAL_SERVER_ERROR = (500, "服務器內部錯誤！")  # 系統異常公共code###################################相關業務返回#############################RESOURCES_IP_EXIST = (1001,"資源IP已存在!")"""在枚舉值初始化時，將元組中的code和message分別賦值給實例屬性self.code和self.message。這樣每個枚舉值都有獨立的code和message屬性。"""def __init__(self, code, message):self.code = codeself.message = message"""@property 是 Python 中用于將類的方法轉換為屬性訪問的裝飾器。使用 @property 裝飾器，你可以像訪問屬性一樣訪問方法，而不需要調用它示例：ResponseCodeEnum.SUCCESS.status_code      """@propertydef status_code(self):return self.code@propertydef status_message(self):return self.message

#####################################CommonResult.py###############################from typing import Any, Optionalfrom icrplat.common.enum.ResponeCodeEnum import ResponseCodeEnumclass CommonResult:""":param res: 類型為 ResponseCodeEnum，用于存儲響應狀態碼和消息:param data: 類型為 Any，用于存儲返回的數據:param pagination: 類型為 Optional[dict]，是一個可選的參數，用于存儲分頁信息（默認為 None）"""def __init__(self, res: ResponseCodeEnum, data: Any, pagination: Optional[dict] = None):self.ResponseCodeEnum = ResponseCodeEnumself.data = dataself.pagination = pagination"""這是一個裝飾器，用于定義靜態方法。靜態方法不需要訪問類實例（self）或類本身（cls），可以直接通過類名調用。CommonResult.success_pagination"""@staticmethoddef success_pagination(data: Any, pagination: Optional[dict] = None):""":param data: Any: 表示返回的數據，類型為 Any（可以是任意類型）。:param pagination: 表示可選的分頁信息，類型為字典（dict），默認值為 None。:return:返回一個字典，包含以下鍵值對：code: 狀態碼，取自 ResponseCodeEnum.SUCCESS.status_code。message: 響應消息，取自 ResponseCodeEnum.SUCCESS.message。data: 傳入的 data 數據。pagination: 傳入的分頁信息（如果未傳入則為 None）"""return {'code': ResponseCodeEnum.SUCCESS.code,'message': ResponseCodeEnum.SUCCESS.message,'data': data,  # 將 QuerySet 序列化為 JSON,'pagination': pagination}@staticmethoddef success_data(data: Any):return {'code': ResponseCodeEnum.SUCCESS.code,'message': ResponseCodeEnum.SUCCESS.message,'data': data,  # 將 QuerySet 序列化為 JSON,}@staticmethoddef error(code,message):return {'code': code,'message': message}@staticmethoddef error():return {'code':ResponseCodeEnum.INTERNAL_SERVER_ERROR.code,'message':ResponseCodeEnum.INTERNAL_SERVER_ERROR.message}

###############################BusinessException.py###############################class BusinessException(Exception):def __init__(self,message,code):self.message = messageself.code = codesuper().__init__(message)

#####################################Products.py##################################from django.db import models# Create your models here.from django.db import modelsclass Products(models.Model):# 定義 name 字段name = models.CharField(max_length=100)# 定義 description 字段description = models.TextField()class Meta:app_label = 'mongodb'db_table = 'products'  #"""自定義 to_dict 實例方法"""def to_dict(self):"""將模型實例轉化為字典"""return {'name': self.name,'description': self.description}

五、nacos配置

#換成自己的地址{"mysql": {"DATABASE_NAME": "micro_aics","DATABASE_USER": "xx","DATABASE_PASSWORD": "xx","DATABASE_HOST": "xx","DATABASE_PORT": "8081"},"mongodb": {"DB_NAME": "action_log","DB_HOST": "xx","DB_PORT": "xx","DB_USERNAME": "xx","DB_PASSWORD": "xx","AUTH_DB_SOURCE": "xx"}
}

六、啟動項目

#通過 daphne 啟動應用
daphne icrplat.asgi:application

七、測試

?