MiniCPM-Llama3-V-2_5-Int4：量化压缩技术赋能的高效多模态推理

from transformers import BitsAndBytesConfig
import torch

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,                    # 启用 4-bit 量化加载
    bnb_4bit_quant_type="nf4",             # 采用 NF4 量化格式
    bnb_4bit_use_double_quant=True,        # 启用双量化优化
    bnb_4bit_compute_dtype=torch.float16,  # 计算时使用 FP16 精度
    bnb_4bit_quant_storage=torch.uint8     # 量化存储使用 UINT8
)

# 量化配置下的注意力机制配置示例
model = AutoModel.from_pretrained(
    model_path,
    quantization_config=BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True
    ),
    attn_implementation='sdpa',  # 使用 SDPA 替代 eager attention
    torch_dtype=torch.bfloat16
)

# 使用 NanoLLM 框架部署
from nanollm import NanoLLM

model = NanoLLM.from_pretrained(
    "openbmb/MiniCPM-Llama3-V-2_5-int4",
    api='mlc',
    quantization='q4f16_ft'
)

# 初始化视觉模块
model.config_vision()
model.init_vision()

# 不同硬件平台的性能测试代码
def benchmark_model(model_path: str, device: str, batch_size: int = 1):
    import time
    import torch
    
    model = AutoModel.from_pretrained(
        model_path,
        quantization_config=BitsAndBytesConfig(load_in_4bit=True),
        device_map=device,
        trust_remote_code=True
    )
    
    # 预热
    for _ in range(3):
        _ = model.chat(image=test_image, msgs=[test_prompt])
    
    # 计时测试
    start = time.time()
    for _ in range(10):
        _ = model.chat(image=test_image, msgs=[test_prompt])
    elapsed = time.time() - start
    
    return {
        "device": device,
        "avg_time": elapsed / 10,
        "tokens_per_second": 50 / (elapsed / 10)  # 假设平均50 tokens
    }

# 批量推理性能测试
def batch_inference_benchmark(model, images: list, prompts: list):
    start = time.time()
    
    # 串行处理
    results_serial = []
    for img, prompt in zip(images, prompts):
        results_serial.append(model.chat(img, prompt))
    
    serial_time = time.time() - start
    
    # 批量处理
    start = time.time()
    results_batch = model.batch_chat(images, prompts)
    batch_time = time.time() - start
    
    return {
        "serial_time_per_sample": serial_time / len(images),
        "batch_time_per_sample": batch_time / len(images),
        "speedup": serial_time / batch_time
    }

from transformers import AutoModel, AutoTokenizer, BitsAndBytesConfig
from PIL import Image
import torch

model_path = "./MiniCPM-Llama3-V-2_5-int4"

model = AutoModel.from_pretrained(
    model_path,
    quantization_config=BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True
    ),
    device_map="auto",
    trust_remote_code=True
)

tokenizer = AutoTokenizer.from_pretrained(
    model_path,
    trust_remote_code=True
)

model.eval()

def analyze_contract(image_path: str) -> dict:
    image = Image.open(image_path).convert('RGB')
    
    prompt = """请分析这份合同，提取以下信息：
    1. 合同双方名称
    2. 合同金额
    3. 关键日期（签署日期、有效期）
    4. 潜在风险条款"""
    
    response = model.chat(
        image=image,
        msgs=[{"role": "user", "content": [image, prompt]}],
        tokenizer=tokenizer
    )
    
    return parse_contract_response(response)

result = analyze_contract("contract.jpg")
print(f"合同金额: {result['amount']}")

import re
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class ContractInfo:
    party_a: Optional[str] = None
    party_b: Optional[str] = None
    amount: Optional[str] = None
    start_date: Optional[str] = None
    end_date: Optional[str] = None
    risk_clauses: List[str] = None
    raw_response: str = ""

class ContractAnalyzer:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
    
    def extract_info(self, image: Image.Image) -> ContractInfo:
        structured_prompt = """请从合同图像中提取以下结构化信息，并以JSON格式输出：
        {
            "party_a": "甲方名称",
            "party_b": "乙方名称", 
            "amount": "合同金额",
            "start_date": "开始日期",
            "end_date": "结束日期",
            "risk_clauses": ["风险条款1", "风险条款2"]
        }
        
        如果某项信息无法确定，请填写"未明确"。"""
        
        response = self.model.chat(
            image=image,
            msgs=[{"role": "user", "content": [image, structured_prompt]}],
            tokenizer=self.tokenizer
        )
        
        # 解析 JSON 响应
        info = self._parse_json_response(response)
        return ContractInfo(**info, raw_response=response)
    
    def _parse_json_response(self, response: str) -> dict:
        import json
        
        # 尝试提取 JSON 块
        json_match = re.search(r'\{.*\}', response, re.DOTALL)
        if json_match:
            try:
                return json.loads(json_match.group())
            except json.JSONDecodeError:
                pass
        
        # 返回默认结构
        return {
            "party_a": "解析失败",
            "party_b": "解析失败",
            "amount": "解析失败",
            "start_date": "解析失败",
            "end_date": "解析失败",
            "risk_clauses": []
        }

