2026版大模型应用源码_RAG检索增强生成系统

import os import pdfplumber from docx import Document from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Chroma import whisper # 1. 初始化工具 # 嵌入模型初始化（中文适配） embeddings = HuggingFaceEmbeddings( model_name=”BAAI/bge-large-zh-v1.5″, model_kwargs={‘device’: ‘cpu’}, # 支持GPU，修改为’cuda’即可 encode_kwargs={‘normalize_embeddings’: True} ) # 文本分割器（递归切分，平衡语义与长度） text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 片段大小，可根据大模型上下文窗口调整 chunk_overlap=200, # 片段重叠长度，防止关键信息割裂 separators=[“\n\n”, “\n”, “. “, “! “, “? “, ” “, “”] # 分隔符优先级 ) # 音视频转文字工具（Whisper） whisper_model = whisper.load_model(“base”) # 2. 多源数据读取函数（支持PDF、Word、TXT、MP3/MP4） def load_data(file_path): “”” 多源数据读取，自动识别文件格式，提取文本内容 :param file_path: 文件路径 :return: 提取的纯文本内容 “”” text = “” # 读取PDF文件（文本版+扫描版） if file_path.endswith(“.pdf”): with pdfplumber.open(file_path) as pdf: for page in pdf.pages: # 尝试提取文本，若为空则视为扫描版，进行OCR识别 page_text = page.extract_text() or “” if not page_text: # 扫描版PDF转图像，OCR识别（需安装Tesseract） img = page.to_image() page_text = img.ocr_text() text += page_text + “\n” # 读取Word文件 elif file_path.endswith(“.docx”): doc = Document(file_path) for para in doc.paragraphs: text += para.text + “\n” # 读取TXT文件 elif file_path.endswith(“.txt”): with open(file_path, “r”, encoding=”utf-8″) as f: text = f.read() # 读取音视频文件（转文字） elif file_path.endswith((“.mp3”, “.mp4″)): result = whisper_model.transcribe(file_path, language=”zh”) text = result[“text”] else: raise ValueError(“不支持的文件格式，请上传PDF、Word、TXT、MP3/MP4格式文件”) # 数据去噪（过滤空白字符、冗余换行） text = text.replace(“\n\n\n”, “\n”).replace(” “, ” “).strip() return text # 3. 文本分割与向量存储（核心函数） def build_knowledge_base(data_dir, persist_dir=”./chroma_db”): “”” 构建知识库，将多源数据分割后，生成向量并存储到Chroma :param data_dir: 数据源文件夹路径（存放各类文件） :param persist_dir: 向量数据库持久化路径 :return: 向量数据库实例 “”” # 遍历文件夹内所有文件 texts = [] metadatas = [] for filename in os.listdir(data_dir): file_path = os.path.join(data_dir, filename) if os.path.isfile(file_path): try: # 读取文件内容 content = load_data(file_path) # 文本分割 split_texts = text_splitter.split_text(content) # 添加元数据（文件名称、来源，用于检索过滤） for text in split_texts: texts.append(text) metadatas.append({“source”: filename, “type”: file_path.split(“.”)[-1]}) except Exception as e: print(f”处理文件{filename}失败：{str(e)}”) continue # 生成向量并存储到Chroma vector_store = Chroma.from_texts( texts=texts, embedding=embeddings, metadatas=metadatas, persist_directory=persist_dir ) # 持久化存储 vector_store.persist() print(f”知识库构建完成，共存储{len(texts)}个文本片段，持久化路径：{persist_dir}”) return vector_store # 测试函数（运行前需创建data文件夹，放入测试文件） if __name__ == “__main__”: build_knowledge_base(data_dir=”./data”)

