从研究到生产：DeepSeek-R1-Distill-Qwen-1.5B模型上线完整路径

1. 为什么需要轻量化模型部署？

如果你正在做AI项目，可能会遇到这样的问题：大模型效果确实好，但部署成本太高，推理速度太慢，普通服务器根本跑不动。这时候，轻量化模型就成了救命稻草。

今天要聊的DeepSeek-R1-Distill-Qwen-1.5B，就是这样一个既轻便又好用的模型。它只有15亿参数，但经过精心优化，在很多任务上的表现不输给那些动辄几十亿参数的大模型。

想象一下，你有一个电商客服系统，需要实时回答用户问题。如果用几百亿参数的大模型，每回答一个问题可能要等好几秒，服务器成本还特别高。但用这个1.5B的模型，响应速度能快好几倍，成本也大幅降低，这就是轻量化模型的价值。

2. 认识DeepSeek-R1-Distill-Qwen-1.5B

2.1 模型的核心特点

这个模型是DeepSeek团队基于Qwen2.5-Math-1.5B基础模型，通过知识蒸馏技术打造的轻量化版本。简单来说，就是让一个小模型学会大模型的本事。

参数效率优化：模型参数量压缩到了1.5B级别，但保持了85%以上的原始模型精度。这就像把一本厚厚的百科全书，压缩成一本精华手册，关键信息都在，但体积小了很多。

任务适配增强：在训练过程中加入了特定领域的数据，比如法律文书、医疗问诊等。这让模型在垂直场景下的表现提升了12-15个百分点。你可以理解为，它不仅是通用助手，还在某些专业领域有特长。

硬件友好性：支持INT8量化部署，内存占用比原来的FP32模式降低了75%。这意味着你可以在普通的NVIDIA T4显卡上就跑起来，甚至一些边缘设备也能用。

2.2 模型的使用建议

根据官方建议，使用这个模型时需要注意几个关键点：

温度设置：建议设置在0.5-0.7之间，推荐0.6。温度太高容易产生重复或不连贯的输出，太低又会让回答太死板。0.6是个比较平衡的值。

提示词格式：所有指令都应该放在用户提示中，不需要单独的系统提示。对于数学问题，建议在提示里加上：“请逐步推理，并将最终答案放在\boxed{}内。”

推理模式：模型有时候会跳过思考过程直接输出答案。为了确保它充分推理，建议在每次输出开始时强制使用“\n”换行符。

3. 环境准备与快速部署

3.1 准备工作

在开始之前，你需要确保环境满足以下要求：

• 操作系统：Linux（Ubuntu 20.04或更高版本）
• 显卡：NVIDIA GPU，至少8GB显存
• Python：3.8或更高版本
• CUDA：11.8或更高版本

如果你用的是云服务器，建议选择带GPU的实例。个人电脑的话，需要有一块不错的显卡。

3.2 安装依赖

首先创建并激活虚拟环境：

# 创建虚拟环境
python -m venv deepseek_env

# 激活环境
source deepseek_env/bin/activate

# 安装PyTorch（根据你的CUDA版本选择）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

# 安装vLLM
pip install vllm

# 安装其他依赖
pip install openai requests

vLLM是一个专门为大规模语言模型推理优化的库，它能显著提升推理速度，减少内存占用。相比直接使用transformers库，vLLM的吞吐量能提升好几倍。

4. 使用vLLM启动模型服务

4.1 启动服务

启动模型服务其实很简单，一行命令就能搞定：

# 进入工作目录
cd /root/workspace

# 启动vLLM服务
python -m vllm.entrypoints.openai.api_server \
    --model DeepSeek-R1-Distill-Qwen-1.5B \
    --served-model-name DeepSeek-R1-Distill-Qwen-1.5B \
    --port 8000 \
    --host 0.0.0.0 \
    --tensor-parallel-size 1 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 4096 \
    --quantization int8

让我解释一下这些参数的含义：

• --model：指定要加载的模型名称
• --port：服务监听的端口号，默认是8000
• --host 0.0.0.0：允许所有IP访问，方便远程调用
• --tensor-parallel-size 1：使用单卡推理，如果你有多张显卡可以调整这个值
• --gpu-memory-utilization 0.9：GPU内存使用率，0.9表示使用90%的显存
• --quantization int8：使用INT8量化，能大幅减少内存占用

启动命令可以保存到脚本里，方便以后使用：

