模型调优不是纸上谈兵，谈谈AI工程化落地的一线真实经验

```
你是一个电商客服助手，请回答用户的问题。
```

```python
SYSTEM_PROMPT = """你是一个专业的电商客服助手，服务于"XX优选"电商平台。

## 你的职责
1. 回答用户关于订单、物流、退换货、商品信息的问题
2. 当用户询问具体订单状态时，引导用户提供订单号
3. 遇到无法确定的问题，诚实告知并建议联系人工客服

## 回答规范
- 基于提供的参考资料回答，不要编造信息
- 回答简洁明了，直接给出用户需要的答案
- 涉及退款/退货等敏感操作时，给出明确的操作步骤
- 语气友善专业，使用"您"称呼用户

## 参考资料使用规则
- 参考资料中有的信息，直接引用
- 参考资料中没有的，说"抱歉，我需要帮您转接人工客服确认"
- 不要把参考资料的原话全部搬出来，提炼关键信息

## 禁止事项
- 不要讨论与电商客服无关的话题（政治、宗教、娱乐等）
- 不要承诺参考资料中没有的优惠政策
- 不要编造订单号或物流信息"""

# 构造Few-shot示例
FEW_SHOT_EXAMPLES = [
    {
        "role": "user",
        "content": "我的订单什么时候到？"
    },
    {
        "role": "assistant",
        "content": "您好！请提供一下您的订单号，我帮您查询物流状态。您可以在"我的订单"页面找到订单号，格式类似：XX20240115XXXX。"
    },
    {
        "role": "user",
        "content": "订单号是XX20240115001，帮我看看到哪了"
    },
    {
        "role": "assistant",
        "content": "好的，已为您查询到订单XX20240115001的物流信息：\n\n当前状态：运输中\n当前位置：杭州转运中心\n预计送达：明天下午\n\n包裹正在派送途中，请您留意短信通知。如超过预计时间未收到，请联系人工客服处理。"
    },
]
```

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
import re

class SmartDocumentSplitter:
    """基于文档结构的智能分块器"""
    
    def __init__(self, chunk_size=512, chunk_overlap=64):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        # 按标题层级分块，优先在高级标题处切割
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=[
                "\n## ",       # 二级标题
                "\n### ",      # 三级标题
                "\n#### ",     # 四级标题
                "\n\n",        # 段落
                "\n",          # 换行
                "。",          # 中文句号
                "；",          # 中文分号
                " ",           # 空格
                "",            # 字符级
            ],
        )
    
    def split(self, documents):
        """分块并增强元数据"""
        chunks = self.splitter.split_documents(documents)
        
        for chunk in chunks:
            # 提取所属章节标题作为元数据
            text = chunk.page_content
            section = self._extract_section(text)
            chunk.metadata["section"] = section
            
            # 计算内容类型
            chunk.metadata["content_type"] = self._detect_content_type(text)
        
        return chunks
    
    def _extract_section(self, text):
        """从文本中提取最近的标题"""
        lines = text.strip().split("\n")
        for line in reversed(lines):
            if line.strip().startswith("#"):
                return line.strip().lstrip("#").strip()
        return "未分类"
    
    def _detect_content_type(self, text):
        """检测内容类型，辅助后续检索"""
        if re.search(r'步骤|流程|操作', text):
            return "procedure"
        elif re.search(r'定义|是指|表示', text):
            return "definition"
        elif re.search(r'价格|费用|收费', text):
            return "pricing"
        elif re.search(r'注意|警告|禁止', text):
            return "warning"
        else:
            return "general"


# 使用示例
splitter = SmartDocumentSplitter(chunk_size=512, chunk_overlap=64)
loader = PyPDFLoader("./docs/退换货政策.pdf")
docs = loader.load()
chunks = splitter.split(docs)

for chunk in chunks[:3]:
    print(f"章节: {chunk.metadata['section']}")
    print(f"类型: {chunk.metadata['content_type']}")
    print(f"内容: {chunk.page_content[:100]}...")
    print("---")
```

```python
from rank_bm25 import BM25Okapi
import jieba

