RAG大模型开发初探 || 动手大模型应用开发

何为RAG?

大型语言模型（LLM）相较于传统的语言模型具有更强大的能力，然而在某些情况下，它们仍可能无法提供准确的答案。为了解决大型语言模型在生成文本时面临的一系列挑战，提高模型的性能和输出质量，研究人员提出了一种新的模型架构：检索增强生成（RAG, Retrieval-Augmented Generation）。该架构巧妙地整合了从庞大知识库中检索到的相关信息，并以此为基础，指导大型语言模型生成更为精准的答案，从而显著提升了回答的准确性与深度

安装 

conda+ ollama

测试环境可以一步安装：深入后再详细了解每一个做什么的

langchain
langchain-community
pypdf
playwright
beautifulsoup4
nest_asyncio
"langchain-unstructured[local]"
scikit-learn
langchain-ollama
chromadb

 例如：pip install langchain   ollama run deepseek-r1:1.5b

 RAG技术路线

检索（Retrieval）

 1.文档加载：

 一：从PDF文件加载

技术：langchain-community，开发文档：How to load PDFs | 🦜️🔗 LangChain

安装：

pip install langchain-community
pip install pypdf

PyPDFLoader

langchain-community包中的pdf文档加载和处理类

参数：

‌file_path（必需参数）：

• ‌作用‌：指定PDF文件路径（本地路径或URL）

loader = PyPDFLoader("./data/document.pdf")  # 本地文件 :ml-citation{ref="1,2" data="citationList"}
loader = PyPDFLoader("http://example.com/doc.pdf")  # 网络文件 :ml-citation{ref="7" data="citationList"}

 ‌password（可选参数）

• ‌作用‌：解密受密码保护的PDF文件 

loader = PyPDFLoader("encrypted.pdf", password="your_password")

extract_images（可选参数）

• ‌作用‌：是否启用OCR提取图像中的文本（需额外安装OCR库）

headers（可选参数）

‌作用‌：自定义HTTP请求头（配合URL加载时使用）

headers = {"User-Agent": "My-App/1.0"}
loader = PyPDFLoader("http://api.example.com/doc.pdf", headers=headers)

🔍 返回对象结构

加载后的文档返回Document对象列表，每个对象包含：

Document(page_content="PDF文本内容",  # 当前页的文本metadata={'source': '文件路径或URL',  # 如 "./data/doc.pdf"'page': 0  # 页码（从0开始） :ml-citation{ref="1,7" data="citationList"}}
)

‌load()：

一次性加载整个PDF文档，返回所有页的Document列表 

docs = loader.load()
print(docs[0].page_content)  # 输出第一页文本

‌‌lazy_load()

• 惰性加载（逐页处理），返回生成器对象

• ‌优势‌：节省内存，适合大文件处理 

for page in loader.lazy_load():process(page)  # 逐页处理

‌‌load_and_split(text_splitter=None)

• 加载文档并自动分块（需提供文本分割器）

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
chunks = loader.load_and_split(text_splitter=splitter) :ml-citation{ref="3,10" data="citationList"}

‌扫描PDF

PyPDFLoader本身无法直接解析图片型PDF（扫描件/图像内容），

使用UnstructuredFileLoader处理扫描件，再与PyPDFLoader结果合并

def load_pdf(file_path: Path):# 先走PyPDFLoader# try:#     loader = PyPDFLoader(file_path)#     pages = loader.load()#     if any(doc.page_content.strip() for doc in pages):#         return pages# except:#     pass# 纯文本走UnstructuredFileLoader:OCRloader = UnstructuredLoader(file_paths=[file_path], api_key=os.getenv("UNSTRUCTURED_API_KEY"))docs = loader.load()return docs

TextLoader

langchain-community包中的text文档加载和处理类

参数和方法和PyPDFLoader基本一样

SQLDatabaseLoader

langchain-community数据库加载处理类

from langchain_community.utilities import SQLDatabase  
from langchain_community.document_loaders import SQLDatabaseLoader  # 连接数据库  
db = SQLDatabase.from_uri("sqlite:///example.db")  
# 加载数据  
loader = SQLDatabaseLoader(db, query="SELECT * FROM table_name")  
documents = loader.load()  

1.语义分块

LangChain的语义分块技术通过理解文本含义实现智能文档划分，主要分为基础规则分块和高级语义分块两类实现方式：

‌字符级分割

