Centos7本地部署阿里Qwen2-7B模型

1.從hagging face下載模型

2.把下載的模型文件，放到/usr/local/Qwen2-7B目錄下

3.創建虛擬環境，安裝依賴

1.環境安裝
sudo yum update -y
sudo yum install -y python3 python3-pip git

2.創建虛擬環境并激活
python3 -m venv qwen2_env
source qwen2_env/bin/activate

conda activate??qwen2_env

conda install numpy transformers torch langchain sentence-transformers jieba requests streamlit

3.將diet_ui.py文件放到根目錄下

import streamlit as st
import requests
import re
import logging
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain
from langchain.llms import HuggingFacePipeline
from langchain.prompts import PromptTemplate
from transformers import pipeline
from sentence_transformers import SentenceTransformer, models
import numpy as np
import jieba# 配置日志記錄
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')# 加載 Qwen2 模型和分詞器
try:tokenizer = AutoTokenizer.from_pretrained("/usr/local/Qwen2-7B", trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained("/usr/local/Qwen2-7B", device_map="auto", trust_remote_code=True).eval()logging.info("分詞器和模型加載成功")
except Exception as e:logging.error(f"加載分詞器或模型時發生錯誤: {e}")raise# 公司業務接口地址
PORTFOLIO_API_URL = "https://xxxxx/plugin/portfolio"# 擴充關鍵詞列表
KEYWORDS = ["方案", "配餐", "計劃", "飲食安排", "食譜", "卡路里", "餐單", "膳食"]
# 允許的話題列表，添加配餐相關關鍵詞
ALLOWED_TOPICS = ["健康", "營養", "飲食", "訂單", "優惠券", "吃飯", "運動", "養生", "客服"] + KEYWORDS# 初始化會話記憶，最多存儲 10 輪對話
memory = ConversationBufferMemory(k=10)# 創建 HuggingFacePipeline
pipe = pipeline("text-generation",model=model,tokenizer=tokenizer,max_new_tokens=500,do_sample=True,top_p=0.85,temperature=0.35
)
llm = HuggingFacePipeline(pipeline=pipe)# 創建一個簡單的提示模板
prompt_template = PromptTemplate(input_variables=["input"],template="{input}"
)# 創建 LLMChain
chain = LLMChain(llm=llm, memory=memory, prompt=prompt_template)# 加載知識庫
def load_knowledge_base(file_path):with open(file_path, 'r', encoding='utf-8') as file:lines = file.readlines()knowledge_base = []question = Noneanswer = ""for line in lines:line = line.strip()if line.startswith("Q:"):if question is not None:knowledge_base.append((question, answer))question = line[2:]answer = ""elif line.startswith("A:"):answer = line[2:]else:answer += line + " "if question is not None:knowledge_base.append((question, answer))return knowledge_base# 加載知識庫
knowledge_base = load_knowledge_base("/root/knowledge_base.txt")# 手動加載模型組件
word_embedding_model = models.Transformer('/root/all-MiniLM-L6-v2')
pooling_model = models.Pooling(word_embedding_model.get_word_embedding_dimension())
encoder = SentenceTransformer(modules=[word_embedding_model, pooling_model])# 編碼知識庫中的問題
question_embeddings = [encoder.encode(question) for question, _ in knowledge_base]# 提取關鍵詞
def extract_keywords(text):return list(jieba.cut(text))# 從知識庫查找答案（結合關鍵詞匹配和向量相似度）
def find_answer_from_knowledge_base(question, threshold=0.5):question_keywords = extract_keywords(question)question_embedding = encoder.encode(question)best_match_index = -1best_similarity = -1for i, (kb_question, _) in enumerate(knowledge_base):kb_keywords = extract_keywords(kb_question)# 檢查關鍵詞是否匹配if any(keyword in kb_keywords for keyword in question_keywords):similarity = np.dot(question_embedding, question_embeddings[i]) / (np.linalg.norm(question_embedding) * np.linalg.norm(question_embeddings[i]))if similarity > best_similarity and similarity >= threshold:best_similarity = similaritybest_match_index = iif best_match_index != -1:return knowledge_base[best_match_index][1]return None# 提取配餐參數
def extract_params(text):calory = Noneday = None# 優化后的卡路里提取正則表達式calory_pattern = re.compile(r'(?:每天|每日|每餐)?\s*(?:卡路里|熱量|千卡|大卡)\s*(?:是|約|大概|為)?\s*(\d+)', re.IGNORECASE)# 優化后的天數提取正則表達式day_pattern = re.compile(r'(?:配|安排|制定)?\s*(\d+)\s*(?:天|日|天的餐|日的餐)', re.IGNORECASE)calory_match = calory_pattern.search(text)if calory_match:try:calory = int(calory_match.group(1))except ValueError:passday_match = day_pattern.search(text)if day_match:try:day = int(day_match.group(1))except ValueError:passreturn calory, day# 調用配餐接口
def get_meal_plan(calory, day):logging.debug(f"準備調用配餐接口，calory: {calory}, day: {day}")try:response = requests.get(PORTFOLIO_API_URL, params={"calory": calory, "day": day})logging.debug(f"接口響應狀態碼: {response.status_code}")logging.debug(f"接口響應內容: {response.text}")if response.status_code == 200:logging.debug("配餐接口調用成功")try:response_data = response.json()# 打印完整的響應數據，用于調試logging.debug(f"完整的接口響應數據: {response_data}")# 提取配餐方案信息sku_list = response_data.get("skuList", [])if sku_list:output_text = "以下是為您生成的配餐方案：\n"output_text += "=" * 50 + "\n"current_day = 0for sku in sku_list:sequential_days = sku.get("sequentialDays")if sequential_days is None:logging.warning("配餐數據中缺少 sequentialDays 字段")continueif sequential_days != current_day:current_day = sequential_daysoutput_text += f"\n第 {current_day} 天：\n"output_text += "-" * 50 + "\n"category_type = sku.get("categoryType", "未知餐別")total_kcal = sku.get("totalKcal", "未知熱量")output_text += f"  {category_type}（{total_kcal} 千卡）：\n"meal_list = sku.get("list", [])for meal in meal_list:meal_name = meal.get("name", "未知套餐")output_text += f"    套餐名稱：{meal_name}\n"output_text += "    包含菜品：\n"category_list = meal.get("getCategoryList", [])for item in category_list:skuname = item.get("skuname", "未知菜品")price = item.get("price", "未知價格")energy = item.get("energy", "未知能量")output_text += f"      - {skuname}（價格：{price} 元，能量：{energy} 千卡）\n"output_text += "=" * 50 + "\n"else:output_text = "獲取配餐方案成功，但無具體信息"return output_textexcept ValueError:logging.error("無法將接口響應內容解析為 JSON 格式")return "調用接口成功，但無法解析響應內容"else:logging.error(f"配餐接口調用失敗，狀態碼: {response.status_code}")return f"調用接口失敗，狀態碼: {response.status_code}"except requests.RequestException as e:logging.error(f"調用配餐接口時發生網絡錯誤: {e}")return f"調用接口時發生網絡錯誤: {e}"# Streamlit UI
st.title("Nutribite 智能營養師聊天機器人")# 初始化聊天歷史
if 'chat_history' not in st.session_state:st.session_state.chat_history = []# 初始化輸入框值
if 'input_value' not in st.session_state:st.session_state.input_value = ""# 獲取用戶輸入
user_input = st.text_input("請輸入你的問題", value=st.session_state.input_value, key="input_key")if user_input:# 拼接歷史對話和當前輸入history_text = "\n".join([f"用戶: {msg['user']}\n機器人: {msg['bot']}" for msg in st.session_state.chat_history]) + "\n" + user_inputlogging.debug(f"拼接后的文本: {history_text}")# 檢查是否處于配餐流程中is_in_meal_process = "配餐" in history_text and ("請問，要配幾天的餐？" in history_text or "請問每天的卡路里大概是多少？" in history_text)# 檢查是否是允許的話題if not any(topic in history_text for topic in ALLOWED_TOPICS) and not is_in_meal_process:bot_response = "我是Nutribite智能營養師，請問我相關問題"else:# 從知識庫查找答案answer = find_answer_from_knowledge_base(user_input)if answer:logging.debug("從知識庫找到答案")bot_response = answerelse:# 檢測用戶問題是否與配餐相關，僅依據當前輸入文本is_meal_related = any(keyword in user_input for keyword in KEYWORDS)# 嘗試從拼接后的文本中提取 calory 和 daycalory, day = extract_params(history_text)if is_meal_related or is_in_meal_process:if day is None:if "請問，要配幾天的餐？" in history_text:# 重新提取當前用戶輸入中的 day_, day = extract_params(user_input)if day is None:logging.debug("仍未提取到 day 參數，再次詢問用戶")bot_response = "抱歉，我沒獲取到配餐天數，請明確告知要配幾天的餐。"else:if calory is None:logging.debug("提取到 day 參數，未提取到 calory 參數，詢問用戶")bot_response = "請問每天的卡路里大概是多少？"else:# 調用配餐接口bot_response = get_meal_plan(calory, day)else:logging.debug("未提取到 day 參數，詢問用戶")bot_response = "請問，要配幾天的餐？"elif calory is None:logging.debug("未提取到 calory 參數，詢問用戶")bot_response = "請問每天的卡路里大概是多少？"else:# 調用配餐接口bot_response = get_meal_plan(calory, day)else:logging.debug("用戶問題與配餐無關，使用 Qwen2 模型生成回復")bot_response = chain.run(user_input)# 更新聊天歷史st.session_state.chat_history.append({"user": user_input, "bot": bot_response})# 清空輸入框st.session_state.input_value = ""# 顯示聊天歷史
for msg in st.session_state.chat_history:st.markdown(f"**用戶**：{msg['user']}")st.markdown(f"**機器人**：{msg['bot']}")

4.執行streamlit run diet_ui.py

這個智能體實現了知識庫、調公司業務接口功能，具體效果圖