使用Langchain生成本地rag知识库并搭载大模型

准备设备： 手机+aidlux2.0个人版

一、下载依赖

pip install langchain langchain-community faiss-cpu pypdf

二、安装ollama并下载模型

1. curl -fsSL https://ollama.com/install.sh | sh #需要VPN
2. ollama serve & #让ollama服务在后台运行

复制代码

安装完毕可以查看ollama版本进行验证，出现版本号之后就可以使用ollama

ollama -v

考虑性能因素，选择下载较小的模型

1. ollama pull phi3:mini
2. ollama pull all-minilm

复制代码

三、构建rag知识库

• 打开手机上的aidlux应用，打开Cloud_ip查看网络ip，输入ip到浏览器+端口号:8000访问
  输入以下命令：
  

1. cd ~
2. touch build_knowledge_base.py

复制代码

• 在文件浏览器中/home/aidlux 下找到对应py文件并打开
  

• 自行准备一个知识库文本（txt或pdf），将文本的路径填入脚本中
  

• 写入以下脚本内容
  

1. from langchain_community.document_loaders import PyPDFLoader, TextLoader
2. from langchain_text_splitters import RecursiveCharacterTextSplitter
3. from langchain_community.embeddings import OllamaEmbeddings
4. from langchain_community.vectorstores import FAISS
5. import os
7. # 1. 设置环境变量优化 Ollama 性能
8. os.environ["OLLAMA_NUM_THREADS"] = "8"  # 设置线程数
9. os.environ["OLLAMA_NUM_CTX"] = "2048"   # 设置上下文长度
11. # 2. 配置嵌入模型 - 移除无效参数
12. embeddings = OllamaEmbeddings(
13.     model="all-minilm"  # 仅保留必要参数
14. )
16. # 3. 加载文档
17. def load_documents(file_path):
18.     if file_path.endswith(".pdf"):
19.         loader = PyPDFLoader(file_path)
20.         print(f"加载 PDF 文档: {file_path}")
21.     elif file_path.endswith(".txt"):
22.         loader = TextLoader(file_path)
23.         print(f"加载文本文档: {file_path}")
24.     else:
25.         raise ValueError(f"不支持的文档格式: {file_path}")
26.     return loader.load()
28. # 4. 文本分割
29. def split_documents(docs):
30.     text_splitter = RecursiveCharacterTextSplitter(
31.         chunk_size=500,
32.         chunk_overlap=80,
33.         separators=["\n\n", "\n", "。", "！", "？", "；"]
34.     )
35.     return text_splitter.split_documents(docs)
37. # 5. 主函数
38. def main():
39.     # 示例文档 - 修改为您的文件路径
40.     document_path = "knowledge.txt"
41.    
42.     # 加载和分割文档
43.     print("开始处理文档...")
44.     documents = load_documents(document_path)
45.     chunks = split_documents(documents)
46.     print(f"文档分割完成: 共 {len(chunks)} 个文本块")
47.    
48.     # 创建向量存储
49.     print("开始生成嵌入向量...")
50.     vector_store = FAISS.from_documents(
51.         documents=chunks,
52.         embedding=embeddings
53.     )
54.    
55.     # 保存知识库索引
56.     save_path = "my_knowledge_base"
57.     vector_store.save_local(save_path)
58.     print(f"知识库构建完成! 保存到: {save_path}")
59.     print(f"向量库大小: {len(vector_store.index_to_docstore_id)} 个向量")
61. if __name__ == "__main__":
62.     main()

复制代码

• 运行脚本
  

python3 build_knowledge_base.py

四、创建 RAG 问答系统

• 创建一个脚本
  

touch rag_query.py

• 写入以下内容
  

1. from langchain_community.llms import Ollama
2. from langchain_community.embeddings import OllamaEmbeddings
3. from langchain_community.vectorstores import FAISS
4. from langchain_core.prompts import ChatPromptTemplate
5. from langchain_core.runnables import RunnablePassthrough
6. from langchain_core.output_parsers import StrOutputParser
7. import os
8. import time
9. import sys
10. import select
12. # 1. 通过环境变量设置优化参数
13. os.environ["OLLAMA_NUM_THREADS"] = "8"  # 设置线程数
14. os.environ["OLLAMA_NUM_CTX"] = "2048"   # 设置上下文长度
16. # 2. 初始化模型
17. llm = Ollama(
18.     model="phi3:mini",        # 轻量级语言模型
19.     temperature=0.3,           # 平衡创造性和准确性
20.     timeout=120.0              # 设置超时时间
21.         )
23. embeddings = OllamaEmbeddings(model="all-minilm")
25. # 3. 加载知识库
26. try:
27.     vector_store = FAISS.load_local(
28.         "my_knowledge_base",
29.         embeddings,
30.         allow_dangerous_deserialization=True
31.     )
32.     retriever = vector_store.as_retriever(search_kwargs={"k": 3})
33.     print("知识库加载成功")
34. except Exception as e:
35.     print(f"加载知识库失败: {str(e)}")
36.     print("请确保已运行 build_knowledge_base.py 构建知识库")
37.     exit(1)
39. # 4. 定义提示模板
40. template = """你是一个专业的知识库助手，请基于以下上下文回答问题。
41. 如果不知道答案，请说"我不知道"，不要编造答案。
43. 上下文：
44. {context}
46. 问题：{question}
48. 请用中文给出详细回答："""
49. prompt = ChatPromptTemplate.from_template(template)
51. # 5. 构建 RAG 链
52. rag_chain = (
53.     {"context": retriever, "question": RunnablePassthrough()}
54.     | prompt
55.     | llm
56.     | StrOutputParser()
57. )
59. # 6. 格式化文档显示
60. def format_docs(docs):
61.     return "\n\n".join(doc.page_content for doc in docs)
63. # 7. 改进的输入函数（解决输入卡住问题）
64. def get_user_input(prompt, timeout=60):
65.     print(prompt, end='', flush=True)
66.    
67.     # 使用 select 检测输入可用性
68.     if select.select([sys.stdin], [], [], timeout)[0]:
69.         return sys.stdin.readline().strip()
70.     return None
72. # 8. 交互式问答
73. print("知识库问答系统已启动（输入 'exit' 退出）")
74. while True:
75.     try:
76.         # 使用改进的输入函数
77.         query = get_user_input("\n你的问题：")
78.         
79.         if query is None:
80.             print("\n输入超时，请重新输入...")
81.             continue
82.             
83.         if query.lower() == "exit":
84.             break
85.         
86.         start_time = time.time()
87.         
88.         # 显示检索到的参考内容
89.         relevant_docs = retriever.invoke(query)
90.         print("\n[检索到的参考内容]")
91.         for i, doc in enumerate(relevant_docs[:2]):  # 显示前2个相关片段
92.             print(f"\n片段 {i+1}:\n{doc.page_content[:200]}...")
93.         
94.         # 生成答案
95.         response = rag_chain.invoke(query)
96.         
97.         end_time = time.time()
98.         
99.         print(f"\n[答案] (耗时：{end_time - start_time:.2f}秒)")
100.         print(response)
101.         
102.         # 确保输出缓冲区刷新
103.         sys.stdout.flush()
104.         
105.     except KeyboardInterrupt:
106.         print("\n退出系统...")
107.         break
108.     except Exception as e:
109.         print(f"处理问题时出错: {str(e)}")
110.         print("请尝试简化问题或稍后重试")
111.         # 清除可能的输入缓冲区残留
112.         sys.stdin.readline()

复制代码

五、测试验证

python3 rag_query.py

根据提示词输入

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！