• 使用CharacterTextSplitter按指定字符（如逗号、换行符）拆分，适合结构化文本5
• 关键参数包括separator（分隔符）和chunk_overlap（重叠字符数）

def char_split(text):# 初始化分割器splitter = CharacterTextSplitter(separator="\n",  # 指定双换行符为分隔符： separator支持任意字符或字符串作为分隔标记（如。、?等标点）chunk_size=25,     # 单块最大字符数chunk_overlap=5,   # 块间重叠字符数length_function=len,is_separator_regex=False
)result = splitter.split_text(text)return result

‌递归分割

• RecursiveCharacterTextSplitter按多级分隔符（段落→句子→符号）逐层拆分，保持语义连贯性7
• 默认优先使用\n\n、\n和空格作为分隔层级

def recursive_char_split(text):splitter = RecursiveCharacterTextSplitter(chunk_size=10,          # 单块最大字符数chunk_overlap=3,        # 块间重叠字符数separators=["\n\n", "\n", "。", "？", "！", " ", ""]  # 优先级递减的分隔符
)result = splitter.split_text(text)return resulttext = "段落1内容。\n\n段落2内容！段落3内容？段落4内容..."
r = recursive_char_split(text)
for i, chunk in enumerate(r):print(chunk)

‌嵌入模型驱动

SemanticChunker通过计算句子向量相似度动态分块，需配合OpenA,deepseek, ollamaI等嵌入模型

from langchain_ollama import OllamaEmbeddings
from sklearn.metrics.pairwise import cosine_similarity
import numpy as npclass OllamaSemanticSplitter:def __init__(self, model_name="deepseek-r1:1.5b"):self.embeddings = OllamaEmbeddings(model=model_name)async def split_text(self, text, threshold=0.7):# 按句分割sentences = [s.strip() for s in text.split('。') if s]# 异步获取嵌入embeddings = await self.embeddings.aembed_documents(sentences)# 计算相邻句子相似度chunks, current_chunk = [], []for i in range(len(sentences)-1):sim = cosine_similarity([embeddings[i]], [embeddings[i+1]])[0][0]current_chunk.append(sentences[i])if sim < threshold:chunks.append("。".join(current_chunk))current_chunk = []# 添加最后部分if current_chunk:chunks.append("。".join(current_chunk))return chunksfrom asyncio import run
splitter = OllamaSemanticSplitter()
text = "这是一段测试文本。语义相似的句子会合并。差异大的句子会分割。"
chunks = run(splitter.split_text(text))  # 
print(chunks)

向量化

文档加载 → 文本分块 → 向量化 → 存储 → 相似度检索

通过嵌入模型（如OllamaEmbeddings）将文本转换为高维向量，捕捉语义特征

def retrieval_chain_vectors(documents):from langchain.vectorstores import Chroma# 初始化嵌入模型embeddings = OllamaEmbeddings(model="deepseek-r1:1.5b")# 加载并向量化文档vector_db = Chroma.from_documents(documents, embeddings)# 执行检索query = "如何优化RAG系统？"results = vector_db.similarity_search(query, k=3)return results
r = retrieval_chain_vectors(load_pdf(dir))
print(r)

向量化索引构建

• 使用嵌入模型（如BAAI/bge-small-zh-v1.5）将文本转换为高维向量

• ‌索引构建‌

  • ‌聚类预处理‌：对向量集进行聚类分簇，控制单簇向量数量（如HNSW层级结构）1
  • ‌索引类型选择‌：
    • HNSW：高召回率，适合动态增删78
    • IVFFlat：需预训练，不支持增量

以下是基于langchain_ollama的向量化索引构建完整方案：

安装nomic-embed-text模型：ollama pull nomic-embed-text

‌向量数据库：轻量级：Chroma（Python嵌入式）

def indexs_chain_vectors(doc):from langchain_community.vectorstores import Chroma# 初始化嵌入模型embeddings = OllamaEmbeddings(model="nomic-embed-text",base_url="http://localhost:11434"
)# 文档处理与分块text_splitter = RecursiveCharacterTextSplitter(chunk_size=20,chunk_overlap=5)documents = text_splitter.split_documents(doc)# 构建向量索引vector_db = Chroma.from_documents(documents=documents,embedding=embeddings,persist_directory="./chroma_db")

增强（Augmentation）

RAG（检索增强生成）中的‌增强（Augmentation）‌是指将检索到的外部信息与用户原始查询结合，作为上下文输入给大语言模型（LLM），以提升生成结果准确性和相关性的关键技术环节。其核心作用与实现逻辑如下：

