资讯详情

深入了解 · 专业解读

RAG 应用开发

发布时间：2026/6/27 8:28:06
RAG 应用开发 数据集一、Naive RAG检索增强生成2. 文档加载Loaders下载依赖uv pip install pypdf2.2 常用文档加载器from langchain_community.document_loaders import PyPDFLoader # 加载单个 PDF 文件 loader PyPDFLoader(documents/report.pdf) pages loader.load() # 查看加载结果 for page in pages: print(f页码: {page.metadata[page]}) print(f内容预览: {page.page_content[:200]}...) print(- * 50)2.2.2 Markdown 文档加载下载依赖uv pip install unstructured uv pip install markdownfrom langchain_community.document_loaders import UnstructuredMarkdownLoader loader UnstructuredMarkdownLoader(docs/README.md) documents loader.load()使用TextLoader# 极简版加载Markdown文件并打印内容 from langchain_community.document_loaders import TextLoader # 1. 初始化加载器指定文件路径和编码 loader TextLoader(docs/LangChain.md, encodingutf-8) # 2. 加载文档 documents loader.load() # 3. 打印核心内容只输出文本不打印原始对象 print(文档内容\n, documents[0].page_content)2.2.3 网页加载from langchain_community.document_loaders import WebBaseLoader # 加载单个网页 loader WebBaseLoader(https://reference.langchain.org.cn/python/langchain_core) docs loader.load() print(docs) # 加载多个网页 loader WebBaseLoader([ https://example.com/page1, https://example.com/page2 ]) docs loader.load()2.2.4 批量目录加载from langchain_community.document_loaders import DirectoryLoader from langchain_community.document_loaders import TextLoader # 加载目录下所有 .txt 文件 loader DirectoryLoader( path./documents/, glob**/*.txt, # 匹配模式 loader_clsTextLoader, # 使用的加载器类 show_progressTrue # 显示进度条 ) documents loader.load() print(f共加载 {len(documents)} 个文档)2.3 加载注意事项# 处理编码问题的示例 from langchain_community.document_loaders import TextLoader loader TextLoader( docs/RAG课程大纲.md, encodingutf-8, # 指定编码 autodetect_encodingTrue # 或自动检测 ) def process_document(doc): print(doc.page_content) # 大文件使用 lazy_load for doc in loader.lazy_load(): # 逐个处理文档减少内存占用 process_document(doc)3. 文本分割Text Splitter3.2.1 RecursiveCharacterTextSplitter核心推荐from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 加载单个 PDF 文件 loader PyPDFLoader(docs/LangChain.pdf) pages loader.load() print(pages) str for page in pages: str page.page_content splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个 chunk 的最大字符数 chunk_overlap50, # chunk 之间的重叠字符数 length_functionlen, # 计算长度的函数 separators[ # 分隔符优先级从高到低 \n\n, # 段落 \n, # 换行 。, # 中文句号 ., # 英文句号 , # 空格 # 字符级别 ] ) chunks splitter.split_text(str) for i, chunk in enumerate(chunks): print(fChunk {i1}: {len(chunk)} 字符) print(chunk[:1000] ...) print(- * 40)3.2.2 TokenTextSplitterToken 感知from langchain_text_splitters import TokenTextSplitter splitter TokenTextSplitter( chunk_size200, # Token 数量 chunk_overlap20, encoding_namecl100k_base # GPT-4 使用的编码 )代码示例对比不同参数效果from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader # 加载单个 PDF 文件 loader TextLoader( docs/LangChain.md, encodingutf-8 ) pages loader.load() print(pages) str for page in pages: str page.page_content # 对比不同 chunk_size 的效果 for size in [200, 500, 1000]: splitter RecursiveCharacterTextSplitter( chunk_sizesize, chunk_overlapint(size * 0.1) ) chunks splitter.split_text(str) print(fchunk_size{size}: 生成 {len(chunks)} 个 chunks)3.4 分割最佳实践from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader # 1. 加载文档 loader PyPDFLoader(docs/LangChain.pdf) documents loader.load() # 2. 创建分割器推荐配置 splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, ., , , , ] ) # 3. 分割文档保留元数据 chunks splitter.split_documents(documents) # 4. 验证分割结果 print(f原始文档数: {len(documents)}) print(f分割后 chunks 数: {len(chunks)}) print(f平均 chunk 长度: {sum(len(c.page_content) for c in chunks) / len(chunks):.0f})4. Embedding嵌入模型4.3 LangChain 调用 Embedding4.3.1 使用 OpenAI Embeddingfrom langchain_openai import OpenAIEmbeddings # 初始化 Embedding 模型 embeddings OpenAIEmbeddings( modeltext-embedding-3-small, # api_key 从环境变量 OPENAI_API_KEY 自动读取 ) # 单条文本嵌入 text LangChain 是一个强大的 LLM 应用开发框架 vector embeddings.embed_query(text) print(f向量维度: {len(vector)}) print(f向量前5维: {vector[:5]}) # 批量文本嵌入 texts [ 什么是机器学习, 深度学习和机器学习的区别, 如何入门人工智能 ] vectors embeddings.embed_documents(texts) print(f生成了 {len(vectors)} 个向量)4.3.2 使用阿里云 DashScope通义千问from dotenv import load_dotenv import os from langchain_community.embeddings import DashScopeEmbeddings load_dotenv() embeddings DashScopeEmbeddings( modeltext-embedding-v1, # 官方文档目前示例用 v1 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 单条文本嵌入 text LangChain 是一个强大的 LLM 应用开发框架 vector embeddings.embed_query(text) print(f向量维度: {len(vector)}) print(f向量前5维: {vector[:5]}) # 批量文本嵌入 texts [ 什么是机器学习, 深度学习和机器学习的区别, 如何入门人工智能 ] vectors embeddings.embed_documents(texts) print(f生成了 {len(vectors)} 个向量)5. Vector Store向量数据库5.3 使用 Chroma上手简单5.3.1 基本使用使用from_texts()加载数据示例代码from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from dotenv import load_dotenv import os from langchain_community.embeddings import DashScopeEmbeddings load_dotenv() # 初始化 Embedding embeddings DashScopeEmbeddings( modeltext-embedding-v1, # 官方文档目前示例用 v1 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 准备文档 texts [ LangChain 是一个用于开发 LLM 应用的框架, RAG 是检索增强生成的缩写, 向量数据库用于存储和检索向量, Embedding 将文本转换为向量表示 ] # 创建向量数据库内存模式 vectorstore Chroma.from_texts( textstexts, embeddingembeddings, collection_namemy_collection, #persist_directory./chroma_db # ← 这样就会往硬盘写入 ) # 执行相似度搜索 similarity_search results vectorstore.similarity_search( query什么是 LangChain, k2 # 返回 top 2 结果 ) for doc in results: print(doc.page_content)使用from_documents()加载数据from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from dotenv import load_dotenv import os from langchain_community.embeddings import DashScopeEmbeddings load_dotenv() from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader # 1. 加载文档 loader PyPDFLoader(docs/LangChain.pdf) documents loader.load() # 2. 创建分割器推荐配置 splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, ., , , , ] ) # 3. 分割文档保留元数据 chunks splitter.split_documents(documents) # 初始化 Embedding embeddings DashScopeEmbeddings( modeltext-embedding-v1, # 官方文档目前示例用 v1 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 创建持久化向量数据库 vectorstore Chroma.from_documents( documentschunks, # 分割后的文档 embeddingembeddings, persist_directory./chroma_db, # 持久化目录 collection_nameknowledge_base ) # 执行相似度搜索 similarity_search results vectorstore.similarity_search( query什么是 LangChain, k2 # 返回 top 2 结果 ) idx 0 for doc in results: print(idx1 ,- ,doc.page_content) idx 15.3.2 持久化存储使用Chroma() 加载已有数据# 后续加载已有数据库 vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings, collection_nameknowledge_base )5.3.3 带 Metadata 的存储与过滤from langchain_core.documents import Document from langchain_chroma import Chroma from dotenv import load_dotenv import os from langchain_community.embeddings import DashScopeEmbeddings load_dotenv() # 创建带元数据的文档 documents [ Document( page_contentPython 是一种编程语言, metadata{source: python.pdf, page: 1, category: programming} ), Document( page_content机器学习基础概念, metadata{source: ml.pdf, page: 1, category: ai} ), ] # 初始化 Embedding embeddings DashScopeEmbeddings( modeltext-embedding-v1, dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 存储到向量库 vectorstore Chroma.from_documents( documentsdocuments, embeddingembeddings ) # 带过滤条件的搜索 results vectorstore.similarity_search( query基础, k5, #filter{category: programming} # 只搜索编程类文档 ) # 打印结果 还有对应的序号 for idx, doc in enumerate(results): print(idx1 ,- ,doc.page_content) idx 16. Retriever检索器6.2 从 VectorStore 创建 Retriever使用.as_retriever()是 LangChain v1.x 的推荐方式# 创建基本检索器 retriever vectorstore.as_retriever() # 配置检索参数 retriever vectorstore.as_retriever( search_typesimilarity, # 检索类型 search_kwargs{ k: 4, # 返回 top-k 结果 } ) # 使用检索器 docs retriever.invoke(什么是 RAG) for doc in docs: print(doc.page_content[:100])7. RAG Runnable Pipelinev1.x 官方推荐方式7.2 Pipeline 分段详解7.2.1 Context Retriever Runnablefrom langchain_core.runnables import RunnablePassthrough # 检索器作为 Runnable retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 格式化检索结果的函数 def format_docs(docs): 将检索到的文档格式化为字符串 return \n\n.join( f[来源: {doc.metadata.get(source, 未知)}]\n{doc.page_content} for doc in docs )7.2.2 Prompt Runnable动态注入 contextfrom langchain_core.prompts import ChatPromptTemplate # RAG 专用 Prompt 模板 RAG_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个专业的问答助手。请根据以下提供的上下文信息回答用户的问题。 要求 1. 只使用上下文中的信息回答问题 2. 如果上下文中没有相关信息请诚实地说根据现有资料我无法回答这个问题 3. 回答要简洁准确并在适当时候引用来源 上下文信息 {context}), (human, {question}) ])7.2.3 LLM Runnablefrom langchain_openai import ChatOpenAI from dotenv import load_dotenv import os load_dotenv() # 使用 DashScope通义千问模型 llm ChatOpenAI( modelos.getenv(DASHSCOPE_MODEL_NAME), api_keyos.getenv(DASHSCOPE_API_KEY), base_urlos.getenv(DASHSCOPE_BASE_URL), temperature0.3, # RAG 场景建议使用较低的 temperature )7.2.4 Parser Runnablefrom langchain_core.output_parsers import StrOutputParser output_parser StrOutputParser()7.3 组装完整 RAG Chainfrom langchain_core.runnables import RunnablePassthrough, RunnableParallel # 方式1使用 RunnableParallel 组织输入 rag_chain ( RunnableParallel( contextretriever | format_docs, questionRunnablePassthrough() ) | RAG_PROMPT | llm | output_parser ) # 方式2更简洁的写法使用字典 rag_chain ( { context: retriever | format_docs, question: RunnablePassthrough() } | RAG_PROMPT | llm | output_parser ) # 调用 RAG Chain question 什么是 LangChain answer rag_chain.invoke(question) print(answer)7.4 完整代码示例 RAG Pipeline 完整实现示例 from langchain_openai import ChatOpenAI from langchain_community.embeddings import DashScopeEmbeddings from langchain_chroma import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader from dotenv import load_dotenv import os load_dotenv() # 1. 模型配置 llm ChatOpenAI( modelos.getenv(DASHSCOPE_MODEL_NAME), api_keyos.getenv(DASHSCOPE_API_KEY), base_urlos.getenv(DASHSCOPE_BASE_URL), temperature0.3, ) embeddings DashScopeEmbeddings( modeltext-embedding-v1, # DashScope 的 embedding 模型 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 2. 文档处理 # 加载文档 loader PyPDFLoader(docs/LangChain.pdf) documents loader.load() # 分割文档 splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, ., , ] ) chunks splitter.split_documents(documents) # 3. 向量存储 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db ) # 4. 创建检索器 retriever vectorstore.as_retriever( search_typemmr, search_kwargs{k: 4, fetch_k: 10} ) # 5. 构建 RAG Chain def format_docs(docs): return \n\n---\n\n.join( f[文档片段 {i 1}]\n{doc.page_content} for i, doc in enumerate(docs) ) RAG_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个专业的文档问答助手。 参考资料 {context} 请根据上述参考资料回答问题。如果资料中没有相关信息请说明。), (human, {question}) ]) rag_chain ( { context: retriever | format_docs, question: RunnablePassthrough() } | RAG_PROMPT | llm | StrOutputParser() ) # 6. 运行问答 if __name__ __main__: while True: question input(\n请输入问题输入 quit 退出) if question.lower() quit: break answer rag_chain.invoke(question) print(f\n回答{answer})7.5 调试 Runnable Chain使用 astream_events 查看执行过程import asyncio async def debug_rag_chain(): question 什么是向量数据库 async for event in rag_chain.astream_events(question, versionv2): kind event[event] name event[name] if kind on_retriever_end: print(f\n 检索完成获取到 {len(event[data][output])} 个文档) elif kind on_chat_model_stream: print(event[data][chunk].content, end, flushTrue) elif kind on_chain_end: print(f\n✅ {name} 执行完成) # 运行调试 asyncio.run(debug_rag_chain())8. RAG 实战构建本地 PDF 阅读助手8.3 requirements.txtlangchain1.1 langchain-openai1.1.0 langchain-chroma0.1.0 langchain-community0.4.1 pypdf6.4.1 python-dotenv1.2.18.4 .env 配置示例# 使用阿里云 DashScope DASHSCOPE_API_KEYyour_api_key_here DASHSCOPE_MODEL_NAMEqwen-plus DASHSCOPE_BASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v18.5 完整项目代码 PDF 阅读助手 - 完整实现 支持多文件加载、来源追踪、CLI 交互 import os from typing import List from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader from langchain_core.messages import HumanMessage, AIMessage from langchain_community.embeddings import DashScopeEmbeddings load_dotenv() class PDFReadingAssistant: PDF 阅读助手类 def __init__(self, pdf_path: str None, pdf_directory: str None): 初始化 PDF 阅读助手 Args: pdf_path: 单个 PDF 文件路径 pdf_directory: PDF 文件目录 # 初始化模型 self.llm ChatOpenAI( modelos.getenv(DASHSCOPE_MODEL_NAME, qwen-plus), api_keyos.getenv(DASHSCOPE_API_KEY), base_urlos.getenv(DASHSCOPE_BASE_URL), temperature0.3, ) self.embeddings DashScopeEmbeddings( modeltext-embedding-v1, # DashScope 的 embedding 模型 dashscope_api_keyos.getenv(DASHSCOPE_API_KEY), ) # 加载和处理文档 self.documents self._load_documents(pdf_path, pdf_directory) self.chunks self._split_documents(self.documents) # 创建向量库和检索器 self.vectorstore self._create_vectorstore(self.chunks) self.retriever self.vectorstore.as_retriever( search_typemmr, search_kwargs{k: 4, fetch_k: 10} ) # 创建 RAG Chain self.rag_chain self._create_rag_chain() # 对话历史 self.chat_history: List [] def _load_documents(self, pdf_path: str, pdf_directory: str) - List[Document]: 加载 PDF 文档 documents [] if pdf_path: loader PyPDFLoader(pdf_path) documents.extend(loader.load()) print(f✅ 已加载文件: {pdf_path}) if pdf_directory: loader DirectoryLoader( pdf_directory, glob**/*.pdf, loader_clsPyPDFLoader, show_progressTrue ) documents.extend(loader.load()) print(f✅ 已加载目录: {pdf_directory}) print(f 共加载 {len(documents)} 个文档页面) return documents def _split_documents(self, documents: List[Document]) - List[Document]: 分割文档 splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap50, separators[\n\n, \n, 。, ., , , , ] ) chunks splitter.split_documents(documents) print(f 文档已分割为 {len(chunks)} 个片段) return chunks def _create_vectorstore(self, chunks: List[Document]) - Chroma: 创建向量数据库 vectorstore Chroma.from_documents( documentschunks, embeddingself.embeddings, persist_directory./pdf_assistant_db ) print( 向量数据库创建完成) return vectorstore def _format_docs_with_source(self, docs: List[Document]) - str: 格式化文档包含来源信息 formatted [] for i, doc in enumerate(docs): source doc.metadata.get(source, 未知来源) page doc.metadata.get(page, ?) formatted.append( f[片段 {i 1} | 来源: {os.path.basename(source)} | 第 {page 1} 页]\n f{doc.page_content} ) return \n\n---\n\n.join(formatted) def _create_rag_chain(self): 创建 RAG Chain prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的 PDF 文档阅读助手。你的任务是根据提供的文档内容回答用户问题。 规则 1. 只使用提供的文档内容回答问题 2. 如果文档中没有相关信息请诚实告知 3. 在回答时请指出信息来源如根据第X页... 4. 回答要准确、简洁、有条理 参考文档 {context}), MessagesPlaceholder(variable_namechat_history), (human, {question}) ]) chain ( { context: self.retriever | self._format_docs_with_source, question: RunnablePassthrough(), chat_history: lambda x: self.chat_history } | prompt | self.llm | StrOutputParser() ) return chain def ask(self, question: str) - str: 向助手提问 Args: question: 用户问题 Returns: 助手回答 answer self.rag_chain.invoke(question) # 更新对话历史 self.chat_history.append(HumanMessage(contentquestion)) self.chat_history.append(AIMessage(contentanswer)) # 保持对话历史在合理长度 if len(self.chat_history) 10: self.chat_history self.chat_history[-10:] return answer def clear_history(self): 清除对话历史 self.chat_history [] print( 对话历史已清除) def main(): 主函数 - CLI 交互 # 初始化助手示例加载当前目录下的 PDF assistant PDFReadingAssistant( pdf_directorydocs # 修改为你的 PDF 目录 ) print(\n 提示输入问题开始对话输入 quit 退出输入 clear 清除历史\n) while True: question input( 你的问题: ).strip() if not question: continue elif question.lower() quit: print( 再见) break elif question.lower() clear: assistant.clear_history() continue try: answer assistant.ask(question) print(f\n 回答:\n{answer}\n) print(- * 60) except Exception as e: print(f❌ 发生错误: {e}) if __name__ __main__: main()二、Advanced RAG进阶 RAG1. 预检索优化上下文增强分块策略1.4 句子窗口检索Sentence Window Retrieval代码实现第一步按句子分割文档

相关新闻