OpenVINO 本地部署 DeepSeek-R1 量化大模型：前端交互与后端服务

主要实现了几个简单控件，输入条，发送按钮和状态栏，需要后端提供三个接口：

const response = await fetch(`${API_BASE}/health`);
const response = await fetch(`${API_BASE}/model_info`);
const response = await fetch(`${API_BASE}/chat`);

分别用于服务器状态检查，获取模型信息和对话。

三、后端模型服务器

OpenVINO 为生成式 AI 优化提供了推理引擎：OpenVINO GenAI。

OpenVINO GenAI 提供简洁的 API 接口，支持 Python 和 C++ 两种编程语言，安装容量不到 200MB。

以我本次使用的 DeepSeek-R1-0528-Qwen3-8B 为例，流水线使用 LLMPipeline。

后端需要用到的包：

pip install Flask Flask-CORS openvino-genai

openvino_server.py

import openvino_genai as ov_genai
from flask import Flask, request, jsonify
from flask_cors import CORS
import threading
import logging
from typing import Dict, Any
from pathlib import Path

class OpenVINOBackend:
    def __init__(self, model_path: str):
        self.model_path = model_path
        self.pipeline = None
        self.is_ready = False
        self.init_lock = threading.Lock()

    def initialize_model(self, device: str):
        with self.init_lock:
            if self.is_ready:
                return
            logger.info("正在加载 OpenVINO 模型...")
            try:
                self.pipeline = ov_genai.LLMPipeline(self.model_path, device)
                self.is_ready = True
                logger.info("OpenVINO 模型加载完成！")
            except Exception as e:
                logger.error(f"模型加载失败：{str(e)}")
                self.is_ready = False

    def generate_response(self, message: str, max_tokens: int = 512, temperature: float = 0.7, do_sample: bool = True) -> Dict[str, Any]:
        if not self.is_ready or self.pipeline is None:
            return {"status": "error", "error": "模型未就绪"}
        try:
            config = ov_genai.GenerationConfig()
            config.max_new_tokens = max_tokens
            config.temperature = temperature
            config.do_sample = do_sample
            self.pipeline.start_chat()
            response = self.pipeline.generate(message, config)
            self.pipeline.finish_chat()
            return {
                "status": "success",
                "response": response,
                "tokens_generated": len(response.split())
            }
        except Exception as e:
            return {"status": "error", "error": str(e)}

def create_app(backend: OpenVINOBackend) -> Flask:
    app = Flask(__name__)
    CORS(app)

    @app.route('/health', methods=['GET'])
    def health_check():
        return jsonify({
            "status": "ready" if backend.is_ready else "loading",
            "backend": "openvino_genai",
            "device": "CPU"
        })

    @app.route('/chat', methods=['POST'])
    def chat_endpoint():
        data = request.json
        if not data or 'message' not in data:
            return jsonify({"status": "error", "error": "缺少 message 参数"})
        message = data['message']
        max_tokens = min(max(int(data.get('max_tokens', 512)), 1), 2048)
        result = backend.generate_response(message, max_tokens)
        return jsonify(result)

    @app.route('/model_info', methods=['GET'])
    def model_info():
        return jsonify({
            "backend": "openvino_genai",
            "status": "ready" if backend.is_ready else "loading",
            "device": "CPU",
            "optimization": "int4_quantization"
        })
    return app

def initialize_backend(backend: OpenVINOBackend, device: str='CPU'):
    def init_task():
        backend.initialize_model(device)
    thread = threading.Thread(target=init_task, daemon=True)
    thread.start()

def main(model_path: str = "./converted_ov_model"):
    logging.basicConfig(
        level=logging.INFO,
        format='-- %(levelname)s - %(message)s',
        force=True
    )
    global logger
    logger = logging.getLogger(__name__)
    logger.info("Starting OpenVINO GenAI server...")
    p = Path(model_path)
    if not p.exists():
        logger.error(f"路径不存在：{model_path}")
        return
    if not p.is_dir():
        logger.error(f"请传入模型文件夹：{model_path}")
        return
    backend = OpenVINOBackend(model_path)
    app = create_app(backend)
    initialize_backend(backend, "CPU")
    logger.info(f"启动 OpenVINO GenAI 服务器，模型路径：{model_path}...")
    app.run(host='0.0.0.0', port=5000, debug=False)

if __name__ == '__main__':
    path = "./DeepSeek-R1-0528-Qwen3-8B_openvino_int4"
    main(path)

为了启动方便，我将模型路径写进了 py 文件，读者可以根据自身情况改为从命令行传参。后端也是主要实现三个接口，以及模型的初始化和推理。

为了适配长期运行环境，启用了 logger 和局域网服务器，并且关闭了 Flask 的 debug，防止 Werkzeug 的自动重载机制（reloader），出现一些莫名的情况。

四、资源占用

启动后端服务是不占用资源的，开始对话以后才会明显占用 CPU 和内存。

可以看到，OpenVINO 优化以及 INT4 量化后的 DeepSeek-R1-0528-Qwen3-8B，内存占用约 10G，CPU 也在可接受范围，每次推理大约需要 40 秒。

根据缩略图情况，启动后内存不会释放，直到后端服务器关闭。目前对话是单次输入输出，也就是每次对话只会有一次请求和回复，因此消息没有逐字生成的效果，未来有机会再继续优化，并且现在有字符上线限制，主要是因为我的设备性能有限，如果能将逐字显示完成，使用体验应该可以流畅很多。