1. RAG系统搭建概述
RAG(Retrieval-Augmented Generation)系统是一种结合检索和生成能力的智能问答架构。它通过将用户问题与知识库中的相关内容进行匹配,再结合大语言模型的生成能力,能够提供更准确、更专业的回答。相比传统规则匹配系统,RAG具有以下优势:
- 知识更新灵活:只需更新文档库,无需修改规则
- 回答自然流畅:利用大语言模型的生成能力
- 可解释性强:回答基于实际文档内容
- 适用范围广:可应用于各种领域的知识问答
2. 环境准备与依赖安装
2.1 基础环境配置
建议使用Linux系统(如Ubuntu 20.04+)进行开发,确保系统已安装:
bash复制# 更新系统包
sudo apt update && sudo apt upgrade -y
# 安装基础开发工具
sudo apt install -y python3-pip python3-dev build-essential git curl
# 安装Redis
sudo apt install -y redis-server
sudo systemctl enable redis-server
sudo systemctl start redis-server
2.2 Python环境配置
推荐使用Python 3.8+版本,创建并激活虚拟环境:
bash复制python3 -m venv rag_env
source rag_env/bin/activate
2.3 安装项目依赖
bash复制pip install playwright dashscope redis numpy openai python-dotenv
playwright install chromium
3. 数据收集与处理
3.1 文档爬取实现
使用Playwright进行网页内容抓取,以下是改进后的爬虫代码:
python复制from playwright.sync_api import sync_playwright
from pathlib import Path
import time
def crawl_faq_page(url, output_dir="data/raw"):
"""
爬取FAQ页面内容并保存为文本文件
参数:
url (str): 目标URL
output_dir (str): 输出目录路径
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context(
locale='zh-CN',
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
)
page = context.new_page()
try:
page.goto(url, timeout=60000)
page.wait_for_selector("#faq-list", timeout=30000)
# 滚动页面确保内容加载
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
time.sleep(2)
content = page.locator("#faq-list").inner_text()
# 生成唯一文件名
timestamp = int(time.time())
output_path = f"{output_dir}/faq_{timestamp}.txt"
with open(output_path, "w", encoding="utf-8") as f:
f.write(content)
print(f"成功保存FAQ内容到: {output_path}")
return output_path
except Exception as e:
print(f"爬取失败: {str(e)}")
return None
finally:
browser.close()
3.2 数据清洗与预处理
对爬取的原始数据进行清洗和结构化处理:
python复制import re
import json
from pathlib import Path
from datetime import datetime
def clean_text(text):
"""
清洗文本内容
"""
# 移除特殊字符和多余空格
text = re.sub(r'[\t\r\n]+', '\n', text)
text = re.sub(r'[ ]{2,}', ' ', text)
return text.strip()
def extract_qa_pairs(text):
"""
从文本中提取问答对
"""
qa_pairs = []
current_q = None
current_a = []
for line in text.split('\n'):
line = line.strip()
if not line:
continue
# 识别问题行
if line.startswith(('Q:', 'Q:', '问题:', '问题:')):
if current_q is not None:
qa_pairs.append({
'question': current_q,
'answer': '\n'.join(current_a).strip()
})
current_q = re.sub(r'^(Q[::]|问题[::]\s*)', '', line).strip()
current_a = []
else:
current_a.append(line)
if current_q is not None:
qa_pairs.append({
'question': current_q,
'answer': '\n'.join(current_a).strip()
})
return qa_pairs
def process_faq_file(input_file, output_file, source_url):
"""
处理单个FAQ文件
"""
with open(input_file, 'r', encoding='utf-8') as f:
raw_text = f.read()
cleaned_text = clean_text(raw_text)
qa_pairs = extract_qa_pairs(cleaned_text)
processed_data = []
for qa in qa_pairs:
processed_data.append({
'question': qa['question'],
'answer': qa['answer'],
'metadata': {
'source': source_url,
'process_time': datetime.now().isoformat()
}
})
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(processed_data, f, ensure_ascii=False, indent=2)
print(f"已处理并保存到: {output_file}")
return output_file
4. 向量数据库搭建
4.1 Redis向量索引配置
python复制import redis
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition
def setup_redis_index(redis_conn, index_name="faq_index", vector_dim=1024):
"""
设置Redis向量搜索索引
"""
try:
# 检查索引是否已存在
redis_conn.ft(index_name).info()
print("索引已存在,跳过创建")
return
except:
pass
# 定义索引schema
schema = (
TextField("question", weight=5.0),
TextField("answer"),
TextField("source"),
VectorField("embedding",
"HNSW",
{
"TYPE": "FLOAT32",
"DIM": vector_dim,
"DISTANCE_METRIC": "COSINE"
})
)
# 创建索引
redis_conn.ft(index_name).create_index(
fields=schema,
definition=IndexDefinition(
prefix=["faq:"],
index_type="HNSW"
)
)
print("成功创建向量索引")
4.2 文档向量化存储
python复制import numpy as np
from typing import List, Dict
def store_documents(redis_conn, documents: List[Dict], embed_func):
"""
将文档存储到Redis向量数据库
"""
for doc in documents:
# 生成文档向量
text_to_embed = f"{doc['question']} {doc['answer']}"
embedding = embed_func(text_to_embed)
# 准备文档数据
doc_id = f"faq:{hash(text_to_embed)}"
doc_data = {
'question': doc['question'],
'answer': doc['answer'],
'source': doc['metadata']['source'],
'embedding': np.array(embedding, dtype=np.float32).tobytes()
}
# 存储到Redis
redis_conn.hset(doc_id, mapping=doc_data)
print(f"成功存储{len(documents)}条文档")
5. 检索与生成实现
5.1 相似性检索优化
python复制def search_similar_questions(redis_conn, index_name, query_embedding, top_k=3):
"""
执行相似性问题检索
"""
query = (
Query(f"*=>[KNN {top_k} @embedding $vec AS score]")
.sort_by("score")
.return_fields("question", "answer", "source", "score")
.dialect(2)
)
results = redis_conn.ft(index_name).search(
query,
query_params={"vec": query_embedding}
)
return [dict(doc) for doc in results.docs]
5.2 提示词工程优化
python复制def build_enhanced_prompt(user_question, retrieved_docs):
"""
构建增强版提示词
"""
context_str = "\n\n".join(
f"相关文档 {i+1}:\n问题: {doc['question']}\n回答: {doc['answer']}\n来源: {doc['source']}\n相关性分数: {doc['score']:.3f}"
for i, doc in enumerate(retrieved_docs)
)
prompt = f"""你是一个专业客服助手,请根据以下参考文档回答用户问题。
用户问题: {user_question}
参考文档:
{context_str}
回答要求:
1. 必须基于参考文档内容
2. 保持专业、友好的语气
3. 如无相关信息,明确告知"未找到相关信息"
4. 回答简洁,不超过3句话
请生成回答:"""
return prompt
5.3 大模型响应生成
python复制def generate_response(llm_client, prompt, model="deepseek-r1-distill-llama-70b"):
"""
调用大模型生成回答
"""
response = llm_client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=200
)
return response.choices[0].message.content
6. 系统集成与测试
6.1 主流程集成
python复制class RAGSystem:
def __init__(self, redis_conn, embed_func, llm_client):
self.redis = redis_conn
self.embed_func = embed_func
self.llm_client = llm_client
self.index_name = "faq_index"
def query(self, user_question, top_k=3):
# 生成问题向量
query_embedding = self.embed_func(user_question)
# 检索相关文档
retrieved_docs = search_similar_questions(
self.redis, self.index_name,
np.array(query_embedding).tobytes(),
top_k=top_k
)
if not retrieved_docs:
return "未找到相关信息"
# 构建提示词
prompt = build_enhanced_prompt(user_question, retrieved_docs)
# 生成回答
response = generate_response(self.llm_client, prompt)
return response
6.2 测试案例
python复制def test_rag_system():
# 初始化组件
redis_conn = redis.Redis()
embed_func = lambda text: get_embedding(text) # 替换为实际的embedding函数
llm_client = OpenAI() # 配置好的客户端
# 创建系统实例
rag_system = RAGSystem(redis_conn, embed_func, llm_client)
# 测试查询
questions = [
"订单取消后多久能收到退款?",
"配送延迟怎么办?",
"如何修改支付方式?"
]
for q in questions:
print(f"\n用户问题: {q}")
answer = rag_system.query(q)
print(f"系统回答: {answer}")
7. 性能优化与扩展
7.1 检索性能优化
-
索引优化:
- 调整HNSW参数(ef_construction, M)
- 对高频查询建立缓存
- 对问题字段添加全文索引
-
查询优化:
- 实现多级检索(先关键词过滤,再向量搜索)
- 支持混合搜索(结合关键词和向量相似度)
7.2 系统扩展方案
-
多源知识库支持:
- 添加文档来源分类
- 支持不同来源的优先级设置
-
动态更新机制:
- 实现增量索引更新
- 设置文档过期时间
-
多模型支持:
- 可配置的embedding模型
- 可切换的大语言模型
8. 实际应用建议
8.1 部署注意事项
-
资源规划:
- Redis内存配置(至少16GB)
- GPU资源(如需本地运行embedding模型)
- 网络带宽(考虑embedding API调用)
-
监控指标:
- 响应时间(目标<500ms)
- 检索召回率
- 大模型使用成本
8.2 常见问题解决
-
检索结果不相关:
- 检查embedding模型是否适合领域
- 调整文本预处理流程
- 优化chunk大小
-
生成回答不准确:
- 优化提示词模板
- 添加回答验证机制
- 设置fallback策略
-
性能瓶颈:
- 实现异步处理流程
- 考虑分布式向量数据库
- 优化批量处理
9. 进阶开发方向
-
多模态支持:
- 处理图片、表格等非文本内容
- 实现跨模态检索
-
对话历史整合:
- 维护对话上下文
- 实现多轮问答
-
反馈学习机制:
- 收集用户反馈
- 优化检索和生成
-
领域自适应:
- 领域特定fine-tuning
- 定制化prompt模板
通过以上步骤,我们构建了一个完整的RAG系统,从数据收集到最终回答生成,每个环节都进行了详细设计和实现。这个系统可以根据实际需求进行扩展和优化,适用于各种知识密集型问答场景。