#!/bin/bash
# start_deepseek.sh

cd /root/workspace

# 将输出重定向到日志文件
python -m vllm.entrypoints.openai.api_server \
    --model DeepSeek-R1-Distill-Qwen-1.5B \
    --served-model-name DeepSeek-R1-Distill-Qwen-1.5B \
    --port 8000 \
    --host 0.0.0.0 \
    --tensor-parallel-size 1 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 4096 \
    --quantization int8 > deepseek_qwen.log 2>&1 &

4.2 查看启动状态

启动后，需要确认服务是否正常运行：

# 查看启动日志
cat deepseek_qwen.log

# 或者实时查看日志
tail -f deepseek_qwen.log

正常启动后，你应该能看到类似这样的信息：

INFO 07-15 14:30:15 llm_engine.py:72] Initializing an LLM engine with config: ...
INFO 07-15 14:30:20 model_runner.py:63] Loading model weights...
INFO 07-15 14:32:10 model_runner.py:168] Model loaded successfully.
INFO 07-15 14:32:11 api_server.py:210] Started server process [12345]
INFO 07-15 14:32:11 api_server.py:211] Uvicorn running on http://0.0.0.0:8000

看到最后一行显示服务在8000端口运行，就说明启动成功了。

5. 测试模型服务

5.1 基础测试代码

服务启动后，我们可以写个简单的测试脚本来验证功能：

from openai import OpenAI
import time


class DeepSeekClient:
    def __init__(self, base_url="http://localhost:8000/v1"):
        """初始化客户端"""
        self.client = OpenAI(
            base_url=base_url,
            api_key="none"  # vLLM本地部署不需要API密钥
        )
        self.model = "DeepSeek-R1-Distill-Qwen-1.5B"
        
    def health_check(self):
        """健康检查"""
        try:
            # 尝试获取模型列表
            models = self.client.models.list()
            print(f"服务连接成功，可用模型: {[m.id for m in models.data]}")
            return True
        except Exception as e:
            print(f"服务连接失败: {e}")
            return False
    
    def simple_chat(self, prompt, system_prompt=None, temperature=0.6):
        """简单对话测试"""
        messages = []
        
        # 注意：根据官方建议，系统提示应该放在用户消息中
        if system_prompt:
            # 将系统提示作为第一条用户消息
            messages.append({"role": "user", "content": system_prompt})
            # 添加一个空的助手回复
            messages.append({"role": "assistant", "content": "好的，我明白了。"})
        
        # 添加真正的用户问题
        messages.append({"role": "user", "content": prompt})
        
        try:
            start_time = time.time()
            
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                max_tokens=512
            )
            
            end_time = time.time()
            elapsed_time = end_time - start_time
            
            if response.choices:
                content = response.choices[0].message.content
                print(f"问题: {prompt}")
                print(f"回答: {content}")
                print(f"耗时: {elapsed_time:.2f}秒")
                print(f"使用token数: {response.usage.total_tokens}")
                print("-" * 50)
                return content
                
        except Exception as e:
            print(f"请求失败: {e}")
            return None


# 测试函数
def run_basic_tests():
    """运行基础测试"""
    client = DeepSeekClient()
    
    # 1. 健康检查
    print("=== 服务健康检查 ===")
    if not client.health_check():
        print("服务不可用，请检查vLLM是否正常启动")
        return
    
    print("\n=== 基础功能测试 ===")
    
    # 2. 测试不同场景
    test_cases = [
        {
            "prompt": "请用中文介绍一下人工智能的发展历史",
            "system": "你是一个知识渊博的AI助手，请用简洁明了的语言回答。"
        },
        {
            "prompt": "计算：15 × 24 + 38 ÷ 2",
            "system": "请逐步推理，并将最终答案放在\\boxed{}内。"
        },
        {
            "prompt": "写一首关于春天的五言绝句",
            "system": "你是一个诗人，请创作优美的诗歌。"
        },
        {
            "prompt": "用Python写一个快速排序算法",
            "system": "你是一个编程专家，请提供清晰可运行的代码。"
        }
    ]
    
    for i, test in enumerate(test_cases, 1):
        print(f"\n测试 {i}:")
        client.simple_chat(test["prompt"], test["system"])


if __name__ == "__main__":
    run_basic_tests()

5.2 流式对话测试

流式对话能让用户体验更好，特别是生成长文本时：

