个人随笔
目录
Agent学习(4):一个简单的RAG例子
2026-01-20 23:35:38

1. 大模型应用开发的三种模式

Prompt 模式

适用场景：轻量需求、快速验证，无需额外数据和训练。比如通用问答、文案生成、简单指令执行（如文本摘要、格式转换），适合对结果个性化要求低、调用频率不高的场景，直接通过精准提示词驱动基础大模型输出结果。

RAG 模式

适用场景：需要外部知识库支撑的场景，比如企业内部文档问答、产品手册咨询、实时信息查询（如新闻、财报解读）。核心解决大模型 “知识过时”“信息不准确” 的问题，通过检索外部文档并拼接上下文给模型，无需训练，兼顾灵活性和准确性，适合数据频繁更新的场景。

Fine-tuning 模式

适用场景：高个性化、高适配性需求，比如垂直领域专属模型（如医疗诊断辅助、法律文书生成）、特定话术风格定制（如品牌客服话术）、复杂任务逻辑对齐。需要标注高质量数据集进行模型微调，开发成本和门槛较高，适合对模型输出稳定性、专业性要求极高的规模化场景。

2. 什么是 RAG ？

RAG（Retrieval-Augmented Generation）

  • 检索增强生成，是一种结合信息检索（Retrieval）和文本生成（Generation）的技术
  • RAG技术通过实时检索相关文档或信息，并将其作为上下文输入到生成模型中，从而提高生成结果的时效性和准确性。

2.1. RAG 的优势是什么？

  • 解决知识时效性问题：大模型的训练数据通常是静态的，无法涵盖最新信息，而RAG可以检索外部知识库实时更新信息

  • 减少模型幻觉：通过引入外部知识，RAG能够减少模型生成虚假或不准确内容的可能性

  • 提升专业领域回答质量：RAG能够结合垂直领域的专业知识库，生成更具专业深度的回答

  • 生成内容的溯源（可解释性）

2.2. RAG 的核心原理与流程

RAG（检索增强生成）的核心原理是 “先检索外部知识库的相关信息，再将检索结果与用户问题拼接成提示词，交给大模型生成回答”，以此解决大模型 “知识过时、信息不准确、幻觉” 的核心痛点，同时避免了 Fine-tuning 的高成本。
其完整流程分为 4 个核心步骤，可分为离线构建阶段和在线查询阶段：

离线构建知识库（一次构建，重复使用）

文档拆分：将原始文档（如 PDF、Word、网页）切割成大小合适的文本块（Chunk），避免因文本过长超出模型上下文窗口。
嵌入向量化：用嵌入模型（如 text-embedding-v4）将每个文本块转换成向量（Embedding），向量能表征文本的语义信息。
向量入库：将所有文本块的向量和对应的原始内容、元数据（如页码、来源）存入向量数据库（如 FAISS、Milvus）。

在线查询（用户交互阶段）

用户问题向量化：将用户的查询问题用同一嵌入模型转换成向量。
相似度检索：在向量数据库中检索与问题向量最相似的 Top-N 个文本块，这些文本块就是回答问题的核心依据。
Prompt 拼接：把检索到的文本块、用户问题、系统指令（如 “仅基于给定文档回答”）拼接成完整的提示词。
模型生成回答：将拼接好的提示词输入大模型，模型基于提示词中的文档信息生成准确、无幻觉的回答。

划重点：RAG 本质上就是重构了一个新的 Prompt！

4. LangChain快速搭建本地知识库检索

4.1.先安装指定版本的依赖库

这里必须指定版本，不然会报各种错误

  1. pip install pypdf2
  2. pip install dashscope
  3. pip install langchain==0.0.286
  4. pip install langchain-community==0.0.9
  5. pip install openai==0.28.1
  6. pip install faiss-cpu

我这里用的是比较旧的版本，但是这个版本是可以跑的

