Qwen2-VL-2B-Instruct部署案例：Docker Compose编排Streamlit+Redis缓存架构

TurquoiseSea98

1014人浏览 · 2026-04-29 03:42:12

TurquoiseSea98 · 2026-04-29 03:42:12 发布

Qwen2-VL-2B-Instruct部署案例：Docker Compose编排Streamlit+Redis缓存架构

1. 项目概述

Qwen2-VL-2B-Instruct是一个基于通义千问多模态模型开发的本地多模态嵌入与比对工具。这个工具采用Sentence-Transformers框架，能够将文本和图片映射到统一的向量空间，实现精准的语义相似度计算。

在实际应用中，我们经常需要处理文本搜图片、图片搜图片等多模态检索任务。传统的单机部署方式虽然简单，但在高并发场景下会遇到性能瓶颈。为了解决这个问题，我们设计了基于Docker Compose的分布式部署方案，集成了Redis缓存和Streamlit前端，显著提升了系统的响应速度和并发处理能力。

2. 环境准备与部署架构

2.1 系统要求

在开始部署之前，请确保你的系统满足以下要求：

Docker Engine 20.10.0 或更高版本
Docker Compose 2.0.0 或更高版本
NVIDIA显卡驱动（如使用GPU加速）
至少8GB内存（推荐16GB）
20GB可用磁盘空间

2.2 部署架构设计

我们的部署架构包含三个核心组件：

Streamlit应用服务：提供Web界面，处理用户请求
模型推理服务：专门负责向量计算和相似度匹配
Redis缓存服务：存储常用向量结果，减少重复计算

这种架构设计使得系统可以水平扩展，轻松应对高并发场景。

3. Docker Compose部署步骤

3.1 编写Docker Compose配置文件

创建docker-compose.yml文件，内容如下：

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    container_name: qwen2vl_redis
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes
    restart: unless-stopped

  model-service:
    build: 
      context: .
      dockerfile: Dockerfile.model
    container_name: qwen2vl_model
    ports:
      - "8000:8000"
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - MODEL_PATH=/app/ai-models/iic/gme-Qwen2-VL-2B-Instruct
    volumes:
      - model_data:/app/ai-models
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    restart: unless-stopped

  web-app:
    build:
      context: .
      dockerfile: Dockerfile.web
    container_name: qwen2vl_web
    ports:
      - "8501:8501"
    environment:
      - MODEL_SERVICE_URL=http://model-service:8000
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - redis
      - model-service
    restart: unless-stopped

volumes:
  redis_data:
  model_data:

3.2 创建模型服务Dockerfile

创建Dockerfile.model文件：