def stream_chat_demo():
    """流式对话演示"""
    client = DeepSeekClient()
    
    messages = [
        {"role": "user", "content": "你是一个旅行规划师"},
        {"role": "assistant", "content": "好的，我将作为您的旅行规划师。"},
        {"role": "user", "content": "请为我规划一个三天的北京行程，要包含故宫、长城和颐和园。"}
    ]
    
    print("AI: ", end="", flush=True)
    
    try:
        # 创建流式响应
        stream = client.client.chat.completions.create(
            model=client.model,
            messages=messages,
            temperature=0.6,
            max_tokens=1024,
            stream=True
        )
        
        full_response = ""
        for chunk in stream:
            if chunk.choices[0].delta.content is not None:
                content = chunk.choices[0].delta.content
                print(content, end="", flush=True)
                full_response += content
        
        print()  # 换行
        return full_response
        
    except Exception as e:
        print(f"流式对话错误: {e}")
        return ""


# 运行流式对话测试
print("=== 流式对话测试 ===")
stream_chat_demo()

5.3 批量测试与性能评估

如果你要评估模型性能，可以运行批量测试：

import json
from concurrent.futures import ThreadPoolExecutor
import statistics


class PerformanceTester:
    def __init__(self):
        self.client = DeepSeekClient()
        
    def single_test(self, prompt, test_id):
        """单次测试"""
        start_time = time.time()
        
        response = self.client.client.chat.completions.create(
            model=self.client.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.6,
            max_tokens=100
        )
        
        end_time = time.time()
        elapsed_time = end_time - start_time
        
        return {
            "test_id": test_id,
            "time": elapsed_time,
            "tokens": response.usage.total_tokens,
            "success": True
        }
    
    def batch_test(self, num_requests=10, concurrent=3):
        """批量性能测试"""
        prompts = [f"测试问题 {i+1}: 人工智能是什么？" for i in range(num_requests)]
        results = []
        
        print(f"开始批量测试，共{num_requests}个请求，并发数{concurrent}")
        
        with ThreadPoolExecutor(max_workers=concurrent) as executor:
            futures = []
            for i, prompt in enumerate(prompts):
                future = executor.submit(self.single_test, prompt, i)
                futures.append(future)
            
            for future in futures:
                try:
                    result = future.result(timeout=30)
                    results.append(result)
                    print(f"请求 {result['test_id']+1} 完成，耗时 {result['time']:.2f}秒")
                except Exception as e:
                    print(f"请求失败: {e}")
        
        # 统计结果
        if results:
            times = [r["time"] for r in results]
            tokens = [r["tokens"] for r in results]
            
            print(f"\n=== 性能统计 ===")
            print(f"总请求数: {len(results)}")
            print(f"平均响应时间: {statistics.mean(times):.2f}秒")
            print(f"最短响应时间: {min(times):.2f}秒")
            print(f"最长响应时间: {max(times):.2f}秒")
            print(f"平均token数: {statistics.mean(tokens):.0f}")
            print(f"QPS (每秒查询数): {len(results)/sum(times):.2f}")
        
        return results


# 运行性能测试
if __name__ == "__main__":
    tester = PerformanceTester()
    
    # 先做健康检查
    if tester.client.health_check():
        # 运行性能测试（测试10个请求，并发3个）
        tester.batch_test(num_requests=10, concurrent=3)

6. 实际应用场景示例

6.1 智能客服系统

这个模型很适合做智能客服，响应快、成本低：

