RexUniNLU代码实例：对接Milvus向量库，实现Schema语义相似度检索与推荐

RexUniNLU代码实例对接Milvus向量库实现Schema语义相似度检索与推荐1. 引言从零样本理解到智能检索想象一下你正在开发一个智能客服系统。用户可能会问“我想订一张明天去上海的机票”也可能问“帮我查一下后天飞北京的航班”。传统的做法是预先定义好一堆“订票”、“查询”这样的意图然后收集大量标注数据去训练模型。但问题来了业务需求总是在变今天要支持订票明天可能要支持退改签后天又加了酒店预订。每变一次就得重新标注、重新训练费时费力。有没有一种方法能让机器像人一样只需要你告诉它几个关键词它就能理解用户的意图并且不需要任何预先的训练这就是RexUniNLU要解决的问题。它就像一个“语言理解通才”你给它一个任务描述我们称之为Schema比如[‘出发地’ ‘目的地’ ‘时间’ ‘订票意图’]它就能直接从用户的句子中找出对应的信息。但是理解只是第一步。在真实的业务场景里比如一个大型的电商知识库、一个内容推荐系统我们面对的不是单一的一句话而是成千上万条用户Query、商品描述或文章标题。当一个新的用户Query进来时我们如何快速地从海量数据中找到语义上最相似、最相关的那几条这就需要引入“向量检索”的能力。本文将带你手把手实现一个完整的流程使用RexUniNLU将文本转化为富含语义的向量然后存入专业的向量数据库Milvus最后构建一个高效的语义相似度检索与推荐系统。你不需要有标注数据也不需要深厚的机器学习背景跟着做就能让应用拥有“智能理解”和“精准推荐”的能力。2. 核心组件与技术选型在开始动手之前我们先快速了解一下要用到的两个核心工具RexUniNLU和Milvus。知道它们各自擅长什么才能更好地让它们协同工作。2.1 RexUniNLU零样本语义理解与向量化引擎RexUniNLU的核心价值在于其“零样本”和“通用性”。零样本Zero-shot这是它最厉害的地方。你不需要准备“订票”相关的成千上万条标注数据。你只需要告诉它“嘿接下来我们要处理‘订票’这件事你帮我留意‘出发地’、‘目的地’、‘时间’这些信息。” 它就能基于预训练好的通用语言知识直接在新的句子上进行识别和向量化。语义向量化除了提取结构化的信息槽位RexUniNLU的底层模型基于Siamese-UIE架构能够将输入的文本无论是用户Query还是你定义的任务Schema转换成一个固定长度的数字序列也就是“向量”。这个向量就像是文本的“语义指纹”语义相近的文本其向量在数学空间里的距离也会很近。简单来说RexUniNLU在本项目中的核心作用有两个理解用户Query提取结构化信息明确用户意图。生成语义向量为文本包括Query和候选条目生成用于相似度比较的“指纹”。2.2 Milvus高性能向量数据库理解了文本生成了向量接下来就是存储和检索。用传统的数据库如MySQL来存向量、做相似度计算比如计算余弦相似度效率极低当数据量达到百万、千万级时根本无法满足实时检索的需求。Milvus就是为解决这个问题而生的。它是一个开源的向量数据库专门为海量向量数据的存储、索引和检索做了深度优化。高效的相似度检索它内置了多种先进的索引算法如IVF_FLAT, HNSW能让你在毫秒级时间内从上亿条向量中找出与目标向量最相似的Top-K个结果。丰富的功能支持标量过滤比如同时满足“类别是电子产品”和“价格小于1000”、动态Schema、数据持久化等。易于集成提供了Python、Java、Go等多种语言的SDK对接起来非常方便。在本项目中Milvus的职责是高效存储所有候选文本如商品标题、文章摘要对应的语义向量并当新的用户Query向量到来时快速找出最相似的候选条目。2.3 系统架构全景图让我们把这两个组件串联起来看看整个系统是如何工作的用户输入 Query ↓ RexUniNLU 处理 ├───→ 提取结构化信息意图、槽位 └───→ 生成 Query 语义向量 ↓ 在 Milvus 中检索 ↓ 返回 Top-K 个最相似的候选条目 ↓ 结合业务逻辑进行推荐或回答这个流程就是现代语义搜索和推荐系统的核心骨架。接下来我们进入实战环节。3. 环境搭建与快速部署工欲善其事必先利其器。我们先来把开发环境准备好。这里假设你已经在本地或服务器上有一个可用的Python环境3.8。3.1 安装RexUniNLU及相关依赖首先我们需要安装RexUniNLU。根据其项目说明它主要依赖ModelScope。# 1. 安装ModelScope库这是RexUniNLU的基础 pip install modelscope # 2. 安装PyTorch请根据你的CUDA版本选择合适命令以下以CPU版为例 # 访问 https://pytorch.org/get-started/locally/ 获取最新安装命令 pip install torch torchvision torchaudio # 3. 安装其他可能需要的工具 pip install numpy pandas3.2 安装并启动Milvus向量数据库Milvus的安装方式有多种为了快速演示我们使用其最方便的Docker Compose方式。请确保你的系统已经安装了Docker和Docker Compose。# 1. 下载Milvus的docker-compose配置文件 wget https://github.com/milvus-io/milvus/releases/download/v2.3.3/milvus-standalone-docker-compose.yml -O docker-compose.yml # 2. 启动Milvus服务这会在后台启动Milvus及其依赖的Etcd和MinIO sudo docker-compose up -d # 3. 检查服务状态看到所有容器都是Up状态即可 sudo docker-compose ps启动成功后Milvus的服务端口默认19530就处于监听状态了。3.3 安装Milvus的Python SDK我们需要在Python代码中连接和操作Milvus。pip install pymilvus现在环境就准备妥当了。我们有了理解文本的“大脑”RexUniNLU也有了存储和检索向量的“仓库”Milvus。4. 实战构建语义检索系统现在我们来一步步实现核心功能。我们将模拟一个“智能问答知识库”的场景知识库里有一些预定义的问答对我们的目标是当用户提出一个新问题时系统能自动从知识库中找到语义最相似的问题并返回对应的答案。4.1 步骤一初始化RexUniNLU与文本向量化首先我们需要加载RexUniNLU模型并编写一个函数来获取文本的向量。RexUniNLU的模型输出中包含了我们需要的向量表示。from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks import numpy as np class RexUniNLUEncoder: RexUniNLU 文本向量化封装类 def __init__(self, model_repodamo/nlp_structbert_siamese-uninlu_chinese-base): 初始化RexUniNLU管道。 注意首次运行会自动从ModelScope下载模型需要一定时间。 print(f正在加载模型: {model_repo}...) # 创建零样本自然语言理解管道 self.nlu_pipeline pipeline( taskTasks.zero_shot_nli, modelmodel_repo, model_revisionv1.0.1 # 指定一个稳定的版本 ) print(模型加载完毕) def get_embedding(self, text, schema_labelsNone): 获取输入文本的语义向量。 参数: text: 输入文本如“如何办理退票” schema_labels: 可选的Schema标签列表如[问题, 意图]。 即使不用于信息抽取Schema也会影响向量生成。 如果为None则使用一个通用Schema。 返回: embedding: 文本的向量表示 (numpy数组) result: 完整的NLU结果包含槽位等信息 if schema_labels is None: # 使用一个通用的、适合QA检索的Schema schema_labels [问题主题, 问题意图] # 调用模型进行推理 # 注意输入需要构造成模型接受的格式通常是一个字典 input_data { text: text, labels: schema_labels } result self.nlu_pipeline(input_data) # 从结果中提取向量。具体字段名需要查看模型输出结构。 # 这里假设向量在result的embedding或output字段中实际需调试确定。 # 以下是一种可能的提取方式请根据实际模型输出调整 if embedding in result: embedding result[embedding] elif output in result and hasattr(result[output], last_hidden_state): # 有时向量是模型最后一层隐藏状态的平均池化 hidden_states result[output].last_hidden_state embedding np.mean(hidden_states, axis1).squeeze() else: # 如果上述都没有我们可以用模型输出的logits或scores作为向量的一种表示 # 这取决于具体模型此处仅为示例需要你根据实际输出调整 print(警告未找到标准embedding字段尝试使用logits...) # 假设result[output]是一个包含各标签分数的字典或列表 # 这里需要你根据实际打印出的result结构来编写提取代码 # embedding ... # 为了示例能运行我们暂时创建一个随机向量实际使用时务必替换 embedding np.random.randn(768) # 假设向量维度是768 # 确保embedding是numpy一维数组 embedding np.array(embedding).flatten().astype(np.float32) return embedding, result # 初始化编码器 encoder RexUniNLUEncoder()关键点说明schema_labels参数很重要。即使你只想做纯语义检索不关心具体的槽位传入一个相关的Schema如[‘问题主题’ ‘问题意图’]也能帮助模型生成更具任务相关性的向量。向量提取部分 (get_embedding函数) 是需要你根据实际模型输出进行调整的。你需要先运行一次打印出result的结构然后找到其中代表文本整体语义的向量表示。可能是embedding字段也可能是最后一个隐藏层的池化结果。4.2 步骤二连接Milvus并定义数据集合接下来我们连接到Milvus服务并创建一个用于存储问答对及其向量的“集合”Collection。from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility class MilvusOperator: Milvus 向量数据库操作封装类 def __init__(self, hostlocalhost, port19530): 连接Milvus服务器 self.host host self.port port connections.connect(aliasdefault, hosthost, portport) print(f已连接到Milvus: {host}:{port}) def create_qa_collection(self, collection_nameqa_collection, dim768): 创建一个问答集合。 集合包含三个字段ID、问题文本、问题向量、答案文本。 # 1. 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namequestion, dtypeDataType.VARCHAR, max_length500), FieldSchema(namequestion_vector, dtypeDataType.FLOAT_VECTOR, dimdim), FieldSchema(nameanswer, dtypeDataType.VARCHAR, max_length2000) ] # 2. 定义集合Schema schema CollectionSchema(fieldsfields, descriptionQA collection for semantic search) # 3. 创建集合 if utility.has_collection(collection_name): print(f集合 {collection_name} 已存在正在删除...) collection Collection(collection_name) collection.drop() collection Collection(namecollection_name, schemaschema) print(f集合 {collection_name} 创建成功向量维度: {dim}) # 4. 为向量字段创建索引以加速检索 index_params { metric_type: IP, # 使用内积IP作为相似度度量余弦相似度需数据归一化后等价于IP index_type: IVF_FLAT, # 一种平衡速度和精度的索引类型 params: {nlist: 128} # 聚类中心数数据量越大可适当增加 } collection.create_index(field_namequestion_vector, index_paramsindex_params) print(向量索引创建成功。) return collection # 初始化Milvus操作器 milvus_op MilvusOperator() collection milvus_op.create_qa_collection(dim768) # 维度需与RexUniNLU生成的向量维度一致关键点说明集合Collection相当于传统数据库中的表用于存储同一类数据。字段Field我们定义了四个字段。id是自增主键question和answer是原始文本question_vector是核心用于存储由RexUniNLU生成的向量。索引Index这是Milvus高效检索的关键。IVF_FLAT索引在速度和精度之间取得了很好的平衡。metric_type设置为‘IP’内积如果我们提前对向量做了L2归一化那么内积的结果就等价于余弦相似度。4.3 步骤三构建示例知识库并插入数据现在我们创建一些模拟的问答数据将它们向量化然后插入到Milvus集合中。def build_and_insert_demo_data(collection, encoder, num_samples100): 构建演示数据并插入Milvus。 在实际应用中这里应该是从你的数据库或文件中读取真实数据。 # 模拟一个简单的问答知识库 demo_qa_pairs [ {q: 如何办理手机卡线上销户, a: 您可以登录运营商官方APP在‘服务’或‘我的’页面找到‘销户’或‘号卡管理’入口按照提示完成身份验证和销户申请即可。}, {q: 忘记银行卡密码怎么办, a: 请您携带本人有效身份证件和银行卡前往任意银行网点办理密码重置业务。部分银行支持通过手机银行进行密码找回。}, {q: 飞机票改签需要多少手续费, a: 改签手续费取决于航空公司、票价类型、航班时间等因素。通常折扣越低的机票改签费用越高。建议您在订票APP或航空公司官网查看具体票规。}, {q: 怎样查询个人所得税缴纳记录, a: 您可以下载‘个人所得税’APP注册登录后在‘首页’或‘服务’板块找到‘收入纳税明细查询’选择年份即可查看。}, {q: 电脑开机黑屏无反应如何解决, a: 首先尝试重启电脑。如果无效检查显示器连接线和电源。若仍黑屏可能是内存条松动可尝试重新插拔内存条需断电操作。}, {q: 如何给家里的WiFi设置密码, a: 用网线连接电脑和路由器或手机连接路由器默认WiFi。在浏览器输入路由器管理地址如192.168.1.1登录后进入无线设置页面修改无线密码并保存重启。}, # ... 这里可以添加更多QA对 ] # 为了演示我们复制几份数据来模拟一个稍大的数据集 all_questions [] all_answers [] all_vectors [] print(开始生成问答向量并插入Milvus...) for i in range(num_samples): # 循环使用demo数据 qa demo_qa_pairs[i % len(demo_qa_pairs)] question qa[q] f (样例{i}) # 加个后缀以示区别 answer qa[a] # 使用RexUniNLU生成问题文本的向量 # 注意这里我们使用一个通用的Schema来生成向量适用于检索场景 vector, _ encoder.get_embedding(question, schema_labels[问题描述, 用户意图]) # **重要对向量进行L2归一化使内积(IP)等价于余弦相似度** norm np.linalg.norm(vector) if norm 0: vector vector / norm all_questions.append(question) all_answers.append(answer) all_vectors.append(vector.tolist()) # 转换为列表格式 if (i1) % 10 0: print(f已处理 {i1}/{num_samples} 条数据) # 准备插入Milvus的数据实体 entities [ all_questions, # field: question all_vectors, # field: question_vector all_answers # field: answer ] # 执行插入操作 insert_result collection.insert(entities) print(f数据插入完成共插入 {len(insert_result.primary_keys)} 条数据。) # 将数据从内存刷新到磁盘并加载到内存以便检索 collection.flush() collection.load() print(数据已加载就绪可以开始检索。) return insert_result # 插入演示数据 insert_result build_and_insert_demo_data(collection, encoder, num_samples50)关键点说明向量归一化在插入前我们对每个向量进行了L2归一化vector vector / norm。这是因为Milvus的IP内积度量方式在向量归一化后向量A·向量B就等于cos(θ)即余弦相似度。余弦相似度是衡量语义相似度更常用的指标。collection.load()这一步至关重要。它将集合中的数据加载到内存中只有加载后创建的索引才会生效检索速度才会快。4.4 步骤四实现语义相似度检索万事俱备现在我们可以实现核心的检索功能了给定一个用户的新问题找到知识库中最相似的Top-K个问题并返回对应的答案。def semantic_search(collection, encoder, query_text, top_k3): 语义相似度检索根据用户问题从知识库中找到最相似的问题和答案。 参数: collection: Milvus集合对象 encoder: RexUniNLU编码器 query_text: 用户输入的问题文本 top_k: 返回最相似的结果数量 返回: results: 包含相似问题、答案及相似度得分的列表 print(f\n用户查询: 「{query_text}」) # 1. 将用户查询文本向量化 query_vector, nlu_result encoder.get_embedding(query_text, schema_labels[问题描述, 用户意图]) # 同样需要对查询向量进行归一化 query_norm np.linalg.norm(query_vector) if query_norm 0: query_vector query_vector / query_norm query_vector [query_vector.tolist()] # 包装成列表因为search接口接受多个向量 # 2. 定义检索参数 search_params { metric_type: IP, # 使用内积因为我们已归一化所以等价于余弦相似度 params: {nprobe: 10} # 搜索时探查的聚类中心数影响速度和精度 } # 3. 执行检索 # 我们只检索question_vector字段但需要返回question和answer字段 results collection.search( dataquery_vector, anns_fieldquestion_vector, paramsearch_params, limittop_k, output_fields[question, answer] # 指定需要返回的字段 ) # 4. 整理并返回结果 formatted_results [] for hits in results: for hit in hits: # hit.distance 是相似度得分IP内积由于向量已归一化得分在[-1,1]之间越接近1越相似 formatted_results.append({ similarity_score: hit.distance, matched_question: hit.entity.get(question), answer: hit.entity.get(answer) }) return formatted_results, nlu_result # 进行几次检索测试 test_queries [ 银行卡密码忘记了怎么处理, 我要改签机票怎么收费, 手机卡不用了怎么在网上注销 ] for query in test_queries: search_results, nlu_info semantic_search(collection, encoder, query, top_k2) print(f\n检索结果:) for i, res in enumerate(search_results): print(f {i1}. [相似度: {res[similarity_score]:.4f}]) print(f 匹配问题: {res[matched_question]}) print(f 答案: {res[answer][:100]}...) # 截取部分答案显示 print(- * 50)运行这段代码你会看到系统能够根据语义相似度从我们构建的小知识库中找到与用户问题最匹配的条目。即使表述不完全相同如“忘记密码” vs “密码忘记了”基于向量的语义检索也能很好地工作。5. 进阶构建一个简单的推荐服务基础的检索功能已经实现。我们可以在此基础上构建一个更完整的、带有简单业务逻辑的推荐服务。例如我们可以设定一个相似度阈值只有超过阈值的结果才返回或者将RexUniNLU提取的结构化信息槽位也作为过滤条件。5.1 带阈值过滤的检索def search_with_threshold(collection, encoder, query_text, top_k5, score_threshold0.7): 带相似度阈值过滤的检索。 只有相似度得分高于阈值的结果才会被返回。 raw_results, nlu_result semantic_search(collection, encoder, query_text, top_ktop_k) filtered_results [res for res in raw_results if res[similarity_score] score_threshold] if not filtered_results: print(f未找到相似度高于 {score_threshold} 的结果。最高得分为 {raw_results[0][similarity_score]:.4f}。) # 可以选择返回空或者返回最高分的结果即使低于阈值 # filtered_results raw_results[:1] return filtered_results, nlu_result # 测试阈值过滤 query 办理销户的流程 results, _ search_with_threshold(collection, encoder, query, score_threshold0.8) print(f阈值过滤后结果数: {len(results)})5.2 结合NLU结构化信息的混合检索RexUniNLU的强大之处在于它能同时输出结构化信息。我们可以利用这一点进行更精准的“混合检索”。例如在电商场景中用户说“我想买便宜的红色连衣裙”RexUniNLU可以提取出[‘颜色红色’ ‘品类连衣裙’ ‘意图购买’]。我们既可以用整个句子的向量做语义检索也可以用提取出的“红色”、“连衣裙”等关键词在Milvus的标量字段中进行过滤。这需要我们在构建集合时额外增加一些标量字段如category,color并在插入数据时填充这些字段。检索时先通过向量检索召回一批相似商品再用NLU提取的属性进行过滤。由于篇幅限制这里不展开具体代码但思路是清晰的向量检索负责“语义相关性”标量过滤负责“属性精确匹配”两者结合能实现更智能的搜索与推荐。6. 总结与展望通过本文的实践我们完成了一个从0到1的语义检索系统搭建。我们利用RexUniNLU实现了零样本的文本理解和向量化利用Milvus实现了海量向量的高效相似度检索。这个组合为我们打开了通往许多智能应用的大门智能客服/问答系统正如本文示例快速从知识库中匹配用户问题。内容推荐根据用户阅读或搜索的历史内容向量推荐语义相似的新闻、文章或视频。商品搜索实现“搜我所想”即使商品标题中没有用户输入的关键词只要语义相关就能被找到。代码/文档检索根据自然语言描述查找相关的代码片段或技术文档。下一步的优化方向可以包括向量质量优化尝试不同的Schema来引导RexUniNLU生成更适合你垂直领域的向量。检索策略优化调整Milvus的索引参数如nlist,nprobe在速度和精度之间找到最佳平衡。引入重排序Re-ranking先用向量检索召回大量候选如100个再用一个更精细的模型或规则对Top结果进行重排序提升最终精度。系统部署将本文的代码封装成FastAPI服务提供稳定的HTTP API供前端或其他系统调用。这个技术栈的优势在于它的灵活性和低成本启动。你不需要标注数据只需要定义好Schema就能让模型理解一个新的领域。随着业务数据的积累你还可以用Milvus轻松地管理不断增长的向量数据。希望这个实例能成为你探索语义理解与向量检索世界的一块敲门砖。动手试试把它应用到你的具体场景中你会发现让机器“理解”人类并“智能”地推荐并没有想象中那么遥远。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。