大家好，我是herosunly。985院校硕士毕业，现担任算法工程师一职，获得CSDN博客之星第一名，热衷于大模型算法的研究与应用。曾担任百度千帆大模型比赛、BPAA算法大赛评委，编写微软OpenAI考试认证指导手册。曾获得多项AI顶级比赛的Top名次，其中包括阿里云、科大讯飞比赛第一名，CCF、开放原子比赛二等奖。在技术创新领域拥有多项授权发明。曾辅导多位非科班出身的同学成功进入算法行业就业。希望和大家一起成长进步。

  本文主要介绍了飞桨框架3.0解锁DeepSeek部署全流程极简体验，希望对使用大语言模型的同学们有所帮助。

一、前言：开启大模型部署的极简时代

  在大模型技术爆发式增长的今天，如何将训练成果高效、低成本地转化为实际应用，成为开发者与企业面临的核心挑战。飞桨（PaddlePaddle）框架 3.0 的推出，正是瞄准这一痛点，以全流程极简体验重新定义大模型部署的边界。

  飞桨框架 3.0，从设计理念上实现了从底层硬件适配到顶层开发体验的全面进化，在训练效率、性能、兼容性等关键指标上建立了新标杆。其中，“动静统一自动并行”、“大模型训推一体”、“科学计算高阶微分”、“神经网络编译器”、“异构多芯适配”这五大技术新特性，系统性解决了当前大模型研发应用面临的分布式策略开发门槛高、训练推理效率低、硬件适配优化难等核心痛点，并为科学智能领域前沿探索提供强大支撑。

  飞桨框架 3.0 通过"动静统一自动并行"技术显著降低大模型训练成本，助力算法高效创新；其"大模型训推一体"设计深度优化训练与推理链路，在文心 4.5、文心 X1、DeepSeek-V3/R1 等模型上实现单机吞吐倍增，推动低时延、高吞吐、低算力推理服务落地。

  科学智能领域创新显著，通过高阶自动微分与编译器技术使微分方程求解速度较 PyTorch 2.6 快 115%，并完成 DeepXDE 等主流工具适配，成为 DeepXDE 默认推荐后端，为气象预测、生命科学、航空航天等前沿领域提供新范式。

  性能层面，自研 CINN 编译器实现算子级突破：A100 平台 RMSNorm 算子经编译优化后提速 4 倍，覆盖超 60 个模型测试中 63% 模型性能提升，平均增幅达 27.4%。硬件生态构建"一次开发全栈部署"体系，已适配 60+ 芯片系列，支持训练集群、自动驾驶等多场景无缝迁移。开发者只需编写一份代码，就可以让程序在不同芯片上顺畅运行，轻松实现业务的跨芯片迁移。

  通过底层架构的全面革新，飞桨框架 3.0 不仅支持 DeepSeek 系列模型的“满血版”高性能推理，更在量化压缩、算子优化、服务部署等关键环节实现突破，让开发者告别复杂的部署调优，真正实现从训练到落地的无缝衔接。

  本次升级中，飞桨框架 3.0 在 Hopper 架构 GPU 上的表现尤为亮眼：单机 4 比特量化部署让 DeepSeek-R1 的吞吐量翻倍，MLA 算子性能领先行业方案最高 23%，MTP 投机解码实现吞吐与解码速度的“双飞跃”。无论是学术研究还是工业级应用，飞桨框架 3.0 都为大模型落地提供了更高效、更经济、更灵活的技术底座。