class CustomerServiceBot:
    def __init__(self):
        self.client = DeepSeekClient()
        self.conversation_history = []
        
    def add_to_history(self, role, content):
        """添加对话历史"""
        self.conversation_history.append({"role": role, "content": content})
        # 保持最近10轮对话
        if len(self.conversation_history) > 20:
            self.conversation_history = self.conversation_history[-20:]
    
    def get_response(self, user_query):
        """获取客服回复"""
        # 添加系统提示（作为第一条用户消息）
        system_prompt = """你是专业的电商客服助手，请用友好、专业的态度回答用户问题。
        如果遇到无法回答的问题，请如实告知并建议用户联系人工客服。
        保持回答简洁明了，重点突出。"""
        
        messages = [
            {"role": "user", "content": system_prompt},
            {"role": "assistant", "content": "好的，我将作为您的电商客服助手。"}
        ]
        
        # 添加上下文历史
        messages.extend(self.conversation_history[-6:])  # 最近3轮对话
        
        # 添加当前问题
        messages.append({"role": "user", "content": user_query})
        
        try:
            response = self.client.client.chat.completions.create(
                model=self.client.model,
                messages=messages,
                temperature=0.5,  # 客服场景温度低一些，更稳定
                max_tokens=256
            )
            
            if response.choices:
                reply = response.choices[0].message.content
                # 记录对话历史
                self.add_to_history("user", user_query)
                self.add_to_history("assistant", reply)
                return reply
                
        except Exception as e:
            return f"抱歉，系统暂时无法处理您的请求。错误: {str(e)}"
    
    def handle_common_questions(self):
        """处理常见问题示例"""
        test_questions = [
            "我的订单什么时候发货？",
            "商品有质量问题怎么退货？",
            "快递运费是多少？",
            "支持哪些支付方式？",
            "会员有什么优惠？"
        ]
        
        print("=== 智能客服测试 ===")
        for question in test_questions:
            print(f"\n用户: {question}")
            response = self.get_response(question)
            print(f"客服: {response}")
            time.sleep(1)  # 避免请求过快


# 运行客服测试
bot = CustomerServiceBot()
bot.handle_common_questions()

6.2 内容生成助手

对于内容创作场景，模型也能提供很好的支持：

class ContentGenerator:
    def __init__(self):
        self.client = DeepSeekClient()
    
    def generate_article(self, topic, style="专业"):
        """生成文章"""
        prompt = f"""请以{style}的风格，写一篇关于{topic}的文章。
        要求：
        1. 文章结构清晰，有引言、正文和结论
        2. 字数在500字左右
        3. 内容要有价值，信息准确
        4. 语言流畅，符合中文表达习惯"""
        
        response = self.client.client.chat.completions.create(
            model=self.client.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,  # 创作场景温度可以稍高
            max_tokens=1024
        )
        
        if response.choices:
            return response.choices[0].message.content
        return "生成失败"
    
    def generate_social_media_post(self, product, platform="微信"):
        """生成社交媒体文案"""
        prompt = f"""为{product}写一个{platform}平台的推广文案。
        要求：
        1. 吸引眼球，引发兴趣
        2. 突出产品卖点
        3. 适合{platform}平台的风格
        4. 包含合适的标签和表情符号建议
        5. 字数在200字以内"""
        
        response = self.client.client.chat.completions.create(
            model=self.client.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.8,  # 创意内容温度可以更高
            max_tokens=300
        )
        
        if response.choices:
            return response.choices[0].message.content
        return "生成失败"


# 测试内容生成
generator = ContentGenerator()

print("=== 文章生成测试 ===")
article = generator.generate_article("人工智能在医疗领域的应用", "科普")
print(article[:500] + "...")  # 只显示前500字

print("\n=== 社交媒体文案测试 ===")
post = generator.generate_social_media_post("智能手环", "小红书")
print(post)

6.3 代码助手

对于开发者来说，代码生成和解释功能也很实用：

class CodeAssistant:
    def __init__(self):
        self.client = DeepSeekClient()
    
    def explain_code(self, code, language="python"):
        """解释代码功能"""
        prompt = f"""请解释以下{language}代码的功能和实现原理：
        
        {code}
        
        请用简单易懂的语言说明：
        1. 这段代码是做什么的？
        2. 关键函数或方法的作用是什么？
        3. 代码的执行流程是怎样的？
        4. 有什么需要注意的地方？"""
        
        response = self.client.client.chat.completions.create(
            model=self.client.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.6,
            max_tokens=512
        )
        
        if response.choices:
            return response.choices[0].message.content
        return "解释失败"
    
    def generate_code(self, requirement, language="python"):
        """根据需求生成代码"""
        prompt = f"""请用{language}实现以下需求：
        
        {requirement}
        
        要求：
        1. 代码要完整，可以直接运行
        2. 添加必要的注释
        3. 考虑错误处理
        4. 代码风格要规范"""
        
        response = self.client.client.chat.completions.create(
            model=self.client.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5,  # 代码生成温度低一些，更准确
            max_tokens=1024
        )
        
        if response.choices:
            return response.choices[0].message.content
        return "生成失败"


# 测试代码助手
assistant = CodeAssistant()

# 测试代码解释
sample_code = """
def quick_sort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    return quick_sort(left) + middle + quick_sort(right)
"""

print("=== 代码解释测试 ===")
explanation = assistant.explain_code(sample_code)
print(explanation)