import cv2
from transformers import AutoModel, AutoTokenizer
from decord import VideoReader, cpu
import torch
from PIL import Image

def real_time_video_understanding(video_path: str, model, tokenizer):
    vr = VideoReader(video_path, ctx=cpu(0))
    
    frame_interval = 30  # 每隔30帧处理一次
    frames = vr[::frame_interval]
    
    descriptions = []
    for idx, frame in enumerate(frames):
        image = Image.fromarray(frame.asnumpy())
        
        prompt = "描述这一帧画面的主要内容"
        
        response = model.chat(
            image=image,
            msgs=[{"role": "user", "content": [image, prompt]}],
            tokenizer=tokenizer,
            sampling=False,
            stream=True
        )
        
        # 流式输出
        for chunk in response:
            print(chunk, end="", flush=True)
        print()
        
        descriptions.append({
            "frame_index": idx * frame_interval,
            "description": response
        })
    
    return descriptions

video_result = real_time_video_understanding(
    "meeting.mp4",
    model=model,
    tokenizer=tokenizer
)

class AdaptiveVideoSampler:
    def __init__(self, model, tokenizer, strategy: str = "uniform"):
        self.model = model
        self.tokenizer = tokenizer
        self.strategy = strategy
    
    def sample_frames(self, video_path: str, max_frames: int = 16) -> list:
        vr = VideoReader(video_path, ctx=cpu(0))
        total_frames = len(vr)
        
        if self.strategy == "uniform":
            indices = np.linspace(0, total_frames - 1, max_frames, dtype=int)
        elif self.strategy == "scene_change":
            indices = self._detect_scene_changes(vr, max_frames)
        else:
            indices = list(range(0, total_frames, total_frames // max_frames))[:max_frames]
        
        return [Image.fromarray(vr[i].asnumpy()) for i in indices]
    
    def _detect_scene_changes(self, vr, target_frames):
        # 基于帧差异的场景切换检测
        scores = []
        frames = vr[:]
        
        for i in range(1, len(frames)):
            diff = np.mean(np.abs(frames[i].asnumpy() - frames[i-1].asnumpy()))
            scores.append(diff)
        
        # 选择差异最大的帧
        indices = np.argsort(scores)[-target_frames:]
        return sorted(indices)

class IndustrialQualityInspector:
    def __init__(self, model_path: str, confidence_threshold: float = 0.85):
        self.model = AutoModel.from_pretrained(
            model_path,
            quantization_config=BitsAndBytesConfig(load_in_4bit=True),
            device_map="cuda:0",
            trust_remote_code=True
        )
        self.tokenizer = AutoTokenizer.from_pretrained(
            model_path,
            trust_remote_code=True
        )
        self.confidence_threshold = confidence_threshold
    
    def inspect(self, product_image: Image.Image) -> dict:
        prompt = """这是一张工业产品的X光图像。
        请进行严格的质量检测，判断是否存在以下缺陷：
        - 裂纹
        - 气孔
        - 夹杂物
        - 尺寸偏差
        
        对于每种缺陷，请给出缺陷类型、位置（如果可见）和置信度。"""
        
        response = self.model.chat(
            image=product_image,
            msgs=[{"role": "user", "content": [product_image, prompt]}],
            tokenizer=self.tokenizer
        )
        
        defects = self.parse_defects(response)
        
        return {
            "pass": len(defects) == 0,
            "defects": defects,
            "raw_response": response
        }
    
    def parse_defects(self, response: str) -> list:
        # 解析模型输出，提取缺陷信息
        defects = []
        # ... 解析逻辑
        return defects

inspector = IndustrialQualityInspector(
    model_path="./MiniCPM-Llama3-V-2_5-int4",
    confidence_threshold=0.85
)

result = inspector.inspect(product_image)
print(f"质检结果: {'合格' if result['pass'] else '不合格'}")

import time
from datetime import datetime
from typing import List, Tuple

class QualityControlSystem:
    def __init__(self, model_path: str):
        self.inspector = IndustrialQualityInspector(model_path)
        self.inspection_log = []
        self.alert_callbacks = []
    
    def register_alert_callback(self, callback):
        """注册告警回调函数"""
        self.alert_callbacks.append(callback)
    
    def inspect_batch(self, images: List[Tuple[str, Image.Image]]) -> List[dict]:
        results = []
        
        for img_id, image in images:
            start_time = time.time()
            result = self.inspector.inspect(image)
            elapsed = time.time() - start_time
            
            inspection_record = {
                "img_id": img_id,
                "timestamp": datetime.now().isoformat(),
                "result": result,
                "processing_time": elapsed,
                "passed": result['pass']
            }
            
            self.inspection_log.append(inspection_record)
            results.append(inspection_record)
            
            # 触发告警
            if not result['pass']:
                for callback in self.alert_callbacks:
                    callback(inspection_record)
        
        return results
    
    def generate_report(self) -> dict:
        total = len(self.inspection_log)
        passed = sum(1 for r in self.inspection_log if r['passed'])
        
        return {
            "total_inspections": total,
            "passed": passed,
            "failed": total - passed,
            "pass_rate": passed / total if total > 0 else 0,
            "avg_processing_time": sum(r['processing_time'] for r in self.inspection_log) / total if total > 0 else 0
        }

# 使用示例
def on_defect_detected(record):
    print(f"告警：检测到缺陷！产品ID: {record['img_id']}")
    print(f"缺陷详情: {record['result']['defects']}")

qc_system = QualityControlSystem("./MiniCPM-Llama3-V-2_5-int4")
qc_system.register_alert_callback(on_defect_detected)

from transformers import AutoModel, AutoTokenizer, AutoProcessor
from PIL import Image
import torch
import numpy as np

class MultimodalRAG:
    def __init__(self, model_path: str, vector_db, embedding_model):
        self.model = AutoModel.from_pretrained(
            model_path,
            quantization_config=BitsAndBytesConfig(load_in_4bit=True),
            device_map="auto",
            trust_remote_code=True
        )
        self.tokenizer = AutoTokenizer.from_pretrained(
            model_path,
            trust_remote_code=True
        )
        self.vector_db = vector_db
        self.embedding_model = embedding_model
    
    def index_images(self, image_folder: str):
        """将图像集合索引到向量数据库"""
        import os
        
        for img_file in os.listdir(image_folder):
            if img_file.endswith(('.jpg', '.png', '.jpeg')):
                img_path = os.path.join(image_folder, img_file)
                image = Image.open(img_path).convert('RGB')
                
                # 提取图像描述
                description = self.model.chat(
                    image=image,
                    msgs=[{"role": "user", "content": [image, "详细描述这张图片"]}],
                    tokenizer=self.tokenizer
                )
                
                # 生成嵌入向量
                embedding = self.embedding_model.encode(description)
                
                # 存储到向量数据库
                self.vector_db.add(
                    id=img_file,
                    vector=embedding,
                    metadata={"path": img_path, "description": description}
                )
    
    def query(self, query_image: Image.Image, query_text: str, top_k: int = 3):
        """基于图像和文本的混合检索"""
        # 提取查询图像特征
        query_description = self.model.chat(
            image=query_image,
            msgs=[{"role": "user", "content": [query_image, query_text]}],
            tokenizer=self.tokenizer
        )
        
        # 检索相似图像
        results = self.vector_db.search(
            query=query_description,
            top_k=top_k
        )
        
        # 生成最终答案
        context = "\n".join([r['metadata']['description'] for r in results])
        final_prompt = f"""基于以下参考图像信息回答问题：
        
        参考信息：
        {context}
        
        问题：{query_text}"""
        
        answer = self.model.chat(
            image=query_image,
            msgs=[{"role": "user", "content": [query_image, final_prompt]}],
            tokenizer=self.tokenizer
        )
        
        return {"answer": answer, "references": results}

rag_system = MultimodalRAG(
    model_path="./MiniCPM-Llama3-V-2_5-int4",
    vector_db=vector_db,
    embedding_model=embedding_model
)

class AdvancedMultimodalRAG(MultimodalRAG):
    def __init__(self, model_path: str, vector_db, embedding_model):
        super().__init__(model_path, vector_db, embedding_model)
    
    def hybrid_search(
        self, 
        query_image: Image.Image, 
        query_text: str,
        image_weight: float = 0.5,
        text_weight: float = 0.5,
        top_k: int = 5
    ):
        # 纯图像检索
        image_description = self.model.chat(
            image=query_image,
            msgs=[{"role": "user", "content": [query_image, "简洁描述图像内容"]}],
            tokenizer=self.tokenizer
        )
        image_results = self.vector_db.search(image_description, top_k=top_k)
        
        # 纯文本检索
        text_results = self.vector_db.search(query_text, top_k=top_k)
        
        # 融合结果
        fused_scores = {}
        for result in image_results:
            fused_scores[result['id']] = result['score'] * image_weight
        for result in text_results:
            if result['id'] in fused_scores:
                fused_scores[result['id']] += result['score'] * text_weight
            else:
                fused_scores[result['id']] = result['score'] * text_weight
        
        # 排序返回
        sorted_ids = sorted(fused_scores.keys(), key=lambda x: fused_scores[x], reverse=True)
        return [self.vector_db.get(id) for id in sorted_ids[:top_k]]

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset
import torch

# 加载量化基础模型
model = AutoModelForCausalLM.from_pretrained(
    "openbmb/MiniCPM-Llama3-V-2_5-int4",
    quantization_config=BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_use_double_quant=True,
        bnb_4bit_quant_type="nf4"
    ),
    device_map="auto",
    trust_remote_code=True
)

# 配置 LoRA 参数
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                          # LoRA 秩
    lora_alpha=32,                 # LoRA 缩放因子
    lora_dropout=0.05,             # Dropout 概率
    target_modules=[               # 目标模块
        "q_proj", "k_proj", 
        "v_proj", "o_proj",
        "gate_proj", "up_proj", 
        "down_proj"
    ],
    bias="none",
    inference_mode=False
)

