使用 LangChain 和 RAG 实现《斗破苍穹》文本问答系

大模型系列文章

在本篇文章中，我将介绍如何使用 LangChain 框架和 RAG（Retrieval-Augmented Generation） 技术构建一个基于《斗破苍穹》小说文本的问答系统。通过这个系统，我们可以对小说内容提出问题，并获得由大模型生成的答案。

---

---

前言

《斗破苍穹》是一部非常受欢迎的中国网络小说，内容丰富、章节众多。为了更高效地检索和理解其中的信息，我们可以通过 RAG 技术，将原始文本与大语言模型结合起来，从而实现自然语言的问题回答。

---

一、技术栈概览

• LangChain：用于构建 LLM 应用流程
• TextLoader：加载本地 TXT 文本数据
• RecursiveCharacterTextSplitter：将长文本分割为适合嵌入的小块
• FAISS：用于向量数据库的构建与存储
• OllamaLLM 和 OllamaEmbeddings：调用本地运行的大语言模型（如 deepseek-r1:1.5b 和 bge-m3）
• create_retrieval_chain：构建 RAG 管道

二、使用步骤

1.加载与预处理文本

我们首先使用 TextLoader 加载本地的 .txt 文件（即《斗破苍穹》全文），并设置编码格式为 gb18030 以支持中文字符。

loader = TextLoader("斗破苍穹.txt", encoding='gb18030')
docs = loader.load()

接着使用 RecursiveCharacterTextSplitter 将文档切分为多个段落（chunk），每个 chunk 最多 500 个字符，重叠部分为 50 个字符，以保留上下文信息。

text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
documents = text_splitter.split_documents(docs)

2.构建嵌入模型与向量数据库

由于 LangChain 并未提供官方的 Ollama 嵌入模型，我们自定义了一个 OllamaEmbeddings 类来调用本地 API 接口获取嵌入向量。

class OllamaEmbeddings(Embeddings):def __init__(self, model: str = "bge-m3", url: str = "http://localhost:11434/api/embeddings"):self.model = modelself.url = url

然后使用 FAISS 构建向量数据库。如果已有索引文件，则直接加载；否则重新构建并保存。

try:vector_store = FAISS.load_local(VECTOR_STORE_PATH, embeddings, allow_dangerous_deserialization=True)
except Exception as e:vector_store = FAISS.from_documents(documents, embeddings)vector_store.save_local(VECTOR_STORE_PATH)

3. 构建提示模板与检索链

我们使用 ChatPromptTemplate 定义输入格式，将检索到的上下文插入到提示词中供 LLM 使用。

prompt = ChatPromptTemplate.from_template("""使用以下上下文回答问题。<context>{context}</context>问题: {input}"""
)

再创建文档链和检索链，连接 LLM 和向量数据库：

document_chain = create_stuff_documents_chain(llm, prompt)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
retrieval_chain = create_retrieval_chain(retriever, document_chain)

4. 执行查询函数

最后定义 query_rag 函数，接受一个问题字符串，返回模型生成的回答：

def query_rag(question: str) -> str:response = retrieval_chain.invoke({"input": question})return response['answer']

示例：

if __name__ == "__main__":answer = query_rag("萧炎是谁")print(answer)

---

完整代码

# 导入必要的模块（保持不变）
from langchain_community.document_loaders import TextLoader  # 替换为 TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.embeddings import Embeddings
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
from langchain_ollama import OllamaLLM
import requests
from typing import List# 加载 TXT 文件并提取内容
file_path = "斗破苍穹.txt"  # 替换为你的 txt 文件路径
loader = TextLoader(file_path, encoding='gb18030')  # 使用 TextLoader 加载 txt 文件
docs = loader.load()
# 分割文本为小块（chunk），以便嵌入处理
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
documents = text_splitter.split_documents(docs)# 自定义 Ollama 嵌入模型类（保持不变）
class OllamaEmbeddings(Embeddings):def __init__(self, model: str = "bge-m3", url: str = "http://localhost:11434/api/embeddings"):self.model = modelself.url = urldef embed_documents(self, texts: List[str]) -> List[List[float]]:embeddings = []for text in texts:emb = self._get_embedding(text)if emb:embeddings.append(emb)return embeddingsdef embed_query(self, text: str) -> List[float]:return self._get_embedding(text)def _get_embedding(self, text: str) -> List[float]:headers = {"Content-Type": "application/json"}data = {"model": self.model,"prompt": f"Represent this sentence for retrieval: {text}"}try:response = requests.post(self.url, headers=headers, json=data)response.raise_for_status()return response.json()['embedding']except Exception as e:print(f"嵌入生成失败: {e}")return []# 初始化嵌入模型并构建向量数据库（保持不变）
embeddings = OllamaEmbeddings(model="bge-m3")
VECTOR_STORE_PATH = "faiss_index_doupo"try:vector_store = FAISS.load_local(VECTOR_STORE_PATH, embeddings, allow_dangerous_deserialization=True)print("✅ 已加载本地向量数据库。")
except Exception as e:print("⚠️ 未找到本地向量数据库，正在构建新的数据库...")vector_store = FAISS.from_documents(documents[:200], embeddings)vector_store.save_local(VECTOR_STORE_PATH)print("✅ 新的向量数据库已保存至本地。")# 构建 LLM 模型及提示模板（保持不变）
llm = OllamaLLM(model="deepseek-r1:1.5b")prompt = ChatPromptTemplate.from_template("""使用以下上下文回答问题。<context>{context}</context>问题: {input}"""
)# 创建文档链和检索链（保持不变）
document_chain = create_stuff_documents_chain(llm, prompt)
retriever = vector_store.as_retriever()
retriever.search_kwargs = {"k": 5}
retrieval_chain = create_retrieval_chain(retriever, document_chain)def query_rag(question: str) -> str:"""执行 RAG 查询流程，返回模型的回答"""response = retrieval_chain.invoke({"input": question})return response['answer']
# 示例查询（保持不变）
if __name__ == "__main__":answer = query_rag("萧炎是谁")print(answer)