Fork me on GitHub

基于LLM与RAG的安全知识问答系统开发笔记

字数统计: 2.7k字   |   阅读时长≈ 10分

RAG是什么

RAG 是 Retrieval-Augmented Generation（检索增强生成） 的缩写，指的是一种把「外部知识检索」和「大模型生成」结合起来的技术：在模型回答问题前，先从数据库、文档或向量库中检索出最相关的资料，再基于这些真实资料来生成回答，这样可以减少“瞎编”、提高准确性，还能让模型使用最新或私有数据，因此常用于知识库问答、企业内部文档助手、论文/代码辅助等场景。

RAG全流程

RAG全流程

alt text

1、准备知识库

收集资料：收集与安全知识相关的文档、手册、FAQ等资料。

  • 1️⃣ 安全操作规程
    起重机安全操作规程、吊装作业规范、设备检查制度

  • 2️⃣ 事故案例
    起重机倾覆事故、违章操作事故、典型事故分析报告

  • 3️⃣ 管理制度
    安全生产责任制、作业许可制度、人员培训制度

  • 4️⃣ 应急处置
    起重机事故应急预案、人员伤害处置流程、设备故障应急方案

2、文本预处理

在将原始文档导入数据库之前，需要对文本进行清洗和标准化处理，以提高检索和生成的准确性。这包括：

  • 剔除页码、页眉页脚等无关信息
  • 修复断句问题（如将非标点结尾的换行符替换为空格）
  • 规范化空白字符（如多个空格合并为一个）
  • 保留关键术语和结构信息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import pdfplumber
import re
import os

class BatchPDFProcessor:
    def __init__(self, output_file):
        self.output_file = output_file
        # 预清理：如果输出文件已存在，先删除
        if os.path.exists(self.output_file):
            os.remove(self.output_file)

    def clean_text(self, text):
        """针对起重机规程优化的清洗逻辑"""
        # 剔除页码噪声
        text = re.sub(r'第\s*\d+\s*页', '', text)
        text = re.sub(r'-\s*\d+\s*-', '', text)
        # 修复断句：将非标点结尾的换行符替换为空格
        text = re.sub(r'(?<![。！？；…：])\n', '', text)
        # 规范化空白
        text = re.sub(r' +', ' ', text)
        text = re.sub(r'\n+', '\n', text)
        return text.strip()

    def process_files(self, file_paths):
        with open(self.output_file, 'a', encoding='utf-8') as f_out:
            for path in file_paths:
                if not os.path.exists(path):
                    print(f"跳过：找不到文件 {path}")
                    continue
                
                print(f"正在提取: {os.path.basename(path)} ...")
                file_text = f"\n\n--- 来源: {os.path.basename(path)} ---\n"
                
                try:
                    with pdfplumber.open(path) as pdf:
                        for page in pdf.pages:
                            raw_text = page.extract_text()
                            if raw_text:
                                file_text += self.clean_text(raw_text) + "\n"
                    
                    f_out.write(file_text)
                except Exception as e:
                    print(f"处理 {path} 时出错: {e}")

        print("-" * 30)
        print(f"全部处理完成！汇总知识库已保存至: {self.output_file}")

if __name__ == "__main__":
    # 你的 8 个文件路径列表
    pdf_list = [
        r"D:\资料\毕业设计\代码\RAG_project\data\安全规程\《起重机械安全技术规程》.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\国家标准与规程\6067_1-2010-gb-e-300.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\国家标准与规程\GBT 28264-2017 起重机械 安全监控管理系统.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\国家标准与规程\gbt14405-2011通用桥式起重机标准.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\事故案例\9839300.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\事故案例\P020210722541036839169.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\事故案例\P020230129571281983048.pdf",
        r"D:\资料\毕业设计\代码\RAG_project\data\事故案例\P020241018549892886455.pdf"
    ]

    processor = BatchPDFProcessor("total_crane_knowledge.txt")
    processor.process_files(pdf_list)

3、文本切分

将清洗后的长文本切分成更小的片段，以便后续的嵌入和检索。常用的方法有基于句子、段落或固定长度的滑动窗口切分。以下是一个示例代码，展示如何使用LangChain进行文本切分：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain_text_splitters import RecursiveCharacterTextSplitter