class HybridRetriever:
    """混合检索器：向量检索 + BM25关键词检索"""
    
    def __init__(self, vector_store, documents, embedder):
        self.vector_store = vector_store
        self.embedder = embedder
        
        # 构建BM25索引
        tokenized_docs = [list(jieba.cut(doc["text"])) for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        self.documents = documents
    
    def retrieve(self, query, top_k=10):
        """双路召回 + RRF融合排序"""
        # 向量检索
        query_vec = self.embedder.embed_query(query)
        vec_results = self.vector_store.search(query_vec, limit=top_k)
        vec_scores = {r["id"]: r["score"] for r in vec_results}
        
        # BM25关键词检索
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)
        
        # RRF (Reciprocal Rank Fusion) 融合
        rrf_k = 60  # RRF平滑参数
        rrf_scores = {}
        
        # 向量检索结果排名
        vec_ranked = sorted(vec_scores.items(), key=lambda x: x[1], reverse=True)
        for rank, (doc_id, _) in enumerate(vec_ranked):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1.0 / (rrf_k + rank + 1)
        
        # BM25检索结果排名
        bm25_ranked = sorted(enumerate(bm25_scores), key=lambda x: x[1], reverse=True)
        for rank, (doc_id, _) in enumerate(bm25_ranked[:top_k]):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1.0 / (rrf_k + rank + 1)
        
        # 按融合分数排序
        final_ranking = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
        return [doc_id for doc_id, _ in final_ranking[:top_k]]
```

```python
from sentence_transformers import CrossEncoder

class Reranker:
    """基于Cross-Encoder的重排序器"""
    
    def __init__(self, model_name="BAAI/bge-reranker-v2-m3"):
        self.model = CrossEncoder(model_name, max_length=512)
    
    def rerank(self, query, documents, top_n=3):
        """对检索结果做精排"""
        pairs = [(query, doc["text"]) for doc in documents]
        scores = self.model.predict(pairs)
        
        for doc, score in zip(documents, scores):
            doc["rerank_score"] = float(score)
        
        documents.sort(key=lambda x: x["rerank_score"], reverse=True)
        return documents[:top_n]
```

```python
import torch
from datasets import Dataset
from peft import LoraConfig, get_peft_model, TaskType
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForSeq2Seq,
)

# ---- 1. 加载模型 ----
model_id = "Qwen/Qwen2.5-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)

model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True,
)

# ---- 2. 配置LoRA ----
# LoRA只训练这些低秩矩阵，原模型参数冻结
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                    # LoRA秩，越大表达能力越强但参数越多
    lora_alpha=32,           # 缩放系数，一般设为r的2倍
    lora_dropout=0.05,       # 防过拟合
    target_modules=[         # 对哪些层加LoRA
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
    bias="none",
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 39,321,600 || all params: 7,621,877,760 || trainable%: 0.52%
# 只训练0.52%的参数！

# ---- 3. 准备训练数据 ----
# 数据格式: [{"instruction": "...", "input": "...", "output": "..."}]
def format_training_data(examples):
    """将业务QA数据格式化为模型训练格式"""
    formatted = []
    for ex in examples:
        conversation = [
            {"role": "system", "content": "你是XX优选电商平台的客服助手。"},
            {"role": "user", "content": ex["instruction"]},
            {"role": "assistant", "content": ex["output"]},
        ]
        text = tokenizer.apply_chat_template(conversation, tokenize=False)
        formatted.append({"text": text})
    return formatted

# 加载训练数据（实际从业务日志中提取的高质量QA对）
train_data = load_training_examples("./data/train_qa.jsonl")
formatted_data = format_training_data(train_data)
dataset = Dataset.from_list(formatted_data)

def tokenize_fn(examples):
    tokenized = tokenizer(
        examples["text"],
        truncation=True,
        max_length=1024,
        padding=False,
    )
    tokenized["labels"] = tokenized["input_ids"].copy()
    return tokenized

dataset = dataset.map(tokenize_fn, batched=True)

# ---- 4. 训练 ----
training_args = TrainingArguments(
    output_dir="./lora_output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,    # 等效batch_size=16
    learning_rate=2e-4,
    lr_scheduler_type="cosine",
    warmup_ratio=0.1,
    logging_steps=10,
    save_strategy="epoch",
    save_total_limit=3,
    bf16=True,                        # 混合精度训练
    gradient_checkpointing=True,      # 省显存
    report_to="none",
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=dataset,
    data_collator=DataCollatorForSeq2Seq(
        tokenizer=tokenizer,
        padding=True,
        return_tensors="pt",
    ),
)

trainer.train()

# ---- 5. 保存LoRA权重 ----
model.save_pretrained("./lora_output/qwen2.5-7b-ecommerce-lora")
tokenizer.save_pretrained("./lora_output/qwen2.5-7b-ecommerce-lora")
print("LoRA微调完成！")
```