大模型落地这一年：QLoRA微调、vLLM加速与Multi-Agent编排的生产级实战笔记

from unsloth import FastLanguageModel
import torch
from trl import SFTTrainer
from transformers import TrainingArguments

# 加载4-bit量化模型
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="Qwen/Qwen2.5-7B-Instruct",
    max_seq_length=4096,
    dtype=None,  # 自动选择Float16/BFloat16
    load_in_4bit=True,
    r=16,  # LoRA秩
    lora_alpha=32,
    lora_dropout=0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
)

# 添加PEFT适配器
model = FastLanguageModel.get_peft_model(
    model,
    use_gradient_checkpointing=True,
    use_rslora=True,  # 秩稳定优化
)

# 训练参数
training_args = TrainingArguments(
    output_dir="./medical_lora",
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,
    num_train_epochs=3,
    learning_rate=2e-4,
    fp16=True,
    logging_steps=10,
    save_strategy="epoch",
    ddp_find_unused_parameters=False,
)

trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=medical_dataset,
    args=training_args,
    dataset_text_field="text",
    max_seq_length=4096,
)
trainer.train()

from vllm import LLM, SamplingParams
import asyncio

# 初始化推理引擎（TP=2，双卡并行）
llm = LLM(
    model="./medical_lora/checkpoint-300",
    tensor_parallel_size=2,
    dtype="float16",
    enforce_eager=True,  # 避免CUDA图编译开销
    max_num_batched_tokens=8192,
    enable_prefix_caching=True,  # 缓存System Prompt前缀
)

sampling_params = SamplingParams(
    temperature=0.1,
    top_p=0.9,
    max_tokens=1024,
    stop=["<|im_end|>", "\n\n"],
    skip_special_tokens=True,
)

# 异步批量处理
async def batch_generate(prompts: list[str]):
    loop = asyncio.get_event_loop()
    outputs = await loop.run_in_executor(
        None, lambda: llm.generate(prompts, sampling_params)
    )
    return [out.outputs[0].text for out in outputs]

from langchain_community.vectorstores import Milvus
from langchain_huggingface import HuggingFaceEmbeddings
from sentence_transformers import CrossEncoder

# 阶段1：稠密向量召回（BGE-M3）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True},
)

vector_store = Milvus(
    embedding_function=embeddings,
    collection_name="medical_knowledge",
    connection_args={"uri": "milvus://localhost:19530"},
)

# 阶段2：精排重排序
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def retrieve_with_rerank(query: str, top_k: int = 20):
    # 第一阶段：ANN召回20篇
    docs = vector_store.similarity_search(query, k=top_k)
    
    # 第二阶段：Cross-Encoder精排取Top-5
    pairs = [[query, doc.page_content] for doc in docs]
    scores = reranker.predict(pairs, activation=True)  # sigmoid概率
    
    sorted_pairs = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in sorted_pairs[:5]]

import autogen
from autogen import ConversableAgent, UserProxyAgent
from autogen.tools import FunctionTool

# 工具函数
def query_database(sql: str) -> str:
    # 沙箱执行SQL
    return f"查询结果: {sql}"

def send_email(recipient: str, content: str) -> str:
    return f"邮件已发送至{recipient}"

tools = [
    FunctionTool(query_database, description="查询业务数据库"),
    FunctionTool(send_email, description="发送邮件通知"),
]

# Planner：负责拆解任务
planner = ConversableAgent(
    name="Planner",
    system_message="将复杂任务分解为子步骤，明确依赖关系。",
    llm_config={"config_list": [{"model": "gpt-4", "api_key": "xxx"}]},
)

# Executor：执行具体操作
executor = ConversableAgent(
    name="Executor",
    system_message="严格按照Planner的步骤执行，调用工具并返回结果。",
    llm_config={"config_list": [{"model": "qwen-7b", "base_url": "http://localhost:8000/v1"}]},
    tools=tools,
)

# Validator：校验结果
validator = ConversableAgent(
    name="Validator",
    system_message="校验执行结果是否符合预期，如有问题反馈修正。",
    llm_config={"config_list": [{"model": "gpt-4"}]},
)

# 群聊编排
groupchat = autogen.GroupChat(
    agents=[planner, executor, validator],
    messages=[],
    max_round=10,
)
manager = autogen.GroupChatManager(groupchat=groupchat)

user_proxy = UserProxyAgent(name="User", code_execution_config=False)
user_proxy.initiate_chat(
    manager,
    message="分析本月销售数据，若增长超10%则邮件通知销售总监。"
)

from prometheus_client import Histogram, Counter
import time

inference_latency = Histogram("llm_inference_seconds", "推理延迟分布")
token_counter = Counter("llm_tokens_total", "Token消耗总量", ["type"])

@inference_latency.time()
def invoke_llm(prompt: str):
    start = time.perf_counter()
    response = llm.generate([prompt], sampling_params)
    latency = time.perf_counter() - start
    tokens_used = len(response[0].outputs[0].token_ids)
    token_counter.labels(type="output").inc(tokens_used)
    return response

import bentoml

# 保存模型到BentoML Store
bentoml.transformers.save_model(
    "qwen_medical",
    model,
    tokenizer=tokenizer,
    metadata={"accuracy": 0.89, "framework": "pytorch", "lora_rank": 16},
)

# 定义服务
@bentoml.service(name="llm_service", resources={"gpu": 1})
class LLMService:
    def __init__(self):
        self.model = bentoml.transformers.load_model("qwen_medical:latest")
        
    @bentoml.api
    def generate(self, request: JSON) -> JSON:
        prompt = request.get("prompt")
        return {"response": self.model.generate(prompt)}