print("\n=== 代码生成测试 ===")
requirement = "写一个函数，接收一个URL，返回该网页的标题和所有链接"
generated_code = assistant.generate_code(requirement)
print(generated_code)

7. 部署优化与监控

7.1 性能优化配置

为了让服务运行更稳定，可以调整一些配置参数：

# 优化后的启动脚本
optimized_start_script = """#!/bin/bash
# optimized_start_deepseek.sh

cd /root/workspace

# 设置环境变量
export VLLM_WORKER_MULTIPROC_METHOD=spawn
export VLLM_ATTENTION_BACKEND=XFORMERS

# 启动服务，使用优化参数
python -m vllm.entrypoints.openai.api_server \\
    --model DeepSeek-R1-Distill-Qwen-1.5B \\
    --served-model-name DeepSeek-R1-Distill-Qwen-1.5B \\
    --port 8000 \\
    --host 0.0.0.0 \\
    --tensor-parallel-size 1 \\
    --gpu-memory-utilization 0.85 \\
    --max-model-len 2048 \\
    --max-num-batched-tokens 4096 \\
    --max-num-seqs 16 \\
    --quantization int8 \\
    --enforce-eager \\
    --disable-custom-all-reduce \\
    > deepseek_qwen.log 2>&1 &

echo "服务启动中，PID: $!"
echo "查看日志: tail -f deepseek_qwen.log"
"""

# 保存优化脚本
with open("optimized_start_deepseek.sh", "w") as f:
    f.write(optimized_start_script)

print("优化启动脚本已生成")

主要优化点：

• --gpu-memory-utilization 0.85：稍微降低内存使用率，留出更多余量
• --max-model-len 2048：根据实际需求设置最大长度
• --max-num-batched-tokens 4096：控制批量处理的token数
• --max-num-seqs 16：限制并发序列数
• --enforce-eager：禁用图优化，提高稳定性
• --disable-custom-all-reduce：禁用自定义all-reduce，避免兼容性问题

7.2 服务监控脚本

部署后需要监控服务状态：

import psutil
import requests
import time
from datetime import datetime


