RAG项目实战：金融问答系统

需求痛点

私有知识很多，如何让大模型只选择跟问题有关的知识进行参考呢？

需求分析

是否可以使用关键词匹配呢？以前的搜索主要使用关键词匹配，这个要求太高了，需要提前抽取准备好关键词，有点像以前SEO的工作。现在是国际化社会，大模型对语言应是无感知理解的，也就是说用户使用英文或中文提问时，它的语义理解应是一样的，要脱离语言本身，看内涵，跨语言、跨表达的习惯，直击这句话的本质，所以我们使用语义化匹配。

实现思路

将私有资料，按逻辑、互相独立的单元来分段；

使用模型将每段知识的核心信息抽取出来，变成向量；

将这些知识片段及其向量存到向量数据库，如chromadb向量数据库；

将用户提的问题转化为向量，去向量数据库中进行相似度匹配，寻找答案。

设置Prompt提示词，让结果更符合预期。

工作流程：

用户提一个问题，把问题通过向量化模型变成一个向量，再使用这个向量到向量库中，查找最相似的上下文，接着把问题和检索到的上下文一起交给模型，让模型回答问题。

这里可能会有一个问题，比如你已经在库里检索到答案了，为何还要让大模型生成答案呢？因为检索出来的结果，有些高度相关，有些低相关，而且比较长，不一定符合预期的格式，需要自己加工。大模型帮我们快速筛选和判断，然后通过prompt设置的格式来综合整理，生成最终的结果。

项目数据

数据集：招股说明书，80个PDF文件

数据集下载地址：https://www.modelscope.cn/datasets/BJQW14B/bs_challenge_financial_14b_dataset/summary

实现效果示例：

用户提问：景顺长城中短债债券C基金在20210331的季报里，前三大持仓占比的债券名称是什么?

回答：景顺长城中短债债券C在20210331的季报中，前三大持仓占比的债券名称分别是21国开01、20农发清发01、20国信03。

提问：上海华铭智能终端设备股份有限公司的首发战略配售结果如何？

回答：上海华铭智能终端设备股份有限公司的首发战略配售具体情况并未在招股意向书中详细说明。

项目实现

1. 规划工程文件

金融问答系统项目涉及前端、后端、数据集和模型等，本文只探讨RAG部分，暂不做前端UI交互。

项目目录规划：

SmartFinance \|- app \ # 该目录用于服务端代码|- conf \|- .qwen # 配置API KEY|- dataset \ # 该目录存放数据集|-pdf \ # 该目录存放PDF文件|- rag \ # 该目录用于保存 rag 相关代码|- chroma_conn.py  # 管理ChromaDB数据库，如添加文档和查询等操作|- pdf_processor.py  # 实现PDF文件加载、文本切分和向量存储等操作|- rag.py # 实现检索功能|- test_framework.py # 测试功能|- utils \ # 该目录用于存放工具配置类文件|- util.py # 加载连接大模型

2.  实现基本连接大模型的 util 库

代码文件及目录：app/utils/util.py

# 导入必要的库
from dotenv import load_dotenv  # 用于加载环境变量
import os  # 用于操作文件路径# 获取当前文件所在的目录
current_dir = os.path.dirname(__file__)# 构建到 conf/.qwen 的相对路径
# 这里通过 os.path.join 拼接路径，确保跨平台兼容性
conf_file_path_qwen = os.path.join(current_dir, '..', 'conf', '.qwen')# 加载 .qwen 文件中的环境变量
# dotenv_path 参数指定了环境变量文件的路径
load_dotenv(dotenv_path=conf_file_path_qwen)def get_qwen_models():"""加载通义千问系列的大模型，包括 LLM、Chat 和 Embedding 模型。"""# 加载 LLM 大模型（语言生成模型）# 使用 Tongyi 类来实例化一个通义千问模型from langchain_community.llms.tongyi import Tongyillm = Tongyi(model="qwen-max",  # 指定模型类型为 qwen-max，适合复杂任务temperature=0.1,   # 控制输出的随机性，值越低越保守top_p=0.7,         # 核采样参数，控制生成文本的多样性max_tokens=1024    # 最大生成的 token 数量)# 加载 Chat 大模型（对话模型）# 使用 ChatTongyi 类来实例化一个通义千问对话模型from langchain_community.chat_models import ChatTongyichat = ChatTongyi(model="qwen-max",  # 指定模型类型为 qwen-max，适合高质量对话temperature=0.01,  # 温度更低以获得更确定的回复top_p=0.2,         # 控制对话生成的多样性max_tokens=1024    # 最大生成的 token 数量)# 加载 Embedding 大模型（嵌入模型）# 使用 DashScopeEmbeddings 类来实例化一个通义千问嵌入模型from langchain_community.embeddings import DashScopeEmbeddingsembed = DashScopeEmbeddings(model="text-embedding-v3"  # 指定嵌入模型版本)# 返回加载的三个模型return llm, chat, embed

