别再对着论文干瞪眼了！这套Hands-On LLM实战项目，手把手带你从理论跨越到生产级应用
为什么这个项目值得关注
在过去的几年里，大语言模型（Large Language Models，简称LLM）已经从学术研究的象牙塔走进了千家万户的日常工作。从ChatGPT的横空出世到各种开源模型的百花齐放，我们见证了一场前所未有的技术革命。然而，对于大多数开发者而言，入门LLM领域仍然面临着诸多挑战：论文里的数学公式晦涩难懂，开源代码又散落在各个角落缺乏系统性整理，想要真正掌握这项技术却不知从何下手。
HandsOnLLM/Hands-On-Large-Language-Models正是为解决这些痛点而生的。这个项目由一群深耕AI领域多年的工程师和研究人员精心打造，它不仅仅是一个代码仓库，更是一套完整的LLM学习体系。无论你是刚刚踏入AI领域的新人，还是希望系统化提升技能的从业者，这个项目都能为你提供一条清晰的学习路径。
这个项目的核心价值体现在几个方面。首先，它采用理论结合实践的教学方式，每一个概念都会配合可直接运行的代码示例，让你不再陷入”懂了很多理论却写不出代码”的困境。其次，项目中的所有代码都经过实际验证，确保能够在你的机器上顺利运行，而不是那种复制粘贴过去就报错的”教学代码”。再者，内容更新速度快，紧跟技术发展趋势，你学到的都是最新的实践经验而非过时的知识。
更重要的是，这个项目覆盖了从模型基础原理到生产环境部署的完整链路。想象一下，当你能够理解Transformer架构的工作原理，能够使用Hugging Face加载和微调模型，能够构建自己的RAG（检索增强生成）系统，甚至还能够优化推理性能的时候，你就已经成为了真正意义上的LLM应用开发者，而不仅仅是API的搬运工。
环境搭建：让你的电脑准备好迎接LLM
在开始之前，我们需要先搭建好开发环境。这个过程可能会让一些初学者感到些许困惑，但别担心，我会把每一步都讲得清清楚楚。
基础环境要求
首先，让我们确认一下你的电脑是否满足基本要求。这个项目主要使用Python作为开发语言，因此你需要确保Python环境已经正确安装。建议使用Python 3.8或更高版本，因为较新的LLM相关库对旧版本Python的支持正在逐渐减少。
对于操作系统，这个项目支持Windows、macOS和Linux三大主流平台。如果你使用的是Windows系统，建议安装Windows Subsystem for Linux（WSL）来获得更好的开发体验，因为很多命令行工具在Linux环境下运行更加顺畅。
关于硬件配置，如果你只是想学习和实验，可以使用CPU进行推理，虽然速度会慢一些，但足以完成大部分学习任务。当然，如果有NVIDIA显卡那就更好了，推荐的显卡显存至少为8GB，这样可以流畅地运行一些中等规模的模型。
安装Anaconda或Miniconda
我们强烈建议使用conda来管理Python环境，这样可以避免不同项目之间的依赖冲突。打开你的终端或命令行工具，执行以下命令来安装Miniconda（一个精简版的Anaconda）：
# 下载Miniconda安装脚本
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
# 执行安装，按照提示完成操作
bash Miniconda3-latest-Linux-x86_64.sh
# 安装完成后，重新加载shell配置
source ~/.bashrc
安装完成后，你可以通过以下命令验证安装是否成功：
conda --version
如果看到类似”conda 24.1.2″这样的版本号输出，说明安装成功。
创建专用的Python环境
现在让我们为这个项目创建一个独立的Python环境。使用conda创建环境非常简单：
# 创建一个名为hands_on_llm的新环境，指定Python版本为3.10
conda create -n hands_on_llm python=3.10
# 激活这个环境
conda activate hands_on_llm
# 验证环境是否激活成功
python --version
你应该能看到Python 3.10的相关信息输出。
安装PyTorch
PyTorch是目前深度学习领域最受欢迎的框架之一，也是这个项目的主要依赖。安装PyTorch需要根据你的硬件情况选择合适的版本：
# 如果你有NVIDIA显卡并希望使用GPU加速（推荐）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
# 如果你使用macOS的Apple Silicon芯片
pip install torch torchvision torchaudio
# 如果你只有CPU或者不确定
pip install torch torchvision torchaudio
验证PyTorch安装是否成功，以及是否能识别你的GPU：
import torch
print(f"PyTorch版本: {torch.__version__}")
print(f"CUDA是否可用: {torch.cuda.is_available()}")
if torch.cuda.is_available():
print(f"GPU设备数量: {torch.cuda.device_count()}")
print(f"当前GPU名称: {torch.cuda.get_device_name(0)}")
安装transformers和其他依赖库
接下来安装项目所需的其他核心库：
# transformers是最重要的库，用于加载和使用各种预训练模型
pip install transformers
# 用于数据处理
pip install datasets accelerate bitsandbytes
# 用于构建Web应用
pip install gradio streamlit
# 用于处理PDF和文档
pip install pypdf langchain
# 用于向量存储和检索
pip install faiss-cpu chromadb
# 用于处理Hugging Face的模型和数据集
pip install huggingface_hub
# Jupyter notebook环境
pip install jupyter notebook ipykernel
安装完成后，我们可以运行一个简单的测试来确保一切正常：
# 创建一个测试脚本
python -c "
from transformers import AutoModel, AutoTokenizer
import torch
print('所有依赖库安装成功！')
print('你现在可以开始你的LLM学习之旅了。')
"
如果看到”All dependencies installed successfully!”的输出，恭喜你，环境搭建完成！
核心功能详解：从原理到实现
HandsOnLLM项目涵盖了LLM应用的方方面面，让我们逐一深入了解每个核心功能模块。
Transformer架构解析
理解Transformer是掌握LLM的基础。Transformer最初由Google的研究团队在2017年提出，最初用于机器翻译任务，随后成为了几乎所有现代大语言模型的基石。
让我们从代码层面来理解Transformer的核心组件：
"""
Transformer架构核心组件解析
这一部分代码展示了Transformer的主要组成部分：
1. 自注意力机制（Self-Attention）
2. 多头注意力（Multi-Head Attention）
3. 前馈神经网络（Feed Forward Network）
4. 层归一化（Layer Normalization）
"""
import torch
import torch.nn as nn
import math
class MultiHeadAttention(nn.Module):
"""
多头注意力机制是Transformer的核心创新点
与传统的循环神经网络不同，Transformer使用注意力机制
来捕获序列中任意两个位置之间的依赖关系，且可以并行计算。
工作原理：
1. 将输入分别投影到Query（查询）、Key（键）和Value（值）空间
2. 计算Query和Key的点积，得到注意力权重
3. 使用softmax将权重归一化
4. 用归一化的权重对Value进行加权求和
"""
def __init__(self, d_model, num_heads):
super().__init__()
assert d_model % num_heads == 0
self.d_model = d_model
self.num_heads = num_heads
self.d_k = d_model // num_heads
# 定义三个线性变换层用于生成Q、K、V
self.W_q = nn.Linear(d_model, d_model)
self.W_k = nn.Linear(d_model, d_model)
self.W_v = nn.Linear(d_model, d_model)
# 输出线性层
self.W_o = nn.Linear(d_model, d_model)
def split_heads(self, x, batch_size):
"""
将输入张量分割成多个注意力头
为什么要分割？
因为每个头可以学习关注不同类型的信息，
有的头可能关注语法结构，有的可能关注语义关系。
"""
x = x.view(batch_size, -1, self.num_heads, self.d_k)
return x.permute(0, 2, 1, 3) # (batch, heads, seq_len, d_k)
def forward(self, query, key, value, mask=None):
batch_size = query.size(0)
# 线性变换并分割成多个头
Q = self.split_heads(self.W_q(query), batch_size)
K = self.split_heads(self.W_k(key), batch_size)
V = self.split_heads(self.W_v(value), batch_size)
# 计算注意力分数
# 公式：Attention(Q,K,V) = softmax(QK^T / sqrt(d_k)) * V
scores = torch.matmul(Q, K.transpose(-1, -2))
scores = scores / math.sqrt(self.d_k)
# 应用掩码（用于处理填充或未来信息）
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
# softmax归一化
attention_weights = torch.softmax(scores, dim=-1)
# 加权求和
context = torch.matmul(attention_weights, V)
# 合并多个头的结果
context = context.permute(0, 2, 1, 3).contiguous()
context = context.view(batch_size, -1, self.d_model)
# 最终线性变换
output = self.W_o(context)
return output, attention_weights
print("=== Transformer核心组件演示 ===")
print("上面我们定义了多头注意力机制，这是现代LLM的基础组件。")
print("在实际使用中，我们通常不会从零实现这些组件，")
print("而是通过transformers库直接加载预训练模型。")
预训练模型加载与使用
现在让我们学习如何使用Hugging Face的transformers库来加载和使用预训练模型：
"""
预训练模型加载与使用指南
这一部分我们将学习：
1. 如何加载预训练的分词器（Tokenizer）
2. 如何加载预训练的语言模型
3. 如何进行文本生成
4. 如何处理批量输入
"""
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
# ============================================
# 第一步：加载分词器
# ============================================
# 分词器负责将文本转换为模型可以处理的数字ID
print("正在加载分词器...")
model_name = "gpt2" # 使用GPT-2作为示例（一个经典的预训练模型）
try:
tokenizer = AutoTokenizer.from_pretrained(model_name)
print(f"✓ 分词器加载成功！词汇表大小: {len(tokenizer)}")
except Exception as e:
print(f"加载分词器时出错: {e}")
# ============================================
# 第二步：加载预训练模型
# ============================================
print("\n正在加载预训练模型...")
try:
model = AutoModelForCausalLM.from_pretrained(model_name)
print("✓ 模型加载成功！")
# 打印模型的基本信息
total_params = sum(p.numel() for p in model.parameters())
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"总参数量: {total_params:,}")
print(f"可训练参数: {trainable_params:,}")
except Exception as e:
print(f"加载模型时出错: {e}")
# ============================================
# 第三步：文本分词
# ============================================
print("\n=== 文本分词演示 ===")
sample_text = "人工智能正在改变我们的生活方式"
print(f"原始文本: {sample_text}")
# 对文本进行分词
inputs = tokenizer(sample_text, return_tensors="pt")
print(f"分词后的输入IDs: {inputs['input_ids']}")
print(f"注意力掩码: {inputs['attention_mask']}")
# 将IDs转换回文本（验证分词是否正确）
decoded_tokens = tokenizer.decode(inputs['input_ids'][0])
print(f"解码后的文本: {decoded_tokens}")
# ============================================
# 第四步：文本生成
# ============================================
print("\n=== 文本生成演示 ===")
# 设置模型为评估模式
model.eval()
# 使用不同的生成方法
prompt = "未来的科技发展将会"
# 方法1：贪婪搜索（Greedy Search）
print("\n--- 方法1：贪婪搜索 ---")
inputs = tokenizer(prompt, return_tensors="pt")
with torch.no_grad():
outputs = model.generate(
inputs['input_ids'],
max_new_tokens=50,
do_sample=False,
pad_token_id=tokenizer.eos_token_id
)
greedy_result = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"输入: {prompt}")
print(f"生成结果: {greedy_result}")
# 方法2：温度采样（Temperature Sampling）
print("\n--- 方法2：温度采样 ---")
inputs = tokenizer(prompt, return_tensors="pt")
with torch.no_grad():
outputs = model.generate(
inputs['input_ids'],
max_new_tokens=50,
do_sample=True,
temperature=0.8,
top_k=50,
pad_token_id=tokenizer.eos_token_id
)
temp_result = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"输入: {prompt}")
print(f"生成结果: {temp_result}")
# 方法3：Top-P采样（Nucleus Sampling）
print("\n--- 方法3：Top-P采样 ---")
inputs = tokenizer(prompt, return_tensors="pt")
with torch.no_grad():
outputs = model.generate(
inputs['input_ids'],
max_new_tokens=50,
do_sample=True,
top_p=0.9,
top_k=0,
pad_token_id=tokenizer.eos_token_id
)
topp_result = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"输入: {prompt}")
print(f"生成结果: {topp_result}")
模型微调实战
预训练模型虽然强大，但在特定任务上往往需要进行微调才能发挥最佳效果。下面让我们学习如何对模型进行微调：
"""
模型微调实战指南
这一部分我们将学习：
1. 如何准备微调数据集
2. 如何进行模型微调训练
3. 如何保存和加载微调后的模型
"""
from transformers import (
AutoTokenizer,
AutoModelForSequenceClassification,
TrainingArguments,
Trainer,
DataCollatorWithPadding
)
from datasets import load_dataset
import torch
# ============================================
# 第一步：准备数据集
# ============================================
print("=== 准备微调数据集 ===")
# 加载一个示例数据集（IMDB电影评论情感分类）
# 这是一个二分类任务：正面/负面评论
dataset = load_dataset("imdb")
print(f"数据集信息:")
print(f"训练集大小: {len(dataset['train'])}")
print(f"测试集大小: {len(dataset['test'])}")
print(f"\n示例数据:")
print(dataset['train'][0])
# ============================================
# 第二步：数据预处理
# ============================================
print("\n=== 数据预处理 ===")
model_name = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)
def tokenize_function(examples):
"""
对文本进行分词处理
"""
return tokenizer(
examples["text"],
truncation=True,
max_length=512,
padding="max_length"
)
# 使用map函数批量处理数据集
print("正在对数据集进行分词...")
tokenized_datasets = dataset.map(tokenize_function, batched=True)
# 删除原始文本列，只保留需要的列
tokenized_datasets = tokenized_datasets.remove_columns(["text"])
# 将标签列重命名为labels（这是Trainer要求的格式）
tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
# 设置数据集格式为PyTorch张量
tokenized_datasets.set_format("torch")
print(f"处理后的数据集列: {tokenized_datasets['train'].column_names}")
# 为了加快演示速度，采样一小部分数据
small_train_dataset = tokenized_datasets["train"].shuffle(seed=42).select(range(1000))
small_test_dataset = tokenized_datasets["test"].shuffle(seed=42).select(range(500))
# ============================================
# 第三步：加载模型
# ============================================
print("\n=== 加载待微调的模型 ===")
model = AutoModelForSequenceClassification.from_pretrained(
model_name,
num_labels=2
)
print(f"模型加载成功！")
print(f"模型结构: {model.__class__.__name__}")
# ============================================
# 第四步：配置训练参数
# ============================================
print("\n=== 配置训练参数 ===")
training_args = TrainingArguments(
output_dir="./results", # 输出目录
num_train_epochs=3, # 训练轮数
per_device_train_batch_size=8, # 每个设备的训练批次大小
per_device_eval_batch_size=16, # 每个设备的评估批次大小
warmup_steps=500, # 学习率预热步数
weight_decay=0.01, # 权重衰减系数
logging_dir="./logs", # 日志目录
logging_steps=100, # 日志记录频率
evaluation_strategy="epoch", # 评估策略
save_strategy="epoch", # 保存策略
load_best_model_at_end=True, # 训练结束后加载最佳模型
metric_for_best_model="accuracy", # 用于选择最佳模型的指标
)
print("训练参数配置完成！")
# ============================================
# 第五步：定义评估指标
# ============================================
from datasets import load_metric
metric = load_metric("accuracy")
def compute_metrics(eval_pred):
"""
计算评估指标的函数
"""
logits, labels = eval_pred
predictions = torch.argmax(torch.tensor(logits), dim=-1)
return metric.compute(predictions=predictions, references=labels)
# ============================================
# 第六步：创建Trainer并开始训练
# ============================================
print("\n=== 开始模型微调训练 ===")
# 数据整理器，用于将不同长度的数据填充到相同长度
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=small_train_dataset,
eval_dataset=small_test_dataset,
compute_metrics=compute_metrics,
data_collator=data_collator,
)
# 开始训练
print("训练进行中，请稍候...")
train_result = trainer.train()
# ============================================
# 第七步：评估微调后的模型
# ============================================
print("\n=== 评估微调后的模型 ===")
metrics = trainer.evaluate()
print(f"测试集准确率: {metrics['eval_accuracy']:.4f}")
# ============================================
# 第八步：测试微调后的模型
# ============================================
print("\n=== 测试微调后的模型 ===")
model.eval()
test_texts = [
"This movie is fantastic! I really enjoyed it.",
"What a waste of time. The plot was terrible.",
]
for text in test_texts:
inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
with torch.no_grad():
outputs = model(**inputs)
predictions = torch.argmax(outputs.logits, dim=-1)
label = "正面评论" if predictions.item() == 1 else "负面评论"
print(f"文本: {text}")
print(f"预测: {label}\n")
# ============================================
# 第九步：保存微调后的模型
# ============================================
print("=== 保存微调后的模型 ===")
trainer.save_model("./fine_tuned_model")
tokenizer.save_pretrained("./fine_tuned_model")
print("✓ 模型和分词器已保存到 ./fine_tuned_model 目录")
实战教程：构建你的第一个LLM应用
现在让我们通过一个完整的实战项目来巩固所学知识。我们将构建一个基于RAG（检索增强生成）的问答系统，这是目前最流行的LLM应用架构之一。
什么是RAG？
RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与语言生成相结合的技术。它的核心思想是：当用户提出问题时，先从知识库中检索相关的文档片段，然后将这些片段作为上下文提供给LLM，让LLM基于这些真实信息来生成答案。
为什么需要RAG？因为LLM的知识是截止到训练日期的，而且可能会产生”幻觉”——即编造看似合理但实际上是错误的信息。RAG通过引入外部知识库来补充LLM的知识，既能让LLM了解最新的信息，又能提供可验证的答案来源。
项目结构
rag_qa_system/
├── data/ # 存放知识库文档
│ └── sample_docs/
├── src/ # 源代码目录
│ ├── __init__.py
│ ├── document_loader.py # 文档加载模块
│ ├── embedding.py # 向量化模块
│ ├── vector_store.py # 向量存储模块
│ ├── retriever.py # 检索模块
│ ├── generator.py # 生成模块
│ └── rag_chain.py # RAG链整合
├── config.py # 配置文件
├── main.py # 主程序入口
└── requirements.txt # 依赖列表
第一步：创建项目结构
# 创建项目目录
mkdir -p rag_qa_system/src
mkdir -p rag_qa_system/data/sample_docs
# 进入项目目录
cd rag_qa_system
第二步：配置文件
# config.py
"""
RAG问答系统配置文件
这个文件包含了系统所需的所有配置参数，
包括模型设置、路径配置、检索参数等。
"""
import os
from pathlib import Path
# ============================================
# 路径配置
# ============================================
BASE_DIR = Path(__file__).parent
DATA_DIR = BASE_DIR / "data"
DOCS_DIR = DATA_DIR / "sample_docs"
VECTOR_STORE_DIR = BASE_DIR / "vector_store"
# 确保必要的目录存在
VECTOR_STORE_DIR.mkdir(parents=True, exist_ok=True)
# ============================================
# 模型配置
# ============================================
# 嵌入模型配置（用于将文本转换为向量）
EMBEDDING_MODEL_CONFIG = {
"model_name": "sentence-transformers/all-MiniLM-L6-v2",
"model_kwargs": {"device": "cpu"},
"encode_kwargs": {"normalize_embeddings": True}
}
# LLM配置（用于生成答案）
LLM_CONFIG = {
"model_name": "gpt2", # 可以换成其他模型，如 "facebook/opt-1.3b"
"max_length": 512,
"temperature": 0.7,
"top_p": 0.9,
"do_sample": True
}
# ============================================
# 检索配置
# ============================================
RETRIEVER_CONFIG = {
"top_k": 3, # 返回最相关的3个文档片段
"score_threshold": 0.5 # 相似度分数阈值
}
# ============================================
# 文档处理配置
# ============================================
DOCUMENT_CONFIG = {
"chunk_size": 500, # 每个文本块的最大字符数
"chunk_overlap": 50, # 文本块之间的重叠字符数
"separators": ["\n\n", "\n", "。", "！", "？", " ", ""] # 分割符优先级
}
print("=== 配置文件加载成功 ===")
print(f"基础目录: {BASE_DIR}")
print(f"数据目录: {DOCS_DIR}")
print(f"向量存储目录: {VECTOR_STORE_DIR}")
第三步：文档加载模块
# src/document_loader.py
"""
文档加载模块
这个模块负责从各种来源加载文档，包括：
- 纯文本文件 (.txt)
- PDF文件 (.pdf)
- Markdown文件 (.md)
- CSV文件 (.csv)
"""
import os
from pathlib import Path
from typing import List, Dict, Any
from langchain_community.document_loaders import (
TextLoader,
PDFLoader,
UnstructuredMarkdownLoader,
CSVLoader
)
class DocumentLoader:
"""
统一文档加载器
这个类提供了统一的接口来加载不同格式的文档，
并将它们转换为LangChain的标准Document格式。
"""
def __init__(self):
self.loader_map = {
".txt": TextLoader,
".md": UnstructuredMarkdownLoader,
".pdf": PDFLoader,
".csv": CSVLoader
}
def load_file(self, file_path: str) -> List[Any]:
"""
加载单个文件
Args:
file_path: 文件路径
Returns:
加载的文档列表
"""
path = Path(file_path)
suffix = path.suffix.lower()
if suffix not in self.loader_map:
print(f"不支持的文件格式: {suffix}")
return []
try:
loader_class = self.loader_map[suffix]
loader = loader_class(file_path, encoding="utf-8")
documents = loader.load()
print(f"✓ 成功加载文件: {path.name} (包含 {len(documents)} 个文档)")
return documents
except Exception as e:
print(f"✗ 加载文件失败 {path.name}: {e}")
return []
def load_directory(self, directory_path: str) -> List[Any]:
"""
加载目录中的所有支持的文件
Args:
directory_path: 目录路径
Returns:
所有加载的文档列表
"""
all_documents = []
directory = Path(directory_path)
if not directory.exists():
print(f"目录不存在: {directory_path}")
return all_documents
for file_path in directory.rglob("*"):
if file_path.is_file() and file_path.suffix.lower() in self.loader_map:
docs = self.load_file(str(file_path))
all_documents.extend(docs)
print(f"\n总计加载了 {len(all_documents)} 个文档片段")
return all_documents
# 示例：创建示例文档用于测试
def create_sample_documents():
"""
创建示例文档用于演示
"""
docs_dir = Path("data/sample_docs")
docs_dir.mkdir(parents=True, exist_ok=True)
sample_docs = [
{
"filename": "人工智能简介.txt",
"content": """
人工智能（Artificial Intelligence，简称AI）是计算机科学的一个分支，
它企图了解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器。
人工智能的研究领域包括机器人、语言识别、图像识别、自然语言处理和专家系统等。
自从chatgpt发布以来，大语言模型成为了人工智能领域最热门的研究方向之一。
大语言模型（Large Language Models，简称LLM）是指参数量巨大的语言模型。
典型的LLM包括GPT系列、BERT、T5、PaLM、Claude等。这些模型通过在海量文本数据
上进行预训练，学习到了丰富的语言知识和世界知识。
"""
},
{
"filename": "机器学习基础.txt",
"content": """
机器学习是人工智能的一个分支，专门研究计算机怎样模拟或实现人类的学习行为，
以获取新的知识或技能，重新组织已有的知识结构使之不断改善自身的性能。
机器学习的基本分类：
1. 监督学习：从标注好的训练数据中学习，常用于分类和回归任务
2. 无监督学习：从无标注的数据中学习，常用于聚类和降维
3. 强化学习：通过与环境交互学习最优策略
深度学习是机器学习的一个子集，使用多层神经网络进行学习。
近年来，深度学习在图像识别、语音识别、自然语言处理等领域取得了突破性进展。
"""
},
{
"filename": "自然语言处理.txt",
"content": """
自然语言处理（Natural Language Processing，简称NLP）是计算机科学领域
与人工智能领域中的一个重要方向。它研究能实现人与计算机之间用自然语言
进行有效通信的各种理论和方法。
NLP的主要任务包括：
- 分词：将文本分割成词语
- 命名实体识别：识别文本中的人名、地名、机构名等
- 情感分析：判断文本的情感倾向
- 机器翻译：将一种语言翻译成另一种语言
- 问答系统：回答用户提出的问题
- 文本生成：根据输入生成相应的文本
近年来，基于Transformer的预训练语言模型（如BERT、GPT等）
大大推动了NLP技术的发展，在各项任务上都取得了state-of-the-art的效果。
"""
}
]
for doc in sample_docs:
file_path = docs_dir / doc["filename"]
with open(file_path, "w", encoding="utf-8") as f:
f.write(doc["content"].strip())
print(f"✓ 创建示例文档: {doc['filename']}")
if __name__ == "__main__":
print("=== 文档加载模块测试 ===\n")
# 创建示例文档
print("创建示例文档...")
create_sample_documents()
# 加载示例文档
print("\n加载示例文档...")
loader = DocumentLoader()
documents = loader.load_directory("data/sample_docs")
if documents:
print(f"\n第一个文档的内容预览:")
print(documents[0].page_content[:200])
第四步：文本分割模块
# src/text_splitter.py
"""
文本分割模块
这个模块负责将长文档分割成较小的文本块（chunks），
以便更好地进行向量化和检索。
"""
from typing import List
from langchain.text_splitter import RecursiveCharacterTextSplitter
class TextSplitter:
"""
文本分割器
使用递归字符分割方法，将文本按照指定的块大小和重叠进行分割。
这种方法会优先按照段落分割，然后是句子，最后是单词，
以保证语义完整性。
"""
def __init__(self, config: dict):
"""
初始化文本分割器
Args:
config: 配置字典，包含chunk_size和chunk_overlap等参数
"""
self.chunk_size = config.get("chunk_size", 500)
self.chunk_overlap = config.get("chunk_overlap", 50)
self.separators = config.get("separators", ["\n\n", "\n", "。", "！", "？", " ", ""])
self.splitter = RecursiveCharacterTextSplitter(
chunk_size=self.chunk_size,
chunk_overlap=self.chunk_overlap,
length_function=len,
separators=self.separators,
add_start_index=True
)
def split_documents(self, documents: List) -> List:
"""
分割文档列表
Args:
documents: LangChain的Document对象列表
Returns:
分割后的文档块列表
"""
print(f"正在分割 {len(documents)} 个文档...")
print(f"参数: chunk_size={self.chunk_size}, chunk_overlap={self.chunk_overlap}")
chunks = self.splitter.split_documents(documents)
print(f"✓ 分割完成，共产生 {len(chunks)} 个文本块\n")
return chunks
def split_text(self, text: str) -> List[str]:
"""
分割单个文本
Args:
text: 待分割的文本
Returns:
分割后的文本块列表
"""
return self.splitter.split_text(text)
# 测试代码
if __name__ == "__main__":
from src.document_loader import DocumentLoader
print("=== 文本分割模块测试 ===\n")
# 加载文档
loader = DocumentLoader()
documents = loader.load_directory("data/sample_docs")
if documents:
# 分割文档
splitter = TextSplitter({
"chunk_size": 200,
"chunk_overlap": 30
})
chunks = splitter.split_documents(documents)
# 显示分割结果
print("分割结果预览:")
for i, chunk in enumerate(chunks[:5]):
print(f"\n--- 文本块 {i+1} ---")
print(f"内容: {chunk.page_content[:150]}...")
print(f"来源: {chunk.metadata.get('source', 'Unknown')}")
第五步：向量化模块
# src/embedding.py
"""
向量化模块
这个模块负责将文本转换为向量表示（embeddings），
这是RAG系统中进行语义检索的基础。
"""
import torch
from typing import List, Dict, Any
from sentence_transformers import SentenceTransformer
class EmbeddingModel:
"""
嵌入模型封装类
使用Sentence Transformers库加载预训练的嵌入模型，
将文本转换为稠密的向量表示。
"""
def __init__(self, config: dict):
"""
初始化嵌入模型
Args:
config: 配置字典，包含model_name等参数
"""
self.model_name = config.get("model_name")
self.model_kwargs = config.get("model_kwargs", {"device": "cpu"})
self.encode_kwargs = config.get("encode_kwargs", {"normalize_embeddings": True})
print(f"正在加载嵌入模型: {self.model_name}...")
self.model = SentenceTransformer(
self.model_name,
**self.model_kwargs
)
print(f"✓ 嵌入模型加载成功！")
print(f" 嵌入维度: {self.model.get_sentence_embedding_dimension()}")
def embed_query(self, query: str) -> List[float]:
"""
对单个查询进行向量化
Args:
query: 查询文本
Returns:
查询的向量表示
"""
embedding = self.model.encode(
query,
**self.encode_kwargs
)
return embedding.tolist()
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""
对多个文档进行批量向量化
Args:
texts: 文档文本列表
Returns:
文档的向量表示列表
"""
embeddings = self.model.encode(
texts,
show_progress_bar=True,
**self.encode_kwargs
)
return embeddings.tolist()
def get_embedding_dimension(self) -> int:
"""
获取嵌入向量的维度
"""
return self.model.get_sentence_embedding_dimension()
# 测试代码
if __name__ == "__main__":
print("=== 向量化模块测试 ===\n")
# 创建示例文档
sample_texts = [
"人工智能是计算机科学的一个分支",
"机器学习是人工智能的核心技术",
"自然语言处理让计算机理解人类语言"
]
# 初始化嵌入模型
config = {
"model_name": "sentence-transformers/all-MiniLM-L6-v2",
"model_kwargs": {"device": "cpu"},
"encode_kwargs": {"normalize_embeddings": True}
}
embedder = EmbeddingModel(config)
# 测试单个查询
print("\n--- 单个查询向量化 ---")
query = "什么是人工智能？"
query_embedding = embedder.embed_query(query)
print(f"查询: {query}")
print(f"向量化维度: {len(query_embedding)}")
print(f"向量前5个元素: {query_embedding[:5]}")
# 测试批量文档向量化
print("\n--- 批量文档向量化 ---")
doc_embeddings = embedder.embed_documents(sample_texts)
print(f"文档数量: {len(doc_embeddings)}")
print(f"每个向量的维度: {len(doc_embeddings[0])}")
# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
print("\n--- 语义相似度计算 ---")
similarities = cosine_similarity([query_embedding], doc_embeddings)[0]
for i, (text, sim) in enumerate(zip(sample_texts, similarities)):
print(f"查询与文档{i+1}的相似度: {sim:.4f}")
第六步：向量存储与检索模块
# src/vector_store.py
"""
向量存储与检索模块
这个模块负责将文档向量存储到向量数据库中，
并提供高效的相似度检索功能。
"""
import os
from pathlib import Path
from typing import List, Dict, Any, Optional
import numpy as np
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
class VectorStore:
"""
向量存储管理器
使用Chroma作为向量数据库，支持：
- 添加文档向量
- 相似度检索
- 持久化存储
"""
def __init__(self, config: dict, embedding_model):
"""
初始化向量存储
Args:
config: 配置字典
embedding_model: 嵌入模型实例
"""
self.persist_directory = config.get("persist_directory", "./vector_store")
self.top_k = config.get("top_k", 3)
# 创建持久化目录
Path(self.persist_directory).mkdir(parents=True, exist_ok=True)
# 初始化嵌入函数
self.embedding_function = HuggingFaceEmbeddings(
model_name=config.get("embedding_model_name", "sentence-transformers/all-MiniLM-L6-v2")
)
self.embedding_model = embedding_model
self.vectorstore = None
self.docstore = [] # 用于存储原始文档
print(f"向量存储初始化完成")
print(f" 持久化目录: {self.persist_directory}")
def create_from_documents(self, documents: List, texts: List[str]):
"""
从文档列表创建向量存储
Args:
documents: LangChain的Document对象列表
texts: 文本内容列表
"""
print(f"正在创建向量存储 (共 {len(texts)} 个文档)...")
# 使用Chroma创建向量数据库
self.vectorstore = Chroma.from_texts(
texts=texts,
embedding=self.embedding_function,
persist_directory=self.persist_directory
)
# 保存原始文档
self.docstore = documents
print(f"✓ 向量存储创建成功！")
def similarity_search(self, query: str, top_k: Optional[int] = None) -> List[Dict]:
"""
相似度检索
Args:
query: 查询文本
top_k: 返回的最相似结果数量
Returns:
检索结果列表，每个结果包含文档内容和相似度分数
"""
if self.vectorstore is None:
print("警告: 向量存储尚未初始化！")
return []
k = top_k or self.top_k
# 执行相似度搜索
results = self.vectorstore.similarity_search_with_score(query, k=k)
# 整理检索结果
formatted_results = []
for i, (doc, score) in enumerate(results):
formatted_results.append({
"rank": i + 1,
"content": doc.page_content,
"score": score,
"metadata": doc.metadata
})
return formatted_results
def similarity_search_by_vector(self, query_vector: List[float], top_k: Optional[int] = None) -> List[Dict]:
"""
基于向量进行相似度检索
Args:
query_vector: 查询向量
top_k: 返回的最相似结果数量
Returns:
检索结果列表
"""
if self.vectorstore is None:
print("警告: 向量存储尚未初始化！")
return []
k = top_k or self.top_k
results = self.vectorstore.similarity_search_by_vector(
embedding=query_vector,
k=k
)
return results
def save(self):
"""
保存向量数据库到磁盘
"""
if self.vectorstore is not None:
self.vectorstore.persist()
print("✓ 向量数据库已保存")
def load(self):
"""
从磁盘加载向量数据库
"""
if os.path.exists(self.persist_directory):
self.vectorstore = Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embedding_function
)
print("✓ 向量数据库已加载")
else:
print("警告: 未找到已保存的向量数据库")
# 测试代码
if __name__ == "__main__":
from src.document_loader import DocumentLoader, create_sample_documents
from src.text_splitter import TextSplitter
from src.embedding import EmbeddingModel
print("=== 向量存储模块测试 ===\n")
# 准备数据
print("准备测试数据...")
create_sample_documents()
# 加载文档
loader = DocumentLoader()
documents = loader.load_directory("data/sample_docs")
# 分割文档
splitter = TextSplitter({"chunk_size": 200, "chunk_overlap": 30})
chunks = splitter.split_documents(documents)
# 提取文本内容
texts = [chunk.page_content for chunk in chunks]
# 创建嵌入模型
embed_config = {
"model_name": "sentence-transformers/all-MiniLM-L6-v2",
"model_kwargs": {"device": "cpu"},
"encode_kwargs": {"normalize_embeddings": True}
}
embedder = EmbeddingModel(embed_config)
# 创建向量存储
vector_config = {
"persist_directory": "./vector_store",
"top_k": 3,
"embedding_model_name": "sentence-transformers/all-MiniLM-L6-v2"
}
vector_store = VectorStore(vector_config, embedder)
vector_store.create_from_documents(chunks, texts)
vector_store.save()
# 测试检索
print("\n--- 检索测试 ---")
queries = [
"什么是人工智能？",
"机器学习有哪些类型？",
"自然语言处理能做什么？"
]
for query in queries:
print(f"\n查询: {query}")
results = vector_store.similarity_search(query, top_k=2)
for result in results:
print(f" [Rank {result['rank']}] 相似度: {result['score']:.4f}")
print(f" 内容: {result['content'][:100]}...")
第七步：生成模块
# src/generator.py
"""
生成模块
这个模块负责使用LLM根据检索到的上下文生成答案。
"""
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
from typing import List, Dict, Optional
class AnswerGenerator:
"""
答案生成器
使用预训练的LLM模型，根据检索到的上下文生成答案。
支持多种生成策略，如贪婪搜索、温度采样、Top-K等。
"""
def __init__(self, config: dict):
"""
初始化生成器
Args:
config: 配置字典，包含模型名称和生成参数
"""
self.model_name = config.get("model_name", "gpt2")
self.max_length = config.get("max_length", 512)
self.temperature = config.get("temperature", 0.7)
self.top_p = config.get("top_p", 0.9)
self.do_sample = config.get("do_sample", True)
self.top_k = config.get("top_k", 50)
print(f"正在加载语言模型: {self.model_name}...")
# 加载分词器和模型
self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
self.model = AutoModelForCausalLM.from_pretrained(self.model_name)
# 设置pad token
if self.tokenizer.pad_token is None:
self.tokenizer.pad_token = self.tokenizer.eos_token
# 创建生成管道
self.generator = pipeline(
"text-generation",
model=self.model,
tokenizer=self.tokenizer,
device=-1 # -1表示CPU，0表示第一个GPU
)
print(f"✓ 语言模型加载成功！")
print(f" 模型参数量: {sum(p.numel() for p in self.model.parameters()):,}")
def generate_with_context(
self,
query: str,
context_documents: List[Dict],
system_prompt: Optional[str] = None
) -> Dict:
"""
基于上下文生成答案
Args:
query: 用户问题
context_documents: 检索到的相关文档列表
system_prompt: 系统提示词
Returns:
包含答案和元数据的字典
"""
# 构建上下文
context_text = "\n\n".join([
f"[文档 {i+1}]\n{doc['content']}"
for i, doc in enumerate(context_documents)
])
# 构建完整的提示词
if system_prompt is None:
system_prompt = """你是一个知识渊博的助手，负责根据提供的上下文信息回答用户的问题。
请仔细阅读上下文内容，如果上下文中包含与问题相关的信息，请基于这些信息给出准确的回答。
如果上下文中没有相关信息，请明确告知用户你无法从提供的上下文中找到答案。"""
prompt = f"""<|system|>
{system_prompt}
<|context|>
{context_text}
</context|>
<|user|>
{query}
</user|>
<|assistant|>"""
# 生成答案
with torch.no_grad():
outputs = self.generator(
prompt,
max_new_tokens=200,
temperature=self.temperature,
top_p=self.top_p,
top_k=self.top_k,
do_sample=self.do_sample,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id,
return_full_text=False
)
generated_text = outputs[0]["generated_text"].strip()
return {
"answer": generated_text,
"context_used": context_documents,
"prompt": prompt
}
def generate_without_context(self, query: str) -> str:
"""
不使用上下文的普通生成（用于对比）
"""
prompt = f"请回答以下问题：{query}"
outputs = self.generator(
prompt,
max_new_tokens=200,
temperature=self.temperature,
top_p=self.top_p,
do_sample=self.do_sample,
pad_token_id=self.tokenizer.pad_token_id,
eos_token_id=self.tokenizer.eos_token_id
)
return outputs[0]["generated_text"].strip()
# 测试代码
if __name__ == "__main__":
print("=== 生成模块测试 ===\n")
# 配置
config = {
"model_name": "gpt2",
"max_length": 512,
"temperature": 0.7,
"top_p": 0.9,
"do_sample": True,
"top_k": 50
}
# 初始化生成器
generator = AnswerGenerator(config)
# 测试生成
print("\n--- 生成测试 ---")
# 示例上下文
sample_context = [
{
"content": "人工智能（AI）是计算机科学的一个分支，致力于开发能够执行通常需要人类智能的任务的系统，如视觉感知、语音识别、决策制定和语言翻译。"
},
{
"content": "机器学习是人工智能的一个子领域，专注于开发能够从数据中学习和改进的算法。"
}
]
query = "什么是人工智能？"
print(f"查询: {query}")
result = generator.generate_with_context(query, sample_context)
print(f"\n生成的答案:\n{result['answer']}")
第八步：RAG链整合
# src/rag_chain.py
"""
RAG链整合模块
这个模块将检索和生成模块整合在一起，
形成一个完整的RAG（检索增强生成）系统。
"""
from typing import List, Dict, Optional, Callable
from src.vector_store import VectorStore
from src.generator import AnswerGenerator
class RAGChain:
"""
RAG链
整合检索和生成两个环节，形成完整的问答流程：
1. 接收用户问题
2. 从向量数据库中检索相关文档
3. 将相关文档作为上下文提供给LLM
4. LLM生成最终答案
"""
def __init__(
self,
vector_store: VectorStore,
generator: AnswerGenerator,
config: Optional[Dict] = None
):
"""
初始化RAG链
Args:
vector_store: 向量存储实例
generator: 答案生成器实例
config: 配置字典
"""
self.vector_store = vector_store
self.generator = generator
self.config = config or {}
self.top_k = self.config.get("top_k", 3)
self.use_rag = self.config.get("use_rag", True)
print("RAG链初始化完成！")
print(f" 检索文档数: {self.top_k}")
print(f" 使用RAG: {self.use_rag}")
def ask(self, question: str, verbose: bool = False) -> Dict:
"""
提问
Args:
question: 用户问题
verbose: 是否输出详细信息
Returns:
包含答案和相关信息的字典
"""
if verbose:
print(f"\n{'='*50}")
print(f"问题: {question}")
print('='*50)
if self.use_rag:
# 步骤1：检索相关文档
if verbose:
print("\n[步骤1] 检索相关文档...")
relevant_docs = self.vector_store.similarity_search(
question,
top_k=self.top_k
)
if verbose:
print(f"找到 {len(relevant_docs)} 个相关文档:")
for i, doc in enumerate(relevant_docs):
print(f" 文档{i+1} [相似度: {doc['score']:.4f}]: {doc['content'][:80]}...")
# 步骤2：基于检索结果生成答案
if verbose:
print("\n[步骤2] 生成答案...")
result = self.generator.generate_with_context(
query=question,
context_documents=relevant_docs
)
else:
# 不使用RAG，直接生成
if verbose:
print("\n不使用RAG，直接生成答案...")
answer = self.generator.generate_without_context(question)
result = {
"answer": answer,
"context_used": [],
"prompt": None
}
if verbose:
print(f"\n[结果]")
print(f"答案: {result['answer']}")
return result
def batch_ask(self, questions: List[str], verbose: bool = False) -> List[Dict]:
"""
批量提问
Args:
questions: 问题列表
verbose: 是否输出详细信息
Returns:
结果列表
"""
results = []
for i, question in enumerate(questions):
print(f"\n处理问题 {i+1}/{len(questions)}")
result = self.ask(question, verbose=verbose)
results.append(result)
return results
# 便捷函数：创建完整的RAG系统
def create_rag_system(
vector_store_config: Dict,
generator_config: Dict,
documents: List,
texts: List[str],
rag_config: Optional[Dict] = None
) -> RAGChain:
"""
创建完整的RAG系统
Args:
vector_store_config: 向量存储配置
generator_config: 生成器配置
documents: 文档对象列表
texts: 文本内容列表
rag_config: RAG链配置
Returns:
RAGChain实例
"""
from src.embedding import EmbeddingModel
# 步骤1：创建嵌入模型
print("步骤1: 初始化嵌入模型...")
embedder = EmbeddingModel(vector_store_config.get("embedding_config", {}))
# 步骤2：创建向量存储
print("步骤2: 创建向量存储...")
vector_store = VectorStore(vector_store_config, embedder)
vector_store.create_from_documents(documents, texts)
vector_store.save()
# 步骤3：创建生成器
print("步骤3: 初始化生成器...")
generator = AnswerGenerator(generator_config)
# 步骤4：创建RAG链
print("步骤4: 创建RAG链...")
rag_chain = RAGChain(vector_store, generator, rag_config)
return rag_chain
# 测试代码
if __name__ == "__main__":
from src.document_loader import DocumentLoader, create_sample_documents
from src.text_splitter import TextSplitter
from src.embedding import EmbeddingModel
print("=== RAG系统完整测试 ===\n")
# 准备数据
print("准备测试数据...")
create_sample_documents()
# 加载文档
loader = DocumentLoader()
documents = loader.load_directory("data/sample_docs")
# 分割文档
splitter = TextSplitter({"chunk_size": 200, "chunk_overlap": 30})
chunks = splitter.split_documents(documents)
texts = [chunk.page_content for chunk in chunks]
# 配置
embed_config = {
"model_name": "sentence-transformers/all-MiniLM-L6-v2",
"model_kwargs": {"device": "cpu"},
"encode_kwargs": {"normalize_embeddings": True}
}
vector_config = {
"persist_directory": "./vector_store",
"top_k": 3,
"embedding_config": embed_config
}
generator_config = {
"model_name": "gpt2",
"max_length": 512,
"temperature": 0.7,
"top_p": 0.9,
"do_sample": True
}
# 创建RAG系统
rag_chain = create_rag_system(
vector_store_config=vector_config,
generator_config=generator_config,
documents=chunks,
texts=texts,
rag_config={"top_k": 2, "use_rag": True}
)
# 测试问答
print("\n" + "="*60)
print("问答测试")
print("="*60)
test_questions = [
"什么是人工智能？",
"机器学习有哪几种类型？",
"自然语言处理能做什么？"
]
for question in test_questions:
print("\n" + "-"*60)
result = rag_chain.ask(question, verbose=True)
第九步：主程序入口
# main.py
"""
RAG问答系统主程序
使用方法:
python main.py # 交互模式
python main.py --query "问题" # 单次查询
python main.py --web # 启动Web界面
"""
import argparse
import sys
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
from src.document_loader import DocumentLoader, create_sample_documents
from src.text_splitter import TextSplitter
from src.embedding import EmbeddingModel
from src.vector_store import VectorStore
from src.generator import AnswerGenerator
from src.rag_chain import RAGChain
def setup_system():
"""
设置RAG系统
"""
print("="*60)
print("初始化RAG问答系统")
print("="*60)
# 创建示例文档（如果不存在）
if not Path("data/sample_docs").exists():
create_sample_documents()
# 加载文档
loader = DocumentLoader()
documents = loader.load_directory("data/sample_docs")
if not documents:
print("错误: 未能加载任何文档！")
sys.exit(1)
# 分割文档
splitter = TextSplitter({
"chunk_size": 300,
"chunk_overlap": 50
})
chunks = splitter.split_documents(documents)
texts = [chunk.page_content for chunk in chunks]
# 配置
embed_config = {
"model_name": "sentence-transformers/all-MiniLM-L6-v2",
"model_kwargs": {"device": "cpu"},
"encode_kwargs": {"normalize_embeddings": True}
}
vector_config = {
"persist_directory": "./vector_store",
"top_k": 3,
"embedding_model_name": "sentence-transformers/all-MiniLM-L6-v2"
}
generator_config = {
"model_name": "gpt2",
"max_length": 512,
"temperature": 0.7,
"top_p": 0.9,
"do_sample": True
}
# 初始化嵌入模型
print("\n初始化嵌入模型...")
embedder = EmbeddingModel(embed_config)
# 初始化向量存储
print("初始化向量存储...")
vector_store = VectorStore(vector_config, embedder)
vector_store.create_from_documents(chunks, texts)
vector_store.save()
# 初始化生成器
print("初始化生成器...")
generator = AnswerGenerator(generator_config)
# 创建RAG链
print("创建RAG链...")
rag_chain = RAGChain(vector_store, generator, {"top_k": 3})
print("\n系统初始化完成！")
return rag_chain
def interactive_mode(rag_chain):
"""
交互模式
"""
print("\n" + "="*60)
print("RAG问答系统 - 交互模式")
print("="*60)
print("输入您的问题，系统将基于知识库给出回答。")
print("输入 'quit' 或 'exit' 退出程序。")
print("输入 'clear' 清空对话历史。")
print("-"*60)
history = []
while True:
try:
query = input("\n您: ").strip()
if query.lower() in ["quit", "exit", "q"]:
print("感谢使用，再见！")
break
if query.lower() == "clear":
history = []
print("对话历史已清空。")
continue
if not query:
continue
history.append({"role": "user", "content": query})
result = rag_chain.ask(query, verbose=False)
print(f"\n助手: {result['answer']}")
history.append({"role": "assistant", "content": result['answer']})
except KeyboardInterrupt:
print("\n\n程序被用户中断。")
break
except Exception as e:
print(f"\n发生错误: {e}")
def single_query_mode(rag_chain, query):
"""
单次查询模式
"""
print(f"\n问题: {query}")
print("-"*60)
result = rag_chain.ask(query, verbose=True)
print("\n" + "="*60)
print("最终答案:")
print("="*60)
print(result["answer"])
def main():
"""
主函数
"""
parser = argparse.ArgumentParser(description="RAG问答系统")
parser.add_argument("--query", type=str, help="单次查询模式")
parser.add_argument("--interactive", action="store_true", help="交互模式")
parser.add_argument("--web", action="store_true", help="启动Web界面")
args = parser.parse_args()
# 初始化系统
rag_chain = setup_system()
if args.query:
single_query_mode(rag_chain, args.query)
elif args.web:
print("\n抱歉，Web界面功能正在开发中...")
print("请使用交互模式或单次查询模式。")
else:
interactive_mode(rag_chain)
if __name__ == "__main__":
main()
运行你的RAG系统
现在让我们运行这个完整的RAG系统：
# 首先确保所有依赖都已安装
pip install -q transformers torch sentence-transformers langchain chromadb faiss-cpu
# 运行交互模式
python main.py --interactive
你应该能看到类似这样的输出：
============================================================
初始化RAG问答系统
============================================================
正在加载文档...
示例文档已创建: 人工智能简介.txt
示例文档已创建: 机器学习基础.txt
示例文档已创建: 自然语言处理.txt
✓ 成功加载文件: 人工智能简介.txt (包含 1 个文档)
✓ 成功加载文件: 机器学习基础.txt (包含 1 个文档)
✓ 成功加载文件: 自然语言处理.txt (包含 1 个文档)
总计加载了 3 个文档片段
正在分割文档...
参数: chunk_size=300, chunk_overlap=50
✓ 分割完成，共产生 15 个文本块
正在加载嵌入模型: sentence-transformers/all-MiniLM-L6-v2...
✓ 嵌入模型加载成功！
嵌入维度: 384
正在创建向量存储 (共 15 个文档)...
✓ 向量存储创建成功！
正在加载语言模型: gpt2...
✓ 语言模型加载成功！
模型参数量: 124,439,808
RAG链初始化完成！
检索文档数: 3
使用RAG: True
系统初始化完成！
============================================================
RAG问答系统 - 交互模式
============================================================
输入您的问题，系统将基于知识库给出回答。
输入 'quit' 或 'exit' 退出程序。
输入 'clear' 清空对话历史。
------------------------------------------------------------
您: 什么是人工智能？
然后系统会检索相关文档并生成答案。
常见应用场景
通过HandsOnLLM项目的学习，你可以掌握多种实际应用场景的开发和部署。
场景一：智能客服系统
智能客服是最常见的LLM应用场景之一。传统的客服系统只能回答预设的问题，而基于LLM的智能客服可以理解用户的自然语言表达，并给出灵活的回复。
"""
智能客服系统示例
这个示例展示了如何构建一个基础的多轮对话客服系统。
"""
class SmartCustomerService:
"""
智能客服系统
支持功能：
- 多轮对话上下文管理
- 基于知识的问答
- 情感识别和礼貌回复
- 转人工判断
"""
def __init__(self, rag_chain, config=None):
self.rag_chain = rag_chain
self.config = config or {}
# 对话历史
self.conversation_history = []
# 最大历史长度
self.max_history_length = self.config.get("max_history_length", 10)
# 转人工关键词
self.human_trigger_keywords = self.config.get(
"human_trigger_keywords",
["人工", "客服", "投诉", "升级", "转接"]
)
def should_escalate_to_human(self, query: str) -> bool:
"""
判断是否需要转人工
检查用户的问题是否包含触发关键词
"""
query_lower = query.lower()
for keyword in self.human_trigger_keywords:
if keyword in query_lower:
return True
return False
def build_context_from_history(self) -> str:
"""
从对话历史构建上下文
"""
context = ""
recent_history = self.conversation_history[-self.max_history_length:]
for turn in recent_history:
context += f"用户: {turn['user']}\n"
context += f"助手: {turn['assistant']}\n"
return context
def chat(self, user_input: str) -> dict:
"""
处理用户输入并返回回复
Args:
user_input: 用户的输入
Returns:
包含回复和元数据的字典
"""
# 检查是否需要转人工
if self.should_escalate_to_human(user_input):
return {
"response": "您的问题比较特殊，我为您转接人工客服，请稍候...",
"escalate": True,
"source": "keyword_match"
}
# 构建带历史的完整查询
history_context = self.build_context_from_history()
full_query = f"{history_context}\n用户新问题: {user_input}"
# 使用RAG系统生成回复
result = self.rag_chain.ask(full_query, verbose=False)
# 更新对话历史
self.conversation_history.append({
"user": user_input,
"assistant": result["answer"]
})
return {
"response": result["answer"],
"escalate": False,
"source": "rag_system"
}
def reset_conversation(self):
"""
重置对话历史
"""
self.conversation_history = []
# 使用示例
print("=== 智能客服系统示例 ===\n")
# 注意：实际使用时需要先初始化rag_chain
# smart_service = SmartCustomerService(rag_chain)
# 示例对话
sample_conversations = [
"我想咨询一下产品退货政策",
"这个产品的价格是多少？",
"我要投诉，你们产品质量太差了"
]
for user_input in sample_conversations:
print(f"用户: {user_input}")
# response = smart_service.chat(user_input)
# print(f"助手: {response['response']}")
# print(f"是否转人工: {response['escalate']}")
print("-" * 40)
场景二：文档分析与摘要
LLM可以帮助我们快速分析和理解长文档，自动提取关键信息并生成摘要。
"""
文档分析与摘要系统
这个模块提供了多种文档分析功能：
- 文档摘要
- 关键信息提取
- 问答式分析
- 结构化信息提取
"""
from typing import List, Dict, Any, Optional
import re
class DocumentAnalyzer:
"""
文档分析器
使用LLM对文档进行深度分析和信息提取
"""
def __init__(self, generator, max_chunk_length: int = 2000):
"""
初始化分析器
Args:
generator: 答案生成器实例
max_chunk_length: 单次处理的最大文本长度
"""
self.generator = generator
self.max_chunk_length = max_chunk_length
def summarize(self, text: str, num_sentences: int = 3) -> str:
"""
生成文档摘要
Args:
text: 待摘要的文档
num_sentences: 摘要句子数
Returns:
生成的摘要
"""
prompt = f"""请为以下文档生成一个简洁的摘要，概括文章的主要内容和核心观点。
文档内容：
{text[:self.max_chunk_length]}
请生成一个{num_sentences}句话的摘要："""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=200,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return outputs[0]["generated_text"].strip()
def extract_key_points(self, text: str) -> List[str]:
"""
提取关键点
Args:
text: 文档文本
Returns:
关键点列表
"""
prompt = f"""请从以下文档中提取5-8个关键点，以列表形式呈现。
文档内容：
{text[:self.max_chunk_length]}
关键点："""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=300,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
result = outputs[0]["generated_text"].strip()
# 解析关键点
points = []
for line in result.split("\n"):
line = line.strip()
if line and (line[0].isdigit() or line.startswith("-") or line.startswith("•")):
# 去除列表标记
cleaned = re.sub(r"^\d+[\.\)]\s*", "", line)
cleaned = re.sub(r"^[-•]\s*", "", cleaned)
if cleaned:
points.append(cleaned)
return points
def answer_questions(self, text: str, questions: List[str]) -> Dict[str, str]:
"""
对文档内容进行问答
Args:
text: 文档文本
questions: 问题列表
Returns:
问题到答案的字典
"""
results = {}
for question in questions:
prompt = f"""基于以下文档内容回答问题。如果文档中没有相关信息，请说明"文档中未提供此信息"。
文档内容：
{text[:self.max_chunk_length]}
问题：{question}
答案："""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=200,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
results[question] = outputs[0]["generated_text"].strip()
return results
def analyze_sentiment_and_themes(self, text: str) -> Dict[str, Any]:
"""
分析文档的情感和主题
Args:
text: 文档文本
Returns:
分析结果字典
"""
prompt = f"""请分析以下文档的情感倾向和主题。
文档内容：
{text[:self.max_chunk_length]}
请提供：
1. 情感倾向（正面/负面/中性）
2. 主要主题（2-3个）
3. 目标受众
4. 文档类型（新闻/评论/技术文档等）
分析结果："""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=300,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return {
"full_analysis": outputs[0]["generated_text"].strip()
}
# 使用示例
print("=== 文档分析系统示例 ===\n")
sample_document = """
人工智能技术的快速发展正在深刻改变我们的社会和生活方式。近年来，
以深度学习为代表的人工智能技术取得了突破性进展，在计算机视觉、自然语言处理、
语音识别等领域都达到了甚至超过了人类的水平。
大语言模型（LLM）是人工智能领域的重要突破。GPT、BERT、T5等预训练语言模型
通过在海量文本数据上进行预训练，学习到了丰富的语言知识和世界知识。这些模型
可以被进一步微调用于各种下游任务，如问答、文本生成、机器翻译等。
检索增强生成（RAG）技术是当前最受欢迎的LLM应用架构之一。RAG通过结合
信息检索和语言生成，可以让LLM基于外部知识库来回答问题，有效减少了
LLM的"幻觉"问题，并提供了答案的可溯源性。
"""
analyzer = DocumentAnalyzer(None) # 实际使用时传入generator
print("原始文档:")
print(sample_document[:200] + "...")
print("\n" + "-"*50)
print("摘要:")
# print(analyzer.summarize(sample_document))
print("\n" + "-"*50)
print("关键点提取:")
# points = analyzer.extract_key_points(sample_document)
# for i, point in enumerate(points, 1):
# print(f" {i}. {point}")
print("\n" + "-"*50)
print("问答分析:")
questions = [
"这篇文章主要讨论什么？",
"什么是大语言模型？",
"RAG技术有什么优势？"
]
# results = analyzer.answer_questions(sample_document, questions)
# for q, a in results.items():
# print(f"Q: {q}")
# print(f"A: {a}\n")
场景三：代码助手
LLM在编程辅助方面也展现出了惊人的能力，可以帮助开发者写代码、调试、解释代码等。
"""
代码助手示例
展示如何使用LLM辅助编程工作：
- 代码生成
- 代码解释
- 代码调试
- 代码优化建议
"""
class CodeAssistant:
"""
代码助手类
提供多种编程辅助功能
"""
def __init__(self, generator):
self.generator = generator
def generate_code(self, description: str, language: str = "python") -> str:
"""
根据描述生成代码
Args:
description: 代码功能描述
language: 编程语言
Returns:
生成的代码
"""
prompt = f"""请为以下功能描述生成{language}代码。要求代码完整、可运行，并添加适当的注释。
功能描述：
{description}
生成的{language}代码：
``` {language}"""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=500,
temperature=0.3, # 较低的temperature以获得更确定的输出
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return outputs[0]["generated_text"].strip()
def explain_code(self, code: str) -> str:
"""
解释代码功能
Args:
code: 待解释的代码
Returns:
代码解释
"""
prompt = f"""请详细解释以下代码的功能、工作原理和关键点。
代码：
```{code}```
代码解释：
"""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=300,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return outputs[0]["generated_text"].strip()
def debug_code(self, code: str, error_message: str = None) -> str:
"""
调试代码
Args:
code: 有问题的代码
error_message: 错误信息（如果有）
Returns:
调试建议
"""
if error_message:
prompt = f"""请分析以下代码的错误原因并提供修复建议。
代码：
```{code}```
错误信息：
{error_message}
调试分析：
"""
else:
prompt = f"""请审查以下代码，找出可能的问题和bug，并提供修复建议。
代码：
```{code}```
代码审查：
"""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=300,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return outputs[0]["generated_text"].strip()
def optimize_code(self, code: str) -> str:
"""
提供代码优化建议
Args:
code: 待优化的代码
Returns:
优化建议和优化后的代码
"""
prompt = f"""请分析以下代码的性能瓶颈，并提供优化建议和优化后的代码。
代码：
```{code}```
优化分析：
"""
with torch.no_grad():
outputs = self.generator.generator(
prompt,
max_new_tokens=400,
temperature=0.5,
do_sample=True,
pad_token_id=self.generator.tokenizer.pad_token_id
)
return outputs[0]["generated_text"].strip()
**使用示例**
print("=== 代码助手示例 ===\n")
**assistant = CodeAssistant(generator)**
**示例1：生成代码**
print("--- 代码生成示例 ---")
description = "写一个函数计算两个数的最大公约数，使用欧几里得算法"
**code = assistant.generate_code(description)**
**print(f"需求: {description}")**
**print(f"生成的代码:\n{code}\n")**
**示例2：解释代码**
print("--- 代码解释示例 ---")
sample_code = """
def quicksort(arr):
if len(arr) <= 1:
return arr
pivot = arr[len(arr) // 2]
left = [x for x in arr if x < pivot]
middle = [x for x in arr if x == pivot]
right = [x for x in arr if x > pivot]
return quicksort(left) + middle + quicksort(right)
"""
**explanation = assistant.explain_code(sample_code)**
**print(f"代码:\n{sample_code}")**
**print(f"\n解释:\n{explanation}\n")**
**示例3：调试代码**
print("--- 代码调试示例 ---")
buggy_code = """
def find_max(nums):
max_num = float('-inf')
for num in nums:
if num > max_num:
max_num = num
return max_num
print(find_max([1, 2, 3, 4, 5]))
"""
**debug_result = assistant.debug_code(buggy_code, "IndexError: list index out of range")**
**print(f"问题代码:\n{buggy_code}")**
**print(f"\n调试结果:\n{debug_result}\n")**
技巧与最佳实践
在长期使用LLM和构建相关应用的过程中，我积累了一些实用的技巧和最佳实践，现在分享给大家。
提示工程技巧
提示工程（Prompt Engineering）是充分发挥LLM能力的关键。一个好的提示词可以显著提升模型的表现。
"""
提示工程最佳实践
这一部分总结了在实际使用中验证有效的提示工程技巧。
"""
class PromptEngineering:
"""
提示工程工具类
提供各种提示词模板和优化技巧
"""
**============================================**
**基础提示模板**
**============================================**
@staticmethod
def basic_template(task: str, context: str, question: str) -> str:
"""
基础提示模板
结构：任务说明 + 上下文 + 问题
"""
return f"""任务：{task}
上下文：
{context}
问题：{question}
回答："""
@staticmethod
def structured_output_template(
task: str,
context: str,
output_format: str
) -> str:
"""
结构化输出模板
用于生成特定格式的输出，如JSON、列表等
"""
return f"""任务：{task}
上下文：
{context}
请严格按照以下格式输出：
{output_format}
输出："""
@staticmethod
def chain_of_thought_template(question: str) -> str:
"""
思维链（Chain of Thought）提示模板
鼓励模型展示推理过程，适用于复杂问题
"""
return f"""请逐步思考并回答问题。在给出最终答案之前，请详细解释你的推理过程。
问题：{question}
让我们一步步来思考：
1."""
@staticmethod
def few_shot_template(task: str, examples: list, query: str) -> str:
"""
少样本（Few-Shot）学习模板
通过提供几个示例来引导模型理解任务
"""
example_str = "\n\n".join([
f"示例 {i+1}:\n输入: {ex['input']}\n输出: {ex['output']}"
for i, ex in enumerate(examples)
])
return f"""任务：{task}
示例：
{example_str}
请按照以上示例的格式处理新的输入：
输入: {query}
输出："""
@staticmethod
def system_prompt_builder(
role: str,
constraints: list,
style: str = "professional"
) -> str:
"""
系统提示构建器
帮助你构建有效的系统提示词
"""
constraints_str = "\n".join([f"- {c}" for c in constraints])
style_guide = {
"professional": "使用正式、专业的语言",
"friendly": "使用友好、轻松的语言，适当使用表情符号",
"technical": "使用技术性语言，提供详细的解释",
"concise": "简洁明了，直接回答问题"
}
return f"""你是一个{role}。
约束条件：
{constraints_str}
回答风格：{style_guide.get(style, style)}
请始终遵循以上设定来回答问题。"""
**提示词优化技巧示例**
print("=== 提示工程技巧演示 ===\n")
**技巧1：明确指定输出格式**
print("--- 技巧1：明确指定输出格式 ---")
task = "从文本中提取人名"
context = "昨天，李明和王小红一起去北京出差，他们见到了客户张总。"
output_format = "以JSON格式输出，例如：{\"names\": [\"姓名1\", \"姓名2\"]}"
prompt = PromptEngineering.structured_output_template(task, context, output_format)
print(f"提示词:\n{prompt}\n")
**技巧2：使用思维链**
print("--- 技巧2：使用思维链 ---")
question = "如果所有的猫都喜欢鱼，小明有一只猫，那么这只猫喜欢鱼吗？"
prompt = PromptEngineering.chain_of_thought_template(question)
print(f"问题: {question}")
print(f"提示词:\n{prompt}\n")
**技巧3：少样本学习**
print("--- 技巧3：少样本学习 ---")
task = "将中文翻译成英文"
examples = [
{"input": "你好", "output": "Hello"},
{"input": "谢谢", "output": "Thank you"},
]
query = "再见"
prompt = PromptEngineering.few_shot_template(task, examples, query)
print(f"提示词:\n{prompt}\n")
**技巧4：构建有效的系统提示**
print("--- 技巧4：构建系统提示 ---")
role = "资深Python程序员"
constraints = [
"代码必须遵循PEP 8规范",
"必须包含适当的文档字符串",
"函数必须有类型注解",
"错误处理必须完善"
]
system_prompt = PromptEngineering.system_prompt_builder(
role, constraints, "technical"
)
print(f"系统提示:\n{system_prompt}\n")
模型选择指南
选择合适的模型对于应用的性能和成本都至关重要。
"""
模型选择指南
这一部分帮助你根据不同场景选择合适的模型。
"""
class ModelSelectionGuide:
"""
模型选择指南
根据不同需求提供模型选择建议
"""
**常用模型对比表**
MODEL_COMPARISON = {
**============================================**
**开源模型**
**============================================**
"gpt2": {
"provider": "OpenAI",
"params": "124M",
"strengths": ["体积小，适合快速实验", "开源免费"],
"weaknesses": ["能力有限", "不适合复杂任务"],
"use_cases": ["学习入门", "快速原型", "简单文本生成"],
"hardware": "CPU可运行"
},
"gpt2-medium": {
"provider": "OpenAI",
"params": "355M",
"strengths": ["比GPT-2基础版更好", "仍可本地运行"],
"weaknesses": ["能力仍然有限"],
"use_cases": ["学习入门", "简单对话"],
"hardware": "CPU/GPU 4GB+"
},
"facebook/opt-1.3b": {
"provider": "Meta",
"params": "1.3B",
"strengths": ["开源", "相对轻量", "英文能力强"],
"weaknesses": ["中文能力一般"],
"use_cases": ["英文对话", "文本生成", "代码辅助"],
"hardware": "GPU 6GB+"
},
"facebook/opt-iml-1.3b": {
"provider": "Meta",
"params": "1.3B",
"strengths": ["指令遵循能力较好", "安全性更高"],
"weaknesses": ["中文能力一般"],
"use_cases": ["指令任务", "对话助手"],
"hardware": "GPU 6GB+"
},
"bigscience/bloom-560m": {
"provider": "BigScience",
"params": "560M",
"strengths": ["多语言支持（46种语言）", "开源"],
"weaknesses": ["性能一般"],
"use_cases": ["多语言任务", "翻译"],
"hardware": "CPU/GPU 4GB+"
},
"bigscience/bloom-1b7": {
"provider": "BigScience",
"params": "1.7B",
"strengths": ["多语言支持好", "开源"],
"weaknesses": ["需要更多计算资源"],
"use_cases": ["多语言对话", "翻译", "摘要"],
"hardware": "GPU 8GB+"
},
"tiiuae/falcon-7b": {
"provider": "TII UAE",
"params": "7B",
"strengths": ["开源最强7B模型之一", "效果好"],
"weaknesses": ["资源需求较高"],
"use_cases": ["复杂对话", "代码生成", "长文本处理"],
"hardware": "GPU 16GB+ (int4量化后8GB)"
},
"meta-llama/Llama-2-7b": {
"provider": "Meta",
"params": "7B",
"strengths": ["效果好", "开源可商用", "活跃社区"],
"weaknesses": ["需要申请", "资源需求高"],
"use_cases": ["生产级对话", "复杂推理", "代码生成"],
"hardware": "GPU 16GB+ (int4量化后8GB)"
},
**============================================**
**API模型（云端）**
**============================================**
"gpt-3.5-turbo": {
"provider": "OpenAI",
"params": "未公开",
"strengths": ["效果好", "成本低", "API稳定"],
"weaknesses": ["需要付费", "依赖网络"],
"use_cases": ["生产环境", "复杂任务", "聊天机器人"],
"hardware": "仅需API调用"
},
"gpt-4": {
"provider": "OpenAI",
"params": "未公开",
"strengths": ["效果最佳", "复杂推理能力强"],
"weaknesses": ["成本高", "速率限制"],
"use_cases": ["高精度任务", "代码生成", "复杂分析"],
"hardware": "仅需API调用"
},
"claude-2": {
"provider": "Anthropic",
"params": "未公开",
"strengths": ["长上下文", "安全性高", "效果好"],
"weaknesses": ["需要付费"],
"use_cases": ["长文档处理", "分析任务", "安全敏感场景"],
"hardware": "仅需API调用"
}
}
@classmethod
def recommend_model(cls, requirements: dict) -> list:
"""
根据需求推荐模型
Args:
requirements: 需求字典，包含以下键：
- language: 主要语言（如"中文"、"英文"、"多语言"）
- complexity: 任务复杂度（"低"、"中"、"高"）
- budget: 预算（"免费"、"低"、"中"、"高"）
- deployment: 部署方式（"本地"、"云端API"）
Returns:
推荐模型列表
"""
language = requirements.get("language", "英文")
complexity = requirements.get("complexity", "中")
budget = requirements.get("budget", "低")
deployment = requirements.get("deployment", "本地")
recommendations = []
for model_name, model_info in cls.MODEL_COMPARISON.items():
score = 0
**检查语言匹配**
if language in ["多语言", "中英双语"]:
if "多语言" in str(model_info.get("strengths", [])):
score += 2
elif language == "中文":
**中文支持较好的模型**
if "中文" in str(model_info.get("strengths", [])):
score += 2
elif model_name in ["gpt-3.5-turbo", "gpt-4", "claude-2"]:
score += 2
**检查复杂度匹配**
if complexity == "低":
if model_info["params"] in ["124M", "355M", "560M"]:
score += 2
elif complexity == "中":
if "B" in model_info["params"]:
score += 2
elif complexity == "高":
if "7B" in model_info["params"] or "未公开" in str(model_info["params"]):
score += 2
**检查预算匹配**
if budget == "免费":
if model_info.get("provider") in ["OpenAI", "Anthropic"]:
score -= 3
else:
score += 2
elif budget in ["中", "高"]:
if "API" in str(model_info.get("use_cases", [])):
score += 1
**检查部署方式**
if deployment == "本地":
if model_info.get("hardware") != "仅需API调用":
score += 2
else:
if model_info.get("hardware") == "仅需API调用":
score += 2
if score >= 4:
recommendations.append((model_name, model_info, score))
**按分数排序**
recommendations.sort(key=lambda x: x[2], reverse=True)
return recommendations
@classmethod
def print_recommendations(cls, requirements: dict):
"""
打印模型推荐
"""
print("="*60)
print("模型选择推荐")
print("="*60)
recs = cls.recommend_model(requirements)
if not recs:
print("没有找到符合条件的模型")
return
print(f"\n找到 {len(recs)} 个推荐模型：\n")
for model_name, info, score in recs[:5]:
print(f"{'-'*50}")
print(f"模型: {model_name}")
print(f"提供商: {info['provider']}")
print(f"参数量: {info['params']}")
print(f"优势: {', '.join(info['strengths'])}")
print(f"适用场景: {', '.join(info['use_cases'])}")
print(f"硬件要求: {info['hardware']}")
print(f"匹配度: {score}分")
print(f"{'-'*50}")
**使用示例**
print("=== 模型选择指南 ===\n")
**示例1：中文对话，预算有限**
print("场景1：中文对话，预算有限，希望本地部署")
ModelSelectionGuide.print_recommendations({
"language": "中文",
"complexity": "中",
"budget": "免费",
"deployment": "本地"
})
**示例2：英文代码生成，追求效果**
print("\n场景2：英文代码生成，追求最好效果，预算充足")
ModelSelectionGuide.print_recommendations({
"language": "英文",
"complexity": "高",
"budget": "高",
"deployment": "云端"
})
**示例3：多语言翻译**
print("\n场景3：多语言翻译任务")
ModelSelectionGuide.print_recommendations({
"language": "多语言",
"complexity": "中",
"budget": "免费",
"deployment": "本地"
})
性能优化技巧
在实际应用中，性能优化是非常重要的。以下是一些实用的优化技巧：
"""
性能优化技巧
这一部分介绍如何优化LLM应用的性能。
"""
import torch
from typing import Optional
class PerformanceOptimizer:
"""
性能优化工具类
提供多种性能优化技术
"""
**============================================**
**量化技术**
**============================================**
@staticmethod
def load_quantized_model(model_name: str, bits: int = 4):
"""
加载量化模型
量化可以显著减少模型的显存占用和推理时间，
同时只会有轻微的精度损失。
Args:
model_name: 模型名称
bits: 量化位数（4或8）
Returns:
量化后的模型和分词器
"""
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
**配置量化**
quantization_config = BitsAndBytesConfig(
load_in_4bit=bits == 4,
load_in_8bit=bits == 8,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_quant_type="nf4",
bnb_4bit_use_double_quant=True
)
print(f"正在加载{bits}位量化模型: {model_name}")
**加载量化模型**
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config,
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
print(f"✓ 量化模型加载成功！")
return model, tokenizer
**============================================**
**缓存优化**
**============================================**
@staticmethod
def enable_kv_cache(model):
"""
启用KV缓存
KV缓存可以避免在生成过程中重复计算已生成token的注意力
"""
model.config.use_cache = True
print("✓ KV缓存已启用")
@staticmethod
def disable_kv_cache(model):
"""
禁用KV缓存（用于训练或需要动态注意力的场景）
"""
model.config.use_cache = False
print("✓ KV缓存已禁用")
**============================================**
**批处理优化**
**============================================**
@staticmethod
def batch_inference(
texts: list,
tokenizer,
model,
batch_size: int = 8,
max_length: int = 512
) -> list:
"""
批量推理
相比逐个处理，批处理可以更好地利用GPU并行能力
Args:
texts: 文本列表
tokenizer: 分词器
model: 模型
batch_size: 批次大小
max_length: 最大长度
Returns:
生成结果列表
"""
results = []
**分批处理**
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
**批量编码**
inputs = tokenizer(
batch_texts,
return_tensors="pt",
padding=True,
truncation=True,
max_length=max_length
)
**批量生成**
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=100,
do_sample=False
)
**解码**
batch_results = tokenizer.batch_decode(outputs, skip_special_tokens=True)
results.extend(batch_results)
print(f"已处理 {min(i+batch_size, len(texts))}/{len(texts)} 条")
return results
**============================================**
**模型加速库**
**============================================**
@staticmethod
def optimize_with_accelerate(model, device: str = "cuda"):
"""
使用Accelerate库优化模型
Accelerate是Hugging Face提供的加速库，
可以自动处理分布式训练和混合精度等优化
"""
from accelerate import Accelerator
accelerator = Accelerator()
model = accelerator.prepare(model)
print(f"✓ 模型已使用Accelerate优化")
print(f" 设备: {accelerator.device}")
print(f" 分布式: {accelerator.num_processes}进程")
return model, accelerator
**性能对比示例**
print("=== 性能优化示例 ===\n")
print("""
| 优化技术 | 显存减少 | 速度提升 | 精度损失 |
|-------------------|-----------|----------|----------|
| FP16混合精度 | ~50% | ~1.5x | 可忽略 |
| INT8量化 | ~75% | ~2x | ~1-2% |
| INT4量化 | ~87% | ~3-4x | ~5-10% |
| KV缓存 | - | ~2x+ | 无 |
| 批处理 | - | ~N倍 | 无 |
| Flash Attention | ~20% | ~2x | 无 |
""")
print("\n--- 量化模型加载示例 ---")
print("""
**加载4位量化模型（以LLAMA-2-7B为例）**
**原始模型大小：约14GB**
**4位量化后：约4GB**
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_quant_type="nf4"
)
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b",
quantization_config=quantization_config,
device_map="auto"
)
""")
print("\n--- 批处理示例 ---")
print("""
**批量处理多个文本**
texts = [
"第一个问题",
"第二个问题",
"第三个问题",
**... 更多文本**
]
**传统方式：逐个处理**
results = []
for text in texts:
result = generate(text) # 每次都要单独调用
results.append(result)
**批处理：一次性处理多个**
inputs = tokenizer(texts, return_tensors="pt", padding=True)
outputs = model.generate(**inputs, max_new_tokens=50)
results = tokenizer.batch_decode(outputs, skip_special_tokens=True)
**批处理的优势：**
**1. GPU利用率更高**
**2. 总时间更短**
**3. 适合离线批量任务**
""")
常见问题与解决方案
在使用HandsOnLLM项目的过程中，你可能会遇到一些常见问题。以下是一些问题及其解决方案。
问题一：模型下载失败
问题描述：在下载模型时遇到网络错误或超时。
解决方案：
**方案1：设置镜像源**
import os
**使用Hugging Face镜像**
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
**使用国内镜像**
os.environ["HF_HUB_OFFLINE"] = "0"
os.environ["HF_HUB_DOWNLOAD_TIMEOUT"] = "300"
**方案2：手动下载后加载**
from huggingface_hub import snapshot_download
**下载到本地目录**
local_dir = "./models/gpt2"
snapshot_download(
repo_id="gpt2",
local_dir=local_dir,
local_dir_use_symlinks=False
)
**从本地加载**
from transformers import AutoModel, AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(local_dir)
model = AutoModel.from_pretrained(local_dir)
**方案3：使用离线模式**
**先在有网络的环境下载好模型**
**然后在目标机器上使用**
import offline_mode
offline_mode.enable() # 确保transformers使用本地缓存
问题二：显存不足
问题描述：运行模型时提示CUDA out of memory。
解决方案：
**方案1：减小批次大小**
batch_size = 1 # 从8减小到1
**方案2：使用更小的模型**
small_model_name = "gpt2" # 从gpt2-large切换到gpt2
**方案3：使用量化模型**
from transformers import AutoModelForCausalLM, BitsAndBytesConfig
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
load_in_8bit=True
)
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config
)
**方案4：启用梯度检查点（用于训练场景）**
model.gradient_checkpointing_enable()
**方案5：清理缓存**
import torch
torch.cuda.empty_cache()
**方案6：使用CPU**
device = "cpu" # 而不是"cuda"
model = model.to(device)
**方案7：卸载不需要的变量**
del large_variable
import gc
gc.collect()
问题三：生成结果质量不佳
问题描述：模型生成的文本质量不高，可能是重复、不连贯或不相关。
解决方案：
**方案1：调整生成参数**
outputs = model.generate(
input_ids,
max_new_tokens=100,
temperature=0.7, # 调整随机性（0.1-1.0）
top_p=0.9, # 核采样
top_k=50, # Top-K采样
repetition_penalty=1.2, # 重复惩罚
no_repeat_ngram_size=2, # 禁止重复的n-gram
do_sample=True # 使用采样而非贪婪
)
**方案2：使用更好的提示词**
prompt = """你是一个专业的AI助手。请根据以下要求生成回答：
1. 语言流畅通顺
2. 内容准确相关
3. 避免重复表达
问题：{user_question}
请给出详细且有条理的回答："""
**方案3：使用思维链提示**
prompt = """请按以下步骤思考：
1. 首先理解问题
2. 分析相关因素
3. 给出推理过程
4. 得出结论
问题：{question}
推理过程："""
**方案4：使用更好的模型**
better_model = "gpt-3.5-turbo" # 从gpt2升级
**方案5：使用RAG提供上下文**
relevant_docs = retriever.search(query, top_k=3)
context = "\n".join([doc.content for doc in relevant_docs])
prompt = f"""基于以下上下文回答问题。如果上下文中没有相关信息，请明确说明。
上下文：
{context}
问题：{query}
回答："""
问题四：处理长文本时超出上下文限制
问题描述：输入文本太长，超过了模型的最大上下文长度。
解决方案：
**方案1：截断文本**
max_length = 2048 # 模型的上下文长度
inputs = tokenizer(text, max_length=max_length, truncation=True)
**方案2：分块处理**
def process_long_text(text, chunk_size=1000, overlap=100):
"""
将长文本分割成重叠的块
"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap # 移动窗口，保持重叠
return chunks
**方案3：摘要后处理**
def summarize_and_process(long_text, summarize_prompt, task_prompt):
"""
先摘要，再处理
"""
**第一步：生成摘要**
summary = summarize(long_text, max_length=500)
**第二步：基于摘要处理任务**
result = process_task(summary, task_prompt)
return result
**方案4：使用支持长上下文的模型**
long_context_model = "gpt-4-32k" # 32k上下文
**或**
long_context_model = "claude-2" # 100k上下文
**方案5：RAG分段检索**
**将文档分成小块存储，检索时返回相关的多个块**
chunks = split_document(doc, chunk_size=500)
for chunk in chunks:
embedding = embed(chunk)
store(embedding, chunk)
**检索时获取多个相关块**
relevant_chunks = retrieve(query, top_k=5)
combined_context = " ".join([c.content for c in relevant_chunks])
进阶学习资源
如果你想继续深入学习LLM相关技术，以下是一些优质的学习资源：
在线课程
斯坦福大学的CS224N《自然语言处理与深度学习》是NLP领域的经典课程，涵盖了从词向量到Transformer再到大型语言模型的完整知识体系。课程视频和课件都可以在网上免费获取。
DeepLearning.AI与OpenAI合作推出的《ChatGPT Prompt Engineering for Developers》课程专门讲解如何有效地使用大型语言模型，特别是针对开发者场景的提示工程技巧。
Fast.ai的《Practical Deep Learning for Coders》虽然不是专门讲LLM的课程，但其面向程序员的教学方法和实用导向非常值得学习。
经典论文
《Attention Is All You Need》是Transformer架构的开山之作，所有想深入理解LLM的人都应该阅读这篇论文。《BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding》介绍了一种革命性的预训练方法。《Language Models are Few-Shot Learners》（GPT-3论文）展示了大规模语言模型的惊人能力。《Training Language Models to Follow Instructions with Human Feedback》（InstructGPT论文）阐述了如何让模型更好地遵循人类指令。
开源项目
Hugging Face的Transformers库是LLM领域最重要的开源库之一，提供了数千个预训练模型和完整的工具链。LangChain是一个用于构建LLM应用的框架，提供了链式调用、代理、RAG等功能。LlamaHub收集了各种LLM应用的提示模板和工作流示例。AutoGPT是一个实验性的项目，探索使用LLM自主完成复杂任务的可能性。
社区和博客
Hugging Face博客定期发布LLM领域的最新研究和应用案例。OpenAI的官方博客发布关于GPT系列模型的详细说明和技术文章。The Batch是Andrew Ng主编的AI Newsletter，每周总结AI领域的重要进展。Reddit的r/MachineLearning和r/LanguageTechnology是活跃的讨论社区。
结语
通过这篇教程，我们一起学习了HandsOnLLM/Hands-On-Large-Language-Models项目的核心内容。从环境搭建到基础概念，从模型使用到实际应用，从代码示例到最佳实践，希望这些内容能够帮助你建立起对LLM技术的系统理解。
记住，学习LLM不需要成为数学天才或深度学习专家。这个领域最迷人的地方在于，它让AI变得更加平民化和可触及。你不需要训练自己的GPT模型，也能够利用现有的开源模型构建出有价值的应用。
现在，是时候把这些知识付诸实践了。我建议你：
首先，从简单的项目开始。可以尝试加载一个预训练模型，进行文本生成或情感分析等基本任务。这能帮助你建立对LLM的直观感受。
其次，动手实现一个完整的应用。可以是问答系统、智能客服、文档分析工具，甚至是代码助手。选择一个你感兴趣的场景，从头到尾实现一遍。
然后，不断优化和改进你的应用。学习使用提示工程、模型量化、RAG等高级技术，提升应用的性能和用户体验。
最后，分享你的成果。将你的项目开源，在社区分享你的经验，与其他学习者和开发者交流。
LLM是一个快速发展的领域，新的模型、新的工具、新的应用场景不断涌现。保持好奇心，持续学习，你一定能在这个激动人心的领域找到属于自己的一片天地。
祝你学习愉快，在LLM的世界里探索出无限可能！
相关链接
HandsOnLLM官方仓库：https://github.com/HandsOnLLM/Hands-On-Large-Language-Models
Hugging Face Transformers：https://github.com/huggingface/transformers
LangChain框架：https://github.com/langchain-ai/langchain
PyTorch深度学习框架：https://github.com/pytorch/pytorch
Sentence Transformers：https://github.com/UKPLab/sentence-transformers
“`
评论区