class ServiceMonitor:
    def __init__(self, service_url="http://localhost:8000"):
        self.service_url = service_url
        self.log_file = "service_monitor.log"
    
    def check_service_health(self):
        """检查服务健康状态"""
        try:
            response = requests.get(f"{self.service_url}/health", timeout=5)
            return response.status_code == 200
        except:
            return False
    
    def check_gpu_memory(self):
        """检查GPU内存使用情况"""
        try:
            import pynvml
            pynvml.nvmlInit()
            handle = pynvml.nvmlDeviceGetHandleByIndex(0)
            info = pynvml.nvmlDeviceGetMemoryInfo(handle)
            return {
                "total": info.total / 1024**3,  # GB
                "used": info.used / 1024**3,
                "free": info.free / 1024**3,
                "usage_percent": (info.used / info.total) * 100
            }
        except:
            return None
    
    def check_system_resources(self):
        """检查系统资源"""
        return {
            "cpu_percent": psutil.cpu_percent(interval=1),
            "memory_percent": psutil.virtual_memory().percent,
            "disk_percent": psutil.disk_usage("/").percent
        }
    
    def log_status(self, status="INFO", message=""):
        """记录状态日志"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"[{timestamp}] [{status}] {message}\n"
        
        with open(self.log_file, "a") as f:
            f.write(log_entry)
        
        print(log_entry.strip())
    
    def monitor_loop(self, interval=60):
        """监控循环"""
        print(f"开始监控服务，检查间隔: {interval}秒")
        
        while True:
            try:
                # 检查服务健康
                is_healthy = self.check_service_health()
                
                # 检查资源使用
                gpu_info = self.check_gpu_memory()
                sys_info = self.check_system_resources()
                
                # 记录状态
                status_msg = f"服务状态: {'健康' if is_healthy else '异常'}"
                if gpu_info:
                    status_msg += f", GPU内存: {gpu_info['usage_percent']:.1f}%"
                status_msg += f", CPU: {sys_info['cpu_percent']:.1f}%, 内存: {sys_info['memory_percent']:.1f}%"
                
                self.log_status("INFO", status_msg)
                
                # 如果服务异常，尝试重启
                if not is_healthy:
                    self.log_status("ERROR", "服务异常，尝试重启...")
                    # 这里可以添加重启逻辑
                
                time.sleep(interval)
                
            except KeyboardInterrupt:
                self.log_status("INFO", "监控停止")
                break
            except Exception as e:
                self.log_status("ERROR", f"监控错误: {str(e)}")
                time.sleep(interval)


# 启动监控
if __name__ == "__main__":
    monitor = ServiceMonitor()
    
    # 检查必要依赖
    try:
        import pynvml
        print("GPU监控可用")
    except:
        print("未安装pynvml，GPU监控不可用")
    
    # 运行监控（每5分钟检查一次）
    monitor.monitor_loop(interval=300)

7.3 自动化部署脚本

对于生产环境，可以编写完整的部署脚本：

# deploy_service.py
import os
import subprocess
import time
import sys


class DeepSeekDeployer:
    def __init__(self, workspace="/root/workspace"):
        self.workspace = workspace
        self.log_file = os.path.join(workspace, "deployment.log")
        
    def log(self, message):
        """记录日志"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"[{timestamp}] {message}"
        print(log_entry)
        
        with open(self.log_file, "a") as f:
            f.write(log_entry + "\n")
    
    def check_environment(self):
        """检查环境"""
        self.log("检查环境...")
        
        # 检查Python版本
        python_version = sys.version_info
        if python_version.major < 3 or (python_version.major == 3 and python_version.minor < 8):
            self.log("错误: Python版本需要3.8或更高")
            return False
        
        # 检查CUDA
        try:
            result = subprocess.run(["nvidia-smi"], capture_output=True, text=True)
            if "NVIDIA-SMI" in result.stdout:
                self.log("CUDA环境正常")
            else:
                self.log("警告: 未检测到NVIDIA GPU")
        except:
            self.log("警告: nvidia-smi命令不可用")
        
        return True
    
    def install_dependencies(self):
        """安装依赖"""
        self.log("安装依赖...")
        
        dependencies = [
            "torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118",
            "vllm",
            "openai",
            "requests",
            "psutil"
        ]
        
        for dep in dependencies:
            self.log(f"安装: {dep}")
            try:
                subprocess.run([sys.executable, "-m", "pip", "install"] + dep.split(), 
                             check=True, capture_output=True)
            except subprocess.CalledProcessError as e:
                self.log(f"安装失败: {dep}, 错误: {e}")
                return False
        
        self.log("依赖安装完成")
        return True
    
    def create_startup_script(self):
        """创建启动脚本"""
        script_content = """#!/bin/bash
# deepseek_service.sh

WORKSPACE="/root/workspace"
LOG_FILE="$WORKSPACE/deepseek_service.log"
PID_FILE="$WORKSPACE/deepseek_service.pid"

cd $WORKSPACE

# 检查是否已运行
if [ -f "$PID_FILE" ]; then
    PID=$(cat $PID_FILE)
    if ps -p $PID > /dev/null 2>&1; then
        echo "服务已在运行 (PID: $PID)"
        exit 0
    fi
fi

# 启动服务
echo "启动DeepSeek服务..."
python -m vllm.entrypoints.openai.api_server \\
    --model DeepSeek-R1-Distill-Qwen-1.5B \\
    --served-model-name DeepSeek-R1-Distill-Qwen-1.5B \\
    --port 8000 \\
    --host 0.0.0.0 \\
    --tensor-parallel-size 1 \\
    --gpu-memory-utilization 0.85 \\
    --max-model-len 2048 \\
    --quantization int8 \\
    > $LOG_FILE 2>&1 &

# 保存PID
echo $! > $PID_FILE
echo "服务启动成功，PID: $!"
echo "日志文件: $LOG_FILE"
echo "测试命令: curl http://localhost:8000/v1/models"
"""

        script_path = os.path.join(self.workspace, "deepseek_service.sh")
        with open(script_path, "w") as f:
            f.write(script_content)
        
        # 添加执行权限
        os.chmod(script_path, 0o755)
        
        self.log(f"启动脚本已创建: {script_path}")
        return script_path
    
    def create_test_script(self):
        """创建测试脚本"""
        test_content = """#!/usr/bin/env python3
# test_service.py

import requests
import json
import time


def test_service():
    '''测试服务是否正常'''
    print("测试DeepSeek服务...")
    
    # 测试1: 健康检查
    print("1. 健康检查...")
    try:
        response = requests.get("http://localhost:8000/health", timeout=10)
        if response.status_code == 200:
            print("   ✓ 服务健康")
        else:
            print(f"   ✗ 服务异常: {response.status_code}")
            return False
    except Exception as e:
        print(f"   ✗ 连接失败: {e}")
        return False
    
    # 测试2: 模型列表
    print("2. 获取模型列表...")
    try:
        response = requests.get("http://localhost:8000/v1/models", timeout=10)
        models = response.json()
        print(f"   ✓ 可用模型: {[m['id'] for m in models['data']]}")
    except Exception as e:
        print(f"   ✗ 获取模型失败: {e}")
        return False
    
    # 测试3: 对话测试
    print("3. 对话测试...")
    try:
        payload = {
            "model": "DeepSeek-R1-Distill-Qwen-1.5B",
            "messages": [
                {"role": "user", "content": "你好，请简单介绍一下自己"}
            ],
            "max_tokens": 100,
            "temperature": 0.6
        }
        
        start_time = time.time()
        response = requests.post(
            "http://localhost:8000/v1/chat/completions",
            json=payload,
            timeout=30
        )
        end_time = time.time()
        
        if response.status_code == 200:
            result = response.json()
            reply = result['choices'][0]['message']['content']
            print(f"   ✓ 响应时间: {end_time-start_time:.2f}秒")
            print(f"   ✓ 模型回复: {reply[:50]}...")
        else:
            print(f"   ✗ 对话失败: {response.status_code}")
            return False
            
    except Exception as e:
        print(f"   ✗ 对话测试失败: {e}")
        return False
    
    print("\n所有测试通过！服务运行正常。")
    return True


if __name__ == "__main__":
    if test_service():
        exit(0)
    else:
        exit(1)
"""

        test_path = os.path.join(self.workspace, "test_service.py")
        with open(test_path, "w") as f:
            f.write(test_content)
        
        os.chmod(test_path, 0o755)
        self.log(f"测试脚本已创建: {test_path}")
        return test_path
    
    def deploy(self):
        """执行部署"""
        self.log("开始部署DeepSeek-R1-Distill-Qwen-1.5B服务")
        
        # 创建workspace
        os.makedirs(self.workspace, exist_ok=True)
        
        # 检查环境
        if not self.check_environment():
            self.log("环境检查失败")
            return False
        
        # 安装依赖
        if not self.install_dependencies():
            self.log("依赖安装失败")
            return False
        
        # 创建脚本
        startup_script = self.create_startup_script()
        test_script = self.create_test_script()
        
        self.log("\n部署完成！")
        self.log(f"启动服务: bash {startup_script}")
        self.log(f"测试服务: python {test_script}")
        self.log(f"查看日志: tail -f {os.path.join(self.workspace, 'deepseek_service.log')}")
        
        return True