def perform_chunking(file_path):
    # 1. 读取汇总后的纯文本
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()

    # 2. 初始化切分器
    # chunk_size: 每个块的最大字符数
    # chunk_overlap: 相邻块之间的重叠字数（保证上下文不丢失）
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=600,       # 建议 500-800 字，适合起重机条款长度
        chunk_overlap=100,    # 10%~20% 的重叠度
        length_function=len,
        separators=["\n\n", "\n", "。", "！", "？", " ", ""]
    )

    # 3. 执行切分
    chunks = text_splitter.create_documents([content])
    
    print(f"原始文件长度: {len(content)} 字符")
    print(f"切分后的块数: {len(chunks)}")
    
    # 4. 预览前两个块的内容
    for i, chunk in enumerate(chunks[:2]):
        print(f"\n--- 第 {i+1} 个知识块内容 ---")
        print(chunk.page_content)
        print("-" * 30)
    
    return chunks

if __name__ == "__main__":
    txt_path = "total_crane_knowledge.txt"
    all_chunks = perform_chunking(txt_path)

以上的代码展示了文本切分的基本流程。使用RecursiveCharacterTextSplitter可以根据指定的块大小和重叠度，将长文本切分成多个知识块，方便后续的嵌入和检索。可以根据实际需求调整 chunk_size 和 chunk_overlap 参数，以获得最佳的切分效果。

4、向量化与存储

将切分后的文本块转换为向量表示，并存储在向量数据库中，方便后续的相似度检索。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import os
from tqdm import tqdm
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from chunk import perform_chunking  # 确保你的 chunk.py 在同目录下

def start_ingest_with_progress():
    # 1. 加载并切分
    txt_path = "total_crane_knowledge.txt"
    documents = perform_chunking(txt_path)
    total_docs = len(documents)

    # 2. 初始化 Embedding 模型
    print("\n[1/3] 正在加载 Embedding 模型 (首次运行需下载)...")
    model_name = "shibing624/text2vec-base-chinese"
    embeddings = HuggingFaceEmbeddings(model_name=model_name)

    # 3. 创建并分批次存入 ChromaDB
    persist_directory = "./chroma_db"
    print(f"[2/3] 正在构建向量数据库，目标位置: {persist_directory}")

    # 先用第一条数据初始化数据库（为了获取正确的 Embedding 维度）
    vector_db = Chroma.from_documents(
        documents=[documents[0]],
        embedding=embeddings,
        persist_directory=persist_directory
    )

    # 剩余数据分批次压入，并显示进度条
    # 建议批次大小为 50-100
    batch_size = 50
    print(f"[3/3] 正在处理并存储知识块...")
    
    # 使用 tqdm 创建进度条
    for i in tqdm(range(1, total_docs, batch_size), desc="入库进度", unit="批次"):
        batch = documents[i : i + batch_size]
        vector_db.add_documents(batch)

    print("-" * 30)
    print(f"✅ 入库成功！共处理 {total_docs} 个知识块。")
    print(f"📂 数据库文件夹: {os.path.abspath(persist_directory)}")

if __name__ == "__main__":
    start_ingest_with_progress()

使用HuggingFaceEmbeddings将文本块转换为向量表示，并使用Chroma向量数据库进行存储。代码中包含了进度条显示，方便监控入库过程。HuggingFaceEmbeddings模型将文本转化为向量的维度为768。

1. Embedding 模型：shibing624/text2vec-base-chinese

这是一个专门针对中文语义优化的开源模型，基于 BERT 架构进行改进.

  • 向量维度：768 维。 这意味着它会将你每一段 500 字左右的起重机规程，编码成一个包含 768 个浮点数的数学列表。这 768 个维度分别代表了文本在语义空间中的不同特征。
  • 核心原理：它通过在大规模中文语料（百科、新闻、论文）上进行预训练，学习到了词与词、句与句之间的“亲密度”。
    • 例如：它能识别出“吊钩”和“索具”在空间坐标上非常接近，而与“冰激凌”相距甚远。
  • 为什么选它？
    • 中文支持极佳：比很多多语言模型更能理解“起重机”、“溜钩”、“下挠”等专业词汇。
    • 性能平衡：768 维是目前工业界的“黄金维度”，既保留了足够的语义精度，计算速度又比 1024 或 1536 维的模型快得多。

