章节6：模块综合实战

---

学习目标

• 综合运用 RAG + LangChain + ChromaDB + Reranker 构建企业私有知识库问答系统
• 运用 PEFT + QLoRA 对开源大模型进行垂直领域微调
• 理解 RAG 与微调两种技术的适用场景与组合策略
• 掌握大模型应用落地的完整工程链路

---

实战项目一：企业私有知识库 RAG 问答系统

项目概述

构建一个面向企业内部的智能问答系统，能够基于公司内部文档（技术手册、产品文档、规章制度）进行精准问答，并支持引用溯源。

技术栈：LangChain + ChromaDB + BGE Embedding + BGE Reranker + Qwen2.5 / GPT-4o

系统架构

用户提问
    │
    ▼
┌──────────────────────────────────────┐
│  1. 查询向量化 (BGE Embedding)       │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  2. 粗排检索 (ChromaDB Top-20)       │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  3. 精排重排序 (BGE Reranker Top-3)  │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  4. 上下文组装 + Prompt 增强         │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  5. LLM 生成回答 (含引用来源)         │
└──────────────┬───────────────────────┘
               │
               ▼
             用户看到最终答案

完整代码实现

"""
企业私有知识库 RAG 问答系统
技术栈：LangChain + ChromaDB + BGE Embedding + BGE Reranker
"""

import os
from typing import List, Dict, Any

# ── 配置 ─────────────────────────────────────────────
CONFIG = {
    "embedding_model": "BAAI/bge-large-zh-v1.5",
    "reranker_model": "BAAI/bge-reranker-v2-m3",
    "chunk_size": 512,
    "chunk_overlap": 64,
    "retrieval_top_k": 20,     # 粗排数量
    "rerank_top_k": 3,         # 精排数量
    "chroma_path": "./enterprise_kb",
    "llm_model": "Qwen/Qwen2.5-7B-Instruct",
    "llm_api_base": "http://localhost:8000/v1",  # 本地部署或云端API
}

# ── 1. 文档加载与分块 ──────────────────────────────────

from langchain_community.document_loaders import (
    PyPDFLoader, TextLoader, DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_and_split_documents(data_dir: str):
    """
    加载指定目录下的所有文档并分块
    支持格式: .txt, .pdf, .md, .csv
    """
    loaders = {
        ".txt": DirectoryLoader(data_dir, glob="**/*.txt", loader_cls=TextLoader),
        ".md": DirectoryLoader(data_dir, glob="**/*.md", loader_cls=TextLoader),
        ".pdf": DirectoryLoader(data_dir, glob="**/*.pdf", loader_cls=PyPDFLoader),
    }
    
    all_docs = []
    for ext, loader in loaders.items():
        try:
            docs = loader.load()
            all_docs.extend(docs)
            print(f"加载 {ext} 文件: {len(docs)} 个")
        except Exception as e:
            print(f"加载 {ext} 文件失败: {e}")
    
    # 分块
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CONFIG["chunk_size"],
        chunk_overlap=CONFIG["chunk_overlap"],
        separators=["\n\n", "\n", "。", ".", " ", ""]
    )
    
    chunks = splitter.split_documents(all_docs)
    print(f"文档分块后总数: {len(chunks)}")
    return chunks

# ── 2. 向量化与向量数据库 ──────────────────────────────

from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_community.vectorstores import Chroma

def create_embeddings():
    """创建 BGE Embedding 模型"""
    return HuggingFaceBgeEmbeddings(
        model_name=CONFIG["embedding_model"],
        model_kwargs={"device": "cuda"},
        encode_kwargs={"normalize_embeddings": True}
    )

def build_vector_store(chunks, embeddings):
    """构建向量数据库"""
    vector_store = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CONFIG["chroma_path"]
    )
    vector_store.persist()
    print(f"向量数据库已保存至: {CONFIG['chroma_path']}")
    return vector_store