# 执行部署
if __name__ == "__main__":
    deployer = DeepSeekDeployer()
    deployer.deploy()

8. 总结

通过上面的步骤，我们完成了DeepSeek-R1-Distill-Qwen-1.5B模型从研究到生产的完整部署流程。整个过程可以分为几个关键阶段：

环境准备阶段：确保有合适的硬件和软件环境，安装必要的依赖包。这个阶段最重要的是选择正确的CUDA版本和PyTorch版本，避免兼容性问题。

服务部署阶段：使用vLLM启动模型服务，这是整个流程的核心。vLLM提供了高效的推理引擎，能充分发挥GPU性能。关键是要根据实际硬件配置调整参数，比如GPU内存使用率、最大序列长度等。

测试验证阶段：编写测试脚本验证服务是否正常工作。包括基础功能测试、性能测试、以及针对不同应用场景的测试。这个阶段要确保服务稳定可靠，能够处理实际请求。

应用开发阶段：基于部署好的服务开发实际应用。我们展示了智能客服、内容生成、代码助手等场景的实现方法。在实际项目中，可以根据需求调整提示词、温度参数等，获得最佳效果。

监控维护阶段：生产环境需要持续的监控和维护。我们提供了监控脚本和自动化部署脚本，帮助大家更好地管理服务。

这个1.5B的模型虽然参数不多，但在很多实际场景中表现相当不错。它的最大优势是部署成本低、推理速度快，特别适合对实时性要求高、预算有限的场景。

如果你刚开始接触模型部署，建议先从基础功能开始，逐步扩展到更复杂的应用。遇到问题时，多查看日志文件，通常能找到解决方案。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。