二、飞桨框架 3.0 的核心优势：为什么选择它部署 DeepSeek？

  飞桨框架 3.0 以技术创新为引擎，为 DeepSeek 的部署提供了全方位的高效解决方案。在性能层面，其突破性的 4 比特单机部署技术彻底颠覆传统多机方案，通过 Weight Only INT4 量化实现单机运行 DeepSeek-R1，不仅将吞吐量提升 100%~128%，更让每秒输出 token 数突破 2000 大关，在降低成本的同时显著提升效率。针对不同硬件场景，飞桨框架 3.0 支持 Hopper 架构 GPU 的 FP8 推理与 A800 的 INT8 部署，在精度无损的前提下，首 Token 推理速度提升 37%，长文本处理能力大幅跃升。尤其值得一提的是动态量化技术 SageAttention 的集成，通过对 Q/K/V 矩阵的智能量化（INT8+FP8），在 64K 长序列输入场景中，Prefill 阶段性能提升 37.4%，真正实现“速度与精度兼得”。

  技术革新方面，飞桨框架 3.0 通过 MLA 算子优化与 MTP 投机解码双管齐下，为大模型推理注入强劲动力。MLA 算子的多级流水线编排与寄存器精细分配，使其性能相比行业标杆 FlashMLA 提升 4%~23%，成为加速推理的“隐形引擎”。而 MTP 投机解码框架的创新设计，不仅统一了草稿模型、多头解码等范式，更通过注意力机制优化实现单批次吞吐提升 144% 或解码速度提升 42%，彻底解决大批次推理性能劣化的行业难题。与此同时，KV Cache 的动态管理与多级流水线并行技术，在 256 并发测试中稳定输出超 1000 token/s，展现工业级部署的硬实力。

  长序列推理，由于 Attention 计算量与序列长度的平方成正比，量化和稀疏都能取得非常好的加速。飞桨框架 3.0 大模型推理在基于 Hopper 架构 GPU 上，集成了 Attention 动态量化方案 SageAttention，并优化了 DeepSeek 模型中 Head Dim 为 192 的情况，在精度近乎无损的基础上，实现了长序列输入 Prefill 阶段的高性能注意力计算实现，64K 长文输入首 token 推理速度提升 37.4%。

  飞桨框架 3.0 的全栈工具链生态，则为开发者提供了从模型压缩到服务部署的一站式支持。其自研的精度无损量化算法覆盖 INT4/INT8/FP8 多精度选择，灵活适配不同硬件需求；高性能推理引擎集成 KV Cache 量化、上下文缓存等黑科技，无论是单机多卡还是多机分布式部署，均可通过一键脚本快速启动。更值得关注的是其跨平台兼容性，除英伟达 GPU 外，昇腾、昆仑芯、海光等国产硬件均能无缝对接，真正实现“一次开发，全场景覆盖”。

  对于开发者而言，飞桨框架 3.0 的极简设计大幅降低技术门槛。开箱即用的 Docker 脚本让单机/多机部署在 5 分钟内完成，OpenAI 兼容的 API 接口支持标准 curl 与 Python SDK 调用，现有开发流程无需重构即可平滑迁移。配合详实的 benchmark 指南、多硬件调优文档及直播课程，即使是初次接触大模型部署的团队，也能快速实现从理论到落地的跨越。这种“技术深度”与“使用轻量化”的平衡，正是飞桨框架 3.0 重新定义大模型部署范式的核心所在。

三、实战部署 DeepSeek-R1-Distill-Qwen-7B

  本次实验是依托潞晨云算力平台展开，硬件环境搭载 NVIDIA A800-80GB 显卡（CUDA 12.4 驱动）。为确保环境配置的可复现性与技术路径透明化，采用手动安装飞桨框架 3.0 环境，从而方便绝大多数同学复现和使用。

  首先进行环境配置，使用 conda 命令创建虚拟环境，这里虚拟环境名称为 paddle_deepseek：

conda create --name paddle_deepseek python=3.10  -y

  在虚拟环境安装完毕后，可使用以下命令安装飞桨框架 3.0 环境所需的 Python 依赖库：

conda activate paddle_deepseek
pip install paddlenlp==3.0.0b4
pip install paddlepaddle-gpu==3.0.0rc1 -i https://www.paddlepaddle.org.cn/packages/stable/cu123/
conda install -c conda-forge gcc=12.1.0

  为了方便同学进行逐步操作，将后续代码将在 Jupyter Notebook 中加以运行，所以需要安装依赖库并添加对应的 Kernel：

