bge-large-zh-v1.5实战教程：结合Milvus构建高并发中文向量数据库

bge-large-zh-v1.5实战教程结合Milvus构建高并发中文向量数据库1. 引言为什么需要高并发向量检索如果你正在构建一个智能问答系统、一个文档搜索引擎或者一个内容推荐平台你可能会遇到一个核心问题如何在海量文本中快速、准确地找到与用户查询最相关的内容传统的基于关键词匹配的方法比如“全文搜索”已经很难满足需求了。比如用户搜索“如何保养笔记本电脑”传统的搜索可能找不到包含“笔记本维护技巧”的文章因为关键词没对上。但这两句话在语义上高度相关。这就是向量数据库和嵌入模型大显身手的地方。简单来说这个过程分三步向量化用一个强大的模型比如我们今天要用的 bge-large-zh-v1.5把所有的文本文章、问题、商品描述都转换成一串数字也就是“向量”。语义相近的文本它们的向量在数学空间里的距离也更近。存储把这些向量和对应的原始文本高效地存到一个专门的数据库里这就是向量数据库比如 Milvus。检索当用户输入一个新问题时同样用模型把它转换成向量然后去数据库里快速找出距离最近的几个向量对应的文本就是最相关的答案。而“高并发”意味着你的系统能同时处理成千上万个这样的查询请求依然保持飞快的速度。本教程的目标就是带你从零开始搭建这样一个高性能的中文语义检索系统。我们将使用性能强劲的bge-large-zh-v1.5模型来生成向量并用专为向量运算设计的Milvus数据库来存储和检索最终构建一个能扛住高并发查询的服务。2. 核心组件介绍在开始动手之前我们先快速了解一下将要使用的两个核心工具。2.1 bge-large-zh-v1.5你的中文语义“转换器”bge-large-zh-v1.5是目前中文社区里第一梯队的文本嵌入模型。你可以把它理解为一个极其擅长理解中文的“翻译官”但它不是把中文翻译成英文而是翻译成计算机更擅长处理的“向量语言”。它的几个特点决定了它非常适合我们的项目专为中文优化在大规模中文语料上训练对中文的语义、句法甚至一些文化语境都有很好的把握。强大的语义表征它能生成1024维的高维向量。维度高意味着它能捕捉更细腻、更复杂的语义差别。比如“苹果公司”和“一种水果”这两个“苹果”它生成的向量会相差很远。处理长文本最多能处理512个token大约相当于300-350个汉字的输入足以覆盖大多数段落文本。通用且强大在各类中文语义相似度、检索任务的标准测试集上它都名列前茅是一个可靠的“优等生”。2.2 Milvus为向量而生的“高速仓库”Milvus 是一个开源的向量数据库。想象一下如果你有1亿条文本向量用传统数据库如MySQL来做“找最近邻”的运算会慢得无法忍受。Milvus 就是为解决这个问题而生的。它的核心能力包括超高速检索内置了多种近似最近邻ANN搜索算法能在毫秒级别从亿级数据中找出最相似的向量。可扩展性支持分布式部署数据可以分散在多个机器上轻松应对数据量和并发量的增长。开发者友好提供了 Python、Java、Go 等多种语言的 SDK接口简单易用。丰富的功能不仅支持存储向量还能关联存储原始文本或其他属性并支持过滤查询。简单来说bge-large-zh-v1.5负责生产高质量的“语义指纹”而Milvus负责海量存储和闪电检索。两者结合就是构建现代AI应用如RAG的基础设施。3. 环境准备与部署接下来我们进入实战环节。我们将在一个Linux工作环境中部署这两个组件。3.1 部署 bge-large-zh-v1.5 嵌入服务为了获得更好的性能和并发处理能力我们使用sglang来部署模型服务。sglang是一个高效的推理运行时特别适合部署此类模型API。首先我们进入工作目录并启动服务。以下操作假设你已在/root/workspace目录下准备好了模型和启动脚本。# 进入工作目录 cd /root/workspace # 使用 sglang 启动 bge-large-zh-v1.5 模型服务 # 这里假设你的启动命令已经写在了某个脚本中例如 # python -m sglang.launch_server --model-path /path/to/bge-large-zh-v1.5 --port 30000 ... # 具体启动命令可能因环境而异启动后日志会输出到 sglang.log服务启动后我们需要确认它是否正常运行。# 查看启动日志 cat sglang.log你需要关注日志末尾寻找类似“Server started on http://0.0.0.0:30000”或“Embedding model loaded successfully”的成功信息。这表示模型服务已经在30000端口就绪等待接收请求。3.2 验证嵌入模型服务服务启动后我们最好立刻验证一下。打开 Jupyter Notebook 或一个 Python 脚本运行以下测试代码import openai # 注意这里使用openai兼容的客户端库 # 初始化客户端指向我们本地启动的sglang服务 client openai.Client( base_urlhttp://localhost:30000/v1, # 服务地址 api_keyEMPTY # 因为是本地服务API Key可以填任意值或EMPTY ) # 尝试生成一个文本的向量 test_text 今天天气真好 response client.embeddings.create( modelbge-large-zh-v1.5, # 指定模型名称 inputtest_text, ) # 打印响应查看向量维度 print(f向量长度: {len(response.data[0].embedding)}) print(f前10个向量值: {response.data[0].embedding[:10]}...)如果一切正常你会看到输出一个长度为1024的列表这就是“今天天气真好”这句话的语义向量。看到这个说明你的嵌入模型服务已经完美运行了。3.3 安装并启动 Milvus 数据库我们使用 Docker 来安装 Milvus这是最快捷的方式。Milvus 单机版包含三个核心组件etcd元数据存储、minio对象存储和milvus-standalone主体。确保你的系统已经安装了 Docker 和 Docker Compose。下载配置文件wget https://github.com/milvus-io/milvus/releases/download/v2.4.0/milvus-standalone-docker-compose.yml -O docker-compose.yml启动 Milvussudo docker-compose up -d检查运行状态sudo docker-compose ps当看到三个服务milvus-standaloneetcdminio的状态都是Up时表示 Milvus 已经成功启动。默认的向量服务端口是19530。4. 构建你的第一个向量数据库应用现在两个引擎都已就位让我们编写一个完整的 Python 程序实现从文本到向量再到存储和检索的全流程。4.1 连接 Milvus 并创建集合在 Milvus 中“集合Collection”类似于传统数据库中的“表”用于存储同一类数据。from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility import openai # 1. 连接到 Milvus 服务 connections.connect(aliasdefault, hostlocalhost, port19530) # 2. 定义集合的字段Schema # id: 主键用于唯一标识每条数据 # text: 存储原始文本 # vector: 存储由 bge-large-zh-v1.5 生成的1024维向量 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length1000), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim1024) # 维度必须与模型输出一致 ] # 3. 创建集合 Schema schema CollectionSchema(fieldsfields, description中文文本向量检索测试集合) # 4. 指定集合名称并创建 collection_name demo_chinese_collection if utility.has_collection(collection_name): # 如果集合已存在先删除仅用于演示生产环境慎用 collection Collection(collection_name) collection.drop() print(f已删除已存在的集合: {collection_name}) # 创建新集合 collection Collection(namecollection_name, schemaschema) print(f集合 {collection_name} 创建成功)4.2 准备数据并生成向量假设我们有一个小的中文文档库需要把它们存入向量数据库。# 初始化嵌入模型客户端复用之前的代码 embed_client openai.Client(base_urlhttp://localhost:30000/v1, api_keyEMPTY) # 我们的示例文档库 documents [ 机器学习是人工智能的一个分支它让计算机能够从数据中学习。, 深度学习是机器学习的一种它使用深度神经网络来处理复杂模式。, Python是一种流行的编程语言广泛用于数据科学和人工智能。, 向量数据库专门用于高效存储和检索向量形式的数据。, 今天上海的天气是晴转多云气温在20到25摄氏度之间。, 如何快速学习一门新的编程语言多写代码阅读优秀项目。 ] # 批量生成向量 def generate_embeddings(texts): 调用 bge-large-zh-v1.5 服务为文本列表生成向量 response embed_client.embeddings.create( modelbge-large-zh-v1.5, inputtexts ) # 返回向量列表 return [item.embedding for item in response.data] print(正在为文档生成向量...) vectors generate_embeddings(documents) print(f已为 {len(documents)} 个文档生成向量每个向量维度为 {len(vectors[0])}。)4.3 将数据插入 Milvus 集合生成向量后我们将文本和向量一起插入到 Milvus 集合中。# 准备要插入的数据格式需要与定义的字段顺序对应 [text, vector] # id 字段是自增的所以我们不需要提供 data_to_insert [ documents, # text 字段的数据 vectors # vector 字段的数据 ] # 执行插入操作 insert_result collection.insert(data_to_insert) print(f成功插入 {len(documents)} 条数据。插入事务ID: {insert_result}) # 重要在执行搜索前需要将数据从内存加载到显存/内存中并创建索引以加速搜索 collection.load() # 为向量字段创建索引这里使用IVF_FLAT索引适合中小规模数据集 index_params { index_type: IVF_FLAT, metric_type: L2, # 使用L2距离欧氏距离来衡量向量相似度 params: {nlist: 128} # 聚类单元数值越大搜索越精确但越慢可根据数据量调整 } collection.create_index(field_namevector, index_paramsindex_params) print(向量索引创建完成已准备好进行检索。)4.4 执行你的第一次语义检索现在最激动人心的部分来了让我们尝试用自然语言提问从数据库中找出最相关的文档。# 用户的查询问题 query_text 人工智能有哪些学习方法 # 1. 将查询文本转换为向量 print(f用户查询: {query_text}) query_vector generate_embeddings([query_text])[0] # 2. 定义搜索参数 search_params { metric_type: L2, params: {nprobe: 10} # 搜索时探查的聚类单元数影响速度和精度 } # 3. 执行搜索 # 我们想搜索 vector 字段限制返回最相似的3条结果 results collection.search( data[query_vector], # 查询向量 anns_fieldvector, # 在哪个字段上搜索 paramsearch_params, limit3, # 返回前3个最相似的结果 output_fields[text] # 同时返回 text 字段的内容 ) # 4. 解析并展示结果 print(\n 语义检索结果 ) for i, hits in enumerate(results): for j, hit in enumerate(hits): print(f结果 {j1}:) print(f 相关文本: {hit.entity.get(text)}) print(f 距离分数 (L2): {hit.distance:.4f} (分数越小越相似)) print(- * 50)运行这段代码你会看到系统返回了与“人工智能学习方法”最相关的句子比如“机器学习是人工智能的一个分支...”和“深度学习是机器学习的一种...”。尽管你的查询没有出现“机器学习”或“深度学习”这些关键词但系统通过语义理解找到了它们。这就是向量检索的魅力5. 构建高并发服务架构单次检索演示成功了但如何支撑成百上千的并发请求呢我们需要一个简单的服务化架构。这里设计一个基于 FastAPI 的轻量级服务。5.1 创建 FastAPI 检索服务# 文件: retrieval_service.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import openai from pymilvus import connections, Collection import asyncio import aiohttp app FastAPI(title中文高并发向量检索服务) # 全局初始化在实际生产中这部分需要更优雅的启动和错误处理 # 1. 连接 Milvus connections.connect(aliasdefault, hostlocalhost, port19530) COLLECTION_NAME demo_chinese_collection collection Collection(COLLECTION_NAME) collection.load() # 确保集合已加载 # 2. 初始化嵌入模型客户端考虑使用连接池 EMBED_BASE_URL http://localhost:30000/v1 # 定义请求和响应模型 class SearchRequest(BaseModel): query: str top_k: int 5 # 默认返回5个结果 class SearchResult(BaseModel): text: str score: float class SearchResponse(BaseModel): results: List[SearchResult] async def async_generate_embedding(text: str, session: aiohttp.ClientSession) - List[float]: 异步调用嵌入模型服务 async with session.post( f{EMBED_BASE_URL}/embeddings, json{model: bge-large-zh-v1.5, input: text}, headers{Authorization: Bearer EMPTY} ) as resp: if resp.status ! 200: raise HTTPException(status_code500, detail嵌入模型服务调用失败) data await resp.json() return data[data][0][embedding] app.post(/search, response_modelSearchResponse) async def semantic_search(request: SearchRequest): 语义搜索端点。 接收一个查询字符串返回最相关的文本列表。 try: # 异步生成查询向量 async with aiohttp.ClientSession() as session: query_vector await async_generate_embedding(request.query, session) # 在 Milvus 中执行搜索 search_params {metric_type: L2, params: {nprobe: 10}} results collection.search( data[query_vector], anns_fieldvector, paramsearch_params, limitrequest.top_k, output_fields[text] ) # 格式化结果 search_results [] for hits in results: for hit in hits: search_results.append( SearchResult(texthit.entity.get(text), scorefloat(hit.distance)) ) return SearchResponse(resultssearch_results) except Exception as e: raise HTTPException(status_code500, detailf搜索过程发生错误: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)5.2 使用并发客户端进行压力测试服务启动后我们可以模拟高并发请求来测试其稳定性。# 文件: stress_test.py import aiohttp import asyncio import time async def single_search(session, query, semaphore): 单个搜索任务 async with semaphore: # 使用信号量控制并发度 async with session.post( http://localhost:8000/search, json{query: query, top_k: 3} ) as resp: return await resp.json() async def main(): # 准备一批测试查询 test_queries [ 怎么学习编程, 天气怎么样, 什么是神经网络, 介绍下Python语言。, 如何做数据分析, 机器学习有什么用, ] * 10 # 重复10次模拟60个请求 semaphore asyncio.Semaphore(50) # 控制最大并发数为50 connector aiohttp.TCPConnector(limit0) # 不限制连接池总数 timeout aiohttp.ClientTimeout(total30) start_time time.time() async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: tasks [single_search(session, q, semaphore) for q in test_queries] results await asyncio.gather(*tasks, return_exceptionsTrue) end_time time.time() # 统计结果 success 0 for r in results: if not isinstance(r, Exception): success 1 print(f压力测试完成) print(f总请求数: {len(test_queries)}) print(f成功请求数: {success}) print(f总耗时: {end_time - start_time:.2f} 秒) print(f平均每秒处理请求数 (QPS): {len(test_queries) / (end_time - start_time):.2f}) if __name__ __main__: asyncio.run(main())运行这个测试脚本你可以看到服务在并发请求下的表现。通过调整semaphore的值和增加测试查询数量你可以初步评估出当前部署架构的并发处理能力瓶颈在哪里是模型推理速度、Milvus检索速度还是网络IO。6. 总结与进阶建议恭喜你你已经成功搭建了一个基于bge-large-zh-v1.5和Milvus的高并发中文向量检索系统原型。让我们回顾一下关键步骤和核心价值模型服务化我们使用sglang部署了bge-large-zh-v1.5嵌入模型提供了一个高效的 HTTP API这是处理高并发嵌入请求的基础。向量数据库集成我们安装并配置了Milvus创建了用于存储文本和向量的集合并建立了索引使得亿级向量的毫秒级检索成为可能。完整应用构建我们编写了从数据准备、向量化、入库到语义检索的完整代码实现了核心业务流程。服务化与并发测试我们通过FastAPI将检索功能封装成 RESTful 服务并进行了简单的压力测试验证了其处理并发请求的潜力。为了让这个系统真正用于生产环境你还需要考虑以下几个进阶方向性能优化批处理在向模型服务请求向量时尽量将多个文本打包成一个批次Batch发送可以极大提高吞吐量。索引调优根据你的数据规模十万、百万、亿级和查询延迟要求选择合适的 Milvus 索引类型如HNSW,SCANN,DISKANN和参数如nlist,nprobe。服务部署将bge-large-zh-v1.5服务、Milvus数据库和FastAPI应用分别容器化Docker并使用 Kubernetes 或 Docker Compose 进行编排实现资源隔离和弹性伸缩。系统可靠性服务健康检查与熔断在FastAPI服务中增加对模型服务和 Milvus 的健康检查如果下游服务失败应快速失败并返回优雅降级响应。异步处理对于数据导入等耗时操作可以引入消息队列如 Redis, RabbitMQ改为异步任务避免阻塞实时查询接口。监控与日志集成 Prometheus 和 Grafana 来监控服务的 QPS、延迟、错误率建立完善的日志收集系统。功能增强多模态支持除了文本是否可以接入图像、音频的嵌入模型Milvus 同样支持存储和检索多模态向量。混合搜索结合传统的关键词搜索BM25和向量语义搜索往往能获得更准确、更全面的结果。RAG 系统这正是构建检索增强生成RAG应用的核心。将本系统检索到的相关文本作为上下文提供给大语言模型如 GPT可以生成更精准、信息量更丰富的答案。这个由bge-large-zh-v1.5和Milvus组成的“黄金搭档”为你打开了构建智能检索应用的大门。从简单的文档搜索到复杂的问答系统它的潜力只受你想象力的限制。现在就基于这个原型去实现你的创意吧获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。