‌解决LLM的固有缺陷‌

• ‌消除幻觉‌：通过注入真实检索结果，约束LLM生成与事实一致的答案，避免虚构信息39。
• ‌补充时效性知识‌：动态注入最新数据（如企业文档、实时资讯），突破LLM训练数据的时间限制411。
• ‌适配私域场景‌：通过专属知识库（如产品手册、内部报告）生成定制化内容1113

‌优化生成质量‌

• 提供精准上下文，使LLM生成的答案更具针对性和完整性18。
• 在专业领域（如医疗、法律）中提升术语准确性和逻辑严谨性

增强的技术实现

1. ‌上下文拼接（Context Concatenation）‌

   • 将用户查询 Q 与检索到的相关文本片段 [D1, D2,..., Dk] 拼接为增强后的输入：

     "基于以下信息：{D1}...{Dk}，回答：{Q}"
     

 ‌  2.提示词工程（Prompt Engineering）‌

• 设计结构化指令，明确要求LLM优先参考检索内容：

• 你是一名专业顾问，请严格根据提供的资料回答：
  资料：{D1}...{Dk}
  问题：{Q}
  

3‌.知识融合（Knowledge Fusion）

对多篇检索结果进行去重、排序、摘要，提炼核心信息后再输入LLM

上下文拼接

def context_augmentation(retrieved_context, user_question):from langchain_core.prompts import ChatPromptTemplatefrom langchain_community.llms import Ollama# 定义提示词模板（显式声明上下文依赖）prompt = ChatPromptTemplate.from_template("请根据以下背景资料回答问题：\n""---背景开始---\n""{context}\n""---背景结束---\n""问题：{question}"
)# 拼接并生成chain = prompt | Ollama(model="deepseek-r1:1.5b")response = chain.invoke({"context": retrieved_context,"question": user_question})return responseprint(context_augmentation("LangChain是一个用于连接大语言模型与外部数据的框架，支持RAG架构", "LangChain的主要作用是什么？"))

提示词工程（Prompt Engineering）

def retrieval_priority():from langchain.prompts import (ChatPromptTemplate,SystemMessagePromptTemplate,HumanMessagePromptTemplate
)from langchain_core.messages import AIMessage# 1. 定义系统级指令模板system_template = """你必须严格遵守以下规则：1. 当检索内容存在时，必须优先使用检索内容回答2. 引用格式要求：- 直接引用：保持原文不变，用「引用」标注- 补充说明：用[知识库补充]标注3. 禁止以下行为：- 修改检索内容的原始事实- 对未检索到内容的问题进行猜测"""system_prompt = SystemMessagePromptTemplate.from_template(system_template)# 2. 构建人类输入模板human_template = """检索到的相关内容：{context}用户问题：{question}"""human_prompt = HumanMessagePromptTemplate.from_template(human_template)# 3. 组合完整提示链full_prompt = ChatPromptTemplate.from_messages([system_prompt,human_prompt,AIMessage(content="我将严格遵循指令要求回答：")
])#   4. 使用示例formatted_prompt = full_prompt.format(context="LangChain的Memory模块包含ConversationBufferMemory...",question="如何实现多轮对话记忆？")print(formatted_prompt)

知识融合（Knowledge Fusion）

from langchain.prompts import ChatPromptTemplate
from langchain_community.graphs import Neo4jGraph
from langchain_community.vectorstores import Chromaclass KnowledgeFusionEngine:def __init__(self):# 1. 连接知识图谱与向量库self.graph = Neo4jGraph(url="bolt://localhost:7687")self.vector_db = Chroma(persist_directory="./vector_db")# 2. 定义融合提示模板self.fusion_prompt = ChatPromptTemplate.from_messages([("system", "你是一个知识融合专家，必须遵守：\n""1. 优先使用检索内容回答\n""2. 图谱实体用【】标注\n""3. 文档引用用『』标注"),("human", "知识图谱结果：\n{graph_data}\n\n""相关文档：\n{docs}\n\n""问题：{question}")])def query_fusion(self, question: str):# 3. 并行检索多源数据graph_data = self.graph.query(f"MATCH (n)-[r]->(m) WHERE n.label CONTAINS '{question}' RETURN n,r,m")docs = self.vector_db.similarity_search(question, k=3)# 4. 执行知识融合chain = self.fusion_prompt | ChatOpenAI(temperature=0)return chain.invoke({"graph_data": str(graph_data),"docs": "\n".join(d.page_content for d in docs),"question": question})