2. 向量数据库：ChromaDB

ChromaDB 是目前 AI 开发社区最流行的开源嵌入式向量数据库。你可以把它想象成一个专门存储“坐标”和“标签”的高级 Excel。
核心特性：

  • 存储机制：它属于持久化本地存储。在你的项目目录下生成的 chroma_db 文件夹，实际上包含了向量数据索引和 SQLite 数据库（用于存储原始文本和元数据）。
  • 检索算法 (HNSW)： 它使用的是 HNSW (Hierarchical Navigable Small World) 算法。这是一种“小世界”图算法。
    • 形象比喻：它不像传统的数据库那样一行行去查，而是像在朋友圈里找人一样，通过层层关联的“节点”快速定位到最接近的向量。
  • 元数据过滤 (Metadata Filtering)： 我们在代码里存入的 source（文件名）就存在这里。这让你不仅能搜到内容，还能瞬间知道它出自哪本 PDF。
组件名称 具体选型 关键参数 作用
Embedding 模型 text2vec-base-chinese 向量维度：768 将规程文本转化为数学向量
向量数据库 ChromaDB 算法：HNSW
距离度量：L2		 存储向量并实现快速语义检索
文本处理 LangChain Chunk Size：600
Overlap：100		 保证知识块语义完整

5、构建问答系统

利用向量数据库实现基于内容的检索，并结合大语言模型生成回答。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import os
from google import genai
from dotenv import load_dotenv
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 加载配置
load_dotenv()
client = genai.Client() # 自动读取环境变量 GEMINI_API_KEY

# 2. 初始化检索器 (与你之前的 ingest 保持一致)
print("正在连接知识库...")
embeddings = HuggingFaceEmbeddings(model_name="shibing624/text2vec-base-chinese")
vector_db = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

def get_answer_from_gemini(query):
    # --- 检索阶段 ---
    # 搜索最相关的 3 条知识
    docs = vector_db.similarity_search(query, k=3)
    
    # 组合上下文
    context_list = []
    for d in docs:
        source = d.metadata.get('source', '未知文件')
        context_list.append(f"【参考来源: {source}】\n内容: {d.page_content}")
    
    context_text = "\n\n".join(context_list)

    # --- 生成阶段 ---
    # 构建适合起重机安全场景的系统提示词
    prompt = f"""你是一名专业的起重机安全工程师。请根据提供的规程和案例回答用户问题。
要求：
1. 回答必须严谨，基于参考资料。
2. 资料中如果没有，请回答“知识库暂无记录”，不要误导用户。
3. 必须在回答中指明你参考了哪个文件。

【参考资料】:
{context_text}

【用户问题】:
{query}
"""

    # 调用你习惯的 client 写法
    response = client.models.generate_content(
        model="gemini-2.5-flash", 
        contents=prompt
    )
    
    return response.text, docs

# --- 运行测试 ---
if __name__ == "__main__":
    question = "电气系统的设计计算或者选型计算，主要包括哪些？"
    print(f"\n用户提问: {question}")
    
    answer, source_docs = get_answer_from_gemini(question)
    
    print("\n" + "="*20 + " AI 安全建议 " + "="*20)
    print(answer)

使用Gemini 2.5 Flash模型进行回答生成。系统提示词中明确要求模型基于检索到的起重机规程和案例进行回答，避免“瞎编”。同时，回答中会附上参考的文件来源，提升可信度。
新建.env文件，添加环境变量。在.env文件中写入：GEMINI_API_KEY=你的API密钥
将用户的问题使用相同的embedding模型转换为向量，并在ChromaDB向量数据库中进行相似度搜索，找到最相关的3个知识块作为上下文。相似度搜索使用的是L2距离度量。

L2距离度量:

公式中参数解释：

  • (d)：表示查询向量 (q) 和存储向量 (v) 之间的 L2 距离。
  • (n)：表示向量的维度（在本例中为768维）。
  • (q_i)：表示查询向量 (q) 在第 (i) 个维度的值。
  • (v_i)：表示存储向量 (v) 在第 (i) 个维度的值。
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

一个努力学习，积极向上的程序员。