pip install ipykernel
python -m ipykernel install --name paddle_deepseek

  在完成环境配置后，接下来我们就要进行实战部署了，请在进入Jupyter Notebook后先将Kernel切换为paddle_deepseek。

  首先定义多个核心模块：1) 日志管理系统(LoggingConfigurator)支持结构化日志记录；2) 系统资源监控工具(SystemMonitor)实时跟踪 CPU/GPU/Paddle 内存使用；3) Paddle 环境检查器(PaddleInspector)验证框架配置；4) 模型推理引擎(ModelInference)和服务器(ModelServer)实现文本生成功能，支持性能分析、参数调优和系统状态监控。系统采用模块化设计，提供完整的模型加载、推理执行、资源监控和日志记录能力，特别针对大语言模型推理场景优化，支持 JSON 格式日志、生成参数调节和 CUDA 加速，适用于生产环境部署。请依次将以下每个代码段复制到 Jupyter 的单元格中运行。：

import os
import time
import json
import logging
import argparse
import inspect
import psutil
import functools
from typing import List, Dict, Any, Union, Optional
from contextlib import contextmanager

import paddle
from paddlenlp.transformers import AutoTokenizer, AutoModelForCausalLM

class LoggingConfigurator:
    """日志系统配置管理类"""
    
    def __init__(self, log_directory: str = "/data/coding/"):
        """初始化日志配置
        
        参数:
            log_directory: 日志文件存储目录
        """
        self.log_dir = log_directory
        self._ensure_log_directory()
        self._configure_loggers()
    
    def _ensure_log_directory(self) -> None:
        """确保日志目录存在"""
        os.makedirs(self.log_dir, exist_ok=True)
    
    def _configure_loggers(self) -> None:
        """配置所有日志记录器"""
        self._setup_server_logger()
        self._setup_infer_logger()
        self._setup_paddle_logger()
    
    def _setup_server_logger(self) -> None:
        """配置服务器日志记录器(JSON格式)"""
        logger = logging.getLogger('server')
        logger.setLevel(logging.INFO)
        
        # 使用lambda动态创建JSON而不是固定格式字符串
        class JsonFormatter(logging.Formatter):
            def format(self, record):
                log_data = {
                    'timestamp': self.formatTime(record, self.datefmt),
                    'severity': record.levelname,
                    'log_message': record.getMessage(),
                    'source': record.module,
                    'line_number': record.lineno
                }
                return json.dumps(log_data)
        
        handler = logging.FileHandler(os.path.join(self.log_dir, 'server.log'))
        handler.setFormatter(JsonFormatter())
        
        # 清除现有处理器避免重复
        if logger.handlers:
            logger.handlers.clear()
            
        logger.addHandler(handler)
    
    def _setup_infer_logger(self) -> None:
        """配置推理日志记录器"""
        self._setup_basic_logger('infer', 'infer.log')
    
    def _setup_paddle_logger(self) -> None:
        """配置Paddle日志记录器"""
        self._setup_basic_logger('paddle', 'paddle.log')
    
    def _setup_basic_logger(self, name: str, filename: str) -> None:
        """配置基础格式日志记录器
        
        参数:
            name: 日志记录器名称
            filename: 日志文件名
        """
        logger = logging.getLogger(name)
        logger.setLevel(logging.INFO)
        
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler = logging.FileHandler(os.path.join(self.log_dir, filename))
        handler.setFormatter(formatter)
        
        # 清除现有处理器避免重复
        if logger.handlers:
            logger.handlers.clear()
            
        logger.addHandler(handler)
    
    @staticmethod
    def get_logger(name: str) -> logging.Logger:
        """获取已配置的日志记录器
        
        参数:
            name: 日志记录器名称
            
        返回:
            已配置的日志记录器
        """
        return logging.getLogger(name)