生成(Generation)

📌 案例1：医疗诊断辅助系统

• ‌检索内容‌：患者症状 + 最新医学期刊 + 电子健康记录12
• ‌生成过程‌：

  1. 输入：检索到“患者胸痛+心电图ST段抬高” 
  2. 生成输出：「根据2025年ACC指南（原文引用）」，ST段抬高提示急性心肌梗死，建议立即PCI干预[知识库补充]。

• ‌价值‌：误诊率降低30%，医生决策效率提升25%12

📌 案例2：金融报告自动撰写

• ‌检索内容‌：实时财报数据 + 历史市场分析 + 宏观经济指标10
• ‌生成过程‌：

  检索到：Q2营收同比增长20%（2025财报）、美联储加息预期 
  生成输出：公司营收增长强劲『2025Q2财报原文』，但需警惕[知识库补充]加息对融资成本的潜在影响。

• ‌价值‌：报告生产耗时减少40%，关键数据引用准确率>95%10

📌 案例3：客户服务多轮对话

• ‌检索内容‌：用户历史工单 + 产品文档 + 政策条款12
• ‌生成示例‌：

用户问：“订单退款为何延迟？”  
生成输出：您的订单ID#123处于『跨境清关流程（条款2.3）』，预计48小时内到账[知识库补充]。

 

例如：案例1

from ollama import Client
from typing import List, Dictclass CardiacDiagnosisSystem:def __init__(self):# 初始化医疗知识库self.chroma_client = chromadb.PersistentClient(path="./medical_db")self.guidelines_col = self.chroma_client.get_or_create_collection("acc_guidelines_2025",metadata={"source": "2025 ACC/AHA指南中文版"})# 连接Ollama医疗专用模型self.llm = Client(host='http://localhost:11434')self.embed_model = "llama3:8b-med"  # 医疗微调版本def retrieve_evidence(self, symptoms: str, ecg: str) -> Dict:# 多条件联合检索query = f"{symptoms} {ecg}"results = self.guidelines_col.query(query_texts=[query],n_results=3,where={"document_type": "diagnosis_guideline"})return {'guidelines': results['documents'][0],'metadatas': results['metadatas'][0]}def generate_alert(self, context: Dict) -> str:# 结构化提示模板prompt = f"""作为心脏科AI助手，请根据以下指南生成诊断建议：
指南内容：
{chr(10).join(context['guidelines'])}生成要求：
1. 必须标注「原文引用」并注明指南版本
2. 补充[临床知识库]关联信息
3. 明确建议干预措施及时限"""response = self.llm.generate(model=self.embed_model,prompt=prompt,options={'temperature': 0.3})return response['response']def diagnose(self, symptoms: str, ecg: str) -> str:evidence = self.retrieve_evidence(symptoms, ecg)return self.generate_alert(evidence)

向量数据库

向量数据库是用于高效计算和管理大量向量数据的解决方案。向量数据库是一种专门用于存储和检索向量数据（embedding）的数据库系统。它与传统的基于关系模型的数据库不同，它主要关注的是向量数据的特性和相似性。

在向量数据库中，数据被表示为向量形式，每个向量代表一个数据项。这些向量可以是数字、文本、图像或其他类型的数据。向量数据库使用高效的索引和查询算法来加速向量数据的存储和检索过程。

向量数据库中的数据以向量作为基本单位，对向量进行存储、处理及检索。向量数据库通过计算与目标向量的余弦距离、点积等获取与目标向量的相似度。当处理大量甚至海量的向量数据时，向量数据库索引和查询算法的效率明显高于传统数据库。

• Chroma：是一个轻量级向量数据库，拥有丰富的功能和简单的 API，具有简单、易用、轻量的优点，但功能相对简单且不支持GPU加速，适合初学者使用。
• Weaviate：是一个开源向量数据库。除了支持相似度搜索和最大边际相关性（MMR，Maximal Marginal Relevance）搜索外还可以支持结合多种搜索算法（基于词法搜索、向量搜索）的混合搜索，从而提高搜索结果的相关性和准确性。
• Qdrant：Qdrant使用 Rust 语言开发，有极高的检索效率和RPS（Requests Per Second），支持本地运行、部署在本地服务器及Qdrant云三种部署模式。且可以通过为页面内容和元数据制定不同的键来复用数据。