1.大模型应用开发的三种模式
- 模型构建阶段： 通过海量知识的预训练（Pre-training），将原始数据转化为具备理解能力的 LLM。
- 应用交互阶段： LLM 被封装成具备各种能力的 AI 应用，用户与其进行交互。
- 错误修正阶段（核心）： 当 AI 给出错误回复时，通过三条路径进行优化： Prompt（提示词工程）： 解决“问题没问清楚”的情况。通过优化指令，引导模型输出正确答案。 RAG（检索增强生成）： 解决“缺乏背景知识”的情况。给模型外接一个知识库（如企业私有文档），让它“看书考试”。 Fine-tuning（微调）： 解决“模型能力不足”的情况。通过特定领域的标注数据，深度改变模型的行为逻辑或专业知识。
2.什么是RAG
RAG（Retrieval-Augmented Generation） 的核心思想是：将大模型的生成能力与外部知识库的检索能力相结合，形成“先查资料，再作答”的智能行为。
- 知识的“入库”阶段
专业知识 -> 向量数据库：第一，将海量的非结构化数据（如 PDF、文档、代码等）进行分段，并利用 Embedding 模型将其转化为数学向量，存储在向量数据库中。这一步相当于为 AI 准备了一本可以快速翻阅的“百科全书”。
- 问题的“检索”阶段
问题 -> 向量数据库：当用户提出一个问题时，系统不会直接把问题丢给大模型，而是先去向量数据库中进行“语义搜索”。系统会寻找与这个问题意思最相近的专业知识片段。
- 信息的“合成”阶段
向量数据库 ->大模型（Prompt）：系统将检索到的相关知识与原始问题拼接在一起，组成一个内容更丰富的 Prompt（提示词） 发送给大模型。 这时，大模型不再是凭空想象，而是基于你提供的“参考资料”进行回答。
- 答案的“生成”阶段
大模型 -> 答案：大模型通过对提示词内容的理解和总结，生成最终的答案。
**检索（Retrieval）:**当用户提出问题时，系统使用上一章学习的 Embedding 模型 将问题转化为向量，然后在 向量数据库 中进行类似度搜索，找出与问题最相关的文档片段。
**增强（Augmentation）：**将检索到的相关文档内容（原文）作为上下文，拼接到 Prompt 中，形成一个“增强版”的输入。这相当于告知模型：“请参考以下信息来回答问题。”
**生成（Generation）：**大模型基于增强后的 Prompt（包含问题 + 上下文）生成最终回答。由于模型有了明确的信息依据，生成的答案更准确、更可信，大幅降低“幻觉”风险。
RAG 本质上就是重构了一个新的 Prompt！
3.LangChain快速搭建本地知识库
LangChain 是一个专门用来开发大模型（LLM）应用的“工具箱”或“粘合剂”，它可以：
- 链式调度（Chain）： 这是它的名字来源。它能把多个任务串联起来。列如：先让 AI 总结网页内容，再根据总结写一封邮件，最后翻译成英文。LangChain 负责把这些步骤自动连接。
- 连接外部数据（Data Connection）： 就像你之前看到的 RAG 架构图，LangChain 可以轻松地把本地文档、数据库或网页连接给 AI，让它具备“查资料”的能力。
- 记忆管理（Memory）： 原生的大模型实则是没有长期记忆的。LangChain 提供了存储聊天记录的机制，让 AI 能够记住你们几分钟前聊过的内容。
- 调用工具（Agents）： 它能让 AI 学会使用计算器、搜索谷歌或运行代码。AI 会根据你的问题，自主决定目前该用什么工具。
3.1 准备环境
- 本地安装好 Conda 环境
conda create -n rag python=3.11 ipykernel
- 准备阿里百炼平台API-KEY :https://bailian.console.aliyun.com/
- 将 API-KEY 配置到环境变量中
- 安装依赖
pip install pypdf2 dashscope langchain langchain-openai langchain-community faiss-cpu
- pypdf2 : 解析pdf文档
- sashscope : 百炼平台SDK
- faiss-cpu: faiss 向量数据库，CPU版本
- langchain langchain-openai langchain-community ：LangChain框架
3.2 搭建流程
- 文档加载，并按照必定条件切割成片段
- 将切割的文本片段灌入检索引擎
- 封装检索接口
- 构建调用流程： Query->检索->Prompt ->LLM->回复
需要两个模型
Embedding 模型 LLM模型
3.2.1 从PDF中提取文本并记录每行文本对应的页码
import logging
from typing import Tuple, List
from PyPDF2 import PdfReader
def extract_text_with_page_numbers(pdf_reader: PdfReader) -> Tuple[str, List[int]]:
"""
从PDF中提取文本并记录每行文本对应的页码
参数:
pdf: PDF文件对象
返回:
text: 提取的文本内容
page_numbers: 每行文本对应的页码列表
"""
text = ""
page_numbers = []
# 从1开始计数，符合人类阅读习惯
# 遍历PDF中的每一页
for page_number, page in enumerate(pdf_reader.pages, start=1):
# 从当前页中提取文本
extracted_text = page.extract_text()
# 如果当前页有文本，则将其添加到总文本中并记录其页码
if extracted_text:
text += extracted_text
# Python 列表（List）的一个内置方法，用于将一个可迭代对象的所有元素添加到列表的末尾。
# [page_number] * 行数 创建一个包含多个一样页码的列表
page_numbers.extend([page_number] * len(extracted_text.split("
")))
else:
logging.warning(f"No text found on page {page_number}.")
# 实际上，返回的是一个 元组，即 Tuple
return text, page_numbers
main函数中读取pdf文件，然后打印每一行所对应的页码：
if __name__ == '__main__':
# 配置日志以便看到警告
logging.basicConfig(level=logging.WARNING)
pdf_path = "mypdf.pdf"
try:
with open(pdf_path, "rb") as f:
pdf_reader = PdfReader(f)
full_text, page_nums = extract_text_with_page_numbers(pdf_reader)
# 所有的行
lines = full_text.split("
")
# zip Python 的内置函数，用于将多个可迭代对象打包成一个迭代器，返回元组序列。
for line, page_num in zip(lines, page_nums):
print(f"{line} (Page {page_num})")
except FileNotFoundError:
print(f"Error: The file '{pdf_path}' was not found.")
except Exception as e:
print(f"An error occurred: {e}")
3.2.2 处理文本并创建向量存储
# 这个函数中需要导入
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
import os
import pickle
def process_text_with_splitter(text: str, page_numbers: List[int], save_path: str = None) -> FAISS:
"""
处理文本并创建向量存储
参数:
text: 提取的文本内容
page_numbers: 每行文本对应的页码列表
save_path: 可选，保存向量数据库的路径
返回:
knowledgeBase: 基于FAISS的向量存储对象
"""
# 创建文本分割器（递归字符文本分割器），用于将长文本分割成小块
text_splitter = RecursiveCharacterTextSplitter(
separators=["
", "
", ".", " ", ""],
chunk_size=512,
chunk_overlap=128,
length_function=len,
)
# 分割文本
chunks = text_splitter.split_text(text)
# logging.debug(f"Text split into {len(chunks)} chunks.")
print(f"文本被分割成 {len(chunks)} 个块。")
# 创建嵌入模型，OpenAI嵌入模型，配置环境变量 OPENAI_API_KEY
# embeddings = OpenAIEmbeddings()
# 调用阿里百炼平台文本嵌入模型，配置环境变量 AI_BAI_LIAN_API_KEY 默认为DASHSCOPE_API_KEY
embeddings = DashScopeEmbeddings(
dashscope_api_key=os.environ["AI_BAI_LIAN_API_KEY"],
model="text-embedding-v4"
)
# 从文本块创建知识库
knowledgeBase = FAISS.from_texts(chunks, embeddings)
print("已从文本块创建知识库...")
# 存储每个文本块对应的页码信息
# 生成形如 {"文本块内容": 页码, ...} 的字典
page_info = {chunk: page_numbers[i] for i, chunk in enumerate(chunks)}
knowledgeBase.page_info = page_info
# 如果提供了保存路径，则保存向量数据库和页码信息
if save_path:
# 确保目录存在
os.makedirs(save_path, exist_ok=True)
# 保存FAISS向量数据库
knowledgeBase.save_local(save_path)
print(f"向量数据库已保存到: {save_path}")
# 保存页码信息到同一目录
with open(os.path.join(save_path, "page_info.pkl"), "wb") as f:
pickle.dump(page_info, f)
print(f"页码信息已保存到: {os.path.join(save_path, 'page_info.pkl')}")
return knowledgeBase
测试
process_text_with_splitter函数：
def test_process_text_with_splitter():
pdf_path = "mypdf.pdf"
try:
with open(pdf_path, "rb") as f:
# 从PDF中提取文本
pdf_reader = PdfReader(f)
# 从PDF中提取文本并记录每行文本对应的页码
full_text, page_nums = extract_text_with_page_numbers(pdf_reader)
# 处理文本并创建向量存储
knowledgeBase = process_text_with_splitter(full_text, page_nums, save_path="knowledgeBase")
except FileNotFoundError:
print(f"Error: The file '{pdf_path}' was not found.")
except Exception as e:
print(f"An error occurred: {e}")
输出：
文本被分割成 10 个块。
已从文本块创建知识库...
向量数据库已保存到: knowledgeBase
页码信息已保存到: knowledgeBasepage_info.pkl
3.2.3 开始检索
在检索之前，第一要加载上面保存到磁盘上的向量数据库和页码信息
def load_knowledge_base(load_path: str, embeddings=None) -> FAISS:
"""
从磁盘加载向量数据库和页码信息
参数:
load_path: 向量数据库的保存路径
embeddings: 可选，嵌入模型。如果为None，将创建一个新的DashScopeEmbeddings实例
返回:
knowledgeBase: 加载的FAISS向量数据库对象
"""
# 如果没有提供嵌入模型，则创建一个新的
if embeddings is None:
embeddings = DashScopeEmbeddings(
dashscope_api_key=os.environ["AI_BAI_LIAN_API_KEY"],
model="text-embedding-v4"
)
# 加载FAISS向量数据库，添加allow_dangerous_deserialization=True参数以允许反序列化
knowledgeBase = FAISS.load_local(load_path, embeddings, allow_dangerous_deserialization=True)
print(f"向量数据库已从 {load_path} 加载。")
# 加载页码信息
page_info_path = os.path.join(load_path, "page_info.pkl")
if os.path.exists(page_info_path):
with open(page_info_path, "rb") as f:
page_info = pickle.load(f)
knowledgeBase.page_info = page_info
print("页码信息已加载。")
else:
print("警告: 未找到页码信息文件。")
return knowledgeBase
- 问题一: “客户经理被投诉，投诉一次扣多少分”
查看文档，正确答案是“一次扣2分”, 下面是测试函数：
# 要导入的包
import logging
from typing import Tuple, List
from PyPDF2 import PdfReader
from langchain_classic.chains.combine_documents import create_stuff_documents_chain
from langchain_community.callbacks import get_openai_callback
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
import os
import pickle
def test_query_knowledge_base():
# 加载知识库
knowledgeBase = load_knowledge_base("knowledgeBase")
query = "客户经理被投诉，投诉一次扣多少分？"
# 1. 执行类似度搜索，找到与查询相关的文档
docs = knowledgeBase.similarity_search(query)
# 2. 初始化对话大模型
chatLLM = ChatOpenAI(
# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",
api_key=os.getenv("AI_BAI_LIAN_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model="deepseek-v3"
)
# 3. 创建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "请基于以下上下文信息回答问题：
{context}"),
("human", "{input}")
])
# 4. 创建文档问答链
chain = create_stuff_documents_chain(chatLLM, prompt)
# 5. get_openai_callback 用于跟踪和监控 OpenAI API 调用
with get_openai_callback() as cost:
# 执行问答链
response = chain.invoke({
"context": docs,
"input": query
})
print(response)
print("="*50)
print(f"查询已处理。成本: {cost}")
先到向量数据库中进行类似度查询，与 query 问题相关联的 文档 初始化对话模型 创建提示模板 创建文档问答链 跟踪 OpenAI API 调用, 获取模型回答结果
向量数据库已从 knowledgeBase 加载。
页码信息已加载。
根据上下文信息中的**工作质量考核标准**第九条（一）服务质量考核第2项规定：
> 客户服务效率低，态度生硬或不及时为客户提供维护服务，有客户投诉的，**每投诉一次扣2分**。
因此，客户经理被投诉一次会**扣2分**。
==================================================
查询已处理。成本: Tokens Used: 1233
Prompt Tokens: 1167
Prompt Tokens Cached: 0
Completion Tokens: 66
Reasoning Tokens: 0
Successful Requests: 1
Total Cost (USD): $0.0
- 问题二:”客户经理每年评聘申报时间是怎样的？” 只需要将上面的query 替换掉，输出为:
**第十一条** 明确说明：
- **每年一月份** 为固定客户经理评聘的申报时间。
- 分行人力资源部和个人业务部会在**每年二月份**组织统一的资格考试，考试合格者由分行颁发有效期为一年的个金客户经理资格证书。
因此，**客户经理评聘的申报时间为每年1月**，需在此期间向分行人力资源部提交申请。
==================================================
查询已处理。成本: Tokens Used: 1305
Prompt Tokens: 1205
Prompt Tokens Cached: 0
Completion Tokens: 100
Reasoning Tokens: 0
Successful Requests: 1
Total Cost (USD): $0.0
4.有效提升RAG质量的方法
4.1 数据准备阶段
- 数据质量差： 企业大部分数据（尤其是非结构化数据）缺乏良好的数据治理，未经标记/评估的非结构化数据可能包含敏感、过时、矛盾或不正确的信息。
- 多模态信息： 提取、定义和理解文档中的不同内容元素，如标题、配色方案、图像和标签等存在挑战。
- 复杂的PDF提取： PDF是为人类阅读而设计的，机器解析起来超级复杂。
提升数据准备阶段的质量：
- 数据评估与分类
- 数据审计：全面审查现有数据，识别敏感、过时、矛盾或不准确的信息。
- 数据分类：按类型、来源、敏感性和重大性对数据进行分类，便于后续处理。
- 数据清洗
- 去重：删除重复数据
- 纠错：修正格式错误、拼写错误等
- 更新：替换过时信息，确保数据时效性
- 一致性检查：解决数据矛盾，确保逻辑一致
- 敏感信息处理
- 识别敏感数据：使用工具或正则表达式识别敏感信息，如个人身份信息
- 脱敏或加密：对敏感数据进行脱敏处理，确保合规。
- 数据标记与标注
- 元数据标记：为数据添加元数据，如来源、创建时间等
- 内容标注：对非结构化数据进行标注，便于后续检索和分析
- 数据治理框架
- 制定政策：明确数据管理、访问控制和更新流程
- 责任分配：指定数据治理负责人，确保政策执行
- 监控与审计：定期监控数据质量，进行审计
各个厂商也提供了智能文档技术：
- 阿里文档智能
- 微软 LayoutLMv3
- MinerU
4.2 知识检索阶段
- 内容缺失： 当检索过程缺少关键内容时，系统会提供不完整、碎片化的答案 => 降低RAG的质量
- 错过排名靠前的文档： 用户查询相关的文档时被检索到，但相关性极低，导致答案不能满足用户需求，这是由于在检索过程中，用户通过主观判断决定检索“文档数量”。理论上所有文档都要被排序并思考进一步处理，但在实践中，一般只有排名top k的文档才会被召回，而k值需要根据经验确定。
- 不在上下文中： 从数据库中检索出包含答案的文档，但未能包含在生成答案的上下文中。这种情况一般发生在返 回大量文件时，需要进行整合以选择最相关的信息。
这里有两种提高质量的办法：
- 通过查询转换澄清用户意图：明确用户意图，提高检索准确性。
- 采用混合检索和重排策略：确保最相关的文档被优先处理，生成更准确的答案。
4.2.1通过查询转换澄清用户意图
- 场景：用户询问 “如何申请信用卡？”
- 问题：用户意图可能模糊，例如不清楚是申请流程、所需材料还是资格条件。
- 解决方法：通过查询转换明确用户意图（调用大模型）。 意图识别：使用自然语言处理技术识别用户意图。例如，识别用户是想了解流程、材料还是资格。 查询扩展：根据识别结果扩展查询。例如：如果用户想了解流程，查询扩展为“信用卡申请的具体步骤”，如果用户想了解材料，查询扩展为“申请信用卡需要哪些材料”，如果用户想了解资格，查询扩展为“申请信用卡的资格条件” 检索：使用扩展后的查询检索相关文档
4.2.2 混合检索和重排策略
- 场景：用户询问“信用卡年费是多少？”
- 问题：直接检索可能返回大量文档，部分相关但排名低，导致答案不准确。
- 解决方法：采用混合检索和重排策略。 混合检索：结合关键词检索和语义检索。列如：关键词检索：“信用卡年费”。 语义检索：使用嵌入模型检索与“信用卡年费”语义相近的文档。 重排：对检索结果进行重排。 生成答案：从重排后的文档中生成答案。
4.3 答案生成阶段
- 未提取： 答案与所提供的上下文相符，但大语言模型却无法准确提取。这种情况一般发生在上下文中存在过多噪音或相互冲突的信息时。
- 不完整： 尽管能够利用上下文生成答案，但信息缺失会导致对用户查询的答复不完整。格式错误：当prompt中的附加指令格式不正确时，大语言模型可能误解或曲解这些指令，从而导致错误的答案。
- 幻觉： 大模型可能会产生误导性或虚假性信息。
4.3.1 改善提示词模板
|
场景 |
原始提示词 |
改善后的提示词 |
|
用户询问“如何申请信用卡？” |
“根据以下上下文回答问题：如何申请信用卡？” |
“根据以下上下文，提取与申请信用卡相关的具体步骤和所需材料：如何申请信用卡？” |
|
用户询问“信用卡的年费是多少？” |
“根据以下上下文回答问题：信用卡的年费是多少？” |
“根据以下上下文，详细列出不同信用卡的年费信息，并说明是否有减免政策：信用卡的年费是多少？” |
|
用户询问“什么是零存整取？” |
“根据以下上下文回答问题：什么是零存整取？” |
“根据以下上下文，准确解释零存整取的定义、特点和适用人群，确保信息真实可靠：什么是零存整取？” |
4.3.2 实施动态防护栏
动态防护栏（Dynamic Guardrails）是一种在生成式AI系统中用于实时监控和调整模型输出的机制，旨在确保生成的内容符合预期、准确且安全。它通过设置规则、约束和反馈机制，动态地干预模型的生成过程，避免生成错误、不完整、不符合格式要求或含有虚假信息（幻觉）的内容。
场景1：防止未提取
用户问题：“如何申请信用卡？”
- 上下文：包含申请信用卡的步骤和所需材料。
- 动态防护栏规则：检查生成的答案是否包含“步骤”和“材料”。如果缺失，提示模型重新生成。
- 示例： 错误输出：“申请信用卡需要提供一些材料。” 防护栏触发：检测到未提取具体步骤，提示模型补充。
场景2：防止不完整
用户问题：“信用卡的年费是多少？”
- 上下文：包含不同信用卡的年费信息。
- 动态防护栏规则：检查生成的答案是否列出所有信用卡的年费。如果缺失，提示模型补充。
- 示例： 错误输出：“信用卡A的年费是100元。” 防护栏触发：检测到未列出所有信用卡的年费，提示模型补充。
场景3：防止幻觉
用户问题：“什么是零存整取？”
- 上下文：包含零存整取的定义和特点。
- 动态防护栏规则：检查生成的答案是否与上下文一致。如果不一致，提示模型重新生成。
- 示例： 错误输出：“零存整取是一种贷款产品。 防护栏触发：检测到与上下文不一致，提示模型重新生成。
如何制定实际性校验规则？
当业务逻辑明确且规则较为固定时，可以人为定义一组规则，列如：
- 规则1：生成的答案必须包含检索到的知识片段中的关键实体（如“年费”、“利率”）。
- 规则2：生成的答案必须符合指定的格式（如步骤列表、表格等）。
- 实施方法： 使用正则表达式或关键词匹配来检查生成内容是否符合规则。 例如，检查生成内容是否包含“年费”这一关键词，或者是否符合步骤格式（如“1. 登录；2. 设置”）。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
暂无评论...
