llm vlm启动器 魔搭下模型 vLLM
·
vllm Qwen2.5-0.5B输出乱码解决办法 用-Instruct版本的-CSDN博客
服务创建代码
import os
import subprocess
import signal
import time
import re
from modelscope import snapshot_download
# ======================
# 配置模型
# ======================
# 添加路径净化函数
def sanitize_model_path(model_name):
"""净化模型名称,防止路径遍历攻击"""
# 移除路径分隔符和其他潜在危险字符
sanitized = re.sub(r'[\/\\:\*\?"<>\|]', '_', model_name)
# 确保不包含 '..' 序列
sanitized = re.sub(r'\.\.', '_', sanitized)
return sanitized
LLM_MODEL = "OpenBMB/MiniCPM4-0.5B-QAT-Int4-GPTQ-format"
VLM_MODEL = "OpenBMB/MiniCPM-V-2_6-int4" # 或 MiniCPM-V-2_6(非量化)
LLM_DIR = f"/root/my_python_server/models/{sanitize_model_path(LLM_MODEL)}"
VLM_DIR = f"/root/my_python_server/models/{sanitize_model_path(VLM_MODEL)}"
# 存储进程对象
processes = []
# ======================
# 下载模型(ModelScope)
# ======================
def download_models():
for name, path in [(LLM_MODEL, LLM_DIR), (VLM_MODEL, VLM_DIR)]:
if not os.path.exists(path) or not os.listdir(path):
print(f"📥 下载模型: {name}")
os.makedirs(path, exist_ok=True)
snapshot_download(model_id=name, local_dir=path)
else:
print(f"✅ 模型已存在: {path}")
def start_process(cmd, description):
print(f"🚀 启动 {description}...")
process = subprocess.Popen(cmd, shell=True, preexec_fn=os.setsid)
processes.append(process)
return process
def cleanup_processes():
print("\n🔄 清理进程...")
for process in processes:
try:
os.killpg(os.getpgid(process.pid), signal.SIGTERM)
process.wait(timeout=5)
except:
try:
os.killpg(os.getpgid(process.pid), signal.SIGKILL)
except:
pass
print("✅ 所有进程已清理完毕")
def signal_handler(sig, frame):
print("\n⚠️ 收到终止信号,正在清理...")
cleanup_processes()
exit(0)
if __name__ == "__main__":
# 注册信号处理
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
download_models()
llm_cmd = (
f"/root/my_python_server/vllm-env/bin/python -m vllm.entrypoints.openai.api_server "
f"--model {LLM_DIR} "
f"--host 0.0.0.0 --port 8000 "
f"--quantization gptq_marlin "
f"--trust-remote-code "
f"--dtype bfloat16 "
f"--gpu-memory-utilization 0.6 "
f"--max-num-batched-tokens 256 "
f"--enable-auto-tool-choice "
f"--tool-call-parser functiongemma"
)
vlm_cmd = (
f"/root/my_python_server/vllm-env/bin/python -m vllm.entrypoints.openai.api_server "
f"--model {VLM_DIR} "
f"--host 0.0.0.0 --port 8001 "
f"--dtype bfloat16 "
f"--trust-remote-code "
f"--gpu-memory-utilization 0.6 "
f"--max-model-len 256 "
f"--max-num-batched-tokens 256"
)
print("\n🌐 访问地址:")
print(f" - LLM: http://localhost:8000/v1")
print(f" - VLM: http://localhost:8001/v1")
vlm_process = start_process(vlm_cmd, "VLMP 服务 (端口 8001)")
#llm_process = start_process(llm_cmd, "LLM 服务 (端口 8000)")
print("\n✅ 所有服务已启动")
print("📝 按 Ctrl+C 停止所有服务...")
# 等待进程结束
try:
for process in processes:
process.wait()
except KeyboardInterrupt:
cleanup_processes()
LLM
开始测试 LLM 服务...
=== 测试 LLM 非流式调用 ===
[LLM] 使用 requests 直接调用 API
LLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM4-0.5B-QAT-Int4-GPTQ-format
开始调用LLM服务(非流式)
LLM请求已保存到: output\llm_request.json
正在调用LLM服务...
LLM服务响应状态码: 200
LLM响应已保存到: output\llm_response.json
LLM服务调用完成
响应状态: 成功
模型: /root/my_python_server/models/OpenBMB_MiniCPM4-0.5B-QAT-Int4-GPTQ-format
生成内容: 你好!我是MiniCPM系列模型,由面壁智能和OpenBMB开源社区贡献。详细信息请访问 https://github.com/OpenBMB/
非流式调用测试完成
=== 测试 LLM 流式调用 ===
[LLM] 使用 requests 直接调用 API
LLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM4-0.5B-QAT-Int4-GPTQ-format
开始流式接收响应...
开始调用LLM服务(流式)
LLM流式请求已保存到: output\llm_stream_request.json
LLM服务流式响应状态码: 200
人工智能(AI)的发展历程是一个充 满创新和突破的旅程,它不仅改变了 我们的生活方式,还推动了科技的进 步。以下是一些关键的发展阶段:
1. **起源(1950-1970)**:20世纪50年代至60年代,AI研究主要集中在解决数学问题,特别是解决数学上的计 算机问题。这一时期的AI主要集中在 解决数学问题,而不是解决实际问题 。
2. **机械化(1970-1980)**:随着 计算机科学的发展,特别是在1970年 代至1980年代,AI研究开始转向解决 实际问题的机器。这一时期,计算机 科学家开始关注如何让计算机能够解 决复杂的数学问题,如线性代数、微 分方程等。
3. **商业化(1990-2000)**:1990 年代,AI研究开始转向商业应用。IBM的艾伦·图灵(Alan Turing)在1990 年代末至2000年初发表的论文《人工 智能》(Artificial Intelligence: A Report from the IBM System Summit in 1997)被广泛认为是AI史上的 里程碑。这篇报告详细探讨了AI的潜 在应用和挑战,包括机器学习、自然 语言处理等。
4. **AI革命(2000至今)**:2000年,Facebook的扎克伯格(Zacharias Facebook)在Twitter上发布了第一条 关于AI的言论,这标志着AI进入了一 个新的时代。随后,Google、微软、 亚马逊等科技巨头也加大了在AI领域 的投入,推动了AI技术的快速发展。
5. **跨领域融合(2010-至今)**: 近年来,AI与生物技术、机器人技术 、增强现实(AR)、自然语言处理(NLP)等领域的交叉融合成为热门话题 。AI在这些领域的应用不仅提高了效 率,还带来了新的可能性和挑战。
人工智能的发展是一个持续的探索和 创新过程,它不仅改变了我们对技术 的理解,还极大地推动了科技的进步 和社会的变革。LLM流式响应已保存到: output\llm_stream_response.json
LLM服务流式调用完成
流式调用测试完成
=== 测试 LLM 带工具调用 ===
[LLM] 使用 requests 直接调用 API
LLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM4-0.5B-QAT-Int4-GPTQ-format
开始调用LLM服务(非流式)
LLM请求已保存到: output\llm_request.json
正在调用LLM服务...
LLM服务响应状态码: 200
LLM响应已保存到: output\llm_response.json
LLM服务调用完成
响应状态: 成功
生成内容: 根据我得到的信息,今天 北京的天气情况是晴朗的,阳光明媚 。天空清澈,没有云层遮挡,阳光普 照,让人感到十分舒适和愉悦。
带工具调用测试完成
=== 测试结果汇总 ===
非流式调用: 通过
流式调用: 通过
带工具调用: 通过
总体结果: 3/3 通过
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
LLM 服务测试脚本
测试 LLMService 类的功能,包括非流式和流式调用
"""
import sys
import os
# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from llm_server.llm_class import LLMService
def test_llm_non_stream():
"""测试 LLM 非流式调用"""
print("=== 测试 LLM 非流式调用 ===")
# 初始化 LLM 服务
llm_service = LLMService()
# 准备测试消息
messages = [
{
"role": "user",
"content": "你好,请介绍一下自己"
}
]
try:
# 调用非流式方法
response = llm_service.create(messages)
# 打印响应结果
print("响应状态: 成功")
print(f"模型: {response.get('model', '未知')}")
# 提取并打印生成的内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"生成内容: {content[:200]}..." if len(content) > 200 else f"生成内容: {content}")
print("非流式调用测试完成\n")
return True
except Exception as e:
print(f"非流式调用测试失败: {str(e)}")
return False
def test_llm_stream():
"""测试 LLM 流式调用"""
print("=== 测试 LLM 流式调用 ===")
# 初始化 LLM 服务
llm_service = LLMService()
# 准备测试消息
messages = [
{
"role": "user",
"content": "请简要介绍一下人工智能的发展历程"
}
]
# 流式回调函数
def stream_callback(chunk_content):
"""流式回调,打印每个 chunk 的内容"""
print(chunk_content, end="", flush=True)
try:
# 调用流式方法
print("开始流式接收响应...")
chunks = []
for chunk in llm_service.create_stream(messages, callback=stream_callback):
chunks.append(chunk)
print("\n流式调用测试完成\n")
return True
except Exception as e:
print(f"流式调用测试失败: {str(e)}")
return False
def test_llm_with_tools():
"""测试 LLM 带工具调用"""
print("=== 测试 LLM 带工具调用 ===")
# 初始化 LLM 服务
llm_service = LLMService()
# 准备测试消息
messages = [
{
"role": "user",
"content": "今天北京的天气怎么样?"
}
]
# 定义工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
},
"required": ["city"]
}
}
}
]
try:
# 调用非流式方法
response = llm_service.create(messages, tools=tools)
# 打印响应结果
print("响应状态: 成功")
# 提取并打印生成的内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
tool_calls = message.get('tool_calls', [])
if tool_calls:
print(f"工具调用: {tool_calls[0].get('function', {}).get('name', '未知')}")
print(f"参数: {tool_calls[0].get('function', {}).get('arguments', '无')}")
else:
print(f"生成内容: {content[:200]}..." if len(content) > 200 else f"生成内容: {content}")
print("带工具调用测试完成\n")
return True
except Exception as e:
error_msg = str(e)
if "enable-auto-tool-choice" in error_msg:
print("⚠️ 工具调用功能未启用")
print("提示: 如需启用工具调用,请在启动 LLM 服务器时添加以下参数:")
print(" --enable-auto-tool-choice --tool-call-parser")
print("工具调用测试已跳过(这是正常的,如果不需要工具调用功能)\n")
return True
else:
print(f"带工具调用测试失败: {error_msg}")
return False
if __name__ == "__main__":
print("开始测试 LLM 服务...\n")
# 运行所有测试
test_results = {
"非流式调用": test_llm_non_stream(),
"流式调用": test_llm_stream(),
"带工具调用": test_llm_with_tools()
}
# 打印测试结果
print("=== 测试结果汇总 ===")
for test_name, result in test_results.items():
status = "通过" if result else "失败"
print(f"{test_name}: {status}")
# 计算总体结果
passed_tests = sum(test_results.values())
total_tests = len(test_results)
print(f"\n总体结果: {passed_tests}/{total_tests} 通过")
VLM
开始测试 VLM 服务...
=== 测试 VLM 单图像识别 ===
VLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM-V-2_6-int4
测试图像已创建: C:\Users\njsgcs\AppData\Local\Temp\tmpkbgl7v2a.png
[VLM调试] 成功收到响应
响应状态: 成功
生成内容: 这张图片包含简单的几何 形状,由两个主要物体组成:一个正 方形和一个圆形。正方形具有鲜艳的 蓝色,并且看起来是填充的。圆形填 充了红色,看起来同样填充着。两个 物体放置在一个普通的白色背景上。 在形状下方有一条水平绿色线。在图 像底部,有一个标签或水印,上面写 着“测试图片”这几个字,这表明该图 像可能是测试绘图能力或分辨率的样 本。
单图像识别测试完成
=== 测试 VLM 多图像识别 ===
VLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM-V-2_6-int4
测试图像 1 已创建: C:\Users\njsgcs\AppData\Local\Temp\tmpelfbz1u1.png
测试图像 2 已创建: C:\Users\njsgcs\AppData\Local\Temp\tmp8x81xh1_.png
[VLM调试] 成功收到响应
响应状态: 成功
生成内容: 当比较这两张图片时,区 别非常明显。第一张图片是一个在蓝 色背景下带有白色文字的矩形,而第 二张图片是一个带有白色文字的纯红 色圆圈。这两张图片都共享相似的布 局和字体,但形状和颜色的显著差异 使它们彼此不同。
多图像识别测试完成
=== 测试 VLM 多轮对话 ===
VLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM-V-2_6-int4
测试图像已创建: C:\Users\njsgcs\AppData\Local\Temp\tmpds0ocw2j.png
第一轮对话:描述图像
[VLM调试] 成功收到响应
AI 回复: 这张图片展示了一个简单的图形组合,由一个蓝色正方形和一个 红色圆形组成,它们位于图片的顶部 ,靠右配对。在下面,有一条水平的 绿色细线,可能表明与形状的对比或 平衡。这张图片看起来像是一个测试 图像,因为底部有文字“测试图像”, 在视觉内容之前提供了一个信息性标 签。
第二轮对话:基于前一轮的问题
[VLM调试] 成功收到响应
AI 回复: 这张图片中的颜色有:
1. 蓝色(正方形)
2. 红色(圆形)
3. 绿色(细线)
多轮对话测试完成
=== 测试 VLM 直接使用图像内容格式 ===
VLM服务初始化完成,模型: /root/my_python_server/models/OpenBMB_MiniCPM-V-2_6-int4
测试图像已创建: C:\Users\njsgcs\AppData\Local\Temp\tmpd2opf2vx.png
[VLM调试] 成功收到响应
响应状态: 成功
生成内容: 这张图片是一个简单的几 何图形构成的测试图,可能用于视觉 测试或显示校准。中间有一个大的蓝 色正方形和一个红色的圆,是这两个 形状中唯一的颜色。蓝色正方形位于 图片的左侧,红色圆位于右侧,两者 的顶部边缘相接触。图片的底部有一 条绿色线,表明测试图像的底部边框 ,也起到用来关闭图像或显示更多信 息的空间。整体的设计简单,缺乏任 何纹理或阴影,确保形状和颜色是最 显著的特征。
直接图像内容格式测试完成
=== 测试结果汇总 ===
单图像识别: 通过
多图像识别: 通过
多轮对话: 通过
直接图像内容格式: 通过
总体结果: 4/4 通过
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
VLM 服务测试脚本
测试 VLMService 类的功能,包括单图像、多图像和多轮对话
"""
import sys
import os
import tempfile
from PIL import Image, ImageDraw, ImageFont
# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from llm_server.llm_class import VLMService
def create_test_image(image_path):
"""
创建一个测试图像
Args:
image_path: 图像保存路径
"""
# 创建一个 400x300 的图像
img = Image.new('RGB', (400, 300), color='white')
d = ImageDraw.Draw(img)
# 绘制一些基本图形
d.rectangle([50, 50, 150, 150], fill='blue')
d.circle([250, 100], 50, fill='red')
d.line([50, 200, 350, 200], fill='green', width=5)
# 添加文本
try:
# 尝试使用默认字体
font = ImageFont.load_default()
d.text((50, 220), "Test Image", fill='black', font=font)
except Exception:
# 如果没有字体,跳过文本
pass
img.save(image_path)
print(f"测试图像已创建: {image_path}")
def test_vlm_single_image():
"""测试 VLM 单图像识别"""
print("=== 测试 VLM 单图像识别 ===")
# 初始化 VLM 服务
vlm_service = VLMService()
# 创建测试图像
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp:
test_image_path = tmp.name
try:
create_test_image(test_image_path)
# 准备测试消息
messages = [
{
"role": "user",
"content": "描述一下这张图片的内容"
}
]
# 调用 VLM 服务
try:
response = vlm_service.create_with_image(messages, image_source=test_image_path)
# 打印响应结果
print("响应状态: 成功")
# 提取并打印生成的内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"生成内容: {content[:300]}..." if len(content) > 300 else f"生成内容: {content}")
print("单图像识别测试完成\n")
return True
except Exception as e:
print(f"单图像识别测试失败: {str(e)}")
return False
finally:
# 清理测试图像
if os.path.exists(test_image_path):
os.unlink(test_image_path)
def test_vlm_multiple_images():
"""测试 VLM 多图像识别"""
print("=== 测试 VLM 多图像识别 ===")
# 初始化 VLM 服务
vlm_service = VLMService()
# 创建测试图像
test_image_paths = []
try:
# 创建两张测试图像
for i in range(2):
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp:
test_image_path = tmp.name
test_image_paths.append(test_image_path)
# 创建不同内容的图像
img = Image.new('RGB', (400, 300), color='white')
d = ImageDraw.Draw(img)
if i == 0:
# 第一张图像:蓝色方块
d.rectangle([50, 50, 350, 250], fill='blue')
d.text((150, 130), "Image 1", fill='white')
else:
# 第二张图像:红色圆形
d.ellipse([100, 50, 300, 250], fill='red')
d.text((150, 130), "Image 2", fill='white')
img.save(test_image_path)
print(f"测试图像 {i+1} 已创建: {test_image_path}")
# 准备测试消息
messages = [
{
"role": "user",
"content": "比较这两张图片的不同之处"
}
]
# 调用 VLM 服务
try:
response = vlm_service.create_with_multiple_images(messages, image_sources=test_image_paths)
# 打印响应结果
print("响应状态: 成功")
# 提取并打印生成的内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"生成内容: {content[:300]}..." if len(content) > 300 else f"生成内容: {content}")
print("多图像识别测试完成\n")
return True
except Exception as e:
print(f"多图像识别测试失败: {str(e)}")
return False
finally:
# 清理测试图像
for img_path in test_image_paths:
if os.path.exists(img_path):
os.unlink(img_path)
def test_vlm_multimodal_conversation():
"""测试 VLM 多轮对话"""
print("=== 测试 VLM 多轮对话 ===")
# 初始化 VLM 服务
vlm_service = VLMService()
# 创建测试图像
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp:
test_image_path = tmp.name
try:
create_test_image(test_image_path)
# 准备历史对话
conversation_history = []
# 第一轮对话:描述图像
print("第一轮对话:描述图像")
current_query = "描述一下这张图片的内容"
try:
# 调用 VLM 服务
response = vlm_service.create_multimodal_conversation(
conversation_history,
current_query,
image_source=test_image_path
)
# 提取响应内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"AI 回复: {content[:200]}..." if len(content) > 200 else f"AI 回复: {content}")
# 添加到历史对话
conversation_history.append({'role': 'user', 'content': current_query})
conversation_history.append({'role': 'assistant', 'content': content})
# 第二轮对话:基于前一轮的问题
print("\n第二轮对话:基于前一轮的问题")
current_query = "图片中有哪些颜色?"
response = vlm_service.create_multimodal_conversation(
conversation_history,
current_query
)
# 提取响应内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"AI 回复: {content[:200]}..." if len(content) > 200 else f"AI 回复: {content}")
print("多轮对话测试完成\n")
return True
except Exception as e:
print(f"多轮对话测试失败: {str(e)}")
return False
finally:
# 清理测试图像
if os.path.exists(test_image_path):
os.unlink(test_image_path)
def test_vlm_direct_image_content():
"""测试 VLM 直接使用图像内容格式"""
print("=== 测试 VLM 直接使用图像内容格式 ===")
# 初始化 VLM 服务
vlm_service = VLMService()
# 创建测试图像
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp:
test_image_path = tmp.name
try:
create_test_image(test_image_path)
# 读取图像并转换为base64
import base64
from io import BytesIO
with Image.open(test_image_path) as img:
buffered = BytesIO()
img.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode('utf-8')
# 构建data URI
image_data = f"data:image/png;base64,{img_str}"
# 准备测试消息(直接包含图像内容)
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": "描述一下这张图片的内容"
},
{
"type": "image_url",
"image_url": {
"url": image_data
}
}
]
}
]
# 调用 VLM 服务
try:
response = vlm_service.create_with_image(messages)
# 打印响应结果
print("响应状态: 成功")
# 提取并打印生成的内容
choices = response.get('choices', [])
if choices:
message = choices[0].get('message', {})
content = message.get('content', '无内容')
print(f"生成内容: {content[:300]}..." if len(content) > 300 else f"生成内容: {content}")
print("直接图像内容格式测试完成\n")
return True
except Exception as e:
print(f"直接图像内容格式测试失败: {str(e)}")
return False
finally:
# 清理测试图像
if os.path.exists(test_image_path):
os.unlink(test_image_path)
if __name__ == "__main__":
print("开始测试 VLM 服务...\n")
# 运行所有测试
test_results = {
"单图像识别": test_vlm_single_image(),
"多图像识别": test_vlm_multiple_images(),
"多轮对话": test_vlm_multimodal_conversation(),
"直接图像内容格式": test_vlm_direct_image_content()
}
# 打印测试结果
print("=== 测试结果汇总 ===")
for test_name, result in test_results.items():
status = "通过" if result else "失败"
print(f"{test_name}: {status}")
# 计算总体结果
passed_tests = sum(test_results.values())
total_tests = len(test_results)
print(f"\n总体结果: {passed_tests}/{total_tests} 通过")
更多推荐
所有评论(0)