极简LLM入门指南5

【LLM实操系列05】RAG实战：知识库问答系统从0到生产

在开始之前，建议先完成第03篇（API调用）和第04篇（Prompt技巧）的学习。你需要理解Embedding（文本向量化）的基本原理，并安装langchain、chromadb等依赖。

10分钟实现最简RAG

什么是RAG？

检索增强生成 = 搜索相关文档 + 结合文档回答问题

Embedding是什么？

Embedding将文本转换为高维向量表示。语义相似的文本会有相似的向量，这使得我们可以用它来计算文本相似度和进行语义搜索。

最小可运行代码

# pip install langchain chromadb openai tiktokenfromlangchain.embeddingsimportOpenAIEmbeddingsfromlangchain.vectorstoresimportChromafromlangchain.chat_modelsimportChatOpenAIfromlangchain.chainsimportRetrievalQA# 1. 准备文档docs=["Python是1991年发布的编程语言","JavaScript主要用于网页开发","Rust注重内存安全和性能"]# 2. 创建向量库embeddings=OpenAIEmbeddings()vectordb=Chroma.from_texts(docs,embeddings)# 3. 创建RAG链llm=ChatOpenAI(temperature=0)qa=RetrievalQA.from_chain_type(llm=llm,retriever=vectordb.as_retriever())# 4. 提问answer=qa.run("Python是什么时候发布的？")print(answer)# 输出：Python是1991年发布的

生产级RAG架构

完整实现

importosfromtypingimportList,Dict,Anyimportnumpyasnpfromdataclassesimportdataclass@dataclassclassDocument:"""文档数据结构"""content:strmetadata:Dict[str,Any]embedding:np.ndarray=NoneclassProductionRAG:"""生产环境RAG系统"""def__init__(self,embedding_model="text-embedding-ada-002",llm_model="gpt-3.5-turbo",vector_db="chroma"):# 初始化组件self.embeddings=OpenAIEmbeddings(model=embedding_model)self.llm=ChatOpenAI(model=llm_model,temperature=0)# 向量数据库ifvector_db=="chroma":self.vectordb=Chroma(persist_directory="./chroma_db",embedding_function=self.embeddings)elifvector_db=="faiss":fromlangchain.vectorstoresimportFAISS self.vectordb=FAISS.load_local("./faiss_index",self.embeddings)# 文档处理器self.text_splitter=self._init_splitter()def_init_splitter(self):"""智能文档分块"""fromlangchain.text_splitterimportRecursiveCharacterTextSplitterreturnRecursiveCharacterTextSplitter(chunk_size=500,# 块大小chunk_overlap=100,# 重叠separators=["\n\n",# 段落"\n",# 行"。",# 句号"！",# 感叹号"？",# 问号"；",# 分号"，",# 逗号" "# 空格])defadd_documents(self,file_paths:List[str]):"""批量添加文档"""fromlangchain.document_loadersimport(PDFLoader,TextLoader,UnstructuredWordDocumentLoader,CSVLoader)loaders={'.pdf':PDFLoader,'.txt':TextLoader,'.docx':UnstructuredWordDocumentLoader,'.csv':CSVLoader}all_docs=[]forpathinfile_paths:ext=os.path.splitext(path)[1].lower()loader_class=loaders.get(ext,TextLoader)loader=loader_class(path)# 加载并分块docs=loader.load()splits=self.text_splitter.split_documents(docs)all_docs.extend(splits)# 添加到向量库self.vectordb.add_documents(all_docs)print(f"已添加{len(all_docs)}个文档块")defsearch(self,query:str,k=3,use_rerank=True):"""增强搜索"""# 1. 初步检索docs=self.vectordb.similarity_search(query,k=k*2ifuse_rerankelsek)# 2. 重排序（可选）ifuse_rerank:docs=self._rerank(query,docs,top_k=k)returndocsdef_rerank(self,query:str,docs:List,top_k:int):"""使用CrossEncoder重排序"""fromsentence_transformersimportCrossEncoder# 加载重排序模型（首次会下载）reranker=CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')# 计算相关性分数pairs=[[query,doc.page_content]fordocindocs]scores=reranker.predict(pairs)# 按分数排序doc_scores=list(zip(docs,scores))doc_scores.sort(key=lambdax:x[1],reverse=True)return[docfordoc,_indoc_scores[:top_k]]defquery(self,question:str,use_memory=False):"""智能问答"""# 检索相关文档docs=self.search(question)# 构建上下文context="\n\n".join([d.page_contentfordindocs])# 生成答案ifuse_memory:fromlangchain.memoryimportConversationBufferWindowMemory memory=ConversationBufferWindowMemory(k=3)fromlangchain.chainsimportConversationalRetrievalChain chain=ConversationalRetrievalChain.from_llm(llm=self.llm,retriever=self.vectordb.as_retriever(),memory=memory)response=chain({"question":question})else:# 自定义Promptprompt=f""" 基于以下信息回答问题。如果信息不足，请说"根据提供的信息无法回答"。 信息：{context}问题：{question}回答："""response=self.llm.predict(prompt)return{"answer":response,"sources":[{"content":d.page_content,"metadata":d.metadata}fordindocs]}