def load_vector_store(embeddings):
    """加载已有向量数据库"""
    return Chroma(
        persist_directory=CONFIG["chroma_path"],
        embedding_function=embeddings
    )

# ── 3. Reranker 重排序 ────────────────────────────────

from sentence_transformers import CrossEncoder

class Reranker:
    """基于 Cross-Encoder 的重排序器"""
    
    def __init__(self, model_name: str = None):
        self.model = CrossEncoder(
            model_name or CONFIG["reranker_model"],
            device="cuda"
        )
    
    def rerank(self, query: str, documents: List[Dict], top_k: int = None) -> List[Dict]:
        """
        对检索结果进行重排序
        documents: [{"content": str, "metadata": dict}, ...]
        """
        pairs = [[query, doc["content"]] for doc in documents]
        scores = self.model.predict(pairs)
        
        # 按分数降序排列
        ranked = sorted(
            zip(documents, scores),
            key=lambda x: x[1],
            reverse=True
        )
        
        top_k = top_k or CONFIG["rerank_top_k"]
        results = [
            {"content": doc["content"], "metadata": doc["metadata"], "score": score}
            for doc, score in ranked[:top_k]
        ]
        return results

# ── 4. RAG 问答链 ──────────────────────────────────────

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

class EnterpriseRAG:
    """企业知识库 RAG 问答系统"""
    
    def __init__(self, vector_store, reranker, llm):
        self.vector_store = vector_store
        self.reranker = reranker
        self.llm = llm
        
        # 带引用溯源的 Prompt
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个企业知识库智能助手。你的回答必须：
1. 严格基于提供的参考文档，不要编造信息
2. 每个观点后标注引用来源编号 [文档X]
3. 如果参考文档不足以回答，请说"参考文档中未找到相关信息"
4. 用中文回答，语言简洁专业"""),
            ("human", "参考文档：\n{context}\n\n问题：{question}")
        ])
    
    def query(self, question: str) -> Dict[str, Any]:
        """执行 RAG 查询"""
        # 1. 向量检索（粗排）
        initial_results = self.vector_store.similarity_search_with_score(
            question,
            k=CONFIG["retrieval_top_k"]
        )
        
        # 格式化
        docs = [
            {"content": doc.page_content, "metadata": doc.metadata}
            for doc, score in initial_results
        ]
        
        # 2. Reranker 重排序（精排）
        reranked = self.reranker.rerank(question, docs)
        
        # 3. 构建上下文（带编号）
        context_parts = []
        for i, doc in enumerate(reranked):
            source = doc["metadata"].get("source", "未知来源")
            context_parts.append(f"[文档{i+1}]（来自: {source}）\n{doc['content']}")
        
        context = "\n\n---\n\n".join(context_parts)
        
        # 4. LLM 生成
        messages = self.prompt.format_messages(
            context=context,
            question=question
        )
        response = self.llm.invoke(messages)
        
        return {
            "answer": response.content,
            "sources": [
                {
                    "content": doc["content"][:200] + "...",
                    "source": doc["metadata"].get("source", "未知"),
                    "relevance_score": doc["score"]
                }
                for doc in reranked
            ]
        }

# ── 5. 主流程 ──────────────────────────────────────────

def main_rag():
    """主函数：初始化与运行"""
    print("=" * 50)
    print("企业私有知识库 RAG 问答系统")
    print("=" * 50)
    
    # 初始化组件
    embeddings = create_embeddings()
    reranker = Reranker()
    
    llm = ChatOpenAI(
        model=CONFIG["llm_model"],
        openai_api_base=CONFIG["llm_api_base"],
        temperature=0.1,  # 低温度，确保答案确定性
    )
    
    # 加载或构建向量库
    if os.path.exists(CONFIG["chroma_path"]):
        print("加载已有向量数据库...")
        vector_store = load_vector_store(embeddings)
    else:
        print("请先运行 load_and_split_documents() 导入文档")
        return
    
    # 创建 RAG 实例
    rag = EnterpriseRAG(vector_store, reranker, llm)
    
    # 交互式问答
    print("\n系统就绪！输入 'exit' 退出。\n")
    while True:
        question = input("\n请输入问题: ").strip()
        if question.lower() in ("exit", "quit", "q"):
            break
        
        print("\n正在检索...")
        result = rag.query(question)
        
        print(f"\n📝 回答:")
        print("-" * 40)
        print(result["answer"])
        print("-" * 40)
        
        print("\n📚 参考来源:")
        for i, src in enumerate(result["sources"]):
            print(f"  {i+1}. [{src['source']}] (相关度: {src['relevance_score']:.4f})")
            print(f"     预览: {src['content'][:100]}...")

# if __name__ == "__main__":
#     main_rag()

---

实战项目二：垂直领域大模型 LoRA 微调

项目概述

针对企业特定业务场景（如客服对话、法律咨询、医疗问答），使用 QLoRA 对开源模型进行垂直领域微调。

技术栈：Transformers + PEFT + QLoRA + bitsandbytes + 指令数据集

完整代码实现

"""
垂直领域大模型 LoRA 微调
技术栈：Transformers + PEFT + QLoRA + bitsandbytes
"""

# ── 1. 环境配置 ──────────────────────────────────────────

"""
# 安装依赖
pip install torch transformers accelerate peft bitsandbytes datasets
pip install transformers[torch]
"""

import torch
import json
from datasets import Dataset
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    BitsAndBytesConfig,
    Trainer,
    DataCollatorForSeq2Seq,
)
from peft import (
    LoraConfig,
    get_peft_model,
    prepare_model_for_kbit_training,
    PeftModel,
)

# ── 配置 ────────────────────────────────────────────────

FT_CONFIG = {
    # 模型配置
    "base_model": "Qwen/Qwen2.5-7B-Instruct",
    "lora_r": 16,
    "lora_alpha": 32,
    "lora_dropout": 0.05,
    "target_modules": ["q_proj", "k_proj", "v_proj", "o_proj"],
    
    # 训练配置
    "output_dir": "./domain-lora-checkpoints",
    "num_epochs": 3,
    "batch_size": 4,
    "gradient_accumulation_steps": 4,
    "learning_rate": 2e-4,
    "max_seq_length": 1024,
    
    # 量化配置
    "use_4bit": True,
    "bnb_4bit_quant_type": "nf4",
    "bnb_4bit_compute_dtype": "bfloat16",
    "bnb_4bit_use_double_quant": True,
}

# ── 2. 数据构造 ──────────────────────────────────────────

def build_instruction_dataset(data_path: str) -> Dataset:
    """
    从 JSONL 文件加载指令数据并格式化
    数据格式: {"instruction": "...", "input": "...", "output": "..."}
    """
    data = []
    with open(data_path, "r", encoding="utf-8") as f:
        for line in f:
            example = json.loads(line.strip())
            data.append(example)
    
    def format_instruction(example):
        """构造指令微调格式"""
        if example.get("input"):
            text = f"""<|im_start|>system
