实战案例库
精选来自提示词工程、LoRA微调和RAG开发三大领域的完整实战案例， 包含详细代码示例、最佳实践和效果对比。
提示词工程
基础提示技巧
掌握基本的提示词编写原则，包括清晰指令、具体上下文和期望输出格式。
实战示例：
### 实战案例：客服对话优化
**问题场景**：AI客服回复过于机械，用户体验差。
**优化前提示**：
```
回答用户问题
```
**优化后提示**：
```
你是一位专业的客服代表，请用友好、专业的语气回答用户问题。
- 保持回答简洁明了（不超过100字）
- 使用中文口语化表达
- 如果无法回答，请礼貌说明并建议联系人工客服
- 当前时间：{{current_time}}
```
**效果对比**：
- 用户满意度提升45%
- 平均对话轮次减少30%
- 转人工率降低25%
**问题场景**：AI客服回复过于机械，用户体验差。
**优化前提示**：
```
回答用户问题
```
**优化后提示**：
```
你是一位专业的客服代表，请用友好、专业的语气回答用户问题。
- 保持回答简洁明了（不超过100字）
- 使用中文口语化表达
- 如果无法回答，请礼貌说明并建议联系人工客服
- 当前时间：{{current_time}}
```
**效果对比**：
- 用户满意度提升45%
- 平均对话轮次减少30%
- 转人工率降低25%
提示词工程
思维链 (Chain-of-Thought)
通过引导模型展示推理过程来提高复杂问题的解决能力。
实战示例：
### 实战案例：数学问题求解
**问题**：小明有5个苹果，给了小红2个，又买了3个，现在有多少个？
**普通提示**：
```
小明有5个苹果，给了小红2个，又买了3个，现在有多少个？
```
**思维链提示**：
```
让我们一步步思考：
1. 小明开始有5个苹果
2. 给了小红2个，剩下 5 - 2 = 3个
3. 又买了3个，现在有 3 + 3 = 6个
4. 所以答案是6个苹果
```
**效果**：准确率从68%提升到92%
**问题**：小明有5个苹果，给了小红2个，又买了3个，现在有多少个？
**普通提示**：
```
小明有5个苹果，给了小红2个，又买了3个，现在有多少个？
```
**思维链提示**：
```
让我们一步步思考：
1. 小明开始有5个苹果
2. 给了小红2个，剩下 5 - 2 = 3个
3. 又买了3个，现在有 3 + 3 = 6个
4. 所以答案是6个苹果
```
**效果**：准确率从68%提升到92%
提示词工程
Few-Shot学习
通过提供少量示例来引导模型学习特定任务模式。
实战示例：
### 实战案例：情感分析
**提示模板**：
```
将以下文本分类为正面或负面情感：
示例1：
文本："这个产品太棒了，完全超出预期！"
情感：正面
示例2：
文本："质量很差，根本不值得购买。"
情感：负面
新文本："{{input_text}}"
情感：
```
**应用场景**：客户评论分析、社交媒体监控
**提示模板**：
```
将以下文本分类为正面或负面情感：
示例1：
文本："这个产品太棒了，完全超出预期！"
情感：正面
示例2：
文本："质量很差，根本不值得购买。"
情感：负面
新文本："{{input_text}}"
情感：
```
**应用场景**：客户评论分析、社交媒体监控
提示词工程
自一致性 (Self-Consistency)
通过多次采样和投票机制提高推理准确性。
实战示例：
### 实战案例：复杂推理问题
**实现方法**：
1. 使用不同的随机种子生成多个推理路径
2. 对每个路径得到最终答案
3. 选择出现频率最高的答案作为最终结果
**代码示例**：
```python
def self_consistency(prompt, n_samples=5):
answers = []
for i in range(n_samples):
response = llm.generate(prompt, temperature=0.7, seed=i)
answer = extract_final_answer(response)
answers.append(answer)
return most_frequent(answers)
```
**效果**：在复杂QA任务上提升15-25%准确率
**实现方法**：
1. 使用不同的随机种子生成多个推理路径
2. 对每个路径得到最终答案
3. 选择出现频率最高的答案作为最终结果
**代码示例**：
```python
def self_consistency(prompt, n_samples=5):
answers = []
for i in range(n_samples):
response = llm.generate(prompt, temperature=0.7, seed=i)
answer = extract_final_answer(response)
answers.append(answer)
return most_frequent(answers)
```
**效果**：在复杂QA任务上提升15-25%准确率
提示词工程
多模态提示工程
设计能够处理图像、文本等多模态输入的提示词。
实战示例：
### 实战案例：图像描述优化
**基础提示**：
```
描述这张图片
```
**优化提示**：
```
请详细描述这张图片，包括：
1. 主要对象及其位置
2. 颜色和材质特征
3. 场景和环境背景
4. 可能的情感或氛围
5. 任何值得注意的细节
使用简洁、准确的语言，避免主观推测。
```
**应用**：无障碍辅助、内容审核、电商产品描述
**基础提示**：
```
描述这张图片
```
**优化提示**：
```
请详细描述这张图片，包括：
1. 主要对象及其位置
2. 颜色和材质特征
3. 场景和环境背景
4. 可能的情感或氛围
5. 任何值得注意的细节
使用简洁、准确的语言，避免主观推测。
```
**应用**：无障碍辅助、内容审核、电商产品描述
LoRA微调
LoRA基础原理与实现
理解LoRA（Low-Rank Adaptation）的核心思想：通过低秩矩阵分解来减少微调参数量，保持预训练模型权重不变的同时实现高效适配。
实战示例：
### Hugging Face Transformers中的LoRA实现
使用PEFT库在Hugging Face Transformers中快速应用LoRA：
```python
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM
# 加载预训练模型
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
# 配置LoRA参数
lora_config = LoraConfig(
r=8, # 秩大小
lora_alpha=16, # 缩放因子
target_modules=["q_proj", "v_proj"], # 目标模块
lora_dropout=0.1, # dropout率
bias="none",
task_type="CAUSAL_LM"
)
# 应用LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters() # 查看可训练参数
```
关键配置说明：
- `r=8`：秩大小，控制参数量
- `lora_alpha=16`：缩放因子，影响学习率
- `target_modules`：指定要应用LoRA的模块
使用PEFT库在Hugging Face Transformers中快速应用LoRA：
```python
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM
# 加载预训练模型
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b-hf")
# 配置LoRA参数
lora_config = LoraConfig(
r=8, # 秩大小
lora_alpha=16, # 缩放因子
target_modules=["q_proj", "v_proj"], # 目标模块
lora_dropout=0.1, # dropout率
bias="none",
task_type="CAUSAL_LM"
)
# 应用LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters() # 查看可训练参数
```
关键配置说明：
- `r=8`：秩大小，控制参数量
- `lora_alpha=16`：缩放因子，影响学习率
- `target_modules`：指定要应用LoRA的模块
LoRA微调
LoRA训练策略优化
掌握LoRA微调的最佳实践，包括数据准备、超参数调优、训练监控等关键技术。
实战示例：
### 使用Trainer API进行LoRA训练
```python
from transformers import TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model
# 配置训练参数
training_args = TrainingArguments(
output_dir="./lora-finetuned",
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
learning_rate=2e-4,
num_train_epochs=3,
logging_steps=10,
save_strategy="epoch",
fp16=True, # 启用混合精度训练
)
# 创建Trainer
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
data_collator=data_collator,
)
# 开始训练
trainer.train()
```
性能优化技巧：
- 梯度累积：在小批量情况下模拟大批量效果
- 混合精度：FP16训练可减少50%显存占用
- 学习率调度：余弦退火或线性预热效果更好
```python
from transformers import TrainingArguments, Trainer
from peft import LoraConfig, get_peft_model
# 配置训练参数
training_args = TrainingArguments(
output_dir="./lora-finetuned",
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
learning_rate=2e-4,
num_train_epochs=3,
logging_steps=10,
save_strategy="epoch",
fp16=True, # 启用混合精度训练
)
# 创建Trainer
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
data_collator=data_collator,
)
# 开始训练
trainer.train()
```
性能优化技巧：
- 梯度累积：在小批量情况下模拟大批量效果
- 混合精度：FP16训练可减少50%显存占用
- 学习率调度：余弦退火或线性预热效果更好
LoRA微调
LoRA与模型量化结合
学习如何将LoRA微调与4-bit/8-bit量化技术结合，在消费级GPU上微调大模型。
实战示例：
### 4-bit量化 + LoRA 微调
```python
from transformers import BitsAndBytesConfig
from peft import LoraConfig, prepare_model_for_kbit_training
# 配置4-bit量化
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.float16,
)
# 加载量化模型
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b-hf",
quantization_config=bnb_config,
device_map="auto",
)
# 准备模型进行k-bit训练
model = prepare_model_for_kbit_training(model)
# 应用LoRA
lora_config = LoraConfig(
r=64,
lora_alpha=16,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.1,
bias="none",
task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)
```
显存占用对比：
- 原始Llama-2-7B：约14GB
- 4-bit量化：约6GB
- 4-bit + LoRA：约6GB + 可训练参数
```python
from transformers import BitsAndBytesConfig
from peft import LoraConfig, prepare_model_for_kbit_training
# 配置4-bit量化
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.float16,
)
# 加载量化模型
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b-hf",
quantization_config=bnb_config,
device_map="auto",
)
# 准备模型进行k-bit训练
model = prepare_model_for_kbit_training(model)
# 应用LoRA
lora_config = LoraConfig(
r=64,
lora_alpha=16,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.1,
bias="none",
task_type="CAUSAL_LM"
)
model = get_peft_model(model, lora_config)
```
显存占用对比：
- 原始Llama-2-7B：约14GB
- 4-bit量化：约6GB
- 4-bit + LoRA：约6GB + 可训练参数
LoRA微调
LoRA模型性能评估
掌握LoRA微调模型的全面评估方法，包括基准测试、指标分析和部署优化。
实战示例：
### 使用Hugging Face Evaluate进行评估
```python
import evaluate
from evaluate import load
# 加载评估指标
accuracy = load("accuracy")
rouge = load("rouge")
# 评估函数
def compute_metrics(eval_pred):
predictions, labels = eval_pred
predictions = np.argmax(predictions, axis=1)
return accuracy.compute(predictions=predictions, references=labels)
# 在Trainer中使用
training_args = TrainingArguments(
evaluation_strategy="epoch",
...
)
trainer = Trainer(
model=model,
args=training_args,
compute_metrics=compute_metrics,
)
```
多维度评估框架：
1. 任务性能：准确率、F1分数等
2. 推理速度：延迟、吞吐量测试
3. 资源消耗：显存、CPU使用率
4. 泛化能力：跨领域测试
```python
import evaluate
from evaluate import load
# 加载评估指标
accuracy = load("accuracy")
rouge = load("rouge")
# 评估函数
def compute_metrics(eval_pred):
predictions, labels = eval_pred
predictions = np.argmax(predictions, axis=1)
return accuracy.compute(predictions=predictions, references=labels)
# 在Trainer中使用
training_args = TrainingArguments(
evaluation_strategy="epoch",
...
)
trainer = Trainer(
model=model,
args=training_args,
compute_metrics=compute_metrics,
)
```
多维度评估框架：
1. 任务性能：准确率、F1分数等
2. 推理速度：延迟、吞吐量测试
3. 资源消耗：显存、CPU使用率
4. 泛化能力：跨领域测试
LoRA微调
多模态LoRA微调
探索LoRA在多模态模型（如LLaVA、Flamingo）中的应用，实现视觉-语言联合微调。
实战示例：
### LLaVA模型的LoRA微调
```python
# LLaVA架构包含视觉编码器和语言模型
# 对语言模型部分应用LoRA
lora_config = LoraConfig(
r=32,
lora_alpha=64,
target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
lora_dropout=0.1,
bias="none",
task_type="CAUSAL_LM"
)
# 只对语言模型部分应用LoRA
language_model = model.language_model
language_model = get_peft_model(language_model, lora_config)
# 冻结视觉编码器，只训练LoRA参数
for param in model.vision_tower.parameters():
param.requires_grad = False
```
训练策略：
- 视觉编码器：通常保持冻结
- 语言模型：应用LoRA进行微调
- 投影层：可选择性微调或冻结
```python
# LLaVA架构包含视觉编码器和语言模型
# 对语言模型部分应用LoRA
lora_config = LoraConfig(
r=32,
lora_alpha=64,
target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
lora_dropout=0.1,
bias="none",
task_type="CAUSAL_LM"
)
# 只对语言模型部分应用LoRA
language_model = model.language_model
language_model = get_peft_model(language_model, lora_config)
# 冻结视觉编码器，只训练LoRA参数
for param in model.vision_tower.parameters():
param.requires_grad = False
```
训练策略：
- 视觉编码器：通常保持冻结
- 语言模型：应用LoRA进行微调
- 投影层：可选择性微调或冻结
RAG开发
向量数据库实战
掌握主流向量数据库（Pinecone、Chroma、Weaviate）的使用方法，包括索引创建、向量存储、相似度搜索等核心功能。
实战示例：
### Pinecone 实战示例
**1. 环境配置**
```bash
pip install pinecone-client
```
**2. 基础索引操作**
```python
import pinecone
from sentence_transformers import SentenceTransformer
# 初始化
pinecone.init(api_key="YOUR_API_KEY", environment="us-west1-gcp")
index = pinecone.Index("your-index-name")
# 文本嵌入
model = SentenceTransformer('all-MiniLM-L6-v2')
texts = ["机器学习是人工智能的子领域", "深度学习使用神经网络"]
vectors = model.encode(texts).tolist()
# 存储向量
index.upsert(vectors=[(str(i), vec) for i, vec in enumerate(vectors)])
# 相似度搜索
query_vector = model.encode(["AI相关技术"]).tolist()
results = index.query(vector=query_vector[0], top_k=2)
print(results)
```
### Chroma 实战示例
**本地向量存储**
```python
import chromadb
from sentence_transformers import SentenceTransformer
# 初始化客户端
client = chromadb.Client()
collection = client.create_collection("my_documents")
# 添加文档
documents = ["机器学习是人工智能的子领域", "深度学习使用神经网络"]
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(documents).tolist()
collection.add(
embeddings=embeddings,
documents=documents,
ids=["doc1", "doc2"]
)
# 查询
query_embedding = model.encode(["AI技术"]).tolist()
results = collection.query(
query_embeddings=[query_embedding],
n_results=2
)
print(results)
```
**1. 环境配置**
```bash
pip install pinecone-client
```
**2. 基础索引操作**
```python
import pinecone
from sentence_transformers import SentenceTransformer
# 初始化
pinecone.init(api_key="YOUR_API_KEY", environment="us-west1-gcp")
index = pinecone.Index("your-index-name")
# 文本嵌入
model = SentenceTransformer('all-MiniLM-L6-v2')
texts = ["机器学习是人工智能的子领域", "深度学习使用神经网络"]
vectors = model.encode(texts).tolist()
# 存储向量
index.upsert(vectors=[(str(i), vec) for i, vec in enumerate(vectors)])
# 相似度搜索
query_vector = model.encode(["AI相关技术"]).tolist()
results = index.query(vector=query_vector[0], top_k=2)
print(results)
```
### Chroma 实战示例
**本地向量存储**
```python
import chromadb
from sentence_transformers import SentenceTransformer
# 初始化客户端
client = chromadb.Client()
collection = client.create_collection("my_documents")
# 添加文档
documents = ["机器学习是人工智能的子领域", "深度学习使用神经网络"]
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(documents).tolist()
collection.add(
embeddings=embeddings,
documents=documents,
ids=["doc1", "doc2"]
)
# 查询
query_embedding = model.encode(["AI技术"]).tolist()
results = collection.query(
query_embeddings=[query_embedding],
n_results=2
)
print(results)
```
RAG开发
检索策略优化
掌握多种检索策略，包括混合检索、重排序、查询扩展等高级技术，提升RAG系统的召回率和准确率。
实战示例：
### 混合检索实现
**结合关键词和向量检索**
```python
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import numpy as np
class HybridRetriever:
def __init__(self, documents):
self.documents = documents
self.bm25 = BM25Okapi([doc.split() for doc in documents])
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
self.doc_embeddings = self.embedding_model.encode(documents)
def retrieve(self, query, top_k=5, alpha=0.5):
# BM25分数
bm25_scores = self.bm25.get_scores(query.split())
# 向量相似度分数
query_embedding = self.embedding_model.encode([query])
vector_scores = np.dot(self.doc_embeddings, query_embedding.T).flatten()
# 归一化分数
bm25_norm = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8)
vector_norm = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min() + 1e-8)
# 加权融合
hybrid_scores = alpha * bm25_norm + (1 - alpha) * vector_norm
# 获取top-k结果
top_indices = np.argsort(hybrid_scores)[-top_k:][::-1]
return [(self.documents[i], hybrid_scores[i]) for i in top_indices]
# 使用示例
documents = ["你的文档列表"]
retriever = HybridRetriever(documents)
results = retriever.retrieve("你的查询", top_k=3, alpha=0.3)
```
**结合关键词和向量检索**
```python
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import numpy as np
class HybridRetriever:
def __init__(self, documents):
self.documents = documents
self.bm25 = BM25Okapi([doc.split() for doc in documents])
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
self.doc_embeddings = self.embedding_model.encode(documents)
def retrieve(self, query, top_k=5, alpha=0.5):
# BM25分数
bm25_scores = self.bm25.get_scores(query.split())
# 向量相似度分数
query_embedding = self.embedding_model.encode([query])
vector_scores = np.dot(self.doc_embeddings, query_embedding.T).flatten()
# 归一化分数
bm25_norm = (bm25_scores - bm25_scores.min()) / (bm25_scores.max() - bm25_scores.min() + 1e-8)
vector_norm = (vector_scores - vector_scores.min()) / (vector_scores.max() - vector_scores.min() + 1e-8)
# 加权融合
hybrid_scores = alpha * bm25_norm + (1 - alpha) * vector_norm
# 获取top-k结果
top_indices = np.argsort(hybrid_scores)[-top_k:][::-1]
return [(self.documents[i], hybrid_scores[i]) for i in top_indices]
# 使用示例
documents = ["你的文档列表"]
retriever = HybridRetriever(documents)
results = retriever.retrieve("你的查询", top_k=3, alpha=0.3)
```
RAG开发
上下文压缩技术
学习如何在保持关键信息的前提下压缩检索到的上下文，解决大语言模型输入长度限制问题。
实战示例：
### 上下文压缩实现
**使用LLM进行摘要压缩**
```python
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
# 压缩提示模板
compression_template = """
请将以下文本压缩为简洁的摘要，保留关键信息：
原文：{context}
摘要：
"""
prompt = PromptTemplate.from_template(compression_template)
compression_chain = LLMChain(llm=your_llm, prompt=prompt)
# 压缩多个文档
def compress_context(documents, max_tokens=1000):
compressed_docs = []
total_tokens = 0
for doc in documents:
if total_tokens >= max_tokens:
break
summary = compression_chain.run(context=doc)
compressed_docs.append(summary)
total_tokens += len(summary.split())
return "\n".join(compressed_docs)
# 使用示例
retrieved_docs = ["文档1内容", "文档2内容", ...]
compressed_context = compress_context(retrieved_docs, max_tokens=800)
```
**基于重要性的选择**
```python
# 使用LLM评估每个句子的重要性
importance_template = """
请为以下句子打分（1-5分），分数越高表示对回答查询越重要：
查询：{query}
句子：{sentence}
分数：
"""
importance_prompt = PromptTemplate.from_template(importance_template)
importance_chain = LLMChain(llm=your_llm, prompt=importance_prompt)
def select_important_sentences(documents, query, max_tokens=1000):
scored_sentences = []
for doc in documents:
sentences = doc.split('.')
for sent in sentences:
if sent.strip():
score = importance_chain.run(query=query, sentence=sent.strip())
try:
score = int(score.strip())
scored_sentences.append((sent.strip(), score))
except:
scored_sentences.append((sent.strip(), 3)) # 默认分数
# 按分数排序并选择
scored_sentences.sort(key=lambda x: x[1], reverse=True)
selected = []
total_tokens = 0
for sent, score in scored_sentences:
if total_tokens + len(sent.split()) <= max_tokens:
selected.append(sent)
total_tokens += len(sent.split())
else:
break
return ". ".join(selected)
```
**使用LLM进行摘要压缩**
```python
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
# 压缩提示模板
compression_template = """
请将以下文本压缩为简洁的摘要，保留关键信息：
原文：{context}
摘要：
"""
prompt = PromptTemplate.from_template(compression_template)
compression_chain = LLMChain(llm=your_llm, prompt=prompt)
# 压缩多个文档
def compress_context(documents, max_tokens=1000):
compressed_docs = []
total_tokens = 0
for doc in documents:
if total_tokens >= max_tokens:
break
summary = compression_chain.run(context=doc)
compressed_docs.append(summary)
total_tokens += len(summary.split())
return "\n".join(compressed_docs)
# 使用示例
retrieved_docs = ["文档1内容", "文档2内容", ...]
compressed_context = compress_context(retrieved_docs, max_tokens=800)
```
**基于重要性的选择**
```python
# 使用LLM评估每个句子的重要性
importance_template = """
请为以下句子打分（1-5分），分数越高表示对回答查询越重要：
查询：{query}
句子：{sentence}
分数：
"""
importance_prompt = PromptTemplate.from_template(importance_template)
importance_chain = LLMChain(llm=your_llm, prompt=importance_prompt)
def select_important_sentences(documents, query, max_tokens=1000):
scored_sentences = []
for doc in documents:
sentences = doc.split('.')
for sent in sentences:
if sent.strip():
score = importance_chain.run(query=query, sentence=sent.strip())
try:
score = int(score.strip())
scored_sentences.append((sent.strip(), score))
except:
scored_sentences.append((sent.strip(), 3)) # 默认分数
# 按分数排序并选择
scored_sentences.sort(key=lambda x: x[1], reverse=True)
selected = []
total_tokens = 0
for sent, score in scored_sentences:
if total_tokens + len(sent.split()) <= max_tokens:
selected.append(sent)
total_tokens += len(sent.split())
else:
break
return ". ".join(selected)
```
RAG开发
多源检索系统
构建能够从多个异构数据源（数据库、API、文件系统等）同时检索信息的复杂RAG系统。
实战示例：
### 多源检索架构
**统一检索接口**
```python
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class DataSource(ABC):
@abstractmethod
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
pass
class DatabaseSource(DataSource):
def __init__(self, connection_string: str):
self.connection = create_connection(connection_string)
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
# SQL查询实现
sql_query = f"SELECT * FROM documents WHERE content LIKE '%{query}%'"
results = self.connection.execute(sql_query)
return [{"content": row.content, "source": "database", "score": 1.0} for row in results]
class APISource(DataSource):
def __init__(self, api_endpoint: str, api_key: str):
self.endpoint = api_endpoint
self.headers = {"Authorization": f"Bearer {api_key}"}
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
response = requests.post(
self.endpoint,
json={"query": query, "top_k": kwargs.get("top_k", 5)},
headers=self.headers
)
return [{"content": item["text"], "source": "api", "score": item["score"]} for item in response.json()]
class FileSource(DataSource):
def __init__(self, file_paths: List[str]):
self.documents = []
for path in file_paths:
with open(path, 'r') as f:
self.documents.extend(f.read().split('\n\n'))
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
# 简单的关键词匹配
results = []
for doc in self.documents:
if query.lower() in doc.lower():
score = doc.lower().count(query.lower()) / len(doc.split())
results.append({"content": doc, "source": "file", "score": score})
return sorted(results, key=lambda x: x["score"], reverse=True)[:kwargs.get("top_k", 5)]
# 统一检索器
class MultiSourceRetriever:
def __init__(self, sources: List[DataSource]):
self.sources = sources
def retrieve(self, query: str, **kwargs) -> List[Dict[str, Any]]:
all_results = []
for source in self.sources:
results = source.search(query, **kwargs)
all_results.extend(results)
# 重新排序所有结果
return sorted(all_results, key=lambda x: x["score"], reverse=True)
# 使用示例
sources = [
DatabaseSource("postgresql://user:pass@localhost/mydb"),
APISource("https://api.example.com/search", "your-api-key"),
FileSource(["/data/docs1.txt", "/data/docs2.txt"])
]
retriever = MultiSourceRetriever(sources)
results = retriever.retrieve("你的查询", top_k=10)
```
**统一检索接口**
```python
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class DataSource(ABC):
@abstractmethod
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
pass
class DatabaseSource(DataSource):
def __init__(self, connection_string: str):
self.connection = create_connection(connection_string)
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
# SQL查询实现
sql_query = f"SELECT * FROM documents WHERE content LIKE '%{query}%'"
results = self.connection.execute(sql_query)
return [{"content": row.content, "source": "database", "score": 1.0} for row in results]
class APISource(DataSource):
def __init__(self, api_endpoint: str, api_key: str):
self.endpoint = api_endpoint
self.headers = {"Authorization": f"Bearer {api_key}"}
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
response = requests.post(
self.endpoint,
json={"query": query, "top_k": kwargs.get("top_k", 5)},
headers=self.headers
)
return [{"content": item["text"], "source": "api", "score": item["score"]} for item in response.json()]
class FileSource(DataSource):
def __init__(self, file_paths: List[str]):
self.documents = []
for path in file_paths:
with open(path, 'r') as f:
self.documents.extend(f.read().split('\n\n'))
def search(self, query: str, **kwargs) -> List[Dict[str, Any]]:
# 简单的关键词匹配
results = []
for doc in self.documents:
if query.lower() in doc.lower():
score = doc.lower().count(query.lower()) / len(doc.split())
results.append({"content": doc, "source": "file", "score": score})
return sorted(results, key=lambda x: x["score"], reverse=True)[:kwargs.get("top_k", 5)]
# 统一检索器
class MultiSourceRetriever:
def __init__(self, sources: List[DataSource]):
self.sources = sources
def retrieve(self, query: str, **kwargs) -> List[Dict[str, Any]]:
all_results = []
for source in self.sources:
results = source.search(query, **kwargs)
all_results.extend(results)
# 重新排序所有结果
return sorted(all_results, key=lambda x: x["score"], reverse=True)
# 使用示例
sources = [
DatabaseSource("postgresql://user:pass@localhost/mydb"),
APISource("https://api.example.com/search", "your-api-key"),
FileSource(["/data/docs1.txt", "/data/docs2.txt"])
]
retriever = MultiSourceRetriever(sources)
results = retriever.retrieve("你的查询", top_k=10)
```
RAG开发
RAG评估指标
掌握RAG系统的全面评估方法，包括检索质量、生成质量、端到端性能等多个维度的指标。
实战示例：
### RAG评估实现
**综合评估框架**
```python
from typing import List, Dict, Any
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class RAGEvaluator:
def __init__(self, embedding_model):
self.embedding_model = embedding_model
def evaluate_retrieval(self, query: str, retrieved_docs: List[str], relevant_docs: List[str]) -> Dict[str, float]:
"""评估检索质量"""
# 计算召回率和精确率
retrieved_set = set(retrieved_docs)
relevant_set = set(relevant_docs)
if len(retrieved_set) == 0:
precision = 0.0
else:
precision = len(retrieved_set & relevant_set) / len(retrieved_set)
if len(relevant_set) == 0:
recall = 0.0
else:
recall = len(retrieved_set & relevant_set) / len(relevant_set)
# F1分数
if precision + recall == 0:
f1 = 0.0
else:
f1 = 2 * (precision * recall) / (precision + recall)
# MRR (Mean Reciprocal Rank)
mrr = 0.0
for i, doc in enumerate(retrieved_docs):
if doc in relevant_docs:
mrr = 1.0 / (i + 1)
break
return {
"precision": precision,
"recall": recall,
"f1": f1,
"mrr": mrr
}
def evaluate_generation(self, generated_answer: str, reference_answers: List[str]) -> Dict[str, float]:
"""评估生成质量"""
# ROUGE分数（简化版）
def calculate_rouge_l(candidate, references):
# 这里应该是完整的ROUGE-L实现
# 简化为字符级别的F1
candidate_chars = set(candidate.lower())
reference_chars = set(" ".join(references).lower())
if len(candidate_chars) == 0:
precision = 0.0
else:
precision = len(candidate_chars & reference_chars) / len(candidate_chars)
if len(reference_chars) == 0:
recall = 0.0
else:
recall = len(candidate_chars & reference_chars) / len(reference_chars)
if precision + recall == 0:
f1 = 0.0
else:
f1 = 2 * (precision * recall) / (precision + recall)
return f1
rouge_l = calculate_rouge_l(generated_answer, reference_answers)
# 语义相似度
gen_embedding = self.embedding_model.encode([generated_answer])
ref_embeddings = self.embedding_model.encode(reference_answers)
semantic_sim = np.mean([
cosine_similarity(gen_embedding, ref_emb.reshape(1, -1))[0][0]
for ref_emb in ref_embeddings
])
return {
"rouge_l": rouge_l,
"semantic_similarity": semantic_sim
}
def evaluate_end_to_end(self, query: str, generated_answer: str, ground_truth: str,
retrieved_docs: List[str], relevant_docs: List[str]) -> Dict[str, float]:
"""端到端评估"""
retrieval_metrics = self.evaluate_retrieval(query, retrieved_docs, relevant_docs)
generation_metrics = self.evaluate_generation(generated_answer, [ground_truth])
# 综合分数
overall_score = (
0.4 * retrieval_metrics["f1"] +
0.4 * generation_metrics["rouge_l"] +
0.2 * generation_metrics["semantic_similarity"]
)
return {
**retrieval_metrics,
**generation_metrics,
"overall_score": overall_score
}
# 使用示例
evaluator = RAGEvaluator(SentenceTransformer('all-MiniLM-L6-v2'))
# 评估单个查询
metrics = evaluator.evaluate_end_to_end(
query="什么是机器学习？",
generated_answer="机器学习是...",
ground_truth="机器学习是...",
retrieved_docs=["文档1", "文档2"],
relevant_docs=["文档1", "文档3"]
)
print(metrics)
```
**综合评估框架**
```python
from typing import List, Dict, Any
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class RAGEvaluator:
def __init__(self, embedding_model):
self.embedding_model = embedding_model
def evaluate_retrieval(self, query: str, retrieved_docs: List[str], relevant_docs: List[str]) -> Dict[str, float]:
"""评估检索质量"""
# 计算召回率和精确率
retrieved_set = set(retrieved_docs)
relevant_set = set(relevant_docs)
if len(retrieved_set) == 0:
precision = 0.0
else:
precision = len(retrieved_set & relevant_set) / len(retrieved_set)
if len(relevant_set) == 0:
recall = 0.0
else:
recall = len(retrieved_set & relevant_set) / len(relevant_set)
# F1分数
if precision + recall == 0:
f1 = 0.0
else:
f1 = 2 * (precision * recall) / (precision + recall)
# MRR (Mean Reciprocal Rank)
mrr = 0.0
for i, doc in enumerate(retrieved_docs):
if doc in relevant_docs:
mrr = 1.0 / (i + 1)
break
return {
"precision": precision,
"recall": recall,
"f1": f1,
"mrr": mrr
}
def evaluate_generation(self, generated_answer: str, reference_answers: List[str]) -> Dict[str, float]:
"""评估生成质量"""
# ROUGE分数（简化版）
def calculate_rouge_l(candidate, references):
# 这里应该是完整的ROUGE-L实现
# 简化为字符级别的F1
candidate_chars = set(candidate.lower())
reference_chars = set(" ".join(references).lower())
if len(candidate_chars) == 0:
precision = 0.0
else:
precision = len(candidate_chars & reference_chars) / len(candidate_chars)
if len(reference_chars) == 0:
recall = 0.0
else:
recall = len(candidate_chars & reference_chars) / len(reference_chars)
if precision + recall == 0:
f1 = 0.0
else:
f1 = 2 * (precision * recall) / (precision + recall)
return f1
rouge_l = calculate_rouge_l(generated_answer, reference_answers)
# 语义相似度
gen_embedding = self.embedding_model.encode([generated_answer])
ref_embeddings = self.embedding_model.encode(reference_answers)
semantic_sim = np.mean([
cosine_similarity(gen_embedding, ref_emb.reshape(1, -1))[0][0]
for ref_emb in ref_embeddings
])
return {
"rouge_l": rouge_l,
"semantic_similarity": semantic_sim
}
def evaluate_end_to_end(self, query: str, generated_answer: str, ground_truth: str,
retrieved_docs: List[str], relevant_docs: List[str]) -> Dict[str, float]:
"""端到端评估"""
retrieval_metrics = self.evaluate_retrieval(query, retrieved_docs, relevant_docs)
generation_metrics = self.evaluate_generation(generated_answer, [ground_truth])
# 综合分数
overall_score = (
0.4 * retrieval_metrics["f1"] +
0.4 * generation_metrics["rouge_l"] +
0.2 * generation_metrics["semantic_similarity"]
)
return {
**retrieval_metrics,
**generation_metrics,
"overall_score": overall_score
}
# 使用示例
evaluator = RAGEvaluator(SentenceTransformer('all-MiniLM-L6-v2'))
# 评估单个查询
metrics = evaluator.evaluate_end_to_end(
query="什么是机器学习？",
generated_answer="机器学习是...",
ground_truth="机器学习是...",
retrieved_docs=["文档1", "文档2"],
relevant_docs=["文档1", "文档3"]
)
print(metrics)
```