# 应用 LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 4,194,304 || all params: 8,061,440,512 || trainable%: 0.052%

# 准备训练数据
train_dataset = load_dataset("json", data_files="train.json")['train']
eval_dataset = load_dataset("json", data_files="eval.json")['eval']

# 训练配置
training_args = TrainingArguments(
    output_dir="./minicpm_finetuned",
    num_train_epochs=3,
    per_device_train_batch_size=2,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    warmup_ratio=0.03,
    lr_scheduler_type="cosine",
    logging_steps=10,
    save_strategy="epoch",
    eval_strategy="epoch",
    fp16=True,
    optim="paged_adamw_32bit",
    max_grad_norm=0.3
)

# 开始训练
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    data_collator=data_collator
)

trainer.train()

# 步骤1: 克隆 llama.cpp 仓库
git clone https://github.com/ggerganov/llama.cpp.git
cd llama.cpp

# 步骤2: 下载量化模型
git clone https://huggingface.co/openbmb/MiniCPM-Llama3-V-2_5-gguf

# 步骤3: 编译
mkdir build && cd build
cmake ..
cmake --build . --config Release

# 步骤4: 运行推理
./bin/llama-cli \
    -m ./MiniCPM-Llama3-V-2_5-gguf/*.gguf \
    -p "请描述这张图片的内容" \
    --image ./test_image.jpg \
    -t 8 \
    -c 4096

# 多线程配置
./bin/llama-cli \
    -m ./model.gguf \
    --image ./test.jpg \
    -t 8              # 8 线程
    -c 4096           # 上下文长度
    --threads 8       # CPU 线程数
    --temp 0.7        # 温度参数
    -p "描述图片"      # 提示词

# 批量处理模式
./bin/llama-bench \
    -m ./model.gguf \
    -t 8 \
    -ngl 0            # 无 GPU 加速

# 安装 Ollama（macOS/Linux）
curl -fsSL https://ollama.com/install.sh | sh

# 运行模型
ollama run openbmb/minicpm-v2.5:latest

# API 服务模式
ollama serve

# Python 调用示例
import ollama

response = ollama.chat(
    model='openbmb/minicpm-v2.5:latest',
    messages=[
        {
            'role': 'user',
            'content': '请描述这张图片',
            'images': ['./test_image.jpg']
        }
    ]
)

print(response['message']['content'])

# 创建自定义 Modelfile
echo '
FROM openbmb/minicpm-v2.5:latest

PARAMETER temperature 0.7
PARAMETER num_predict 512
PARAMETER top_p 0.9

SYSTEM "你是一个专业的图像分析助手，请详细描述图片内容。"
' > Modelfile

# 构建自定义模型
ollama create minicpm-custom -f Modelfile

# 运行自定义模型
ollama run minicpm-custom

import gradio as gr
from transformers import AutoModel, AutoTokenizer, BitsAndBytesConfig
from PIL import Image

model_path = "./MiniCPM-Llama3-V-2_5-int4"
model = AutoModel.from_pretrained(
    model_path,
    quantization_config=BitsAndBytesConfig(load_in_4bit=True),
    device_map="auto",
    trust_remote_code=True
)
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)

def multimodal_chat(image, text, history=[]):
    if image is None:
        return "请上传一张图片"
    
    response = model.chat(
        image=image,
        msgs=[{"role": "user", "content": [image, text]}],
        tokenizer=tokenizer,
        sampling=True,
        stream=True
    )
    
    return response

demo = gr.Interface(
    fn=multimodal_chat,
    inputs=[
        gr.Image(type="pil", label="上传图片"),
        gr.Textbox(label="输入问题", placeholder="请描述这张图片...")
    ],
    outputs=gr.Textbox(label="模型回复"),
    title="MiniCPM-Llama3-V-2_5-Int4 多模态对话系统",
    description="基于 Int4 量化的端侧多模态大模型，支持 OCR、图像理解、多语言等能力"
)

demo.launch(server_name="0.0.0.0", server_port=7860)

import gradio as gr
import threading
import queue
from typing import Iterator

class ProductionGradioApp:
    def __init__(self, model_path: str):
        self.model = AutoModel.from_pretrained(
            model_path,
            quantization_config=BitsAndBytesConfig(load_in_4bit=True),
            device_map="auto",
            trust_remote_code=True
        )
        self.tokenizer = AutoTokenizer.from_pretrained(
            model_path, trust_remote_code=True
        )
        self.request_queue = queue.Queue()
        self.response_queues = {}
        self.response_id = 0
    
    def stream_predict(self, image: Image.Image, text: str) -> Iterator[str]:
        request_id = self._enqueue_request(image, text)
        response_queue = queue.Queue()
        self.response_queues[request_id] = response_queue
        
        while True:
            chunk = response_queue.get()
            if chunk is None:  # 生成结束
                break
            yield chunk
    
    def _enqueue_request(self, image, text) -> int:
        request_id = self.response_id
        self.response_id += 1
        self.request_queue.put((request_id, image, text))
        return request_id
    
    def process_requests(self):
        """后台处理请求"""
        while True:
            request_id, image, text = self.request_queue.get()
            response_queue = self.response_queues[request_id]
            
            try:
                response = self.model.chat(
                    image=image,
                    msgs=[{"role": "user", "content": [image, text]}],
                    tokenizer=self.tokenizer,
                    sampling=True,
                    stream=True
                )
                
                for chunk in response:
                    response_queue.put(chunk)
            finally:
                response_queue.put(None)  # 标记结束
                del self.response_queues[request_id]
    
    def build_interface(self):
        demo = gr.Interface(
            fn=self.stream_predict,
            inputs=[
                gr.Image(type="pil", label="上传图片"),
                gr.Textbox(label="输入问题", lines=3)
            ],
            outputs=gr.Textbox(label="模型回复", lines=5),
            title="MiniCPM-V 2.5 生产级服务"
        )
        return demo

# 启动
app = ProductionGradioApp("./MiniCPM-Llama3-V-2_5-int4")
threading.Thread(target=app.process_requests, daemon=True).start()
app.build_interface().launch(server_name="0.0.0.0", server_port=7860)

# requirements.txt for MiniCPM-Llama3-V-2_5-Int4
transformers>=4.40.0
accelerate>=0.30.0
bitsandbytes>=0.41.0
torch>=2.1.0
torchvision>=0.16.0
Pillow>=10.1.0
sentencepiece>=0.1.99
decord>=0.6.0
peft>=0.10.0
trl>=0.7.0
gradio>=4.0.0
protobuf>=3.20.0

# 1. 对比量化前后的中间结果
def debug_quantization_effect(model, test_input):
    # 获取 FP16 模型输出
    model_fp16 = model.to(torch.float16)
    output_fp16 = model_fp16(test_input)
    
    # 获取 Int4 模型输出
    model_int4 = model.to(BitsAndBytesConfig(load_in_4bit=True))
    output_int4 = model_int4(test_input)
    
    # 计算差异
    diff = torch.abs(output_fp16 - output_int4)
    print(f"最大差异: {diff.max()}")
    print(f"平均差异: {diff.mean()}")

# 2. 逐层分析量化误差
def analyze_layerwise_error(model, test_input):
    for name, module in model.named_modules():
        if hasattr(module, 'weight'):
            weight = module.weight.data
            # 计算权重分布统计
            print(f"{name}: mean={weight.mean():.4f}, std={weight.std():.4f}, min={weight.min():.4f}, max={weight.max():.4f}")