在前文中我们结合langchain和fastapi搭建了一个端到端的问答API这个agent可以调用已经封装好的工具函数可以获取本地数据库有记忆功能但是这样的模型训练好了过后只是就固定了如果没有获取或更新相应的知识agent很可能会产生幻觉即返回一个看似合理实则漏洞百出的答案因此我们要为agent提供一个随时可查阅的“外部数据库”。RAG是什么RAG检索增强生成检索Retrieval从外部知识库中寻找与问题相关的信息。增强Augmented将检索到的信息作为上下文提供给大模型LLM。生成GenerationLLM结合检索到的信息生成更准确、更可靠的回答。在此过程中用户给agent一个输入agent会拿着这个输入去外部知识库中比对将与输入问题相关的信息作为上下文提供给LLM从而得到更准确的回答。但是在RAG之前我们有一些准备工作要做。加载与分割loadsplit加载器Document loaderllm只能接受document格式的对象但是我们存储在文件里的数据可能是各种各样格式的txt、csv、pdf...Document loader的作用就是把这些不同格式的数据转换成langchain认识的document对象。常见的加载器loaderTextLoader(基础篇首选) 对应 .txt 文件。PyPDFLoder对应.pdf文件需要 pypdf 库。WebBaseLoader对应html内容CSVloader对应.csv文件它会将每一行视为一个独立的 Document。有了Document loader我们就可以统一使用loader.load()接口内部处理格式差异。import os from langchain_community.document_loaders import TextLoader # 创建一个演示txt文件 knowledge_base_content 《战争与和平》人物及背景研究笔记 1. 核心人物介绍 - 皮埃尔·别祖霍夫Pierre Bezukhov他是莫斯科最有钱的伯爵的私生子外表笨拙但内心善良一直在寻找人生的意义。 - 安德烈·尼古拉耶维奇·博尔孔斯基Andrei Bolkonsky一名勇敢且富有荣誉感的公爵在奥斯特利茨战役中受过重伤。 - 娜塔莎·罗斯托娃Natasha Rostova罗斯托夫伯爵的女儿充满活力是整部作品的灵魂人物。 2. 关键历史事件 - 1805年奥斯特利茨战役俄国与奥地利联军对抗拿破仑的法军结果惨败。 - 1812年博罗季诺战役拿破仑入侵俄国期间最惨烈的战斗。 3. 艺术评价 列夫·托尔斯泰通过这部作品展现了历史规律与个人意志之间的博弈。助手评价道“这是一部宏大的史诗喵读完需要很多耐心喵” with open(knowledge_base.txt, w, encodingutf8) as f: f.write(knowledge_base_content) # 加载 loader TextLoader(knowledge_base.txt, encodingutf8) docs loader.load() # 打印第一个 Document 对象查看结构 print(f成功加载了 {len(docs)} 个文档对象) print(f内容预览: {docs[0].page_content[:50]}...) print(f元数据信息: {docs[0].metadata})首先我们引入TextLoader工具然后通过loader TextLoader(knowledge_base.txt, encodingutf8)创建一个 TextLoader 的加载器实例对象这个对象知道我们要加载的文本已经编码格式用utf8编码格式可以处理中文建议以后都用此编码格式然后通过loader.load()启动加载并得到加载后的文本。分块器Text Splitter由于llm的输入长度是有上限的我们需要对加载好的document对象进行分块把它们分成一个个小的chunks最推荐、最常用的文本分块器是RecursiveCharacterTextSplitter它会自动按[\n\n, \n, , ]的优先级列表来分割最大限度地保留语义的完整性。这里涉及两个很重要的参数chunk_size、chunk_overlapchunk_size 是每个文本块的最大长度按token计算它规定了每个chunk最大的长度实际上每个chunk分割时都会小于这个数。chunk_overlap 是相邻两个文本块之间的重叠长度为了防止分块导致的上下文语义断裂它规定了每个chunk可以与其他chunk的重叠程度一般是chunk_size的10%。from langchain_text_splitters import RecursiveCharacterTextSplitter # 分割 text_splitter RecursiveCharacterTextSplitter( chunk_size250, chunk_overlap40 ) splits text_splitter.split_documents(docs) print(f分块结果:{len(splits)}) for i,doc in enumerate(splits): print(f片段{i1}(长度:{len(doc.page_content)})) print(doc.page_content) print(-*100\n)我们首先实例化一个分块器text_splitter并规定它的chunk_size与chunk_overlap然后将文本传入分块器最后得到document对象列表向量索引Embedding我们把文本分块过后chunk还要对它贴上“语义标签”这里类似于人工智能里的词嵌入把文本块变成高维向量。在向量空间中向量就是语义语义相近的文本会在向量空间中挨得很近比如马铃薯和土豆而语义相反的文本会在向量空间中离得很远比如热和冷不同 Embedding 模型产生的向量在语义表达能力、维度、上下文感知能力、长度支持等方面是不同的有些embedding模型在通用领域表现很好比如百科而有些模型垂直领域表现很好比如法律、医疗在这里我们使用免费的HuggingFaceEmbeddings本地模型完成embedding第一次运行需要在本地下载该模型如果不想占用本地资源也可以使用付费的OpenAIEmbeddings等模型。#获取embedding向量化模型 from pathlib import Path from langchain_huggingface import HuggingFaceEmbeddings def get_embeddings(model_nameBAAI/bge-small-zh-v1.5,devicecpu,**kwargs): # 支持更换其他向量化模型 local_dir Path(models)/model_name.replace(/, _) if not local_dir.exists(): print(f⚠️ 首次使用嵌入模型正在下载到{local_dir.absolute()}) print( 提示需要联网(必需梯子)完成后可离线使用) from huggingface_hub import snapshot_download # 模型下载工具 snapshot_download( repo_id model_name, local_dir local_dir, ) print(✅ 下载完成) # 构造参数字典 model_kwargs { device:device, local_files_only: True, # 仅使用本地文件 } # 此处实例化时把kwargs传入 _EMBEDDINGS HuggingFaceEmbeddings( model_name str(local_dir), # 使用本地已下载的模型 model_kwargs model_kwargs, **kwargs # 允许传入参数 ) return _EMBEDDINGS这段代码用于首次下载或加载向量化模型需要科学上网把这段代码存储的embedding.py文件中首次下载后就保存在本地中不需要重复下载。在这里get_embeddings函数封装了 Embedding 模型的获取逻辑首次调用时自动从 HuggingFace 下载指定模型到本地models/目录并缓存之后直接使用本地文件加载避免了重复下载。存储Vector Store我们把文本转换成向量过后最后一步就是为这些向量构建一个高效的索引。在向量空间中用户的问题被转换成一个查询向量Query不断与我们所拥有的文本向量进行相似度匹配并在向量空间中找到语义相似度最高的那几个向量。传统的数据库存储的是结构化数据将数据组织成数据表而向量数据库能够处理非结构化数据数据以向量的形式存储。向量数据库的核心是专门为高维空间优化的向量索引技术向量索引工具会基于数据库向量分布建立一个内部结构每当有查询向量出现时向量数据库会使用优化算法HNSW、IVF高效检索语义关系最相近的知识向量。FAISSFAISS是一个高效的向量相似度搜索库提供极致的搜索性能和灵活性支持大规模数据集FAISS关注的是在给定的硬件资源下提高搜索速度与结果质量但是FAISS有几个明显的缺点本质上是个内存向量搜索库主要在内存中操作因此缺乏持久化和数据管理能力FAISS本质上只是一个库不是一个数据库。它不负责数据存储、管理、备份无法对数据进行增删改如果知识库需要更新必须重建整个索引。Chorma由于上述FAISS的缺点我们可以看看Chroma。Chroma是一个轻量级的向量数据库内置了元数据过滤和CRUD增删改查操作通过persist_directory将向量、原始文档和元数据统一持久化到磁盘数据以文件形式保存在本地。简化原型开发与AI框架无缝集成简单易上手。所以在这里我们选择Chroma建立向量数据库这里谈一下向量数据库和一般数据库在人工智能中的区别一般数据库mySQL是硬寻址当KeyQuery时取出Key地址对应存储器中的Value值向量数据库Chroma是软寻址通过构建高维空间中的近似最近邻图来寻找相似向量下面是初始化向量数据库import os from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from embeddings import get_embeddings knowledge_base_file war_and_peace.txt # 持久化目录: Chroma会把所有数据(向量文本元数据)都存到这个文件夹 persist_directory ./chroma_db_war_and_peace_bge_small_en_v1.5 model_name_str BAAI/bge-small-en-v1.5 # 如果愿意等待可以换成模型BAAI/bge-m3效果更好更适合长文但下载时间也更久(2.2G) chunk_size 500 chunk_overlap 75 # 检查是否已创建 if os.path.exists(persist_directory): print(f检测到已存在的向量数据库: {persist_directory}) print(跳过索引构建。如需重新构建请手动删除该目录。) exit() if not os.path.exists(knowledge_base_file): print(f错误: 知识库文件 {knowledge_base_file} 未找到。) print(请从 https://www.gutenberg.org/ebooks/2600.txt.utf-8 下载) print(并重命名为 war_and_peace.txt 放在当前目录。) exit() print(---正在构建索引---) # 1. 加载 loader TextLoader(knowledge_base_file, encodingutf8) docs loader.load() print(加载完成...\n) # 2. 分割 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap ) splits text_splitter.split_documents(docs) print(分割完成...\n) # 3. 向量化 -- 第一次运行会下载模型,预计耗时2分钟 print(f正在加载/下载模型{model_name_str}...) embedding_model get_embeddings( model_namemodel_name_str, devicecpu, # 强制模型在cpu上运行 encode_kwargs{batch_size: 64} # 每次处理64个文本片段 ) print(Embedding模型加载完成...\n) # 4. 存储 print(正在构建Chroma索引...(注此步耗时较久预计要3min)\n) db Chroma( persist_directorypersist_directory, embedding_functionembedding_model ) # 分批添加切片chunks每批不超过 5000 batch_size 5000 # 必须 5461 for i in range(0, len(splits), batch_size): batch splits[i:i batch_size] db.add_documents(batch) print(f已插入 {min(i batch_size, len(splits))} / {len(splits)} 条) print(f✅ 索引构建完毕共 {len(splits)} 条已保存到 {persist_directory})把这个代码存储到一个文件中以构建向量数据库。注意如果直接调用Chroma.from_documents(docs) langchain会一次性把所有chunk全部传给Chroma而Chroma 默认使用SQLite存储数据时一次至多接受5461条记录所以需要分批次传入chroma在这里浅谈一下另外一个向量数据库MilvusMilvus是一个功能强大的生产级向量数据库拥有向量搜索、标量过滤、混合搜索向量关键词、范围搜索等查询能力通过在Docker部署Milvus也能实现增删改查操作CURD。Reranker 重排机制如果说检索retrieval对海量数据进行初步筛选那么reranker重排机制就是对数百个数据进行精准筛选。检索它只能理解“字面”的相似难以把握用户真正的意图如果你问python如何学习它可能会给你返回python书籍也可能返回c和java因为python和这些语言在文本中经常一起出现检索容易引入很多不相关的文本导致llm输出垃圾内容如果你问一本书的主角这本书主角名字贯穿全文检索很可能返回多个不相关的故事情节。为了解决这类问题我们引入了re-ranker重排序机制这是基础检索器我们从向量数据库中快速召回 50 个候选文档base_retriever db.as_retriever(search_kwargs{k: 50})这是重排序模型使用 HuggingFace 的交叉编码器Cross-Encoderencoder_model_name BAAI/bge-reranker-base encoder HuggingFaceCrossEncoder(model_nameencoder_model_name, model_kwargs{device: cpu}) reranker CrossEncoderReranker(modelencoder, top_n6)这里的top_n表示从海量初筛的50个向量里面精选6个向量出来Cross-Encoder的具体机制是将查询和文档拼接成一个整体然后使用自注意力机制让查询中的每个词都与文档中的所有词进行交互捕捉它们之间最细微的语义关联最后进行打分。model_kwargs是一个可选参数它是一个字典负责指定运行设备CPU / GPU设置模型精度半精度 / 全精度你也可以改成这样指定使用GPUmodel_kwargs{device: cuda}最后是封装retriever ContextualCompressionRetriever(base_retrieverbase_retriever, base_compressorreranker)使用ContextualCompressionRetriever把检索和重排联系在一起RAG封装总结一下在完成RAG之前我们需要对数据进行加载分块、对分块进行向量化编码、构建向量数据库进行存储。RAG中retrieval就是检索和重排进行结合Augmentation将检索到的信息与用户的输入整合成一个上下文来构建提示词Generation就是把增强后的提示词输出给llm最终得到结果。def build_rag_chain(llm_instance): #初始化RAG链 rag_chain_instance build_rag_chain(llm_instancellm) tool def search_data(query: str): 查询内容 return rag_chain_instance.invoke(query)格式转换这里注意一下retrieval检索最后输出的是document对象列表但是prompt的构建需要一个str的输入因此我们需要把document对象转换成str字符def format_docs(docs): return \n.join(doc.page_content for doc in docs)这里docs是一个列表列表元素是langchain的document对象doc.page_content for doc in docs这是一个生成器表达式。它遍历docs列表中的每一个doc对象并取出每个对象的page_content属性\n.join(...)字符串的join方法。它使用换行符\n作为连接符将生成器表达式取出的所有文本片段拼接成一个完整的字符串完整代码如下import os import uvicorn from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel # 导入配置 from config import OPENAI_API_KEY, SERVER_HOST, SERVER_PORT from embeddings import get_embeddings from langchain_chroma import Chroma from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough from langchain_classic.retrievers import ContextualCompressionRetriever from langchain_community.cross_encoders import HuggingFaceCrossEncoder from langchain_classic.retrievers.document_compressors import CrossEncoderReranker from langchain_core.tools import tool from langchain_classic.agents import AgentExecutor, create_tool_calling_agent from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.runnables import RunnableWithMessageHistory def build_rag_chain(llm_instance) #持久化目录为《飘》的数据库 persist_directory ./chroma_db_gone_with_the_wind_bge embedding_model_name BAAI/bge-small-en-v1.5 encoder_model_name BAAI/bge-reranker-base # 检查数据库是否存在 if not os.path.exists(persist_directory): print(f警告未找到数据库目录 {persist_directory}请确认索引已创建喵) embeddings_model get_embeddings(model_nameembedding_model_name, devicecpu) db Chroma(persist_directorypersist_directory, embedding_functionembeddings_model) # Reranker 配置 base_retriever db.as_retriever(search_kwargs{k: 50}) encoder HuggingFaceCrossEncoder(model_nameencoder_model_name, model_kwargs{device: cpu}) reranker CrossEncoderReranker(modelencoder, top_n6) retriever ContextualCompressionRetriever(base_retrieverbase_retriever, base_compressorreranker) #系统提示词背景 sys_prompt 你是一个精通美国内战历史和文学的专家尤其对小说《飘》Gone with the Wind有深入研究。 请根据以下关于《飘》的上下文回答问题。 [上下文]: {context} [问题]: {question} prompt ChatPromptTemplate.from_messages([(system, sys_prompt), (human, {question})]) def format_docs(docs): return \n.join(doc.page_content for doc in docs) return ( {context: retriever | format_docs, question: RunnablePassthrough()} | prompt | llm_instance | StrOutputParser() ) #Agent 组装 def create_agent_with_memory(): llm ChatOpenAI(modeldeepseek-chat, api_keyOPENAI_API_KEY, base_urlhttps://api.deepseek.com) prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手。你能查天气也能通过 search_gone_with_the_wind 工具查询小说《飘》的相关内容。), MessagesPlaceholder(variable_namehistory), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad) ]) rag_chain_instance build_rag_chain(llm_instancellm) tool def search_gone_with_the_wind(query: str): 查询小说《飘》(Gone with the Wind) 中的内容包括斯嘉丽、瑞德等人物情节和美国内战背景 return rag_chain_instance.invoke(query) tool def get_weather(location: str): 模拟获得天气信息 return f{location}当前天气23℃晴喵~ tools [get_weather, search_gone_with_the_wind] agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) store {} def get_session_history(session_id: str): if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] return RunnableWithMessageHistory( runnableagent_executor, get_session_historyget_session_history, input_messages_keyinput, history_messages_keyhistory, output_messages_keyoutput ) #FastAPI 初始化与路由 app FastAPI(titleRAG 助手) app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) final_agent create_agent_with_memory() class ChatRequest(BaseModel): message: str session_id: str class ChatResponse(BaseModel): output: str session_id: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: response final_agent.invoke( {input: request.message}, config{configurable: {session_id: request.session_id}} ) return ChatResponse(outputresponse[output], session_idrequest.session_id) except Exception as e: import traceback print(traceback.format_exc()) raise HTTPException(status_code500, detailf坏掉了喵: {str(e)}) if __name__ __main__: uvicorn.run(app, host127.0.0.1, port8000)这里解释一下为什么会出现两个prompt在langchain中输出链条是prompt|llm|parser,一个prompt用于传递给llm告诉它可以干什么可以怎么做同时可以修改llm的语言习惯维持llm的记忆和工具调用功能。另一个prompt封装在llm中让llm在执行RAG时基于检索到的信息进行严谨的回答防止模型产生幻觉这两个prompt相互作用外层 Prompt接收到请求它看到工具描述里有《飘》于是决定调用search_gone_with_the_wind。内层 Prompt此时启动拿到检索回来的文本分析出结果并返回给 Agent。外层 Prompt拿到这个结果把它包装成回答最后输出给你。