在app/conf/.qwen中添加对应的API KEY

DASHSCOPE_API_KEY = sk-xxxxxx

3. 实现向量库的基础功能

文本使用ChromaDB实现，用于基本的向量库连接、数据入库操作

代码文件及目录：app/rag/chroma_conn.py

import chromadb  # 导入 ChromaDB 的核心库
from chromadb import Settings  # 导入 ChromaDB 的配置类
from langchain_community.vectorstores.chroma import Chroma  # 导入 LangChain 的 Chroma 向量存储模块class ChromaDB:"""自定义的 ChromaDB 类，用于管理 Chroma 向量数据库。支持本地模式和 HTTP 模式连接 ChromaDB。"""def __init__(self,chroma_server_type="local",  # 服务器类型：http 表示通过 HTTP 连接，local 表示本地文件方式host="localhost", port=8000,  # 如果是 HTTP 模式，需要指定服务器地址和端口persist_path="chroma_db",  # 数据库路径：如果是本地模式，需要指定数据库的存储路径collection_name="langchain",  # 数据库中的集合名称embed=None  # 数据库使用的向量化函数):"""初始化 ChromaDB 对象。:param chroma_server_type: 连接类型（"local" 或 "http"）:param host: HTTP 服务器地址（仅在 chroma_server_type="http" 时有效）:param port: HTTP 服务器端口（仅在 chroma_server_type="http" 时有效）:param persist_path: 数据库存储路径（仅在 chroma_server_type="local" 时有效）:param collection_name: 数据库集合名称:param embed: 向量化函数"""self.host = host  # HTTP 服务器地址self.port = port  # HTTP 服务器端口self.path = persist_path  # 数据库存储路径self.embed = embed  # 向量化函数self.store = None  # 初始化时未创建向量存储对象# 如果是 HTTP 协议方式连接数据库if chroma_server_type == "http":# 创建一个 HTTP 客户端client = chromadb.HttpClient(host=host, port=port)# 初始化 Chroma 向量存储对象self.store = Chroma(collection_name=collection_name,embedding_function=embed,client=client)# 如果是本地模式连接数据库elif chroma_server_type == "local":# 初始化 Chroma 向量存储对象self.store = Chroma(collection_name=collection_name,embedding_function=embed,persist_directory=persist_path)# 如果初始化失败，抛出异常if self.store is None:raise ValueError("Chroma store initialization failed!")def add_with_langchain(self, docs):"""将文档添加到 ChromaDB 数据库中。:param docs: 要添加的文档列表"""# 使用 LangChain 提供的接口将文档添加到数据库self.store.add_documents(documents=docs)def get_store(self):"""返回 Chroma 向量数据库的对象实例。:return: Chroma 向量存储对象"""return self.store

4. 实现入库功能

代码文件及目录：app/rag/pdf_processor.py