4.2.代码如下

  1. import os
  2. import logging
  3. import pickle
  4. from PyPDF2 import PdfReader
  5. from langchain.chains.question_answering import load_qa_chain
  6. from langchain.chat_models import ChatOpenAI as OldChatOpenAI
  7. from langchain_community.embeddings import DashScopeEmbeddings
  8. from langchain_community.callbacks.manager import get_openai_callback
  9. from langchain.text_splitter import RecursiveCharacterTextSplitter
  10. from langchain_community.vectorstores import FAISS
  11. from typing import List, Tuple
  13. def read_and_split_pdf(pdf_path):
  14.     # 初始化文本分割器（保持你的原有配置）
  15.     text_splitter = RecursiveCharacterTextSplitter(
  16.         separators=["\n\n", "\n", ".", " ", ""],
  17.         chunk_size=512,
  18.         chunk_overlap=128,
  19.         length_function=len,
  20.     )
  21.     """
  22.     读取PDF并分割文本，返回：
  23.     - chunks：分割后的文本块列表
  24.     - chunk_page_numbers：每个文本块对应的页码列表
  25.     """
  26.     # 初始化PDF阅读器
  27.     pdf = PdfReader(pdf_path)
  28.     chunks = []  # 存储分割后的所有文本块
  29.     chunk_page_numbers = []  # 存储每个文本块对应的页码
  31.     # 遍历每一页PDF（保留页码）
  32.     for page_number, page in enumerate(pdf.pages, start=1):
  33.         extracted_text = page.extract_text()
  34.         if not extracted_text:
  35.             logging.warning(f"No text found on page {page_number}.")
  36.             continue
  38.         # 关键：对当前页的文本单独分割（而非拼接后整体分割）
  39.         page_chunks = text_splitter.split_text(extracted_text)
  41.         # 为当前页的每个分割块绑定页码
  42.         chunks.extend(page_chunks)
  43.         chunk_page_numbers.extend([page_number] * len(page_chunks))
  45.     return chunks, chunk_page_numbers
  46. def create_knowledge_base(chunks, chunk_page_numbers, save_path) -> FAISS:
  47.     """
  48.     处理文本并创建向量存储 
  49.     参数:
  50.         chunks：分割后的文本块列表
  51.         chunk_page_numbers：每个文本块对应的页码列表
  52.         save_path: 可选，保存向量数据库的路径
  53.     """
  54.     # 过滤空文本块（避免嵌入接口报错）
  55.     valid_data = [(chunk, num) for chunk, num in zip(chunks, chunk_page_numbers) if chunk.strip()]
  56.     valid_chunks = [item[0] for item in valid_data]
  57.     valid_page_nums = [item[1] for item in valid_data]
  59.     if not valid_chunks:
  60.         raise ValueError("没有有效文本块可创建向量库，请检查PDF文本提取结果")
  62.     # 打印有效文本块数量
  63.     print(f"过滤空文本后，有效文本块数量: {len(valid_chunks)}")
  65.     # 创建嵌入模型
  66.     # 调用阿里百炼平台文本嵌入模型，配置环境变量 DASHSCOPE_API_KEY
  67.     embeddings = DashScopeEmbeddings(
  68.         model = "text-embedding-v4"
  69.     )
  71.     print("从文本块创建知识库（分批处理）...")
  72.     # 关键修复：分批处理文本块（DashScope v4 批量限制≤10）
  73.     batch_size = 10
  74.     knowledgeBase = None
  76.     # 按批次遍历文本块
  77.     for i in range(0, len(valid_chunks), batch_size):
  78.         # 截取当前批次的文本和页码
  79.         batch_chunks = valid_chunks[i:i+batch_size]
  80.         batch_page_nums = valid_page_nums[i:i+batch_size]
  82.         # 为当前批次创建向量库
  83.         batch_db = FAISS.from_texts(batch_chunks, embeddings)
  85.         # 合并批次到总向量库
  86.         if knowledgeBase is None:
  87.             knowledgeBase = batch_db
  88.         else:
  89.             knowledgeBase.merge_from(batch_db)
  91.         print(f"已处理第 {i//batch_size + 1} 批，共处理 {min(i+batch_size, len(valid_chunks))}/{len(valid_chunks)} 个文本块")
  93.     print("已从文本块创建完整知识库...")
  95.     # 存储每个文本块对应的页码信息（仅保留有效文本块）
  96.     page_info = {chunk: valid_page_nums[i] for i, chunk in enumerate(valid_chunks)}
  97.     knowledgeBase.page_info = page_info
  99.     # 如果提供了保存路径，则保存向量数据库和页码信息
  100.     if save_path:
  101.         # 确保目录存在
  102.         os.makedirs(save_path, exist_ok=True)
  104.         # 保存FAISS向量数据库
  105.         knowledgeBase.save_local(save_path)
  106.         print(f"向量数据库已保存到: {save_path}")
  108.         # 保存页码信息到同一目录
  109.         with open(os.path.join(save_path, "page_info.pkl"), "wb") as f:
  110.             pickle.dump(page_info, f)
  111.         print(f"页码信息已保存到: {os.path.join(save_path, 'page_info.pkl')}")
  113.     return knowledgeBase
  115. def load_knowledge_base(load_path: str, embeddings = None) -> FAISS:
  116.     """
  117.     从磁盘加载向量数据库和页码信息
  119.     参数:
  120.         load_path: 向量数据库的保存路径
  121.         embeddings: 可选，嵌入模型。如果为None，将创建一个新的DashScopeEmbeddings实例
  123.     返回:
  124.         knowledgeBase: 加载的FAISS向量数据库对象
  125.     """
  126.     # 如果没有提供嵌入模型，则创建一个新的
  127.     if embeddings is None:
  128.         embeddings = DashScopeEmbeddings(
  129.             model="text-embedding-v4"
  130.         )
  132.     # 加载FAISS向量数据库，添加allow_dangerous_deserialization=True参数以允许反序列化
  133.     #langchain==0.0.286 版本不支持 allow_dangerous_deserialization 参数
  134.     knowledgeBase = FAISS.load_local(load_path, embeddings)
  135.     print(f"向量数据库已从 {load_path} 加载。")
  137.     # 加载页码信息
  138.     page_info_path = os.path.join(load_path, "page_info.pkl")
  139.     if os.path.exists(page_info_path):
  140.         with open(page_info_path, "rb") as f:
  141.             page_info = pickle.load(f)
  142.         knowledgeBase.page_info = page_info
  143.         print("页码信息已加载。")
  144.     else:
  145.         print("警告: 未找到页码信息文件。")
  147.     return knowledgeBase
  149. def queryInfo(query:str,knowledgeBase):
  150.     # 设置查询问题
  151.     # 执行相似度搜索，找到与查询相关的文档
  152.     docs = knowledgeBase.similarity_search(query)
  153.     # 2. 初始化 ChatOpenAI 客户端（兼容百炼）
  154.     # 初始化对话大模型（适配 langchain==0.0.286 的旧版参数）
  155.     chatLLM = OldChatOpenAI(
  156.         openai_api_key=os.getenv("DASHSCOPE_API_KEY"),  # 旧版参数名
  157.         openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 旧版参数名
  158.         model_name="deepseek-v3",  # 旧版参数名（关键！不能写 model）
  159.         temperature=0,#建议设为0，保证回答稳定
  160.         max_tokens=2048,
  161.         # 补充旧版必要参数：确保实例完整
  162.         request_timeout=60,
  163.         n=1
  164.     )
  166.     # 加载问答链
  167.     chain = load_qa_chain(chatLLM, chain_type="stuff")
  169.     # 准备输入数据
  170.     input_data = {"input_documents": docs, "question": query}
  172.     # 执行问答链并捕获原始响应
  173.     response = chain.run(**input_data) 
  174.     print("回答：", response)
  175.     print("来源:")
  177.     # 记录唯一的页码
  178.     unique_pages = set()
  180.     # 显示每个文档块的来源页码
  181.     for doc in docs:
  182.         text_content = getattr(doc, "page_content", "")
  183.         source_page = knowledgeBase.page_info.get(
  184.             text_content.strip(), "未知"
  185.         )
  186.         if source_page not in unique_pages:
  187.             unique_pages.add(source_page)
  188.             print(f"文本块页码: {source_page}")
  193. # 程序入口：只有直接运行这个文件时，才执行游戏
  194. if __name__ == "__main__":
  196.     """
  197.     #先创建知识库
  198.     chunks, chunk_page_numbers = read_and_split_pdf("G:\\AI\\workspace\\vscode\\202601\\123.pdf")
  199.     print(chunks)
  200.     print()
  201.     print(chunk_page_numbers)
  202.     create_knowledge_base(chunks,chunk_page_numbers,"./vector_db");
  204.     """
  205.     #查询知识库
  206.     knowledgeBase = load_knowledge_base("./vector_db")
  207.     query = "客户经理被投诉了，投诉一次扣多少分"
  208.     queryInfo(query,knowledgeBase);
  209.     print()
  210.     query = "总结下可越级聘用的情况"
  211.     queryInfo(query,knowledgeBase);

上面的代码刚好对应了RAG 的流程

  1. read_and_split_pdf：拆分文档
  2. create_knowledge_base：向量化和入库
  3. load_knowledge_base：查询知识库
  4. docs = knowledgeBase.similarity_search(query)：用户问题向量化和相似度检索
  5. chain = load_qa_chain(chatLLM, chain_type="stuff")：prompt拼接
  6. response = chain.run(**input_data) ：模型生成回答
  7. 最后的页面还溯源了
 10

啊！这个可能是世界上最丑的留言输入框功能~


当然，也是最丑的留言列表

有疑问发邮件到 : suibibk@qq.com 侵权立删
Copyright : 个人随笔   备案号 : 粤ICP备18099399号-2