class SystemMonitor:
    """系统资源监控类"""
    
    @staticmethod
    def get_gpu_memory_usage() -> Union[List[int], str]:
        """获取GPU内存使用情况
        
        返回:
            每个GPU的内存使用量(MB)列表或错误信息
        """
        try:
            import subprocess
            cmd_output = subprocess.check_output(
                ["nvidia-smi", "--query-gpu=memory.used", "--format=csv,noheader"],
                encoding='utf-8'
            )
            return [int(x.strip().split()[0]) for x in cmd_output.strip().split('\n')]
        except Exception as e:
            return f"获取GPU内存失败: {e}"
    
    @staticmethod
    def get_cpu_memory_usage() -> float:
        """获取CPU内存使用百分比
        
        返回:
            CPU内存使用百分比
        """
        return psutil.virtual_memory().percent
    
    @staticmethod
    def get_paddle_memory_usage() -> int:
        """获取PaddlePaddle已分配内存
        
        返回:
            已分配内存(字节)
        """
        return paddle.device.cuda.memory_allocated()
    
    @staticmethod
    def get_system_stats() -> Dict[str, Any]:
        """获取综合系统统计信息
        
        返回:
            包含系统统计信息的字典
        """
        stats = {
            'cpu': {
                'percent': psutil.cpu_percent(interval=0.1),
                'cores': psutil.cpu_count(),
                'memory_percent': SystemMonitor.get_cpu_memory_usage()
            },
            'gpu': SystemMonitor.get_gpu_memory_usage(),
            'paddle_memory': SystemMonitor.get_paddle_memory_usage() if paddle.is_compiled_with_cuda() else None
        }
        return stats

class PaddleInspector:
    """PaddlePaddle环境和配置检查类"""
    
    @staticmethod
    def get_build_config() -> Dict[str, Any]:
        """获取PaddlePaddle构建配置
        
        返回:
            构建配置信息
        """
        config = {
            'version': paddle.__version__,
        }
        
        try:
            config.update({
                'cuda_support': paddle.is_compiled_with_cuda(),
                'cuda_version': paddle.version.cuda() if paddle.is_compiled_with_cuda() else None,
                'cudnn_version': paddle.version.cudnn() if paddle.is_compiled_with_cuda() else None
            })
        except AttributeError:
            config['cuda_support'] = "未知"
            
        return config
    
    @staticmethod
    def get_runtime_config() -> Dict[str, Any]:
        """获取PaddlePaddle运行时配置
        
        返回:
            运行时配置信息
        """
        config = {}
        
        # 获取可用设备
        config['available_devices'] = paddle.device.get_available_device()
        
        # 获取当前设备
        try:
            config['current_device'] = paddle.device.get_device()
        except AttributeError:
            config['current_device'] = "未知"
            
        return config
    
    @staticmethod
    def inspect_optimizer(optimizer: Any) -> Dict[str, Any]:
        """检查优化器配置
        
        参数:
            optimizer: Paddle优化器实例
            
        返回:
            优化器配置信息
        """
        if optimizer is None:
            return {}
        
        config = {'type': type(optimizer).__name__}
        
        # 使用内省获取优化器属性
        for attr_name in dir(optimizer):
            if attr_name.startswith('_') and not attr_name.startswith('__'):
                clean_name = attr_name.lstrip('_')
                if hasattr(optimizer, attr_name):
                    attr_value = getattr(optimizer, attr_name)
                    # 处理不可序列化的值
                    try:
                        json.dumps({clean_name: attr_value})
                        config[clean_name] = attr_value
                    except (TypeError, OverflowError):
                        config[clean_name] = str(attr_value)
        
        return config

