Python使用vllm处理多模态数据的预处理技巧

from vllm import LLM, SamplingParams
import re
from transformers import AutoTokenizer

def preprocess_text(text):
    # 清洗文本
    text = re.sub(r'[^\w\s]', '', text)  # 去除标点
    text = text.lower()  # 转为小写

    # 使用vLLM兼容的分词器
    tokenizer = AutoTokenizer.from_pretrained("facebook/opt-1.3b")
    tokens = tokenizer(text, return_tensors="pt", padding=True, truncation=True)

    return tokens

# 示例使用
text_data = "This is an example of text preprocessing with vLLM."
processed_text = preprocess_text(text_data)

from PIL import Image
import torchvision.transforms as transforms

def preprocess_image(image_path, target_size=(224, 224)):
    # 加载图像
    img = Image.open(image_path)

    # 定义预处理流程
    transform = transforms.Compose([
        transforms.Resize(target_size),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406],
                             std=[0.229, 0.224, 0.225]),
    ])

    return transform(img)

# 示例使用
image_data = preprocess_image("example.jpg")

import torchaudio
import torch

def preprocess_audio(audio_path, sample_rate=16000):
    # 加载音频文件
    waveform, sr = torchaudio.load(audio_path)

    # 重采样
    if sr != sample_rate:
        resampler = torchaudio.transforms.Resample(sr, sample_rate)
        waveform = resampler(waveform)

    # 提取Mel频谱特征
    mel_transform = torchaudio.transforms.MelSpectrogram(
        sample_rate=sample_rate,
        n_mels=128,
        n_fft=2048,
        hop_length=512
    )

    return mel_transform(waveform)

# 示例使用
audio_data = preprocess_audio("example.wav")

conda create -n vllm-multimodal python=3.9
conda activate vllm-multimodal
pip install vllm torch torchvision torchaudio transformers pillow

import os
from typing import Dict, Any
from dataclasses import dataclass
import torch
from transformers import AutoTokenizer

@dataclass
class MultiModalSample:
    text: str = None
    image_path: str = None
    audio_path: str = None

class MultiModalPreprocessor:
    def __init__(self, model_name="facebook/opt-1.3b"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.image_transform = self._get_image_transform()
        self.audio_transform = self._get_audio_transform()

    def _get_image_transform(self):
        return transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                               std=[0.229, 0.224, 0.225]),
        ])

    def _get_audio_transform(self):
        return torchaudio.transforms.MelSpectrogram(
            sample_rate=16000,
            n_mels=128,
            n_fft=2048,
            hop_length=512
        )

    def preprocess(self, sample: MultiModalSample) -> Dict[str, Any]:
        processed = {}

        if sample.text:
            processed["text"] = self._process_text(sample.text)

        if sample.image_path and os.path.exists(sample.image_path):
            processed["image"] = self._process_image(sample.image_path)

        if sample.audio_path and os.path.exists(sample.audio_path):
            processed["audio"] = self._process_audio(sample.audio_path)

        return processed

    def _process_text(self, text: str) -> Dict[str, torch.Tensor]:
        return self.tokenizer(
            text,
            return_tensors="pt",
            padding=True,
            truncation=True,
            max_length=512
        )

    def _process_image(self, image_path: str) -> torch.Tensor:
        img = Image.open(image_path).convert("RGB")
        return self.image_transform(img)

    def _process_audio(self, audio_path: str) -> torch.Tensor:
        waveform, sr = torchaudio.load(audio_path)
        if sr != 16000:
            resampler = torchaudio.transforms.Resample(sr, 16000)
            waveform = resampler(waveform)
        return self.audio_transform(waveform)