你是一个专业领域的AI助手。<|im_end|>
<|im_start|>user
{example['instruction']}

{example['input']}<|im_end|>
<|im_start|>assistant
{example['output']}<|im_end|>"""
        else:
            text = f"""<|im_start|>system
你是一个专业领域的AI助手。<|im_end|>
<|im_start|>user
{example['instruction']}<|im_end|>
<|im_start|>assistant
{example['output']}<|im_end|>"""
        return {"text": text}
    
    dataset = Dataset.from_list(data)
    dataset = dataset.map(format_instruction)
    return dataset


def data_augmentation(dataset):
    """简单的数据增强：指令同义改写"""
    import random
    
    def augment(example):
        # 20% 概率进行增强
        if random.random() < 0.2:
            return example
        # 实际应用中可接入 LLM 进行指令改写
        return example
    
    return dataset.map(augment)

# ── 3. 模型加载与 QLoRA 配置 ─────────────────────────────

def create_qlora_model():
    """加载 4-bit 量化模型并应用 LoRA"""
    
    # 1. 量化配置
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=FT_CONFIG["use_4bit"],
        bnb_4bit_quant_type=FT_CONFIG["bnb_4bit_quant_type"],
        bnb_4bit_compute_dtype=torch.bfloat16,
        bnb_4bit_use_double_quant=FT_CONFIG["bnb_4bit_use_double_quant"],
    )
    
    # 2. 加载量化模型
    model = AutoModelForCausalLM.from_pretrained(
        FT_CONFIG["base_model"],
        quantization_config=bnb_config,
        device_map="auto",
        trust_remote_code=True,
    )
    
    tokenizer = AutoTokenizer.from_pretrained(
        FT_CONFIG["base_model"],
        trust_remote_code=True,
    )
    tokenizer.pad_token = tokenizer.eos_token
    
    # 3. 为 k-bit 训练准备
    model = prepare_model_for_kbit_training(model)
    
    # 4. LoRA 配置
    lora_config = LoraConfig(
        r=FT_CONFIG["lora_r"],
        lora_alpha=FT_CONFIG["lora_alpha"],
        target_modules=FT_CONFIG["target_modules"],
        lora_dropout=FT_CONFIG["lora_dropout"],
        bias="none",
        task_type="CAUSAL_LM",
    )
    
    model = get_peft_model(model, lora_config)
    
    # 打印可训练参数
    model.print_trainable_parameters()
    # 输出: trainable params: ~8M || all params: ~7B || trainable%: ~0.11%
    
    return model, tokenizer

# ── 4. 训练 ─────────────────────────────────────────────

def train_model(model, tokenizer, dataset):
    """执行 LoRA 微调"""
    
    def tokenize_function(examples):
        return tokenizer(
            examples["text"],
            truncation=True,
            padding=False,
            max_length=FT_CONFIG["max_seq_length"],
        )
    
    tokenized_dataset = dataset.map(
        tokenize_function,
        remove_columns=dataset.column_names,
        batched=True,
    )
    
    # 数据整理器
    data_collator = DataCollatorForSeq2Seq(
        tokenizer=tokenizer,
        model=model,
        padding=True,
    )
    
    # 训练参数
    training_args = TrainingArguments(
        output_dir=FT_CONFIG["output_dir"],
        num_train_epochs=FT_CONFIG["num_epochs"],
        per_device_train_batch_size=FT_CONFIG["batch_size"],
        gradient_accumulation_steps=FT_CONFIG["gradient_accumulation_steps"],
        learning_rate=FT_CONFIG["learning_rate"],
        warmup_ratio=0.03,
        logging_steps=10,
        save_strategy="epoch",
        save_total_limit=2,
        fp16=True,
        report_to="none",
        remove_unused_columns=False,
        dataloader_num_workers=2,
    )
    
    # Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset,
        tokenizer=tokenizer,
        data_collator=data_collator,
    )
    
    # 开始训练
    trainer.train()
    
    # 保存 LoRA 适配器权重
    model.save_pretrained("./domain-lora-final")
    tokenizer.save_pretrained("./domain-lora-final")
    print("LoRA 适配器已保存至 ./domain-lora-final")
    
    return trainer

# ── 5. 推理与对比 ────────────────────────────────────────

def inference_with_lora(base_model_path, lora_path, prompt):
    """加载 LoRA 适配器进行推理"""
    
    from peft import PeftModel
    
    # 加载基础模型
    model = AutoModelForCausalLM.from_pretrained(
        base_model_path,
        device_map="auto",
        torch_dtype=torch.bfloat16,
    )
    tokenizer = AutoTokenizer.from_pretrained(base_model_path)
    
    # 加载 LoRA 适配器
    model = PeftModel.from_pretrained(model, lora_path)
    
    # 推理
    messages = [
        {"role": "system", "content": "你是一个专业领域的AI助手。"},
        {"role": "user", "content": prompt},
    ]
    
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    
    outputs = model.generate(
        **inputs,
        max_new_tokens=512,
        temperature=0.3,
        top_p=0.9,
        do_sample=True,
    )
    
    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return response

# ── 6. 主流程 ──────────────────────────────────────────

def main_finetune():
    """微调主流程"""
    print("=" * 50)
    print("垂直领域大模型 LoRA 微调")
    print("=" * 50)
    
    # 1. 准备数据
    print("\n[1/4] 加载训练数据...")
    dataset = build_instruction_dataset("domain_train_data.jsonl")
    dataset = data_augmentation(dataset)
    print(f"训练数据量: {len(dataset)} 条")
    
    # 2. 创建模型
    print("\n[2/4] 创建 QLoRA 模型...")
    model, tokenizer = create_qlora_model()
    
    # 3. 训练
    print("\n[3/4] 开始训练...")
    trainer = train_model(model, tokenizer, dataset)
    
    # 4. 测试推理
    print("\n[4/4] 测试推理效果...")
    test_prompts = [
        "请介绍一下我们的主要业务流程",
        "客户投诉处理的标准流程是什么？",
        "解释一下这个专业术语的含义",
    ]
    
    for prompt in test_prompts:
        print(f"\n用户: {prompt}")
        response = inference_with_lora(
            FT_CONFIG["base_model"],
            "./domain-lora-final",
            prompt,
        )
        print(f"助手: {response[:200]}...")

# if __name__ == "__main__":
#     main_finetune()

---

实战对比：RAG vs 微调 vs 混合方案

选型决策

场景	推荐方案	理由
企业内部知识库问答	RAG	知识频繁更新，需要引用溯源
客服对话风格统一	微调	需要统一的回复风格和话术
法律/医疗专业回答	RAG + 微调	微调学专业术语，RAG 提供最新法规
代码生成助手	微调	需要学习特定的代码规范和框架
产品文档问答	RAG	文档持续更新，无法频繁重训

组合策略架构

用户问题
    │
    ├──▶ 微调后的模型初步理解
    │    （掌握了专业术语和业务逻辑）
    │
    ▼
RAG 检索增强
    │
    ├──▶ 检索最新知识库文档
    ├──▶ Reranker 精排
    │
    ▼
微调 + RAG 联合生成
    │
    └──▶ 输出：专业 + 准确 + 可溯源

---

小结与练习

小结

• RAG 系统适合知识频繁更新的场景，核心在于分块策略、Embedding 质量、Reranker 精排
• LoRA 微调适合需要统一风格和专业术语的场景，QLoRA 使消费级 GPU 也能微调 7B 模型
• RAG + 微调混合方案是工业界的主流最佳实践，两者互补而非替代
• 完整的大模型应用工程链路包括：数据准备 → 模型选型 → 检索/微调 → 评估 → 部署

练习题

1. 基础题：用你自己的话描述 RAG 和微调各自的优缺点，以及分别在什么场景下使用。
2. 实践题（RAG）：收集 10 篇你感兴趣的技术文章，构建一个 RAG 问答系统并测试至少 5 个问题。
3. 实践题（微调）：构造 50 条领域指令数据，用 QLoRA 微调 Qwen2.5-1.5B，对比微调前后在领域问题上的表现。
4. 综合题：设计一个"企业智能客服"系统架构方案，说明哪些模块用 RAG、哪些用微调、为什么，并画出架构图（文字描述即可）。
5. 拓展题：调研当前主流的 RAG 框架（LangChain / LlamaIndex / Haystack），对比它们的核心特点和适用场景，写一份 800 字的技术选型报告。