测试与优化
测试文档准备
准备几个不同类型的测试文档:
1. 纯文本文档
创建 test_simple.txt:
人工智能(Artificial Intelligence,简称 AI)是计算机科学的一个分支,
它企图了解智能的实质,并生产出一种新的能以人类智能相似的方式做出反应的智能机器。
机器学习是人工智能的一个子集,它使用算法来解析数据、从中学习,
然后对世界上的某事做出决定或预测。
深度学习是机器学习的一个子集,它模仿人脑的工作方式,使用多层神经网络。测试问题:
- "什么是人工智能?"
- "机器学习和深度学习的关系是什么?"
2. 结构化文档
创建 test_structured.txt:
# 产品手册
## 系统要求
- 操作系统:Windows 10/11, macOS 11+, Ubuntu 20.04+
- 内存:至少 8GB RAM
- 存储:500MB 可用空间
## 安装步骤
1. 下载安装包
2. 运行安装程序
3. 按照提示完成安装
## 常见问题
Q: 支持 Linux 吗?
A: 支持 Ubuntu 20.04 及以上版本。
Q: 需要联网吗?
A: 首次激活需要联网,之后可离线使用。测试问题:
- "支持哪些操作系统?"
- "如何安装?"
- "需要联网吗?"
3. 技术文档(可选)
准备一份你熟悉的技术文档或产品手册。
功能测试
测试脚本
创建 test_functions.py:
python
from rag import *
import os
def test_document_loading():
"""测试文档加载"""
print("测试 1: 文档加载")
# 测试不同格式
test_files = [
"test_simple.txt",
# "test.pdf",
# "test.docx"
]
for file_path in test_files:
if os.path.exists(file_path):
try:
docs = load_document(file_path)
print(f"✓ {file_path}: {len(docs)} 个片段")
except Exception as e:
print(f"✗ {file_path}: {str(e)}")
def test_splitting():
"""测试文本切分"""
print("\n测试 2: 文本切分")
docs = load_document("test_simple.txt")
splits = split_documents(docs)
print(f"原始片段: {len(docs)}")
print(f"切分后: {len(splits)}")
print(f"第一块长度: {len(splits[0].page_content)} 字")
# 检查是否有重叠
if len(splits) > 1:
overlap = splits[0].page_content[-50:]
print(f"第一块末尾: {overlap}")
def test_retrieval():
"""测试检索"""
print("\n测试 3: 检索测试")
docs = load_document("test_simple.txt")
splits = split_documents(docs)
vectorstore = create_vector_store(splits)
# 测试问题
questions = [
"什么是机器学习?",
"AI 和机器学习的关系",
"深度学习的特点"
]
for q in questions:
results = retrieve_documents(vectorstore, q, k=2)
print(f"\n问题: {q}")
print(f"检索到 {len(results)} 个相关片段")
for i, doc in enumerate(results, 1):
print(f" {i}. {doc.page_content[:80]}...")
def test_qa():
"""测试完整问答"""
print("\n测试 4: 问答测试")
docs = load_document("test_simple.txt")
splits = split_documents(docs)
vectorstore = create_vector_store(splits)
questions = [
"什么是人工智能?",
"机器学习的定义是什么?"
]
for q in questions:
result = answer_question(q, vectorstore)
print(f"\n问题: {q}")
print(f"答案: {result['answer']}")
print(f"来源数: {len(result['sources'])}")
# 运行所有测试
if __name__ == "__main__":
test_document_loading()
test_splitting()
test_retrieval()
test_qa()运行测试:
bash
python test_functions.py准确率评估
评估方法
准备 10-20 个测试问题,手动评估答案质量:
| 问题 | 标准答案 | AI 答案 | 准确性 | 相关性 | 完整性 |
|---|---|---|---|---|---|
| 什么是 AI? | ... | ... | ✓ | ✓ | ✓ |
| ... | ... | ... | ... | ... | ... |
评分标准:
- 准确性:答案是否正确
- 相关性:是否回答了问题
- 完整性:信息是否完整
计算准确率
准确率 = (正确答案数 / 总问题数) × 100%目标:> 70%
常见问题排查
问题 1:答案不准确
症状:答非所问或内容错误
排查步骤:
- 检查检索是否准确
python
# 测试检索
results = retrieve_documents(vectorstore, "你的问题", k=5)
for i, doc in enumerate(results, 1):
print(f"{i}. {doc.page_content}\n")如果检索结果不相关:
- 调整 chunk_size(尝试 300/500/800)
- 增加 TOP_K(从 3 改成 5)
- 使用更好的 embedding 模型
- 检查 Prompt
python
# 查看实际发送给 LLM 的 Prompt
qa_chain = create_qa_chain(vectorstore)
# 打印 chain 的 prompt 模板
print(qa_chain.combine_documents_chain.llm_chain.prompt.messages[0].prompt.template)- 优化 Prompt
自定义 Prompt 模板:
python
from langchain.prompts import PromptTemplate
prompt_template = """你是一个专业的问答助手。
请根据以下参考文档回答问题。
参考文档:
{context}
问题:{question}
要求:
1. 只基于参考文档回答,不要编造信息
2. 如果文档中没有答案,直接说"文档中没有提到这个内容"
3. 回答要简洁明了
答案:"""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT}
)问题 2:速度太慢
排查:
- 找出瓶颈
python
import time
start = time.time()
# 执行操作
end = time.time()
print(f"耗时: {end - start:.2f} 秒")- 优化方向
- 文档加载慢:换更快的 PDF 解析器(如 pdfplumber)
- 向量化慢:减小 chunk_size 或用更快的 embedding
- 检索慢:考虑用 Pinecone 等托管服务
- 生成慢:换更快的模型(如 qwen-turbo)
问题 3:内存占用大
解决:
- 使用持久化向量库
python
# 向量化后保存到磁盘
vectorstore = create_vector_store(splits)
# 后续直接加载,不需要重新向量化
vectorstore = load_vector_store()- 限制文档大小
python
MAX_DOC_SIZE = 10 * 1024 * 1024 # 10MB
if os.path.getsize(file_path) > MAX_DOC_SIZE:
st.warning("文档太大,请上传小于 10MB 的文件")
return性能优化
1. 缓存机制
python
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_embedding(text):
"""缓存 embedding 结果"""
return embeddings.embed_query(text)2. 批量处理
python
def batch_embed(texts, batch_size=32):
"""批量 embedding"""
results = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
results.extend(embeddings.embed_documents(batch))
return results3. 并发处理
python
from concurrent.futures import ThreadPoolExecutor
def process_multiple_files(file_paths):
"""并发处理多个文件"""
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(load_document, file_paths)
return list(results)用户体验优化
1. 添加加载提示
python
with st.spinner("加载文档中..."):
# 处理逻辑
time.sleep(1)
st.success("加载完成!")2. 错误处理
python
try:
result = answer_question(question, vectorstore)
except Exception as e:
st.error(f"出错了: {str(e)}")
st.info("试试换个问题或者重新上传文档")3. 输入验证
python
if len(question) < 5:
st.warning("问题太短了,多写一点吧")
elif len(question) > 500:
st.warning("问题太长了,精简一下吧")
else:
# 处理问题
pass下一步
测试通过后,准备部署上线。