class ModelInference:
    """模型推理执行类"""
    
    def __init__(self, model_name: str):
        """初始化推理引擎
        
        参数:
            model_name: 模型标识符
            cache_dir: 模型文件缓存目录
        """
        self.model_name = model_name
        self.tokenizer = None
        self.model = None
        self.logger = LoggingConfigurator.get_logger('infer')
    
    def initialize(self, dtype: str = "float16") -> None:
        """初始化模型和分词器
        
        参数:
            dtype: 模型权重的数据类型
        """
        self.logger.info(f"正在初始化模型: {self.model_name}")
        
        # 初始化分词器(带有更好的错误处理)
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(
                pretrained_model_name_or_path=self.model_name,
            )
        except Exception as e:
            self.logger.error(f"加载分词器失败: {e}")
            raise
        
        # 初始化模型(带有更好的错误处理)
        try:
            self.model = AutoModelForCausalLM.from_pretrained(
                pretrained_model_name_or_path=self.model_name,
                dtype=dtype
            )
        except Exception as e:
            self.logger.error(f"加载模型失败: {e}")
            raise
            
        # 如果有GPU则将模型移到GPU上
        if paddle.is_compiled_with_cuda():
            self.model.eval()
            self.logger.info("模型已加载并移至GPU")
    
    def generate_text(
        self, 
        input_text: str, 
        enable_profiling: bool = False,
        max_new_tokens: int = 200,
        temperature: float = 0.2,
        top_k: int = 20,
        top_p: float = 0.9,
        repetition_penalty: float = 1.1
    ) -> Dict[str, Any]:
        """生成文本响应
        
        参数:
            input_text: 输入提示
            enable_profiling: 是否分析性能
            max_new_tokens: 生成的最大token数
            temperature: 采样温度
            top_k: Top-k采样参数
            top_p: Top-p采样参数
            repetition_penalty: 重复token惩罚
            
        返回:
            包含生成文本和指标的字典
        """
        if not self.model or not self.tokenizer:
            raise RuntimeError("模型未初始化。请先调用initialize()方法")
            
        inputs = self.tokenizer(input_text, return_tensors="pd")
        start_time = time.time()
        
        if enable_profiling:
            return self._profile_generation(
                inputs, 
                start_time,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                top_k=top_k,
                top_p=top_p,
                repetition_penalty=repetition_penalty
            )
        else:
            return self._basic_generation(
                inputs, 
                start_time,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                top_k=top_k,
                top_p=top_p,
                repetition_penalty=repetition_penalty
            )
    
    def _basic_generation(
        self, 
        inputs: Dict[str, Any], 
        start_time: float,
        **generation_kwargs
    ) -> Dict[str, Any]:
        """执行基本生成(无性能分析)
        
        参数:
            inputs: 分词后的输入
            start_time: 生成开始时间
            generation_kwargs: 生成参数
            
        返回:
            结果字典
        """
        with paddle.no_grad():
            outputs = self.model.generate(
                **inputs,
                **generation_kwargs
            )
        
        return self._prepare_results(inputs, outputs, start_time)
    
    @contextmanager
    def _wrap_paddle_apis(self):
        """用于包装Paddle API以进行分析的上下文管理器"""
        api_stats = {'count': {}, 'time': {}}
        original_functions = {}
        
        # 存储原始函数并替换为包装版本
        for name, func in inspect.getmembers(paddle, inspect.isfunction):
            original_functions[name] = func
            
            @functools.wraps(func)
            def wrapped_func(wrapped_name=name, original=func, *args, **kwargs):
                api_stats['count'][wrapped_name] = api_stats['count'].get(wrapped_name, 0) + 1
                call_start = time.time()
                result = original(*args, **kwargs)
                api_stats['time'][wrapped_name] = api_stats['time'].get(wrapped_name, 0) + (time.time() - call_start)
                return result
            
            setattr(paddle, name, wrapped_func)
            
        try:
            yield api_stats
        finally:
            # 恢复原始函数
            for name, func in original_functions.items():
                setattr(paddle, name, func)
    
    def _profile_generation(
        self, 
        inputs: Dict[str, Any], 
        start_time: float,
        **generation_kwargs
    ) -> Dict[str, Any]:
        """执行带性能分析的生成
        
        参数:
            inputs: 分词后的输入
            start_time: 生成开始时间
            generation_kwargs: 生成参数
            
        返回:
            包含分析信息的结果字典
        """
        # 使用更长的序列以获得更有意义的分析
        generation_kwargs['max_new_tokens'] = generation_kwargs.get('max_new_tokens', 200) * 2
        
        profiler_path = os.path.join(os.path.dirname(LoggingConfigurator().log_dir), "profiler_report.json")
        
        with self._wrap_paddle_apis() as api_stats:
            with paddle.profiler.Profiler(targets=["cpu", "gpu"]) as prof:
                with paddle.no_grad():
                    outputs = self.model.generate(
                        **inputs,
                        **generation_kwargs
                    )
                prof.export(profiler_path)
        
        results = self._prepare_results(inputs, outputs, start_time)
        results['api_stats'] = api_stats
        results['profiler_path'] = profiler_path
        return results
    
    def _prepare_results(
        self, 
        inputs: Dict[str, Any], 
        outputs: Any, 
        start_time: float
    ) -> Dict[str, Any]:
        """准备结果字典
        
        参数:
            inputs: 分词后的输入
            outputs: 模型输出
            start_time: 生成开始时间
            
        返回:
            结果字典
        """
        end_time = time.time()
        output_ids = outputs[0].numpy().tolist()
        
        # 更高效地展平output_ids
        flat_output_ids = []
        self._flatten_list_into(output_ids, flat_output_ids)
        
        # 计算输入和输出的token数量
        input_token_count = len(inputs["input_ids"][0])
        output_token_count = len(flat_output_ids)
        inference_time = end_time - start_time
        
        return {
            'input_text': self.tokenizer.decode(inputs["input_ids"][0].numpy().tolist(), skip_special_tokens=True),
            'output_text': self.tokenizer.decode(flat_output_ids, skip_special_tokens=True),
            'input_tokens': input_token_count,
            'output_tokens': output_token_count,
            'inference_time': inference_time,
            'tokens_per_second': (input_token_count + output_token_count) / inference_time if inference_time > 0 else 0,
            'system_stats': SystemMonitor.get_system_stats(),
            'start_time': start_time,
            'end_time': end_time
        }
    
    @staticmethod
    def _flatten_list_into(nested_list: List, output_list: List) -> None:
        """将嵌套列表展平到输出列表中(原地修改)
        
        参数:
            nested_list: 嵌套列表结构
            output_list: 输出平列表(原地修改)
        """
        for item in nested_list:
            if isinstance(item, list):
                ModelInference._flatten_list_into(item, output_list)
            else:
                output_list.append(item)

