目录[-]

一、引言:测试工作中的痛点

 

在软件测试的日常工作中,经常遇到以下场景:

  1. 测试A发现了一个疑似重复的 Bug ,但是他忘记以前有没有提过。
  2. 新同事入职,想了解某个模块的历史 Bug 都有哪些。
  3. 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系统,实现了:

  1. RAG 完整流程:从文档加载到生成答案
  2. 核心技术组件:向量化、检索、LLM、Prompt
  3. 实战经验:检索策略设计、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)

  1. 向量检索(语义):将文本映射到高纬向量空间。理解同义词和语义相近的内容。但是对稀有关键词不敏感,如【束腰】、【铁鞋】等。
  2. 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
    )

分割优先级:

  1. 段落分隔符:\n\n
  2. 不符合就按照换行:\n
  3. 还不符合就按照句子结束符:。! ?
  4. 最后按照词语或字符

重叠机制: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。。。