FROM nvidia/cuda:11.8-runtime-ubuntu22.04

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    python3.10 \
    python3-pip \
    && rm -rf /var/lib/apt/lists/*

# 复制模型文件和代码
COPY requirements_model.txt .
COPY model_service.py .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements_model.txt

# 创建模型目录
RUN mkdir -p /app/ai-models/iic/gme-Qwen2-VL-2B-Instruct

EXPOSE 8000

CMD ["python3", "model_service.py"]

3.3 创建Web应用Dockerfile

创建Dockerfile.web文件：

FROM python:3.10-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    libgl1 \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*

# 复制应用代码
COPY requirements_web.txt .
COPY app.py .
COPY temp_images/ ./temp_images/

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements_web.txt

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

4. 核心服务实现

4.1 模型推理服务实现

创建model_service.py文件，实现基于FastAPI的模型服务：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
from sentence_transformers import SentenceTransformer
import numpy as np
import redis
import json
from typing import Union
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Qwen2-VL Model Service")

# 连接Redis
redis_client = redis.Redis(
    host=os.getenv('REDIS_HOST', 'localhost'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    decode_responses=True
)

class EmbeddingRequest(BaseModel):
    text: Union[str, None] = None
    image_path: Union[str, None] = None
    instruction: str = "Find an image that matches the given text."

class SimilarityRequest(BaseModel):
    query: EmbeddingRequest
    target: EmbeddingRequest

# 加载模型
def load_model():
    model_path = os.getenv('MODEL_PATH', './ai-models/iic/gme-Qwen2-VL-2B-Instruct')
    model = SentenceTransformer(model_path, device='cuda' if torch.cuda.is_available() else 'cpu')
    model.eval()
    return model

model = load_model()

def generate_cache_key(request: EmbeddingRequest) -> str:
    """生成缓存键"""
    import hashlib
    key_data = f"{request.text}_{request.image_path}_{request.instruction}"
    return hashlib.md5(key_data.encode()).hexdigest()

@app.post("/embed")
async def get_embedding(request: EmbeddingRequest):
    """获取文本或图片的向量表示"""
    try:
        # 检查缓存
        cache_key = generate_cache_key(request)
        cached_result = redis_client.get(cache_key)
        
        if cached_result:
            logger.info("命中缓存")
            return json.loads(cached_result)
        
        # 计算嵌入向量
        with torch.no_grad():
            if request.text and request.image_path:
                # 多模态输入
                embedding = model.encode([{
                    'text': request.text,
                    'image': request.image_path
                }], instruction=request.instruction)
            elif request.text:
                # 纯文本输入
                embedding = model.encode([request.text], instruction=request.instruction)
            elif request.image_path:
                # 纯图片输入
                embedding = model.encode([request.image_path], instruction=request.instruction)
            else:
                raise HTTPException(status_code=400, detail="必须提供文本或图片输入")
        
        # 转换为列表并归一化
        embedding_list = embedding.tolist()[0]
        embedding_norm = np.linalg.norm(embedding_list)
        normalized_embedding = (np.array(embedding_list) / embedding_norm).tolist()
        
        result = {
            "embedding": normalized_embedding,
            "dimension": len(normalized_embedding),
            "normalized": True
        }
        
        # 缓存结果（有效期1小时）
        redis_client.setex(cache_key, 3600, json.dumps(result))
        
        return result
        
    except Exception as e:
        logger.error(f"嵌入计算错误: {str(e)}")
        raise HTTPException(status_code=500, detail=f"嵌入计算错误: {str(e)}")

@app.post("/similarity")
async def calculate_similarity(request: SimilarityRequest):
    """计算两个输入的相似度"""
    try:
        # 获取查询向量
        query_embedding_response = await get_embedding(request.query)
        query_vector = np.array(query_embedding_response["embedding"])
        
        # 获取目标向量
        target_embedding_response = await get_embedding(request.target)
        target_vector = np.array(target_embedding_response["embedding"])
        
        # 计算余弦相似度
        similarity = float(np.dot(query_vector, target_vector))
        
        return {
            "similarity": similarity,
            "query_dimension": query_embedding_response["dimension"],
            "target_dimension": target_embedding_response["dimension"]
        }
        
    except Exception as e:
        logger.error(f"相似度计算错误: {str(e)}")
        raise HTTPException(status_code=500, detail=f"相似度计算错误: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

4.2 Streamlit应用优化

更新app.py文件，集成Redis缓存和模型服务调用：

import streamlit as st
import requests
import json
import os
from PIL import Image
import tempfile
import redis

# 配置页面
st.set_page_config(
    page_title="GME-Qwen2-VL 多模态相似度计算",
    page_icon="🖼️",
    layout="wide"
)

# 初始化Redis连接
def init_redis():
    try:
        return redis.Redis(
            host=os.getenv('REDIS_HOST', 'localhost'),
            port=int(os.getenv('REDIS_PORT', 6379)),
            decode_responses=True
        )
    except:
        return None

redis_client = init_redis()

# 模型服务URL
MODEL_SERVICE_URL = os.getenv('MODEL_SERVICE_URL', 'http://localhost:8000')

def save_uploaded_file(uploaded_file, temp_dir="temp_images"):
    """保存上传的文件到临时目录"""
    if not os.path.exists(temp_dir):
        os.makedirs(temp_dir)
    
    file_path = os.path.join(temp_dir, uploaded_file.name)
    with open(file_path, "wb") as f:
        f.write(uploaded_file.getbuffer())
    return file_path

def call_model_service(endpoint, data):
    """调用模型服务"""
    try:
        response = requests.post(
            f"{MODEL_SERVICE_URL}/{endpoint}",
            json=data,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        st.error(f"服务调用失败: {str(e)}")
        return None

# 界面布局
st.title("🖼️ GME-Qwen2-VL 多模态相似度计算工具")

col1, col2 = st.columns(2)

with col1:
    st.header("输入 A (查询/Query)")
    query_type = st.radio("查询类型", ["文本", "图片"], key="query_type")
    
    if query_type == "文本":
        query_text = st.text_area("输入查询文本", height=100)
        query_image = None
    else:
        query_image = st.file_uploader("上传查询图片", type=["jpg", "jpeg", "png"])
        query_text = None
        
    instruction = st.text_input(
        "引导指令",
        value="Find an image that matches the given text.",
        help="这个指令会引导模型如何理解你的查询"
    )

with col2:
    st.header("输入 B (目标/Target)")
    target_type = st.radio("目标类型", ["文本", "图片"], key="target_type")
    
    if target_type == "文本":
        target_text = st.text_area("输入目标文本", height=100)
        target_image = None
    else:
        target_image = st.file_uploader("上传目标图片", type=["jpg", "jpeg", "png"])
        target_text = None

# 计算按钮
if st.button("🔍 计算相似度", type="primary"):
    if (not query_text and not query_image) or (not target_text and not target_image):
        st.error("请确保查询和目标都有输入内容")
    else:
        with st.spinner("计算中..."):
            # 处理文件上传
            query_image_path = save_uploaded_file(query_image) if query_image else None
            target_image_path = save_uploaded_file(target_image) if target_image else None
            
            # 准备请求数据
            query_data = {
                "text": query_text,
                "image_path": query_image_path,
                "instruction": instruction
            }
            
            target_data = {
                "text": target_text,
                "image_path": target_image_path,
                "instruction": instruction
            }
            
            request_data = {
                "query": query_data,
                "target": target_data
            }
            
            # 调用模型服务
            result = call_model_service("similarity", request_data)
            
            if result:
                similarity = result["similarity"]
                
                # 显示结果
                st.success(f"相似度得分: {similarity:.4f}")
                
                # 可视化进度条
                st.progress(similarity)
                
                # 语义解读
                if similarity > 0.8:
                    interpretation = "极高匹配"
                elif similarity > 0.6:
                    interpretation = "高度相关"
                elif similarity > 0.4:
                    interpretation = "中等相关"
                elif similarity > 0.2:
                    interpretation = "低度相关"
                else:
                    interpretation = "几乎不相关"
                
                st.info(f"语义解读: {interpretation}")
                
                # 显示调试信息
                with st.expander("调试信息"):
                    st.json(result)

# 侧边栏功能
with st.sidebar:
    st.header("系统信息")
    
    # 显示缓存状态
    if redis_client:
        try:
            cache_info = redis_client.info()
            st.write(f"缓存使用内存: {cache_info['used_memory_human']}")
            st.write(f"缓存键数量: {cache_info['keyspace_hits']}")
        except:
            st.write("缓存状态: 不可用")
    
    # 清理临时文件按钮
    if st.button("清理临时文件"):
        import shutil
        if os.path.exists("temp_images"):
            shutil.rmtree("temp_images")
            os.makedirs("temp_images")
            st.success("临时文件已清理")

5. 部署与运行

5.1 创建依赖文件

创建requirements_model.txt：

torch>=2.0.0
sentence-transformers>=2.2.0
fastapi>=0.95.0
uvicorn>=0.21.0
redis>=4.5.0
numpy>=1.24.0
Pillow>=9.5.0

创建requirements_web.txt：

streamlit>=1.22.0
requests>=2.28.0
redis>=4.5.0
Pillow>=9.5.0
numpy>=1.24.0

5.2 启动部署

在项目根目录下执行以下命令启动服务：

# 构建并启动所有服务
docker-compose up -d

# 查看服务状态
docker-compose ps

# 查看日志
docker-compose logs -f

5.3 验证部署

访问以下地址验证服务是否正常：

Streamlit Web界面：http://localhost:8501
模型服务API文档：http://localhost:8000/docs
Redis监控（可选）：可以使用redis-cli或RedisInsight工具连接

6. 性能优化与监控

6.1 Redis缓存策略优化

在我们的架构中，Redis缓存发挥了关键作用。以下是优化的缓存策略：

# 高级缓存策略实现
def advanced_cache_strategy(request, compute_function):
    """高级缓存策略"""
    cache_key = generate_cache_key(request)
    
    # 先检查本地内存缓存（如果有）
    if hasattr(st, 'session_state') and cache_key in st.session_state:
        return st.session_state[cache_key]
    
    # 检查Redis缓存
    cached_result = redis_client.get(cache_key)
    if cached_result:
        result = json.loads(cached_result)
        # 更新本地缓存
        if hasattr(st, 'session_state'):
            st.session_state[cache_key] = result
        return result
    
    # 计算新结果
    result = compute_function(request)
    
    # 根据结果重要性设置不同的过期时间
    if 'similarity' in result and result['similarity'] > 0.7:
        # 高相似度结果缓存更久
        expire_time = 86400  # 24小时
    else:
        expire_time = 3600   # 1小时
    
    # 缓存到Redis
    redis_client.setex(cache_key, expire_time, json.dumps(result))
    
    # 更新本地缓存
    if hasattr(st, 'session_state'):
        st.session_state[cache_key] = result
    
    return result

6.2 监控与日志

添加监控端点到模型服务：

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {
        "status": "healthy",
        "model_loaded": model is not None,
        "redis_connected": redis_client.ping(),
        "gpu_available": torch.cuda.is_available()
    }

@app.get("/stats")
async def get_stats():
    """获取服务统计信息"""
    cache_info = redis_client.info()
    return {
        "cache_keys": redis_client.dbsize(),
        "cache_memory_usage": cache_info.get('used_memory_human', 'N/A'),
        "gpu_memory_allocated": torch.cuda.memory_allocated() if torch.cuda.is_available() else 0,
        "gpu_memory_reserved": torch.cuda.memory_reserved() if torch.cuda.is_available() else 0
    }