from langchain.retrievers import VectorStoreRetriever, EnsembleRetriever from langchain.retrievers.keyword import KeywordRetriever from langchain.vectorstores import Chroma from langchain.embeddings import HuggingFaceEmbeddings # 初始化向量数据库和嵌入模型（与预处理模块一致） embeddings = HuggingFaceEmbeddings( model_name=”BAAI/bge-large-zh-v1.5″, model_kwargs={‘device’: ‘cpu’}, encode_kwargs={‘normalize_embeddings’: True} ) vector_store = Chroma( persist_directory=”./chroma_db”, embedding_function=embeddings ) # 1. 构建混合检索器（向量检索+关键词检索） def build_hybrid_retriever(): # 向量检索器（核心，负责语义匹配） vector_retriever = VectorStoreRetriever( vector_store=vector_store, search_kwargs={“k”: 5} # 返回5个最相似文本片段 ) # 关键词检索器（辅助，负责精确匹配关键词、专业术语） keyword_retriever = KeywordRetriever( vectorstore=vector_store, k=3 # 返回3个关键词匹配度最高的片段 ) # 混合检索器（融合两种检索结果，权重可调整） hybrid_retriever = EnsembleRetriever( retrievers=[vector_retriever, keyword_retriever], weights=[0.7, 0.3] # 向量检索权重70%，关键词检索权重30% ) return hybrid_retriever # 2. 多轮检索优化（根据用户追问，动态调整检索策略） def multi_round_retrieval(query, history_queries=None): “”” 多轮检索，结合历史查询记录，优化检索结果 :param query: 当前用户查询 :param history_queries: 历史查询记录（列表） :return: 优化后的检索结果 “”” hybrid_retriever = build_hybrid_retriever() # 如果有历史查询，拼接上下文，提升检索精准度 if history_queries: context_query = ” “.join(history_queries) + ” ” + query retrieved_docs = hybrid_retriever.get_relevant_documents(context_query) else: retrieved_docs = hybrid_retriever.get_relevant_documents(query) # 去重处理（避免重复片段） unique_docs = [] seen_content = set() for doc in retrieved_docs: if doc.page_content not in seen_content: seen_content.add(doc.page_content) unique_docs.append(doc) return unique_docs # 测试检索功能 if __name__ == “__main__”: # 单轮检索测试 query = “2026年RAG技术的核心突破是什么？” docs = multi_round_retrieval(query) print(“单轮检索结果：”) for i, doc in enumerate(docs, 1): print(f”{i}. 来源：{doc.metadata[‘source’]}”) print(f”内容：{doc.page_content[:200]}…\n”) # 多轮检索测试（模拟用户追问） history_queries = [“2026年RAG技术的核心突破是什么？”] new_query = “这些突破在医疗领域有哪些应用？” multi_docs = multi_round_retrieval(new_query, history_queries) print(“多轮检索结果：”) for i, doc in enumerate(multi_docs, 1): print(f”{i}. 来源：{doc.metadata[‘source’]}”) print(f”内容：{doc.page_content[:200]}…\n”)

from langchain.chat_models import ChatQwen from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler # 初始化大模型（Qwen 3 Max，支持本地部署） llm = ChatQwen( model_name=”qwen-max”, temperature=0.3, # 温度越低，答案越严谨，减少幻觉 streaming=True, callbacks=[StreamingStdOutCallbackHandler()], # 流式输出，提升用户体验 # 本地部署需添加model_path参数，指向本地模型路径 # model_path=”./qwen-max”, ) # 1. 定制Prompt（幻觉抑制+依据引用） prompt_template = “”” 你是一个基于检索增强的智能回答助手，严格按照以下规则生成答案： 1. 答案必须完全基于提供的检索结果（上下文），不得编造任何未提及的信息，若检索结果不足，直接说明“暂无相关信息”。 2. 回答时需分点清晰，关键信息加粗，每一条观点都要标注对应的来源（格式：【来源：XXX】）。 3. 语言简洁专业，避免冗余，若用户问题与检索结果无关，直接拒绝回答，不发散话题。 4. 若检索结果中有多个来源的信息，需整合分析，避免重复表述。 上下文： {context} 用户问题：{question} 请生成符合要求的回答： “”” PROMPT = PromptTemplate( template=prompt_template, input_variables=[“context”, “question”] ) # 2. 构建检索-生成链条（整合检索结果与大模型生成） def build_rag_chain(): hybrid_retriever = build_hybrid_retriever() rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type=”stuff”, # 适合短文本检索结果，直接将所有片段传入大模型 retriever=hybrid_retriever, chain_type_kwargs={“prompt”: PROMPT}, return_source_documents=True # 返回检索来源，便于验证答案 ) return rag_chain # 3. 核心生成函数（调用链条，生成答案） def generate_answer(query, history_queries=None): “”” 生成最终答案，整合检索结果，抑制幻觉 :param query: 用户问题 :param history_queries: 历史查询记录 :return: 生成的答案 + 检索来源 “”” rag_chain = build_rag_chain() # 多轮检索，获取优化后的上下文 retrieved_docs = multi_round_retrieval(query, history_queries) # 拼接上下文 context = “\n”.join([f”【来源：{doc.metadata[‘source’]}】\n{doc.page_content}” for doc in retrieved_docs]) # 调用链条生成答案 result = rag_chain({“query”: query, “context”: context}) # 提取答案和来源 answer = result[“result”] sources = [f”【来源：{doc.metadata[‘source’]}】{doc.page_content[:100]}…” for doc in result[“source_documents”]] return answer, sources # 测试生成功能 if __name__ == “__main__”: query = “2026版RAG系统相比2025版，有哪些核心升级？” answer, sources = generate_answer(query) print(“\n最终答案：”) print(answer) print(“\n检索来源：”) for source in sources: print(source)