找人做网站去哪里找,深圳市建筑工程,wordpress左右主题,强力搜索引擎gte-base-zh RAG预处理教程#xff1a;为LangChainLlama3构建高质量向量数据库 1. 环境准备与模型部署 1.1 模型简介与下载 gte-base-zh是由阿里巴巴达摩院训练的中文文本嵌入模型#xff0c;基于BERT框架构建。这个模型专门针对中文文本进行了优化训练#xff0c;能够将…gte-base-zh RAG预处理教程为LangChainLlama3构建高质量向量数据库1. 环境准备与模型部署1.1 模型简介与下载gte-base-zh是由阿里巴巴达摩院训练的中文文本嵌入模型基于BERT框架构建。这个模型专门针对中文文本进行了优化训练能够将文本转换为高质量的向量表示非常适合用于RAG检索增强生成系统的预处理环节。模型已经在包含大量相关文本对的大规模语料库上进行了训练涵盖了广泛的领域和场景。这使得gte-base-zh模型能够应用于多种下游任务包括信息检索、语义文本相似性计算、文本重排序等。模型本地地址/usr/local/bin/AI-ModelScope/gte-base-zh1.2 使用Xinference部署模型服务Xinference是一个高效的模型推理框架我们可以用它来部署gte-base-zh模型的服务接口。启动Xinference服务xinference-local --host 0.0.0.0 --port 9997这个命令会启动一个本地推理服务监听9997端口为后续的模型调用提供基础环境。启动gte-base-zh模型服务python /usr/local/bin/launch_model_server.py这个脚本会通过Xinference接口发布模型服务使其能够被外部应用调用。1.3 验证服务状态部署完成后我们需要确认模型服务是否正常启动。初次加载模型可能需要一些时间具体取决于硬件配置。检查服务状态cat /root/workspace/model_server.log如果服务启动成功日志中会显示相应的成功信息。你也可以通过Web界面来验证服务状态打开Xinference的Web UI界面找到gte-base-zh模型对应的服务入口点击进入模型测试页面在测试页面中你可以输入文本进行相似度比对测试验证模型是否正常工作。2. 构建向量数据库基础2.1 准备文本数据在构建向量数据库之前我们需要准备待处理的文本数据。这些数据可以来自各种来源如文档、网页内容、知识库等。创建文本预处理函数import re import jieba from typing import List def preprocess_text(text: str, max_length: int 512) - str: 预处理中文文本包括清洗、分词和长度控制 # 去除多余空白字符 text re.sub(r\s, , text.strip()) # 使用jieba进行分词 words jieba.cut(text) # 重新组合分词结果 processed_text .join(words) # 控制文本长度 if len(processed_text) max_length: processed_text processed_text[:max_length] return processed_text def batch_preprocess(texts: List[str]) - List[str]: 批量预处理文本 return [preprocess_text(text) for text in texts]2.2 配置嵌入模型客户端我们需要配置与Xinference服务的连接以便调用gte-base-zh模型生成文本嵌入。设置模型客户端import requests import numpy as np from typing import List class GTEEmbeddingClient: def __init__(self, host: str localhost, port: int 9997): self.base_url fhttp://{host}:{port} self.model_name gte-base-zh def get_embedding(self, text: str) - List[float]: 获取单个文本的嵌入向量 try: response requests.post( f{self.base_url}/v1/embeddings, json{ model: self.model_name, input: text } ) response.raise_for_status() return response.json()[data][0][embedding] except Exception as e: print(f获取嵌入向量失败: {e}) return None def get_embeddings_batch(self, texts: List[str]) - List[List[float]]: 批量获取文本嵌入向量 embeddings [] for text in texts: embedding self.get_embedding(text) if embedding is not None: embeddings.append(embedding) return embeddings # 初始化客户端 embedding_client GTEEmbeddingClient()3. 构建FAISS向量数据库3.1 安装必要依赖首先确保安装了所需的Python包pip install faiss-cpu langchain sentence-transformers或者使用GPU版本的FAISS如果可用pip install faiss-gpu3.2 创建向量数据库使用FAISS来存储和管理文本嵌入向量import faiss import numpy as np import pickle from pathlib import Path class VectorDatabase: def __init__(self, dimension: int 768): self.dimension dimension self.index faiss.IndexFlatL2(dimension) self.texts [] self.metadata [] def add_documents(self, texts: List[str], embeddings: List[List[float]], metadata: List[dict] None): 添加文档到向量数据库 if len(texts) ! len(embeddings): raise ValueError(文本数量和嵌入向量数量不匹配) # 转换为numpy数组 embeddings_array np.array(embeddings).astype(float32) # 添加到FAISS索引 self.index.add(embeddings_array) # 保存文本和元数据 self.texts.extend(texts) if metadata: self.metadata.extend(metadata) else: self.metadata.extend([{}] * len(texts)) def search(self, query_embedding: List[float], k: int 5): 在向量数据库中搜索相似文档 query_array np.array([query_embedding]).astype(float32) distances, indices self.index.search(query_array, k) results [] for i, idx in enumerate(indices[0]): if idx len(self.texts): results.append({ text: self.texts[idx], metadata: self.metadata[idx], distance: distances[0][i] }) return results def save(self, filepath: str): 保存向量数据库到文件 path Path(filepath) path.parent.mkdir(parentsTrue, exist_okTrue) # 保存FAISS索引 faiss.write_index(self.index, str(path.with_suffix(.index))) # 保存文本和元数据 with open(path.with_suffix(.pkl), wb) as f: pickle.dump({ texts: self.texts, metadata: self.metadata, dimension: self.dimension }, f) classmethod def load(cls, filepath: str): 从文件加载向量数据库 path Path(filepath) # 加载FAISS索引 index faiss.read_index(str(path.with_suffix(.index))) # 加载文本和元数据 with open(path.with_suffix(.pkl), rb) as f: data pickle.load(f) # 创建实例 vdb cls(data[dimension]) vdb.index index vdb.texts data[texts] vdb.metadata data[metadata] return vdb3.3 批量处理文档并构建数据库现在我们可以将预处理、嵌入生成和数据库构建整合在一起from tqdm import tqdm import time def build_vector_database(documents: List[str], batch_size: int 32, output_path: str vector_db): 构建向量数据库的完整流程 print(开始预处理文档...) processed_docs batch_preprocess(documents) print(生成文本嵌入向量...) embeddings [] # 分批处理避免内存溢出 for i in tqdm(range(0, len(processed_docs), batch_size)): batch_texts processed_docs[i:ibatch_size] batch_embeddings embedding_client.get_embeddings_batch(batch_texts) embeddings.extend(batch_embeddings) # 添加延迟避免服务过载 time.sleep(0.1) print(创建向量数据库...) vdb VectorDatabase(dimensionlen(embeddings[0]) if embeddings else 768) vdb.add_documents(documents, embeddings) print(保存数据库...) vdb.save(output_path) print(f向量数据库构建完成共处理 {len(documents)} 个文档) return vdb # 示例使用 if __name__ __main__: # 示例文档 sample_documents [ 机器学习是人工智能的一个重要分支, 深度学习使用神经网络进行特征学习, 自然语言处理让计算机理解人类语言, 计算机视觉使机器能够看懂图像和视频 ] # 构建向量数据库 vector_db build_vector_database(sample_documents, output_pathmy_vector_db)4. 集成LangChain和Llama34.1 创建LangChain检索器现在我们将向量数据库与LangChain集成创建可用于RAG系统的检索器from langchain.schema import Document from langchain.vectorstores import FAISS from langchain.embeddings.base import Embeddings class GTEEmbeddings(Embeddings): LangChain兼容的GTE嵌入类 def __init__(self, client: GTEEmbeddingClient): self.client client def embed_documents(self, texts: List[str]) - List[List[float]]: return self.client.get_embeddings_batch(texts) def embed_query(self, text: str) - List[float]: return self.client.get_embedding(text) def create_langchain_retriever(vector_db_path: str, embedding_client: GTEEmbeddingClient, k: int 5): 创建LangChain检索器 # 加载向量数据库 vdb VectorDatabase.load(vector_db_path) # 创建LangChain文档对象 documents [] for text, metadata in zip(vdb.texts, vdb.metadata): documents.append(Document(page_contenttext, metadatametadata)) # 创建FAISS向量存储 embeddings GTEEmbeddings(embedding_client) vector_store FAISS.from_documents(documents, embeddings) # 创建检索器 retriever vector_store.as_retriever(search_kwargs{k: k}) return retriever # 创建检索器示例 retriever create_langchain_retriever(my_vector_db, embedding_client, k3)4.2 与Llama3集成实现RAG最后我们将检索器与Llama3语言模型集成构建完整的RAG系统from langchain.llms import LlamaCpp from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate def setup_rag_system(retriever, model_path: str): 设置完整的RAG系统 # 加载Llama3模型 llm LlamaCpp( model_pathmodel_path, temperature0.1, max_tokens2000, top_p1, verboseTrue ) # 创建自定义提示模板 prompt_template 使用以下上下文信息来回答问题。如果你不知道答案就说不知道不要编造答案。 上下文 {context} 问题{question} 答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 创建RAG链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue ) return qa_chain # 使用示例 def ask_question(question: str, qa_chain): 使用RAG系统回答问题 result qa_chain({query: question}) print(f问题: {question}) print(f答案: {result[result]}) print(\n参考文档:) for i, doc in enumerate(result[source_documents]): print(f{i1}. {doc.page_content[:200]}...) return result # 完整流程示例 if __name__ __main__: # 初始化组件 embedding_client GTEEmbeddingClient() retriever create_langchain_retriever(my_vector_db, embedding_client) # 设置RAG系统需要提前下载Llama3模型 # qa_chain setup_rag_system(retriever, /path/to/llama3/model) # 提问示例 # result ask_question(什么是机器学习, qa_chain)5. 优化技巧和最佳实践5.1 性能优化建议批量处理优化def optimized_batch_processing(texts: List[str], batch_size: int 64, max_workers: int 4): 使用多线程优化批量处理 from concurrent.futures import ThreadPoolExecutor def process_batch(batch): return embedding_client.get_embeddings_batch(batch) results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [] for i in range(0, len(texts), batch_size): batch texts[i:ibatch_size] futures.append(executor.submit(process_batch, batch)) for future in futures: results.extend(future.result()) return results5.2 质量保证措施嵌入质量验证def validate_embedding_quality(text_pairs: List[tuple], threshold: float 0.8): 验证嵌入模型的质量 from sklearn.metrics.pairwise import cosine_similarity similarities [] for text1, text2 in text_pairs: emb1 embedding_client.get_embedding(text1) emb2 embedding_client.get_embedding(text2) if emb1 and emb2: sim cosine_similarity([emb1], [emb2])[0][0] similarities.append(sim) avg_similarity sum(similarities) / len(similarities) if similarities else 0 print(f平均相似度: {avg_similarity:.4f}) if avg_similarity threshold: print(警告嵌入质量可能不理想) return avg_similarity6. 总结通过本教程我们学习了如何使用gte-base-zh模型为LangChain和Llama3构建高质量的向量数据库。整个过程包括模型部署、文本预处理、嵌入生成、向量数据库构建以及与LangChain的集成。关键要点包括模型部署使用Xinference高效部署gte-base-zh嵌入模型服务文本处理对中文文本进行适当的预处理和分词向量化将文本转换为高质量的嵌入向量数据库构建使用FAISS存储和管理向量数据系统集成将向量数据库与LangChain和Llama3集成构建完整的RAG系统这种方法能够显著提升基于大语言模型的问答系统的准确性和相关性特别是在处理中文内容时。通过使用专门针对中文优化的gte-base-zh模型我们可以获得比通用多语言模型更好的语义理解效果。在实际应用中你可以根据具体需求调整批量处理大小、检索数量等参数并在生产环境中添加适当的监控和日志记录功能。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。