class ModelServer:
    """用于推理任务的模型服务器类"""
    
    def __init__(
        self, 
        model_name: str = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B", 
        log_dir: str = "/data/coding/"
    ):
        """初始化模型服务器
        
        参数:
            model_name: 模型标识符
            log_dir: 日志目录
        """
        # 设置日志
        self.log_config = LoggingConfigurator(log_dir)
        self.server_log = self.log_config.get_logger('server')
        self.infer_log = self.log_config.get_logger('infer')
        self.paddle_log = self.log_config.get_logger('paddle')
        
        # 初始化模型
        self.model_name = model_name
        self.model_engine = None
        self.optimizer = None
    
    def initialize(self) -> None:
        """初始化服务器和模型"""
        self.server_log.info(json.dumps({
            'event': 'server_start', 
            'details': {
                'model': self.model_name,
            }
        }))
        
        # 记录PaddlePaddle信息
        paddle_config = PaddleInspector.get_build_config()
        self.paddle_log.info(f"PaddlePaddle版本: {paddle_config['version']}")
        self.paddle_log.info(f"构建配置: {paddle_config}")
        self.paddle_log.info(f"运行时配置: {PaddleInspector.get_runtime_config()}")
        
        # 初始化模型
        self.model_engine = ModelInference(
            model_name=self.model_name,
        )
        self.model_engine.initialize()
        
        # 创建优化器(仅当需要微调时)
        self._setup_optimizer()
    
    def _setup_optimizer(self, learning_rate: float = 1e-5) -> None:
        """设置模型参数优化器
        
        参数:
            learning_rate: 优化器的学习率
        """
        if self.model_engine and self.model_engine.model:
            self.optimizer = paddle.optimizer.Adam(
                learning_rate=learning_rate, 
                parameters=self.model_engine.model.parameters()
            )
            self.paddle_log.info(f"优化器配置: {PaddleInspector.inspect_optimizer(self.optimizer)}")
    
    def run_inference(
        self, 
        input_query: str, 
        enable_profiling: bool = False,
        generation_params: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """在输入查询上运行推理
        
        参数:
            input_query: 输入文本
            enable_profiling: 是否启用性能分析
            generation_params: 生成参数
            
        返回:
            结果字典
        """
        if not self.model_engine:
            raise RuntimeError("服务器未初始化。请先调用initialize()方法")
            
        # 合并默认和自定义生成参数
        gen_params = {
            'max_new_tokens': 200,
            'temperature': 0.2,
            'top_k': 20,
            'top_p': 0.9,
            'repetition_penalty': 1.1
        }
        if generation_params:
            gen_params.update(generation_params)
            
        # 记录输入信息
        self.infer_log.info(f"输入查询: {input_query}")
        self.infer_log.info(f"生成参数: {gen_params}")
        
        # 运行推理
        start_time = time.time()
        try:
            results = self.model_engine.generate_text(
                input_text=input_query, 
                enable_profiling=enable_profiling,
                **gen_params
            )
            
            # 记录性能指标
            self.infer_log.info(f"输入长度: {results['input_tokens']} tokens")
            self.infer_log.info(f"输出长度: {results['output_tokens']} tokens")
            self.infer_log.info(f"推理时间: {results['inference_time']:.2f} 秒")
            self.infer_log.info(f"Token处理速度: {results['tokens_per_second']:.2f} tokens/秒")
            self.infer_log.info(f"输出文本: {results['output_text']}")
            
            # 记录系统统计信息
            self.infer_log.info(f"系统统计: {results['system_stats']}")
            
            return results
            
        except Exception as e:
            self.infer_log.exception(f"推理过程中出错: {e}")
            self.server_log.error(json.dumps({
                'event': 'inference_error',
                'error': str(e)
            }))
            raise
    
    def shutdown(self, exit_status: int = 0) -> None:
        """优雅地关闭服务器
        
        参数:
            exit_status: 退出状态码
        """
        self.server_log.info(json.dumps({
            'event': 'server_shutdown', 
            'exit_status': exit_status
        }))

  定义多个路径并启动模型服务：

model = "deepseek-ai/DeepSeek-R1-Distill-Qwen-7B"
cache_dir = "/root/filesystem/disk/"
log_dir = "/root/filesystem/disk/logs"

# 初始化模型服务
server = ModelServer(
    model_name=model,
    cache_dir=cache_dir,
    log_dir=log_dir
)
server.initialize()

  然后我们对不同场景的问题进行逐一测试，首先看一下部署后模型的写作能力：

enable_profiling = False
max_tokens = 500
temperature = 0.00000001
    
# 运行推理
generation_params = {
    'max_new_tokens': max_tokens,
    'temperature': temperature
}

input_query = "帮我写一篇100字的文章"
results = server.run_inference(
    input_query=input_query,
    enable_profiling=enable_profiling,
    generation_params=generation_params
)

# 打印结果到标准输出
print(f"\n生成的文本:\n{results['output_text']}\n")
print(f"推理时间: {results['inference_time']:.2f} 秒")
print(f"Token处理速度: {results['tokens_per_second']:.2f} tokens/秒")

  写作能力测试的截图如下：

  再来看看个人场景题目的测试：

enable_profiling = False
max_tokens = 500
temperature = 0.00000001
    
# 运行推理
generation_params = {
    'max_new_tokens': max_tokens,
    'temperature': temperature
}

input_query = "明天我的同学要过生日，他比较喜欢文学，该送什么样的生日礼物呢？"
results = server.run_inference(
    input_query=input_query,
    enable_profiling=enable_profiling,
    generation_params=generation_params
)

# 打印结果到标准输出
print(f"\n生成的文本:\n{results['output_text']}\n")
print(f"推理时间: {results['inference_time']:.2f} 秒")
print(f"Token处理速度: {results['tokens_per_second']:.2f} tokens/秒")

  个人场景题目的测试的截图如下：

  最后再来看看推理类大模型的核心能力之一，即强大的编程和反思能力，由于该过程需要输出更多的 token，所以这里将 max_tokens 设置为了 2000：

enable_profiling = False
max_tokens = 2000
temperature = 0.00000001
    
# 运行推理
generation_params = {
    'max_new_tokens': max_tokens,
    'temperature': temperature
}

input_query = "请编写一个Python的函数，判断给定的整数是否为质数"
results = server.run_inference(
    input_query=input_query,
    enable_profiling=enable_profiling,
    generation_params=generation_params
)

# 打印结果到标准输出
print(f"\n生成的文本:\n{results['output_text']}\n")
print(f"推理时间: {results['inference_time']:.2f} 秒")
print(f"Token处理速度: {results['tokens_per_second']:.2f} tokens/秒")

  编程题目测试的完整截图如下：

  最终生成代码和代码解释如下所示：

  经过多轮的数据测试，飞桨框架 3.0 的 Token 处理速度稳定维持在 26 tokens/秒，实际业务场景测试表明，该处理速度可满足大多数对话式 AI 应用的实时性需求。

  由于资源有限，后续同时也简单尝试了下 DeepSeek-R1-INT4 版本的部署，需要提前安装 docker 环境，具体部署命令如下，其中 MODEL_PATH 指的是模型下载的本地路径（由于模型文件较大，请设置为数据盘的路径）

export MODEL_PATH=/home/models/DeepSeek-R1/weight_only_int4
export model_name=${model_name:-"deepseek-ai/DeepSeek-R1/weight_only_int4"}

docker run --gpus all --shm-size 32G --network=host --privileged --cap-add=SYS_PTRACE \
-v $MODEL_PATH:/models -e "model_name=${model_name}" \
-dit ccr-2vdh3abv-pub.cnc.bj.baidubce.com/paddlepaddle/paddlenlp:llm-serving-cuda124-cudnn9-v2.1 /bin/bash \
-c -ex 'export CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 && export MP_NUM=8 && start_server $model_name && tail -f /dev/null'

  通过以上命令下载 docker 环境以及模型文件后，可使用以下命令测试是否顺利完成：

curl -i http://127.0.0.1:8510/v2/health/ready

  当看到 200 OK 则说明已经成功搭建：

  然后可使用以下命令来进行 Query 请求：

curl 127.0.0.1:9965/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
      "model":"default",
      "text":"Hello, how are you?"
  }'

