File size: 9,022 Bytes
17e971c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | # -*- coding: utf-8 -*-
# +-------------------------------------------------------------------
# | 宝塔Linux面板
# +-------------------------------------------------------------------
# | Copyleft (c) 2015-2099 宝塔软件(http://bt.cn) All lefts reserved.
# +-------------------------------------------------------------------
# | Author: wzz
# | email : wzz@bt.cn
# +-------------------------------------------------------------------
# +-------------------------------------------------------------------
# | docker sub_app 管理模型 -
# +-------------------------------------------------------------------
import json
import sys
import time
from collections import deque
if "/www/server/panel/class" not in sys.path:
sys.path.append('/www/server/panel/class')
import public
def download_model(service_name, model_name, model_version, ollama_url, app_cmd_log):
"""
下载Ollama模型的具体实现
@param service_name: 服务名称
@param model_name: 模型名称
@param model_version: 模型版本
@param ollama_url: Ollama API URL
@param app_cmd_log: 日志文件路径
"""
def start_download():
url = ollama_url + "/api/pull"
# 准备请求数据
data = {
"model": "{}:{}".format(model_name, model_version),
"stream": True
}
try:
import requests
response = requests.post(url, json=data, stream=True)
with open(app_cmd_log, 'a') as log_file:
log_file.write('正在下载 {} 模型，可能需要等待1-30分钟以上...\n'.format(model_name))
download_tag = None
last_completed = 0
last_time = time.time()
# 使用双端队列存储最近10秒的速度
speed_history = deque(maxlen=60)
count_sum = 0
for line in response.iter_lines():
if line:
json_response = json.loads(line)
status = json_response.get("status", "")
# 记录下载进度
if "pulling" in status:
status = status.split(" ")
if download_tag is None or status[1] != download_tag:
download_tag = status[1]
last_completed = 0
last_time = time.time()
speed_history.clear()
completed = json_response.get("completed", 0)
total = json_response.get("total", 0)
if total > 0:
# 计算下载速度
current_time = time.time()
time_diff = current_time - last_time
if time_diff >= 1: # 每秒更新一次
bytes_diff = completed - last_completed
speed = bytes_diff / time_diff # bytes per second
# 存储当前速度
count_sum += 1
if count_sum > 5:
speed_history.append(speed)
# 检查速度是否异常
avg_speed = None
if len(speed_history) >= 10:
avg_speed = sum(list(speed_history)[:-1]) / (len(speed_history) - 1)
current_speed = speed_history[-1]
if current_speed < 1024000 and avg_speed < 1536000: # 当前速度小于1.2MB/s且平均速度小于1.5MB/s
log_file.write('检测到下载速度过低，正在尝试重置下载...\n')
log_file.flush()
return False # 返回False表示需要重新下载
if current_speed < (avg_speed / 4) and avg_speed > 1024: # 确保有足够的平均速度
log_file.write('检测到下载速度异常或CF降速，正在尝试重置下载...\n')
log_file.flush()
return False # 返回False表示需要重新下载
# 转换速度单位
speed_str = ""
if speed < 1024:
speed_str = "{:.2f} B/s".format(speed)
elif speed < 1024 * 1024:
speed_str = "{:.2f} KB/s".format(speed / 1024)
else:
speed_str = "{:.2f} MB/s".format(speed / (1024 * 1024))
avg_speed_str = ""
if not avg_speed is None:
if avg_speed < 1024:
avg_speed_str = "{:.2f} B/s".format(avg_speed)
elif avg_speed < 1024 * 1024:
avg_speed_str = "{:.2f} KB/s".format(avg_speed / 1024)
else:
avg_speed_str = "{:.2f} MB/s".format(avg_speed / (1024 * 1024))
progress = (completed / total) * 100
log_file.write('文件: {}, 下载进度: {:.2f}%, 平均速度: {}, 当前速度: {}\n'.format(
download_tag,
progress,
avg_speed_str,
speed_str
))
log_file.flush()
# 更新上次的数据
last_completed = completed
last_time = current_time
else:
log_file.write(status + '\n')
log_file.flush()
# 下载完成后验证模型是否存在
verify_cmd = "docker-compose -p {service_name} exec -it {service_name_} ollama list | grep {model_name}:{model_version}".format(
service_name=service_name.lower(),
service_name_=service_name,
model_name=model_name,
model_version=model_version
)
result = public.ExecShell(verify_cmd)[0]
if model_name in result:
log_file.write('bt_successful\n')
return True
else:
public.writeFile("/tmp/{model_name}:{model_version}.failed".format(
model_name=model_name,
model_version=model_version,
), "failed")
log_file.write('bt_failed\n')
return False
except Exception as e:
# 发生异常时记录错误并标记失败
with open(app_cmd_log, 'a') as log_file:
log_file.write('下载失败: {}\n'.format(str(e)))
log_file.write('bt_failed\n')
public.writeFile("/tmp/{model_name}:{model_version}.failed".format(
model_name=model_name,
model_version=model_version,
), "failed")
return False
# 设置下载状态标记
public.ExecShell("echo 'downloading' > /tmp/{model_name}:{model_version}.pl".format(
model_name=model_name,
model_version=model_version
))
public.ExecShell("echo 'downloading' > /tmp/nocandown.pl")
public.ExecShell("rm -f /tmp/{model_name}:{model_version}.failed".format(
model_name=model_name,
model_version=model_version,
))
try:
max_retries = 30
retry_count = 0
while retry_count < max_retries:
if retry_count > 0:
with open(app_cmd_log, 'a') as log_file:
log_file.write('\n正在进行第{}次重试...\n'.format(retry_count + 1))
if start_download():
break
retry_count += 1
time.sleep(3) # 重试前等待3秒
finally:
# 清理状态文件
public.ExecShell("rm -f /tmp/{model_name}:{model_version}.pl".format(
model_name=model_name,
model_version=model_version,
))
public.ExecShell("rm -f /tmp/nocandown.pl") |