目录[-]
一、引言:测试工作中的痛点
在软件测试的日常工作中,经常遇到以下场景:
- 测试A发现了一个疑似重复的 Bug ,但是他忘记以前有没有提过。
- 新同事入职,想了解某个模块的历史 Bug 都有哪些。
- Bug 量太多,每次都要手动调整筛选条件,检索时间长,需要手动筛选几百条记录,挨个点进去查看。
这些问题导致一个问题:数据存了,查询麻烦。
二、解决方案:RAG技术
RAG = Retrieval Augmented Generation(检索增强生成),它的核心思想是:
- 让 AI 先检索相关知识
- 基于检索答案回答问题
三、技术架构
┌─────────┐ ┌─────────┐ ┌─────────┐
1.用户问题 -> 2.向量检索 -> 3.相关文档
└─────────┘ └─────────┘ └────┬────┘
▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ 6.答案 <- 5.LLM <- 4.上下文
└─────────┘ └─────────┘ └─────────┘
四、完整代码(可直接运行)
import os
import shutil
import numpy as np
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
# LangChain 框架
#- OllamaEmbeddings(向量化)
#- Chroma(向量数据库)
#- Ollama(LLM接入)
#- RetrievalQA(问答链)
#- PromptTemplate(提示词工程)
print("🚀 开始最简单的RAG Demo")
# 1. 加载文档 - 按行读取,每行一条独立句子
with open('data.txt', 'r', encoding='utf-8') as f:
texts = [line.strip() for line in f if line.strip()]
print(f"📄 加载了 {len(texts)} 条独立句子:")
for i, text in enumerate(texts):
print(f" 句子{i + 1}: {text}")
# 2. 初始化向量模型(把文字变成向量)
embeddings = OllamaEmbeddings(model="nomic-embed-text")
# 3. 删除旧的向量库(确保全新的)
if os.path.exists("./chroma_db"):
print("🗑️ 删除旧的向量库")
shutil.rmtree("./chroma_db")
# 4. 创建向量数据库(存向量)
vectorstore = Chroma.from_texts(
texts=texts,
embedding=embeddings,
persist_directory="./chroma_db"
)
print("💾 向量数据库已创建")
# 5. 创建检索器(召回率比精确率更重要,宁多勿少)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": len(texts)} # 返回所有文档
)
# 6. 初始化语言模型
llm = Ollama(
model="qwen2.5:3b",
temperature=0.1,
base_url="http://localhost:11434"
)
# 7. 自定义Prompt(防止LLM乱回答)
strict_prompt_template = """你是一个严格基于事实回答的AI助手。
【重要规则】
1. 只使用下面【上下文】中提供的信息来回答问题
2. 如果【上下文】中没有相关信息,必须说"未找到相关信息"
3. 不要添加任何外部知识
4. 不要猜测
【上下文】
{context}
【问题】
{question}
【你的回答】"""
PROMPT = PromptTemplate(
template=strict_prompt_template,
input_variables=["context", "question"]
)
# 8. 创建问答链(LangChain将所有组件串起来)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # 把所有文档一起塞给LLM
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT}
)
print("✅ RAG系统初始化完成!\n")
# 9. 预计算所有文档的向量(用于显示相似度)
print("🔄 预计算文档向量...")
doc_vectors = embeddings.embed_documents(texts)
print("✅ 向量计算完成\n")
def cosine_similarity(v1, v2):
"""手动计算余弦相似度(结果在-1到1之间)"""
v1 = np.array(v1)
v2 = np.array(v2)
dot_product = np.dot(v1, v2)
norm_v1 = np.linalg.norm(v1)
norm_v2 = np.linalg.norm(v2)
if norm_v1 == 0 or norm_v2 == 0:
return 0
return dot_product / (norm_v1 * norm_v2)
# 10. 开始问答
questions = [
"苹果是什么颜色?",
"北京在哪里?",
"Python是什么?"
]
for q in questions:
print(f"\n❓ 问题: {q}")
# 计算问题向量
q_vector = embeddings.embed_query(q)
# 手动计算所有相似度(为了展示)
print("检索结果:")
similarities = []
for i, doc_vector in enumerate(doc_vectors):
sim = cosine_similarity(q_vector, doc_vector)
similarities.append((sim, texts[i]))
# 按相似度排序(从高到低)
similarities.sort(reverse=True)
for sim, text in similarities:
print(f" 相似度: {sim:.4f} | {text}")
# 用QA Chain回答
answer = qa_chain.invoke(q)
print(f"回答: {answer['result'] if isinstance(answer, dict) else answer}")
print("-" * 50)
五、关键技术解析
1. 向量化与相似度计算
# 将文本转换成向量
doc_vectors = embeddings.embed_documents(texts)
# 相似度计算(结果范围:-1到1)
sim = cosine_similarity(q_vector, doc_vector)
def cosine_similarity(v1, v2):
"""
手动计算余弦相似度(结果在-1到1之间)
向量化就是把“文字”翻译成“方向”,余弦相似度是测量这两个方向是否一致
"""
# v1 = [3,4] # 问题向量
# v2 = [6,8] # 文档向量
# 第一步,将数组转换成numpy数据——》array【3,4】
v1 = np.array(v1)
v2 = np.array(v2)
# 第二步,计算点积——》3*6+4*8=18+32=50
dot_product = np.dot(v1, v2)
# 第三步,计算范数 norm_v1(向量的长度/模)
norm_v1 = np.linalg.norm(v1) # 根号(3^2+4*2)=根号(9+16)=根号(25)=5
norm_v2 = np.linalg.norm(v2) # 同上计算=10
if norm_v1 == 0 or norm_v2 == 0:
return 0
# 最后一步,计算余弦相似度=50/(5*10)=1.0,越接近1越相似
return dot_product / (norm_v1 * norm_v2)
结论:这是一个简单的检索,将“问题”和“文档”中的内容,做了向量相除,返回一个数字,类似0.667,代表二者的相似程度。
2. 检索策略设计
# 5. 创建检索器(在RAG系统中,召回率比精确率更重要,所以,宁肯多找信息,也不能漏掉相关信息)
retriever = vectorstore.as_retriever(
search_type="similarity",
# 返回所有文档
search_kwargs={"k": len(texts)}
)
结论:search_kwargs字段中,默认k=4,你可以这么简单的理解,它默认查找四块内容,但是当你的内容存在第5块或者更远的位置时,就查不到了。由于我们的数据量较少,采取了全部获取,但是有一个隐患,当数据量特别大时,可能报错,这也是后续可优化的点。
3. LLM约束设计
# 7. 自定义Prompt
strict_prompt_template = """你是一个严格基于事实回答的AI助手。
【重要规则】
1. 只使用下面【上下文】中提供的信息来回答问题
2. 如果【上下文】中没有相关信息,必须说"未找到相关信息"
3. 不要添加任何外部知识
4. 不要猜测
【上下文】
{context}
【问题】
{question}
【你的回答】"""
PROMPT = PromptTemplate(
template=strict_prompt_template,
input_variables=["context", "question"]
)
结论:为何采取约束?因为就算定义了LLM的temperature属性,让它保守回答,它还是可能凭感觉回答问题,也就是“幻觉”。而我们使用了这个字段,就是为了消除幻觉。
六、运行效果分析
❓ 问题: Python是什么?
检索结果:
相似度: 0.6647 | 猫是一种宠物,会喵喵叫。
相似度: 0.6576 | Python是一种编程语言,不是蛇。
相似度: 0.6547 | 狗是人类的好朋友,会汪汪叫。
相似度: 0.6460 | 香蕉是黄色的水果,富含钾元素。
相似度: 0.6187 | 苹果是一种水果,通常红色或绿色,口感酸甜。
相似度: 0.6162 | 上海是中国的经济中心,东方明珠很出名。
相似度: 0.5694 | 北京是中国的首都,有很多历史古迹。
回答: Python是一种编程语言,不是蛇。
关键观察:正确的结果是第2条,但是它的分数却排行第2,那么为什么LLM返回的值,是正确的?
这就是 RAG 的核心优势——》检索的广度 与 LLM 的理解能力相结合。
七、扩展 Bug 知识库查询
如果将其用于查找 bug ,那么可以得到下面的效果。
❓ 问题: 二维码 bug
📊 检索到 30 条相关文档
🔑 问题关键词: {'二维码'}
- 找到 Bug BUG-041: 二维码扫描成功后页面无反应 (匹配度: 1)
- 找到 Bug BUG-038: 默认在账号密码页却调用二维码接口 (匹配度: 1)
- 找到 Bug BUG-042: 微信扫描二维码返回乱码 (匹配度: 1)
- 找到 Bug BUG-040: 二维码过期后无提示 (匹配度: 1)
- 找到 Bug BUG-039: 二维码刷新后倒计时不重置 (匹配度: 1)
- 找到 Bug BUG-043: 二维码图片加载失败无占位图 (匹配度: 1)
- 找到 Bug BUG-045: 二维码扫描后,页面无确认提示 (匹配度: 1)
- 找到 Bug BUG-044: 二维码刷新频率过高 (匹配度: 1)
💡 回答: 📋 共找到 8 个相关的Bug:
1. **Bug BUG-038**:
- 模块:登录-二维码
- 标题:默认在账号密码页,却调用二维码接口
- 严重程度:P2
- 状态:已修复
- 发现步骤:打开登录页,抓包
- 实际结果:发现status接口被调用
- 预期结果:不应调用二维码接口
- 根因分析:接口调用时机错误
2. **Bug BUG-039**:
- 模块:登录-二维码
- 标题:二维码刷新后倒计时不重置
- 严重程度:P3
- 状态:已修复
- 发现步骤:点击刷新二维码
- 实际结果:二维码更新,倒计时未重置
- 预期结果:倒计时应重新开始
- 根因分析:未重置定时器
📌 涉及到的Bug编号: BUG-038, BUG-039, BUG-040, BUG-041, BUG-042, BUG-043, BUG-044, BUG-045 (共8个)
八、总结
本文用核心代码实现了一个初步的RAG系统,实现了:
- RAG 完整流程:从文档加载到生成答案
- 核心技术组件:向量化、检索、LLM、Prompt
- 实战经验:检索策略设计、LLM约束、相似度计算
这个系统可直接应用于Bug知识库、文档问答、智能客服等场景,帮助团队提升信息检索效率。表面上是让AI能‘查资料’。但更深层的价值,是把测试团队沉淀的经验(历史Bug、用例设计思路)变成AI可以持续学习、调用的知识资产。
当新人入职时,他面对的不再是Excel表格,而是一个能‘告诉他历史故事’的智能助手。这,才是AI赋能测试的真正意义。
九、优化框架
本文所有的代码,放在了同一个文件里,但是后续维护就存在很大的问题,所以,优化的第一步,就是重构一个更加规范、方便维护的项目结构。
将本文中的代码分别放入以下文件中
src/config.py:集中管理所有可调整的配置项
# src/config.py
import os
# --- 路径配置 ---
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 项目根目录
DATA_PATH = os.path.join(BASE_DIR, "data", "data.txt")
CHROMA_DB_PATH = os.path.join(BASE_DIR, "chroma_db")
# --- 模型配置 ---
EMBEDDING_MODEL = "nomic-embed-text"
LLM_MODEL = "qwen2.5:3b"
OLLAMA_BASE_URL = "http://localhost:11434"
# --- RAG 参数配置 ---
LLM_TEMPERATURE = 0.1
RETRIEVER_SEARCH_TYPE = "similarity"
# 注意:这里的 K 值后续会改,但第一步先保持原样
# RETRIEVER_K = 4 # 我们可以先注释掉,等第二步再用
# --- Prompt 模板 ---
STRICT_PROMPT_TEMPLATE = """你是一个严格基于事实回答的AI助手。
【重要规则】
1. 只使用下面【上下文】中提供的信息来回答问题
2. 如果【上下文】中没有相关信息,必须说"未找到相关信息"
3. 不要添加任何外部知识
4. 不要猜测
【上下文】
{context}
【问题】
{question}
【你的回答】"""
src/utils.py:存放通用的、与核心业务无关的工具函数。
# src/utils.py
import numpy as np
def cosine_similarity(v1, v2):
"""
手动计算余弦相似度(结果在-1到1之间)
向量化就是把“文字”翻译成“方向”,余弦相似度是测量这两个方向是否一致
"""
v1 = np.array(v1)
v2 = np.array(v2)
dot_product = np.dot(v1, v2)
norm_v1 = np.linalg.norm(v1)
norm_v2 = np.linalg.norm(v2)
if norm_v1 == 0 or norm_v2 == 0:
return 0
return dot_product / (norm_v1 * norm_v2)
src/embedding_utils.py:封装与向量化和向量库相关的操作。
# src/embedding_utils.py
import os
import shutil
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from src import config
def initialize_embeddings():
"""初始化并返回向量化模型"""
print("🔄 正在初始化向量化模型...")
return OllamaEmbeddings(model=config.EMBEDDING_MODEL)
def create_vector_store(texts, embeddings):
"""创建或重建向量数据库"""
# 删除旧的向量库
if os.path.exists(config.CHROMA_DB_PATH):
print(f"🗑️ 删除旧的向量库: {config.CHROMA_DB_PATH}")
shutil.rmtree(config.CHROMA_DB_PATH)
print("💾 正在创建新的向量数据库...")
vectorstore = Chroma.from_texts(
texts=texts,
embedding=embeddings,
persist_directory=config.CHROMA_DB_PATH
)
print("✅ 向量数据库已创建")
return vectorstore
src/rag_chain.py:核心文件,负责组装整个RAG问答链。
# src/rag_chain.py
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from src import config
def load_documents(file_path):
"""从文件加载文档,按行读取并清洗"""
print(f"📄 正在从 {file_path} 加载文档...")
with open(file_path, 'r', encoding='utf-8') as f:
texts = [line.strip() for line in f if line.strip()]
print(f"📄 加载了 {len(texts)} 条独立句子:")
for i, text in enumerate(texts):
print(f" 句子{i + 1}: {text}")
return texts
def initialize_llm():
"""初始化并返回语言模型"""
print("🤖 正在初始化语言模型...")
return Ollama(
model=config.LLM_MODEL,
temperature=config.LLM_TEMPERATURE,
base_url=config.OLLAMA_BASE_URL
)
def create_qa_chain(vectorstore, llm):
"""创建并返回问答链"""
print("🔗 正在创建问答链...")
# 创建检索器
# 注意:这里为了保持与原代码一致,暂时将所有文档返回。后续会优化。
retriever = vectorstore.as_retriever(
search_type=config.RETRIEVER_SEARCH_TYPE,
search_kwargs={"k": vectorstore._collection.count()} # 动态获取文档总数
)
# 准备 Prompt
prompt = PromptTemplate(
template=config.STRICT_PROMPT_TEMPLATE,
input_variables=["context", "question"]
)
# 组装链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": prompt}
)
print("✅ RAG系统初始化完成!\n")
return qa_chain
main.py:程序入口,负责调配所有模块并运行。
# main.py
from src import config
from src.embedding_utils import initialize_embeddings, create_vector_store
from src.rag_chain import load_documents, initialize_llm, create_qa_chain
from src.utils import cosine_similarity
def main():
print("🚀 开始最简单的RAG Demo (重构版)")
# 1. 加载文档
texts = load_documents(config.DATA_PATH)
# 2. 初始化向量模型与数据库
embeddings = initialize_embeddings()
vectorstore = create_vector_store(texts, embeddings)
# 3. 初始化LLM与问答链
llm = initialize_llm()
qa_chain = create_qa_chain(vectorstore, llm)
# 4. 预计算文档向量(用于演示相似度)
print("🔄 预计算文档向量...")
doc_vectors = embeddings.embed_documents(texts)
print("✅ 向量计算完成\n")
# 5. 问答测试
questions = [
"苹果是什么颜色?",
"北京在哪里?",
"Python是什么?"
]
for q in questions:
print(f"\n❓ 问题: {q}")
q_vector = embeddings.embed_query(q)
print("检索结果:")
similarities = []
for i, doc_vector in enumerate(doc_vectors):
sim = cosine_similarity(q_vector, doc_vector)
similarities.append((sim, texts[i]))
similarities.sort(reverse=True, key=lambda x: x[0])
for sim, text in similarities:
print(f" 相似度: {sim:.4f} | {text}")
answer = qa_chain.invoke(q)
print(f"回答: {answer['result'] if isinstance(answer, dict) else answer}")
print("-" * 50)
if __name__ == "__main__":
main()
再次运行后,结果如下,重构后,效果完美。
十、优化文档加载与智能分块
data.txt 文档里,是按照每一行都是独立一行的形式录入向量库,但如果不是加工好的一句话,而是小说、长文本呢?按照现有的逻辑,每一行都是独立的内容,完全不行。所以我们需要引入新的库来解决语音问题。
调整文档内容,我们再这里下载一个白雪公主的故事,作为后续代码使用:https://pan.baidu.com/s/1NeS7hXCyclFcQC95myjuuQ?pwd=gjht
第一步:引用 langchain_text_splitters 库,可以这么理解,这一段先读取本地文档,将文本按照语义切分成多个块,编辑块的信息后,返回一个 Document 对象列表。
def load_and_split_documents(file_path):
"""
改进的文档加载函数:
1. 使用 TextLoader 加载文档,自动处理编码
2. 使用 RecursiveCharacterTextSplitter 按语义智能切分
3. 为每个文本块添加元数据(如来源文件名)
"""
print(f"📄 正在加载并智能切分文档: {file_path}")
# 1、使用 TextLoader 加载文档(比手动 open 更健壮),自动处理编码问问难题
loader = TextLoader(file_path, encoding='utf-8')
# documents:LangChain 统一的文档数据格式,里面存【文本内容+元数据(来源、序号等)】
# 2、loader.load():返回 Document 对象列表,此时还是一整个文档,没有切分
documents = loader.load()
# 3、创建递归文本分割器
# RecursiveCharacterTextSplitter 最常用的智能文本切分工具,按语义、段落、句子一层层切分。
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=config.CHUNK_SIZE, # 每个文本块最大长度
chunk_overlap=config.CHUNK_OVERLAP, # 块与块之间的重叠长度
separators=config.SEPARATORS, # 分隔符:按段落、换行、空格、字符切分
length_function=len, # 使用字符数计算长度(最常用)
add_start_index=True # 记录每个块在原文档中的起始位置(便于调试)
)
# 4、执行分割,将一整个 Document 对象,切割为多个,并返回列表
chunks = text_splitter.split_documents(documents)
# 5、为每个块添加或完善元数据
base_filename = os.path.basename(file_path)
for i, chunk in enumerate(chunks):
# 保留原有的 source 信息,并添加自定义元数据
chunk.metadata["source"] = base_filename # 标记来源文件
chunk.metadata["chunk_index"] = i # 标记是第几块
# 每个 chunk 都有两个核心内容
# chunk.page_content:文本内容
# chunk.metadata:来源、序号等信息
print(f"📄 文档已切分为 {len(chunks)} 个语义块。")
# 为了适配后续向量库的创建,我们需要提取纯文本列表和元数据列表
# 注意:Chroma.from_documents 可以直接接受 Document 对象列表
return chunks
第二步:embedding 向量化模型,将上一步提供的语义块,切成小块,转换成向量,存储到 Chroma 数据库。
def create_vector_store(documents: List[Document], embeddings):
"""从 Document 列表创建或重建向量数据库"""
# 删除旧的向量库
if os.path.exists(config.CHROMA_DB_PATH):
print(f"🗑️ 删除旧的向量库: {config.CHROMA_DB_PATH}")
shutil.rmtree(config.CHROMA_DB_PATH)
print("💾 正在从文档块创建新的向量数据库...")
# 使用 from_documents 方法,它会自动处理文档内容和元数据
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory=config.CHROMA_DB_PATH
)
print("✅ 向量数据库已创建,元数据已保存")
return vectorstore
第三步:main.py 调整代码, 完整的链路就是,读文档——》建索引——》组装问答系统,最后一步就是问答。
def main():
print("🚀 开始RAG Demo (检索速度优化版)")
# 1. 加载并智能分割文档
documents = load_and_split_documents(config.DATA_PATH)
# texts = [doc.page_content for doc in documents]
print(f"📊 文档被切分为 {len(documents)} 个块")
# 2. 初始化向量模型与数据库
embeddings = initialize_embeddings()
'''
1、create_vector_store 传入2个参数
documents:src/rag_chain.py 切分好的文档块列表,每个块是一个 Document 对象
embeddings:向量化模型(用于将文本转换成数学向量)
2、函数内部执行向量化并存储,返回一个 Chroma 向量库对象
3、将返回的对象赋值给 vectorstore 变量,供后续检索使用
'''
vectorstore = create_vector_store(documents, embeddings)
# 3. 初始化LLM与问答链,启动模型,将“检索器(vectorstore)”与“生成器(llm)”串联,形成完整的问答
llm = initialize_llm()
qa_chain = create_qa_chain(vectorstore, llm)
第四步:运行结果如下,接近300秒也就是5分钟,一个简单的回答,为什么会耗费这么长时间?
十一、优化检索时长
检索时间太长,很明显超出我们的预计,优化方向我们依次说明。
第一:采用更轻量、更小的模型(0.5b)
本文中采用的 7b ,但是我们要做的,并不是让模型扩展的回答答案,而是根据文档检索内容,所以我们换成0.5b,查看时间。如下图所示,直接变成了 37.65秒,比起之前的 290.84秒,缩短了约 80% 时间,此方法成功。
注意:本文中给出的时间,是“清空缓存、重启服务后”多次运行后的时间参考,并不是独立的数字。
第二:添加模型优化参数
# src/config.py
# ... 原有配置 ...
# 新增:模型推理优化参数
LLM_NUM_PREDICT = 256 # 限制最大生成token数(答案不需要太长)
LLM_NUM_CTX = 2048 # 上下文窗口大小
LLM_NUM_THREADS = 4 # CPU线程数(根据你的CPU核心数调整)
# src/rag_chain.py
def initialize_llm():
"""初始化并返回语言模型(优化版)"""
print("🤖 正在初始化语言模型...")
return Ollama(
model=config.LLM_MODEL,
temperature=config.LLM_TEMPERATURE,
base_url=config.OLLAMA_BASE_URL,
# 新增优化参数
num_predict=config.LLM_NUM_PREDICT, # 限制生成长度
num_ctx=config.LLM_NUM_CTX, # 上下文窗口
num_thread=config.LLM_NUM_THREADS # CPU线程数
)
LLM_NUM_PREDICT = 256:限制模型最多生成 256个token,约180汉字,生成越短,速度越快。
LLM_NUM_CTX = 2048:限制模型的“记忆长度”,默认很大,吃内存和CPU,这里减小电脑负担。
LLM_NUM_THREADS = 4:CPU 利用率拉满。
但是返回的结果不尽如人意,因为 0.5b 模型已足够小,Ollama 已经做了最好的配置,此方法失败。
第三:减少文档块大小
# --- 新增:文档分割器配置 ---
CHUNK_SIZE = 300 # 每个文本块的目标大小(字符数)
CHUNK_OVERLAP = 30 # 相邻块之间的重叠字符数,用于保持上下文连贯
RETRIEVER_K = 1 # 从2减到1,只返回最相关的1个块
缩短文档块的自述,按理说应该能提高效率,但是看返回结果的时间,仍然没有更大的进步,此方法失败。
第四:极简版 prompt
STRICT_PROMPT_TEMPLATE = """上下文:{context}
问题:{question}
回答(只基于上下文):"""
缩短描述信息,但是响应时间,仍然没有更好的优化,此方法失败。
第五:更换更快的模型 phi3:mini
反而时间更长,又回到了类似 qwen2.5:3b 模型的返回时间,此方法失败。
第六:使用 GPU 加速(如果有显卡)
输入命令:ollama ps
如下所示,当前我的计算机并没有独立显卡,无法采用这个优化方法。
第七:架构层面优化(添加答案缓存)
在 src 文件夹下, 新增 cache.py 文件,下面代码写入。
import hashlib
import json
import os
class AnswerCache:
def __init__(self, cache_dir="./cache"):
self.cache_dir = cache_dir
os.makedirs(cache_dir, exist_ok=True)
def _key(self, question):
return hashlib.md5(question.encode()).hexdigest()
def get(self, question):
path = os.path.join(self.cache_dir, f"{self._key(question)}.json")
if os.path.exists(path):
with open(path, 'r', encoding='utf-8') as f:
return json.load(f).get('answer')
return None
def set(self, question, answer):
path = os.path.join(self.cache_dir, f"{self._key(question)}.json")
# ensure_ascii=False 让中文直接显示,indent=2 让格式美观
with open(path, 'w', encoding='utf-8') as f:
json.dump(
{'question': question, 'answer': answer},
f,
ensure_ascii=False, # 关键:不转义Unicode
indent=2 # 格式化,便于阅读
)
在 main.py 中调用
cache = AnswerCache()
for q in questions:
print(f"{'=' * 60}")
print(f"❓ 问题: {q}")
print("\n📊 检索到的文档块:")
try:
# 使用混合检索器打印,与LLM看到的保持一致
docs = debug_retriever.invoke(q)
if not docs:
print(" ⚠️ 没有检索到任何文档块!")
for i, doc in enumerate(docs):
preview = doc.page_content.replace('\n', ' ')[:300]
print(f" {i + 1}. {preview}...")
except Exception as e:
print(f" 检索失败: {e}")
start_time = time.time()
# 先查缓存
cached_answer = cache.get(q)
if cached_answer:
if isinstance(cached_answer, dict):
result = cached_answer.get('result', str(cached_answer))
else:
result = str(cached_answer)
if result and result != "未找到相关信息":
print(f"\n📦 缓存命中: {result}")
elapsed_time = time.time() - start_time
print(f"⏱️ 耗时: {elapsed_time:.2f} 秒")
continue
else:
print("\n🗑️ 缓存内容无效,重新查询...")
# 正常问答
answer = qa_chain.invoke(q)
result = answer['result'] if isinstance(answer, dict) else answer
elapsed_time = time.time() - start_time
print(f"\n💡 回答: {result}")
print(f"⏱️ 耗时: {elapsed_time:.2f} 秒")
# 只缓存有效答案
if result and result != "未找到相关信息":
cache.set(q, answer) # 需要时取消注释
print("✅ 已存入缓存")
十二、优化检索精度
检索的精准度,取决于算法和模型,打个比方:
- 检索算法 = 图书管理员的找书能力(知道书在哪个书架,哪一层,哪一本)决定性因素
- LLM模型 = 读者的理解能力(从找到的书中找到答案)
- 检索决定了LLM能看到什么信息,是上限
- 而LLM的理解能力决定了能从这些信息中提取到什么,是下限。两者同样重要
如果在检索环节,不能把真正相关联的文档块找出来,LLM再强大也无济于事,本文接着详细说明如何优化检索精度。
第一步:混合检索(Hybird Search)
- 向量检索(语义):将文本映射到高纬向量空间。理解同义词和语义相近的内容。但是对稀有关键词不敏感,如【束腰】、【铁鞋】等。
- BM25(关键词):精确匹配专有名词和罕见词。但是无法理解语义。
本文将它们二者结合
class SimpleHybridRetriever(BaseRetriever):
def __init__(self, vectorstore, documents):
self.vectorstore = vectorstore
# 从原始文档构建 BM25 检索器
self.bm25 = BM25Retriever.from_documents(documents)
self.bm25.k = 15 # BM25 召回 15 个
def _get_relevant_documents(self, query: str):
# 1. 向量检索:召回 10 个语义相关文档
vector_docs = self.vectorstore.similarity_search(query, k=10)
# 2. BM25 检索:召回 15 个关键词匹配文档
bm25_docs = self.bm25.invoke(query)
# 3. 合并去重,向量结果优先(因为语义相关性更高)
seen = set()
merged = []
for doc in vector_docs + bm25_docs:
if doc.page_content not in seen:
seen.add(doc.page_content)
merged.append(doc)
# 4. 返回前 5 个给 LLM
return merged[:5]
第二步:参数调优
从 config.py 中配置可调节的参数
HYBRID_RETRIEVER_VECTOR_K = 10 # 向量检索召回数(多一些,避免遗漏)
HYBRID_RETRIEVER_BM25_K = 15 # BM25 召回数(多一些,补充关键词)
HYBRID_RETRIEVER_FINAL_K = 5 # 最终返回给 LLM 的块数
经过测试,返回给 LLM 5个信息块,在信息覆盖和长度限制之间比较平衡,作为本地项目来说,效果很好。
第三步:智能文档切割
原始文档是一篇近 2 万字的完整故事。如果整篇喂给 LLM,会超出上下文窗口(4096 token)。如果切得太碎,关键信息可能被切断。
# 2、创建递归文本分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=config.CHUNK_SIZE,
chunk_overlap=config.CHUNK_OVERLAP,
separators=config.SEPARATORS,
length_function=len,
add_start_index=True
)
分割优先级:
- 段落分隔符:\n\n
- 不符合就按照换行:\n
- 还不符合就按照句子结束符:。! ?
- 最后按照词语或字符
重叠机制:50 字符的重叠确保被切断的句子,在相邻块中也有完整的信息。
第四步:Prompt 工程引导检索
PROMPT_TEMPLATE = """根据以下上下文回答问题。
【上下文】
{context}
【问题】
{question}
【回答要求】
1. 基于上下文信息回答,可以进行合理的总结和推断
2. 如果上下文有线索,即使没有直接说明,也可以推断
3. 注意区分不同角色:国王、王子、皇后、白雪公主是不同的,不要混淆
4. 回答要简洁准确
5. 如果上下文明确描述了某个角色的死亡方式和过程,请如实回答
6. 如果上下文中提到"力竭而倒",说明该角色已经死亡
回答:"""
隐式的检索质量兜底机制:
- 要求 1-2 允许 LLM 在信息不完整时进行推断,避免直接返回【不知道】
- 要求 3 防止角色混淆
- 要求 5-6 是对特定问题的精准指令
十三、测试结果总结
第一个问题:白雪公主的相貌。
第二个问题:白雪公主肌肤的颜色。
第三个问题:皇后为什么要伤害白雪公主?
第四个问题:七个小矮人是做什么工作的?
第五个问题:王子有什么特殊的癖好?
第六个问题: 国王喜欢哪个人,白雪公主、皇后、年轻女孩等等,你判断一下?
第七个问题:皇后用了什么手段、方法、策略来伤害白雪公主?
我们统计一下结果,准确率评估:
- 完全正确:5个
- 部分正确(遗漏细节):2个,例如国王喜欢谁,报仇是王子允许的,与国王无关,LLM产生了幻觉;皇后手段回答,遗漏了【束腰】、【毒苹果】
- 错误:0个
- 结论:核心事实准确率100%,细节完整率70%,后续可通过优化检索提升细节完整度。
准确率详细评估(基于7个基本问题)
| 问题 | 预期答案要点 | 实际回答 | 是否正确 | 备注 |
|---|---|---|---|---|
| 白雪公主的相貌 | 肌肤洁白如雪、嘴唇红润如血、头发黑如檀木 | 完全相同 | 正确 | |
| 白雪公主的肌肤颜色 | 白色 | 白色 | 正确 | |
| 皇后为什么要伤害白雪公主 | 嫉妒、威胁地位、派猎人杀害 | 完整覆盖 | 正确 | |
| 七个小矮人是做什么工作的 | 挖掘铜矿、打造成武器 | 完整覆盖 | 正确 | |
| 王子有什么特殊的癖好 | 恋尸癖 | 恋尸癖 | 正确 | |
| 国王喜欢谁 | 白雪公主 | 白雪公主 | 部分幻觉 | |
| 皇后用了什么手段伤害白雪公主 | 毒药、猎人、吃心肝、毒梳子、毒苹果、束腰、铁鞋 | 列出了4种(毒药、猎人、吃心肝、再次使用毒药) | 部分正确 | 遗漏 |
完全正确:5个
部分正确:2个(国王喜欢谁,回答包含了王子相关内容,属于过度扩展,与本人无关)、皇后手段(遗漏束腰、毒苹果)。
完全错误:0个
十四、后续优化方向
1、重排序(Re-ranking)
现状问题:本文虽然使用了混合检索,但排序只是简单的“向量结果在前”,不一定最相关的排列在最前面。
优化方案:引入其他模型对召回的文档进行重新打分,计算更精确的相关性分数。
2、问题重写
现状问题:用户提交的问题可能表达不清晰或者过于简短。
优化方案:用 LLM 先将用户问题改写为更规范的检索语句。
3、检索融合
增加更多检索源,目前还不知道采用哪个。
十五、项目运行说明
如果本地运行本项目,需按照以下步骤操作
# 1. 安装Ollama并拉取模型
ollama pull qwen2.5:0.5b # 轻量级模型(推荐,速度快)
ollama pull bge-m3 # 向量化模型
# 2. 安装Python依赖
pip install langchain langchain-community chromadb sentence-transformers
# 3. 运行项目
python main.py
环境要求:Python 3.9+,内存 8G或以上,如使用7b以上模型,建议更高内存。
to be continued。。。