四、实战部署总结

  飞桨框架 3.0 的突破性实践正在重塑大模型落地的技术范式。本次 DeepSeek-R1 的部署历程生动诠释了这一点：其创新的混合精度量化体系（如 FP8-WINT4 协同）使模型显存占用锐减 70%，而动态编译与自动并行技术的深度耦合，则让模型在 A800 等异构硬件上实现了"开箱即用"的部署体验。这种将尖端研究工程化的能力，标志着国产 AI 框架已从技术追随者蜕变为标准制定者。

  更值得关注的是其带来的产业级变革：

• 效能革命：MLA 算子的三级流水线设计使长文本推理吞吐量提升 2.3 倍，配合 MTP 解码可将单 Token 生成成本降低 40%
• 普惠价值：自动硬件适配技术使模型迁移效率提升 90%，让中小企业也能驾驭百亿参数模型
• 生态赋能：从科研机构的算法创新到工业界的场景落地，形成完整的技术闭环

  这不仅是工具迭代，更是一场 AI 民主化运动。当开发者无需再为分布式部署、量化调参耗费周级时间，当单张消费级显卡也能流畅运行大模型，创新的重心自然转向场景挖掘与价值创造。飞桨框架 3.0 正以这样的技术担当，为中国 AI 产业的"最后一公里"铺设高速通道。