import os  # 文件操作相关模块
import logging  # 日志记录模块
import time  # 时间操作模块
from tqdm import tqdm  # 进度条显示模块
from langchain_community.document_loaders import PyMuPDFLoader  # PDF文件加载器
from langchain_text_splitters import RecursiveCharacterTextSplitter  # 文本切分工具
from rag.chroma_conn import ChromaDB  # 自定义的ChromaDB类class PDFProcessor:"""PDFProcessor 类用于处理PDF文件并将其内容存储到ChromaDB中。"""def __init__(self,directory,  # PDF文件所在目录chroma_server_type,  # ChromaDB服务器类型（"local" 或 "http"）persist_path,  # ChromaDB持久化路径embed):  # 向量化函数self.directory = directory  # PDF文件存放目录self.file_group_num = 80  # 每组处理的文件数self.batch_num = 6  # 每次插入的批次数量self.chunksize = 500  # 切分文本的大小self.overlap = 100  # 切分文本的重叠大小# 初始化ChromaDB对象self.chroma_db = ChromaDB(chroma_server_type=chroma_server_type,persist_path=persist_path,embed=embed)# 配置日志logging.basicConfig(level=logging.INFO,  # 设置日志级别为INFOformat='%(asctime)s - %(levelname)s - %(message)s',  # 日志格式datefmt='%Y-%m-%d %H:%M:%S'  # 时间格式)def load_pdf_files(self):"""加载指定目录下的所有PDF文件。"""pdf_files = []for file in os.listdir(self.directory):  # 遍历目录中的所有文件if file.lower().endswith('.pdf'):  # 筛选PDF文件pdf_files.append(os.path.join(self.directory, file))  # 构建完整路径logging.info(f"Found {len(pdf_files)} PDF files.")  # 记录日志return pdf_filesdef load_pdf_content(self, pdf_path):"""使用PyMuPDFLoader读取PDF文件的内容。"""pdf_loader = PyMuPDFLoader(file_path=pdf_path)  # 创建PDF加载器docs = pdf_loader.load()  # 加载PDF内容logging.info(f"Loading content from {pdf_path}.")  # 记录日志return docsdef split_text(self, documents):"""使用RecursiveCharacterTextSplitter将文档切分为小段。"""text_splitter = RecursiveCharacterTextSplitter(chunk_size=self.chunksize,  # 每段文本的最大长度chunk_overlap=self.overlap,  # 段与段之间的重叠长度length_function=len,  # 使用字符串长度作为分割依据add_start_index=True,  # 添加起始索引信息)docs = text_splitter.split_documents(documents)  # 切分文档logging.info("Split text into smaller chunks with RecursiveCharacterTextSplitter.")  # 记录日志return docsdef insert_docs_chromadb(self, docs, batch_size=6):"""将文档分批插入到ChromaDB中。"""logging.info(f"Inserting {len(docs)} documents into ChromaDB.")  # 记录日志start_time = time.time()  # 记录开始时间total_docs_inserted = 0  # 已插入的文档总数# 计算总批次total_batches = (len(docs) + batch_size - 1) // batch_size# 使用tqdm显示进度条with tqdm(total=total_batches, desc="Inserting batches", unit="batch") as pbar:for i in range(0, len(docs), batch_size):batch = docs[i:i + batch_size]  # 获取当前批次的文档self.chroma_db.add_with_langchain(batch)  # 插入到ChromaDBtotal_docs_inserted += len(batch)  # 更新已插入的文档数量# 计算TPM（每分钟插入的文档数）elapsed_time = time.time() - start_timeif elapsed_time > 0:  # 防止除以零tpm = (total_docs_inserted / elapsed_time) * 60pbar.set_postfix({"TPM": f"{tpm:.2f}"})  # 更新进度条的后缀信息pbar.update(1)  # 更新进度条def process_pdfs_group(self, pdf_files_group):"""处理一组PDF文件，包括加载内容、切分文本和插入数据库。"""pdf_contents = []  # 存储所有PDF文件的内容for pdf_path in pdf_files_group:documents = self.load_pdf_content(pdf_path)  # 加载PDF内容pdf_contents.extend(documents)  # 将内容添加到列表中docs = self.split_text(pdf_contents)  # 切分文本self.insert_docs_chromadb(docs, self.batch_num)  # 插入到ChromaDBdef process_pdfs(self):"""批量处理目录下的所有PDF文件。"""pdf_files = self.load_pdf_files()  # 加载所有PDF文件group_num = self.file_group_num  # 每组处理的文件数# 按组处理PDF文件for i in range(0, len(pdf_files), group_num):pdf_files_group = pdf_files[i:i + group_num]self.process_pdfs_group(pdf_files_group)print("PDFs processed successfully!")  # 提示处理完成