向量数据库选型

对比表

快速集成代码

# 1. Chroma（最简单）fromlangchain.vectorstoresimportChroma vectordb=Chroma.from_texts(texts,embeddings,persist_directory="./db")# 2. FAISS（最快）fromlangchain.vectorstoresimportFAISS vectordb=FAISS.from_texts(texts,embeddings)vectordb.save_local("./faiss_index")# 3. Pinecone（云服务）importpineconefromlangchain.vectorstoresimportPinecone pinecone.init(api_key="xxx",environment="us-east-1")vectordb=Pinecone.from_texts(texts,embeddings,index_name="my-index")# 4. Weaviate（混合搜索）fromlangchain.vectorstoresimportWeaviateimportweaviate client=weaviate.Client("http://localhost:8080")vectordb=Weaviate(client,index_name="Documents",text_key="content")

性能优化技巧

1. 智能分块策略

defsmart_chunking(text:str,max_tokens=500):"""基于语义的智能分块"""importtiktoken enc=tiktoken.encoding_for_model("gpt-3.5-turbo")# 按段落分割paragraphs=text.split('\n\n')chunks=[]current_chunk=[]current_tokens=0forparainparagraphs:para_tokens=len(enc.encode(para))ifcurrent_tokens+para_tokens>max_tokens:# 保存当前块ifcurrent_chunk:chunks.append('\n\n'.join(current_chunk))current_chunk=[para]current_tokens=para_tokenselse:current_chunk.append(para)current_tokens+=para_tokensifcurrent_chunk:chunks.append('\n\n'.join(current_chunk))returnchunks

2. 混合检索

classHybridSearch:"""结合关键词和语义搜索"""def__init__(self,vectordb):self.vectordb=vectordbfromrank_bm25importBM25Okapi self.bm25=Noneself.corpus=[]defindex_documents(self,documents):"""构建双索引"""# 向量索引self.vectordb.add_documents(documents)# BM25索引tokenized_corpus=[doc.page_content.split()fordocindocuments]self.bm25=BM25Okapi(tokenized_corpus)self.corpus=documentsdefsearch(self,query:str,k=5):"""混合搜索"""# 语义搜索semantic_results=self.vectordb.similarity_search(query,k=k)# 关键词搜索tokenized_query=query.split()bm25_scores=self.bm25.get_scores(tokenized_query)top_indices=np.argsort(bm25_scores)[-k:][::-1]keyword_results=[self.corpus[i]foriintop_indices]# 合并去重seen=set()final_results=[]fordocinsemantic_results+keyword_results:ifdoc.page_contentnotinseen:seen.add(doc.page_content)final_results.append(doc)iflen(final_results)>=k:breakreturnfinal_results

3. 缓存优化

fromfunctoolsimportlru_cacheimporthashlibclassCachedRAG:"""带缓存的RAG系统"""def__init__(self,rag_system):self.rag=rag_system self.cache={}@lru_cache(maxsize=1000)def_get_embedding_cached(self,text:str):"""缓存Embedding"""returnself.rag.embeddings.embed_query(text)defquery_with_cache(self,question:str):"""缓存查询结果"""# 生成查询键query_hash=hashlib.md5(question.encode()).hexdigest()# 检查缓存ifquery_hashinself.cache:print("Cache hit!")returnself.cache[query_hash]# 执行查询result=self.rag.query(question)# 保存缓存self.cache[query_hash]=resultreturnresult

实战部署

Docker部署

# docker-compose.ymlversion:'3.8'services:rag-api:build:.ports:-"8000:8000"environment:-OPENAI_API_KEY=${OPENAI_API_KEY}volumes:-./data:/app/data-./vectordb:/app/vectordbchromadb:image:chromadb/chromaports:-"8001:8000"volumes:-./chroma_data:/chroma/chroma

FastAPI服务

fromfastapiimportFastAPI,UploadFile,FilefrompydanticimportBaseModelimportuvicorn app=FastAPI()rag_system=ProductionRAG()classQuery(BaseModel):question:strk:int=3@app.post("/upload")asyncdefupload_document(file:UploadFile=File(...)):"""上传文档"""# 保存文件file_path=f"./uploads/{file.filename}"withopen(file_path,"wb")asf:content=awaitfile.read()f.write(content)# 添加到RAGrag_system.add_documents([file_path])return{"message":"文档已上传并索引"}@app.post("/query")asyncdefquery(q:Query):"""查询接口"""result=rag_system.query(q.question,k=q.k)returnresult@app.get("/stats")asyncdefstats():"""系统统计"""return{"total_documents":len(rag_system.vectordb.get()['ids']),"model":"gpt-3.5-turbo","embedding_model":"text-embedding-ada-002"}if__name__=="__main__":uvicorn.run(app,host="0.0.0.0",port=8000)

常见问题解决