vLLM 高性能推理系列——LoRA 动态加载

# 根据请求动态切换 LoRA
if request.scene == "客服":
    model.load_adapter("/path/to/customer_service_lora")
elif request.scene == "代码":
    model.load_adapter("/path/to/code_lora")

from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

# 调用 SQL 专家
response = client.chat.completions.create(
    model="sql_expert",  # 对应启动时注册的名字
    messages=[
        {"role": "system", "content": "你是一个 SQL 专家，请将自然语言转换为 SQL 查询。"},
        {"role": "user", "content": "查询销售额最高的前10名员工"}
    ],
    max_tokens=256
)
print(response.choices[0].message.content)

# 调用客服助手
response = client.chat.completions.create(
    model="customer_service",
    messages=[
        {"role": "user", "content": "我的订单什么时候能到？"}
    ]
)

# 调用 Base Model（不使用任何 LoRA）
response = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",  # 使用原始模型名
    messages=[{"role": "user", "content": "你好"}]
)

import requests

# 动态注册新的 LoRA
response = requests.post(
    "http://localhost:8000/v1/load_lora_adapter",
    json={
        "lora_name": "new_adapter",
        "lora_path": "/data/loras/new-lora"
    }
)

if response.status_code == 200:
    print("LoRA loaded successfully")
    # 现在可以使用 model="new_adapter" 了

# 卸载 LoRA 以释放显存
response = requests.post(
    "http://localhost:8000/v1/unload_lora_adapter",
    json={"lora_name": "old_adapter"}
)

# 推荐的统一配置
peft_config = LoraConfig(
    r=32,                    # rank，建议 16/32/64
    lora_alpha=64,           # 通常设为 2×r
    target_modules=["q_proj", "v_proj", "k_proj", "o_proj", 
                   "gate_proj", "up_proj", "down_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

import random

def get_model_for_user(user_id: str, scene: str) -> str:
    """根据用户 ID 决定使用哪个版本的 LoRA"""
    
    # 10% 用户使用新版本
    if hash(user_id) % 100 < 10:
        return f"{scene}_v2"  # 新版本
    else:
        return f"{scene}_v1"  # 稳定版本

# 按 LoRA 统计请求量
from collections import defaultdict
import time

class LoRAMetrics:
    def __init__(self):
        self.request_count = defaultdict(int)
        self.latency_sum = defaultdict(float)
    
    def record(self, lora_name: str, latency: float):
        self.request_count[lora_name] += 1
        self.latency_sum[lora_name] += latency
    
    def get_stats(self):
        stats = {}
        for name in self.request_count:
            count = self.request_count[name]
            stats[name] = {
                "requests": count,
                "avg_latency": self.latency_sum[name] / count if count > 0 else 0
            }
        return stats

# 先卸载旧版本
requests.post("http://localhost:8000/v1/unload_lora_adapter",
              json={"lora_name": "my_adapter"})

# 再加载新版本（路径可以相同）
requests.post("http://localhost:8000/v1/load_lora_adapter",
              json={"lora_name": "my_adapter", "lora_path": "/path/to/updated-lora"})