5. 测试导入功能

测试从 PDF 文件加载内容并将其导入到 ChromaDB 向量数据库的过程

代码文件及目录：app/test_framework.py

# 测试将 PDF 文件内容导入到向量数据库的主流程def test_import():"""测试从 PDF 文件加载内容并将其导入到 ChromaDB 向量数据库的过程"""# 导入 PDFProcessor 类，用于处理 PDF 文件并将其内容存储到向量数据库中from rag.pdf_processor import PDFProcessor# 导入工具函数 get_qwen_models，用于获取通义千问的模型实例from utils.util import get_qwen_models# 获取通义千问的 LLM、Chat 和 Embedding 模型# 这里只使用 Embedding 模型来生成文本的向量表示llm, chat, embed = get_qwen_models()# 如果需要使用其他嵌入模型（如 Hugging Face 的 Embeddings），可以取消注释以下行# embed = get_huggingface_embeddings()# 定义 PDF 文件所在的目录路径directory = "./app/dataset/pdf"  # 存放 PDF 文件的目录# 定义 ChromaDB 数据库的持久化路径persist_path = "chroma_db"  # ChromaDB 数据库文件的存储路径# 定义 ChromaDB 的服务器类型（"local" 表示本地模式）server_type = "local"  # 使用本地模式连接 ChromaDB# 创建 PDFProcessor 实例，用于处理 PDF 文件并将其内容存储到 ChromaDB 中pdf_processor = PDFProcessor(directory=directory,  # PDF 文件目录chroma_server_type=server_type,  # ChromaDB 服务器类型persist_path=persist_path,  # ChromaDB 数据库的持久化路径embed=embed  # 嵌入模型，用于生成文本向量)# 调用 process_pdfs 方法，开始处理 PDF 文件并将内容插入到 ChromaDBpdf_processor.process_pdfs()# 如果此脚本作为主程序运行，则执行测试导入流程
if __name__ == "__main__":test_import()

（1）通过命令行启动ChromaDB服务端

chroma run --path chroma_db --port 8000 --host localhost

（2）运行 test_framework.py

 

6. 实现检索功能

代码文件及目录：app/rag/rag.py

通过这段代码，可以实现一个基本的 RAG 系统，从向量数据库中检索知识并生成答案

import logging  # 日志记录模块
from langchain_core.prompts import ChatPromptTemplate  # 用于构建对话提示模板
from langchain_core.runnables import RunnablePassthrough  # 用于传递输入数据的工具
from langchain_core.runnables.base import RunnableLambda  # 用于包装自定义函数为可运行对象
from langchain_core.output_parsers import StrOutputParser  # 用于解析输出为字符串
from .chroma_conn import ChromaDB  # 自定义的 ChromaDB 类# 配置日志记录
logging.basicConfig(level=logging.INFO,  # 设置日志级别为 INFOformat='%(asctime)s - %(levelname)s - %(message)s'  # 日志格式
)class RagManager:"""RAG（检索增强生成）管理器类，用于管理和执行基于向量数据库的知识检索与生成任务。"""def __init__(self,chroma_server_type="http",  # ChromaDB 服务器类型（"http" 或 "local"）host="localhost", port=8000,  # ChromaDB HTTP 服务器地址和端口persist_path="chroma_db",  # ChromaDB 数据库持久化路径llm=None, embed=None):  # LLM 模型和嵌入模型self.llm = llm  # 大语言模型（LLM）self.embed = embed  # 嵌入模型# 初始化 ChromaDB 并获取存储对象chrom_db = ChromaDB(chroma_server_type=chroma_server_type,host=host, port=port,persist_path=persist_path,embed=embed)self.store = chrom_db.get_store()  # 获取 ChromaDB 的存储实例def get_chain(self, retriever):"""构建并返回 RAG 查询链。:param retriever: 向量数据库的检索器:return: RAG 查询链"""# 定义 RAG 系统的经典 Prompt 模板prompt = ChatPromptTemplate.from_messages([("human", """You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.Question: {question} Context: {context} Answer:""")])# 将 `format_docs` 方法包装为 Runnable 对象format_docs_runnable = RunnableLambda(self.format_docs)# 构建 RAG 查询链rag_chain = ({"context": retriever | format_docs_runnable,  # 使用检索器获取上下文"question": RunnablePassthrough()}  # 直接传递问题| prompt  # 应用 Prompt 模板| self.llm  # 使用 LLM 模型生成答案| StrOutputParser()  # 解析输出为字符串)return rag_chaindef format_docs(self, docs):"""格式化检索到的文档内容。:param docs: 检索到的文档列表:return: 格式化后的文档内容"""# 记录检索到的文档数量logging.info(f"检索到资料文件个数：{len(docs)}")# 提取文档来源信息retrieved_files = "\n".join([doc.metadata["source"] for doc in docs])logging.info(f"资料文件分别是:\n{retrieved_files}")# 提取文档内容retrieved_content = "\n\n".join(doc.page_content for doc in docs)logging.info(f"检索到的资料为:\n{retrieved_content}")return retrieved_content  # 返回格式化后的文档内容def get_retriever(self, k=4, mutuality=0.3):"""获取向量数据库的检索器。:param k: 检索返回的文档数量:param mutuality: 相似度阈值（分数阈值）:return: 检索器对象"""retriever = self.store.as_retriever(search_type="similarity_score_threshold",  # 使用相似度分数阈值检索search_kwargs={"k": k, "score_threshold": mutuality}  # 检索参数)return retrieverdef get_result(self, question, k=4, mutuality=0.3):"""执行 RAG 查询并返回结果。:param question: 用户提出的问题:param k: 检索返回的文档数量:param mutuality: 相似度阈值（分数阈值）:return: 查询结果（生成的答案）"""# 获取检索器retriever = self.get_retriever(k, mutuality)# 获取 RAG 查询链rag_chain = self.get_chain(retriever)# 执行查询并返回结果return rag_chain.invoke(input=question)

7. 测试检索效果

在 app/framework.py 中添加 RAG 的测试调用函数

def test_rag():"""测试 RAG（检索增强生成）系统的主流程。"""# 导入 RagManager 类，用于管理和执行 RAG 查询任务from rag.rag import RagManager# 导入工具函数 get_qwen_models，用于获取通义千问的模型实例from utils.util import get_qwen_models# 获取通义千问的 LLM、Chat 和 Embedding 模型llm, chat, embed = get_qwen_models()# 创建 RagManager 实例，用于管理 RAG 系统# 参数说明：# - host: ChromaDB HTTP 服务器地址# - port: ChromaDB HTTP 服务器端口# - llm: 大语言模型（LLM）# - embed: 嵌入模型（Embedding Model）rag = RagManager(host="localhost", port=8000, llm=llm, embed=embed)# 执行 RAG 查询，传入用户提出的问题question = "景顺长城中短债债券C基金在20210331的季报里，前三大持仓占比的债券名称是什么?"result = rag.get_result(question)# 输出查询结果print(result)if __name__ == "__main__":# 执行 RAG 测试函数test_rag()# 批量导入 PDF 测试函数（可选）# 如果需要测试 PDF 文件批量导入功能，请取消以下行的注释# test_import()

总结：

Step 1：创建一个 ChromaDB 类，封装基础的Chroma连接、插入和检索；

Step2：实现 PDFProcessor 类，该类调用 ChromaDB 类的插入函数，将批量读取的PDF文件进行切分后保存至向量库；

Step3：实现 RAGManager 类，该类封装了RAG的检索链，并定义检索参数；

Step4：实现一个测试函数，用于测试RAG的检索功能。