5步搞定Qwen3-Embedding-0.6B从部署到语义相似度判断全流程你是不是经常遇到这样的问题想快速搭建一个文本语义相似度判断系统但面对复杂的模型部署和代码编写就头疼或者想在自己的项目中集成文本嵌入功能却不知道从何下手今天我来分享一个超简单的解决方案——用Qwen3-Embedding-0.6B模型只需5个步骤就能从零开始搭建一个完整的语义相似度判断系统。无论你是AI新手还是有经验的开发者都能轻松上手。1. 为什么选择Qwen3-Embedding-0.6B在开始动手之前我们先简单了解一下这个模型。Qwen3-Embedding-0.6B是通义千问团队推出的文本嵌入模型专门用于将文本转换成向量表示。简单来说它能把一段文字变成一个数字列表向量然后通过比较这些向量的相似度就能判断两段文字的意思是不是相近。这个模型有几个特别实用的特点轻量高效只有6亿参数相比动辄几十亿的大模型它更省资源、运行更快多语言支持能处理超过100种语言包括各种编程语言上手简单提供了标准的API接口调用方式和OpenAI的接口几乎一样效果不错在文本检索、分类、聚类等任务上表现都很好最重要的是它特别适合我们这种想要快速搭建应用的场景。不需要复杂的配置不需要深厚的AI知识跟着下面的步骤走你就能拥有自己的语义相似度判断系统。2. 第一步环境准备与快速部署2.1 获取镜像资源首先我们需要获取Qwen3-Embedding-0.6B的镜像。这里我推荐使用CSDN星图镜像广场里面已经预置好了各种AI镜像包括我们今天要用的这个。访问CSDN星图镜像广场搜索Qwen3-Embedding-0.6B就能找到对应的镜像。点击一键部署系统会自动为你创建好运行环境。如果你在其他平台使用也可以直接从ModelScope或Hugging Face下载模型# 从ModelScope下载 git clone https://www.modelscope.cn/Qwen/Qwen3-Embedding-0.6B.git # 或者从Hugging Face下载 git clone https://huggingface.co/Qwen/Qwen3-Embedding-0.6B2.2 启动嵌入服务部署好环境后我们需要启动嵌入服务。这里使用sglang来启动服务它是个专门为AI模型设计的服务框架用起来特别简单。打开终端输入以下命令sglang serve --model-path /usr/local/bin/Qwen3-Embedding-0.6B --host 0.0.0.0 --port 30000 --is-embedding这个命令的意思是--model-path指定模型的位置--host 0.0.0.0让服务在所有网络接口上都能访问--port 30000使用30000端口--is-embedding告诉sglang这是个嵌入模型执行后如果看到类似下面的输出就说明服务启动成功了INFO: Started server process [12345] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:30000 (Press CTRLC to quit)服务启动后会一直运行在后台等待我们的调用。3. 第二步基础调用验证服务启动好了我们先来测试一下它能不能正常工作。这里我们用Jupyter Notebook来写测试代码当然你也可以用普通的Python脚本。3.1 连接嵌入服务首先我们需要安装必要的Python包pip install openai然后创建一个Python文件写入以下代码import openai # 创建客户端连接 client openai.Client( base_urlhttp://localhost:30000/v1, # 注意这里要改成你实际的服务地址 api_keyEMPTY # 因为我们是本地服务不需要真正的API密钥 ) # 测试文本嵌入 response client.embeddings.create( modelQwen3-Embedding-0.6B, input今天天气真好适合出去散步 ) # 查看返回结果 print(f嵌入向量长度{len(response.data[0].embedding)}) print(f前10个向量值{response.data[0].embedding[:10]})运行这段代码如果一切正常你会看到类似这样的输出嵌入向量长度1024 前10个向量值[0.0234, -0.0456, 0.0123, 0.0678, -0.0345, 0.0891, -0.0123, 0.0567, -0.0789, 0.0345]这说明模型已经成功把今天天气真好适合出去散步这句话转换成了一个1024维的向量。这个向量就是文本的数学表示我们可以用它来做各种有趣的事情。3.2 处理多个文本有时候我们需要一次性处理多个文本比如批量计算相似度。Qwen3-Embedding-0.6B也支持这个功能# 批量处理多个文本 texts [ 我喜欢吃苹果, 苹果是一种水果, 今天要去超市买东西, 天气不错适合外出 ] response client.embeddings.create( modelQwen3-Embedding-0.6B, inputtexts ) print(f处理了{len(response.data)}个文本) for i, embedding in enumerate(response.data): print(f文本{i1}的向量长度{len(embedding.embedding)})4. 第三步计算文本相似度有了文本向量我们就能计算相似度了。相似度的计算原理很简单两个向量的夹角越小说明它们越相似。数学上我们用余弦相似度来表示这个关系。4.1 计算余弦相似度我们来写一个计算相似度的函数import numpy as np from numpy.linalg import norm def cosine_similarity(vec1, vec2): 计算两个向量的余弦相似度 # 转换为numpy数组 vec1 np.array(vec1) vec2 np.array(vec2) # 计算点积 dot_product np.dot(vec1, vec2) # 计算模长 norm1 norm(vec1) norm2 norm(vec2) # 计算余弦相似度 similarity dot_product / (norm1 * norm2) return similarity # 测试相似度计算 text1 我喜欢吃苹果 text2 苹果是一种水果 text3 今天天气真好 # 获取三个文本的嵌入向量 embeddings [] for text in [text1, text2, text3]: response client.embeddings.create( modelQwen3-Embedding-0.6B, inputtext ) embeddings.append(response.data[0].embedding) # 计算相似度 sim12 cosine_similarity(embeddings[0], embeddings[1]) sim13 cosine_similarity(embeddings[0], embeddings[2]) print(f{text1} 和 {text2} 的相似度{sim12:.4f}) print(f{text1} 和 {text3} 的相似度{sim13:.4f})运行结果可能类似这样我喜欢吃苹果 和 苹果是一种水果 的相似度0.8567 我喜欢吃苹果 和 今天天气真好 的相似度0.1234可以看到前两个句子都提到了苹果语义上更接近所以相似度更高0.8567。而第一个句子和第三个句子主题完全不同相似度就很低0.1234。4.2 批量计算相似度在实际应用中我们经常需要计算一个查询文本和多个候选文本的相似度。比如在搜索引擎中用户输入一个查询我们需要从大量文档中找到最相关的。def find_most_similar(query, candidates): 在候选文本中找到与查询最相似的文本 # 获取查询文本的嵌入 query_response client.embeddings.create( modelQwen3-Embedding-0.6B, inputquery ) query_embedding query_response.data[0].embedding # 批量获取候选文本的嵌入 candidates_response client.embeddings.create( modelQwen3-Embedding-0.6B, inputcandidates ) # 计算相似度并排序 results [] for i, candidate_embedding in enumerate(candidates_response.data): similarity cosine_similarity(query_embedding, candidate_embedding.embedding) results.append({ text: candidates[i], similarity: similarity, rank: i }) # 按相似度降序排序 results.sort(keylambda x: x[similarity], reverseTrue) return results # 测试批量相似度计算 query 如何学习编程 candidates [ 编程入门教程, 学习Python的方法, 今天天气不错, 编程语言的选择, 晚餐吃什么好 ] results find_most_similar(query, candidates) print(f查询{query}) print(最相关的文本) for i, result in enumerate(results[:3]): # 显示前3个最相关的 print(f{i1}. {result[text]} (相似度{result[similarity]:.4f}))5. 第四步构建语义相似度判断系统现在我们已经掌握了基础操作接下来构建一个完整的语义相似度判断系统。这个系统能判断两个句子是否表达相同的意思。5.1 系统架构设计我们的系统包含以下几个部分文本预处理清洗和标准化输入文本嵌入生成将文本转换为向量相似度计算计算两个向量的余弦相似度阈值判断根据相似度得分判断是否语义相似5.2 完整实现代码import numpy as np from numpy.linalg import norm import openai from typing import List, Tuple, Dict class SemanticSimilaritySystem: 语义相似度判断系统 def __init__(self, base_url: str http://localhost:30000/v1): 初始化系统 Args: base_url: 嵌入服务的URL地址 self.client openai.Client( base_urlbase_url, api_keyEMPTY ) self.model_name Qwen3-Embedding-0.6B def preprocess_text(self, text: str) - str: 文本预处理 # 去除首尾空格 text text.strip() # 将多个连续空格替换为单个空格 text .join(text.split()) return text def get_embedding(self, text: str) - List[float]: 获取单个文本的嵌入向量 text self.preprocess_text(text) response self.client.embeddings.create( modelself.model_name, inputtext ) return response.data[0].embedding def get_embeddings_batch(self, texts: List[str]) - List[List[float]]: 批量获取文本嵌入向量 processed_texts [self.preprocess_text(text) for text in texts] response self.client.embeddings.create( modelself.model_name, inputprocessed_texts ) return [item.embedding for item in response.data] def cosine_similarity(self, vec1: List[float], vec2: List[float]) - float: 计算余弦相似度 vec1_array np.array(vec1) vec2_array np.array(vec2) dot_product np.dot(vec1_array, vec2_array) norm1 norm(vec1_array) norm2 norm(vec2_array) # 避免除零错误 if norm1 0 or norm2 0: return 0.0 similarity dot_product / (norm1 * norm2) return float(similarity) def judge_similarity(self, text1: str, text2: str, threshold: float 0.8) - Dict: 判断两个文本是否语义相似 Args: text1: 第一个文本 text2: 第二个文本 threshold: 相似度阈值默认0.8 Returns: 包含判断结果和详细信息的字典 # 获取嵌入向量 embedding1 self.get_embedding(text1) embedding2 self.get_embedding(text2) # 计算相似度 similarity self.cosine_similarity(embedding1, embedding2) # 判断是否相似 is_similar similarity threshold return { text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似 } def batch_judge(self, pairs: List[Tuple[str, str]], threshold: float 0.8) - List[Dict]: 批量判断多对文本的语义相似度 # 提取所有文本 all_texts [] for text1, text2 in pairs: all_texts.extend([text1, text2]) # 批量获取嵌入向量 embeddings self.get_embeddings_batch(all_texts) results [] idx 0 for text1, text2 in pairs: embedding1 embeddings[idx] embedding2 embeddings[idx 1] idx 2 similarity self.cosine_similarity(embedding1, embedding2) is_similar similarity threshold results.append({ text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似 }) return results # 使用示例 def main(): # 创建系统实例 system SemanticSimilaritySystem() # 测试用例 test_cases [ (我喜欢吃苹果, 苹果是一种水果), (今天天气真好, 外面阳光明媚), (学习编程很难, 晚餐吃什么), (如何学习Python, Python编程入门指南), (这个电影很好看, 这部电影很精彩) ] print(语义相似度判断结果) print( * 60) for text1, text2 in test_cases: result system.judge_similarity(text1, text2) print(f文本1: {text1}) print(f文本2: {text2}) print(f相似度: {result[similarity_score]:.4f}) print(f判断: {result[judgment]}) print(- * 40) if __name__ __main__: main()5.3 系统测试与优化运行上面的代码你会看到类似这样的输出语义相似度判断结果 文本1: 我喜欢吃苹果 文本2: 苹果是一种水果 相似度: 0.8567 判断: 语义相似 ---------------------------------------- 文本1: 今天天气真好 文本2: 外面阳光明媚 相似度: 0.9123 判断: 语义相似 ---------------------------------------- 文本1: 学习编程很难 文本2: 晚餐吃什么 相似度: 0.1234 判断: 语义不相似 ----------------------------------------这个系统现在已经可以工作了但我们可以进一步优化优化建议1调整阈值相似度阈值threshold是个很重要的参数。默认0.8可能不适合所有场景你可以根据实际需求调整严格场景如法律文档比对设为0.85-0.9一般场景如客服问答设为0.75-0.8宽松场景如话题聚类设为0.65-0.7优化建议2添加缓存机制如果经常处理相同的文本可以添加缓存来提升性能import hashlib from functools import lru_cache class OptimizedSemanticSimilaritySystem(SemanticSimilaritySystem): def __init__(self, base_url: str http://localhost:30000/v1): super().__init__(base_url) self.embedding_cache {} def _get_text_hash(self, text: str) - str: 生成文本的哈希值作为缓存键 return hashlib.md5(text.encode()).hexdigest() lru_cache(maxsize1000) def get_embedding_cached(self, text: str) - List[float]: 带缓存的嵌入获取 text_hash self._get_text_hash(text) if text_hash in self.embedding_cache: return self.embedding_cache[text_hash] embedding self.get_embedding(text) self.embedding_cache[text_hash] embedding return embedding def judge_similarity_optimized(self, text1: str, text2: str, threshold: float 0.8) - Dict: 优化版的相似度判断使用缓存 embedding1 self.get_embedding_cached(text1) embedding2 self.get_embedding_cached(text2) similarity self.cosine_similarity(embedding1, embedding2) is_similar similarity threshold return { text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似, from_cache: True # 可以添加缓存命中信息 }6. 第五步实际应用案例现在我们的系统已经搭建好了来看看它能用在哪些实际场景中。6.1 智能客服问答匹配在客服系统中用户的问题可能有很多种表达方式但核心意思是一样的。我们的系统能帮助匹配用户问题与标准答案def customer_service_application(): 智能客服问答匹配应用 system SemanticSimilaritySystem() # 知识库中的标准问答对 knowledge_base { 如何重置密码: 请访问设置页面点击忘记密码按照提示操作即可重置。, 账户被锁定怎么办: 请联系客服解锁账户或等待24小时自动解锁。, 如何查看账单: 登录后进入我的账户-账单详情即可查看。, 支持哪些支付方式: 我们支持支付宝、微信支付、银行卡支付。 } # 用户可能的各种提问方式 user_questions [ 密码忘了怎么弄, 账号被锁了怎么处理, 我想查一下我的消费记录, 可以用微信付款吗, 今天天气怎么样 # 这个不在知识库中 ] print(智能客服问答匹配) print( * 60) for question in user_questions: best_match None best_score 0 for standard_question, answer in knowledge_base.items(): result system.judge_similarity(question, standard_question, threshold0.7) if result[similarity_score] best_score: best_score result[similarity_score] best_match (standard_question, answer) if best_match and best_score 0.7: print(f用户问题{question}) print(f匹配到{best_match[0]} (相似度{best_score:.4f})) print(f回答{best_match[1]}) else: print(f用户问题{question}) print(未找到匹配答案将转接人工客服) print(- * 40) # 运行客服应用 customer_service_application()6.2 文档去重与聚类在处理大量文档时经常需要去除重复内容或相似内容def document_deduplication(): 文档去重应用 system SemanticSimilaritySystem() # 假设我们有一批文档 documents [ 人工智能是计算机科学的一个分支, AI技术正在快速发展, 机器学习是人工智能的重要部分, 计算机科学包含人工智能领域, 今天的天气非常不错, 人工智能研究如何让机器智能, 外面阳光明媚适合外出 ] print(文档去重结果) print( * 60) # 用于存储唯一文档 unique_docs [] for i, doc in enumerate(documents): is_duplicate False # 与已有唯一文档比较 for unique_doc in unique_docs: result system.judge_similarity(doc, unique_doc, threshold0.8) if result[is_similar]: print(f文档{i1}: {doc[:30]}...) print(f与文档{unique_doc[:30]}...相似 (相似度{result[similarity_score]:.4f})) print(→ 判定为重复文档) is_duplicate True break if not is_duplicate: unique_docs.append(doc) print(f文档{i1}: {doc[:30]}...) print(→ 添加为唯一文档) print(- * 40) print(f\n去重后剩余 {len(unique_docs)} 个唯一文档) for doc in unique_docs: print(f • {doc}) # 运行文档去重 document_deduplication()6.3 内容推荐系统根据用户历史喜欢的内容推荐相似的新内容def content_recommendation(): 内容推荐系统 system SemanticSimilaritySystem() # 用户历史喜欢的内容 user_likes [ Python编程入门教程, 机器学习基础概念, 深度学习实战指南 ] # 可推荐的内容库 content_library [ Python数据分析实战, Java编程从入门到精通, 机器学习算法详解, 深度学习框架比较, 人工智能发展趋势, Web前端开发教程, 数据库设计原理 ] print(为您推荐的内容) print( * 60) # 计算每个内容的推荐分数 recommendations [] for content in content_library: total_similarity 0 for liked_content in user_likes: result system.judge_similarity(content, liked_content) total_similarity result[similarity_score] # 平均相似度作为推荐分数 avg_similarity total_similarity / len(user_likes) recommendations.append({ content: content, score: avg_similarity }) # 按推荐分数排序 recommendations.sort(keylambda x: x[score], reverseTrue) # 输出前3个推荐 for i, rec in enumerate(recommendations[:3]): print(f{i1}. {rec[content]}) print(f 推荐分数{rec[score]:.4f}) print() # 运行内容推荐 content_recommendation()7. 总结通过这5个步骤我们完成了一个完整的语义相似度判断系统的搭建环境准备获取并部署Qwen3-Embedding-0.6B模型服务启动用sglang启动嵌入服务基础调用验证服务正常运行获取文本嵌入向量相似度计算实现余弦相似度计算和阈值判断系统构建创建完整的语义相似度判断系统这个系统的优势很明显部署简单几个命令就能跑起来使用方便API接口标准化调用简单效果实用能满足大多数语义相似度判断需求扩展性强可以轻松集成到各种应用中在实际使用中你可能会遇到一些常见问题这里也给你一些建议如果相似度判断不准怎么办尝试调整阈值参数检查文本预处理是否充分考虑使用更大的Qwen3-Embedding模型4B或8B版本如果响应速度慢怎么办启用上面提到的缓存机制考虑批量处理请求确保服务器资源充足如果想处理更长的文本怎么办Qwen3-Embedding支持长文本但要注意token限制对于超长文本可以考虑分段处理再合并结果这个系统现在已经可以投入使用了。你可以把它集成到你的网站、APP或者内部系统中用来做智能客服、内容去重、文档检索等各种有趣的应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。 5步搞定Qwen3-Embedding-0.6B：从部署到语义相似度判断全流程
5步搞定Qwen3-Embedding-0.6B从部署到语义相似度判断全流程你是不是经常遇到这样的问题想快速搭建一个文本语义相似度判断系统但面对复杂的模型部署和代码编写就头疼或者想在自己的项目中集成文本嵌入功能却不知道从何下手今天我来分享一个超简单的解决方案——用Qwen3-Embedding-0.6B模型只需5个步骤就能从零开始搭建一个完整的语义相似度判断系统。无论你是AI新手还是有经验的开发者都能轻松上手。1. 为什么选择Qwen3-Embedding-0.6B在开始动手之前我们先简单了解一下这个模型。Qwen3-Embedding-0.6B是通义千问团队推出的文本嵌入模型专门用于将文本转换成向量表示。简单来说它能把一段文字变成一个数字列表向量然后通过比较这些向量的相似度就能判断两段文字的意思是不是相近。这个模型有几个特别实用的特点轻量高效只有6亿参数相比动辄几十亿的大模型它更省资源、运行更快多语言支持能处理超过100种语言包括各种编程语言上手简单提供了标准的API接口调用方式和OpenAI的接口几乎一样效果不错在文本检索、分类、聚类等任务上表现都很好最重要的是它特别适合我们这种想要快速搭建应用的场景。不需要复杂的配置不需要深厚的AI知识跟着下面的步骤走你就能拥有自己的语义相似度判断系统。2. 第一步环境准备与快速部署2.1 获取镜像资源首先我们需要获取Qwen3-Embedding-0.6B的镜像。这里我推荐使用CSDN星图镜像广场里面已经预置好了各种AI镜像包括我们今天要用的这个。访问CSDN星图镜像广场搜索Qwen3-Embedding-0.6B就能找到对应的镜像。点击一键部署系统会自动为你创建好运行环境。如果你在其他平台使用也可以直接从ModelScope或Hugging Face下载模型# 从ModelScope下载 git clone https://www.modelscope.cn/Qwen/Qwen3-Embedding-0.6B.git # 或者从Hugging Face下载 git clone https://huggingface.co/Qwen/Qwen3-Embedding-0.6B2.2 启动嵌入服务部署好环境后我们需要启动嵌入服务。这里使用sglang来启动服务它是个专门为AI模型设计的服务框架用起来特别简单。打开终端输入以下命令sglang serve --model-path /usr/local/bin/Qwen3-Embedding-0.6B --host 0.0.0.0 --port 30000 --is-embedding这个命令的意思是--model-path指定模型的位置--host 0.0.0.0让服务在所有网络接口上都能访问--port 30000使用30000端口--is-embedding告诉sglang这是个嵌入模型执行后如果看到类似下面的输出就说明服务启动成功了INFO: Started server process [12345] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:30000 (Press CTRLC to quit)服务启动后会一直运行在后台等待我们的调用。3. 第二步基础调用验证服务启动好了我们先来测试一下它能不能正常工作。这里我们用Jupyter Notebook来写测试代码当然你也可以用普通的Python脚本。3.1 连接嵌入服务首先我们需要安装必要的Python包pip install openai然后创建一个Python文件写入以下代码import openai # 创建客户端连接 client openai.Client( base_urlhttp://localhost:30000/v1, # 注意这里要改成你实际的服务地址 api_keyEMPTY # 因为我们是本地服务不需要真正的API密钥 ) # 测试文本嵌入 response client.embeddings.create( modelQwen3-Embedding-0.6B, input今天天气真好适合出去散步 ) # 查看返回结果 print(f嵌入向量长度{len(response.data[0].embedding)}) print(f前10个向量值{response.data[0].embedding[:10]})运行这段代码如果一切正常你会看到类似这样的输出嵌入向量长度1024 前10个向量值[0.0234, -0.0456, 0.0123, 0.0678, -0.0345, 0.0891, -0.0123, 0.0567, -0.0789, 0.0345]这说明模型已经成功把今天天气真好适合出去散步这句话转换成了一个1024维的向量。这个向量就是文本的数学表示我们可以用它来做各种有趣的事情。3.2 处理多个文本有时候我们需要一次性处理多个文本比如批量计算相似度。Qwen3-Embedding-0.6B也支持这个功能# 批量处理多个文本 texts [ 我喜欢吃苹果, 苹果是一种水果, 今天要去超市买东西, 天气不错适合外出 ] response client.embeddings.create( modelQwen3-Embedding-0.6B, inputtexts ) print(f处理了{len(response.data)}个文本) for i, embedding in enumerate(response.data): print(f文本{i1}的向量长度{len(embedding.embedding)})4. 第三步计算文本相似度有了文本向量我们就能计算相似度了。相似度的计算原理很简单两个向量的夹角越小说明它们越相似。数学上我们用余弦相似度来表示这个关系。4.1 计算余弦相似度我们来写一个计算相似度的函数import numpy as np from numpy.linalg import norm def cosine_similarity(vec1, vec2): 计算两个向量的余弦相似度 # 转换为numpy数组 vec1 np.array(vec1) vec2 np.array(vec2) # 计算点积 dot_product np.dot(vec1, vec2) # 计算模长 norm1 norm(vec1) norm2 norm(vec2) # 计算余弦相似度 similarity dot_product / (norm1 * norm2) return similarity # 测试相似度计算 text1 我喜欢吃苹果 text2 苹果是一种水果 text3 今天天气真好 # 获取三个文本的嵌入向量 embeddings [] for text in [text1, text2, text3]: response client.embeddings.create( modelQwen3-Embedding-0.6B, inputtext ) embeddings.append(response.data[0].embedding) # 计算相似度 sim12 cosine_similarity(embeddings[0], embeddings[1]) sim13 cosine_similarity(embeddings[0], embeddings[2]) print(f{text1} 和 {text2} 的相似度{sim12:.4f}) print(f{text1} 和 {text3} 的相似度{sim13:.4f})运行结果可能类似这样我喜欢吃苹果 和 苹果是一种水果 的相似度0.8567 我喜欢吃苹果 和 今天天气真好 的相似度0.1234可以看到前两个句子都提到了苹果语义上更接近所以相似度更高0.8567。而第一个句子和第三个句子主题完全不同相似度就很低0.1234。4.2 批量计算相似度在实际应用中我们经常需要计算一个查询文本和多个候选文本的相似度。比如在搜索引擎中用户输入一个查询我们需要从大量文档中找到最相关的。def find_most_similar(query, candidates): 在候选文本中找到与查询最相似的文本 # 获取查询文本的嵌入 query_response client.embeddings.create( modelQwen3-Embedding-0.6B, inputquery ) query_embedding query_response.data[0].embedding # 批量获取候选文本的嵌入 candidates_response client.embeddings.create( modelQwen3-Embedding-0.6B, inputcandidates ) # 计算相似度并排序 results [] for i, candidate_embedding in enumerate(candidates_response.data): similarity cosine_similarity(query_embedding, candidate_embedding.embedding) results.append({ text: candidates[i], similarity: similarity, rank: i }) # 按相似度降序排序 results.sort(keylambda x: x[similarity], reverseTrue) return results # 测试批量相似度计算 query 如何学习编程 candidates [ 编程入门教程, 学习Python的方法, 今天天气不错, 编程语言的选择, 晚餐吃什么好 ] results find_most_similar(query, candidates) print(f查询{query}) print(最相关的文本) for i, result in enumerate(results[:3]): # 显示前3个最相关的 print(f{i1}. {result[text]} (相似度{result[similarity]:.4f}))5. 第四步构建语义相似度判断系统现在我们已经掌握了基础操作接下来构建一个完整的语义相似度判断系统。这个系统能判断两个句子是否表达相同的意思。5.1 系统架构设计我们的系统包含以下几个部分文本预处理清洗和标准化输入文本嵌入生成将文本转换为向量相似度计算计算两个向量的余弦相似度阈值判断根据相似度得分判断是否语义相似5.2 完整实现代码import numpy as np from numpy.linalg import norm import openai from typing import List, Tuple, Dict class SemanticSimilaritySystem: 语义相似度判断系统 def __init__(self, base_url: str http://localhost:30000/v1): 初始化系统 Args: base_url: 嵌入服务的URL地址 self.client openai.Client( base_urlbase_url, api_keyEMPTY ) self.model_name Qwen3-Embedding-0.6B def preprocess_text(self, text: str) - str: 文本预处理 # 去除首尾空格 text text.strip() # 将多个连续空格替换为单个空格 text .join(text.split()) return text def get_embedding(self, text: str) - List[float]: 获取单个文本的嵌入向量 text self.preprocess_text(text) response self.client.embeddings.create( modelself.model_name, inputtext ) return response.data[0].embedding def get_embeddings_batch(self, texts: List[str]) - List[List[float]]: 批量获取文本嵌入向量 processed_texts [self.preprocess_text(text) for text in texts] response self.client.embeddings.create( modelself.model_name, inputprocessed_texts ) return [item.embedding for item in response.data] def cosine_similarity(self, vec1: List[float], vec2: List[float]) - float: 计算余弦相似度 vec1_array np.array(vec1) vec2_array np.array(vec2) dot_product np.dot(vec1_array, vec2_array) norm1 norm(vec1_array) norm2 norm(vec2_array) # 避免除零错误 if norm1 0 or norm2 0: return 0.0 similarity dot_product / (norm1 * norm2) return float(similarity) def judge_similarity(self, text1: str, text2: str, threshold: float 0.8) - Dict: 判断两个文本是否语义相似 Args: text1: 第一个文本 text2: 第二个文本 threshold: 相似度阈值默认0.8 Returns: 包含判断结果和详细信息的字典 # 获取嵌入向量 embedding1 self.get_embedding(text1) embedding2 self.get_embedding(text2) # 计算相似度 similarity self.cosine_similarity(embedding1, embedding2) # 判断是否相似 is_similar similarity threshold return { text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似 } def batch_judge(self, pairs: List[Tuple[str, str]], threshold: float 0.8) - List[Dict]: 批量判断多对文本的语义相似度 # 提取所有文本 all_texts [] for text1, text2 in pairs: all_texts.extend([text1, text2]) # 批量获取嵌入向量 embeddings self.get_embeddings_batch(all_texts) results [] idx 0 for text1, text2 in pairs: embedding1 embeddings[idx] embedding2 embeddings[idx 1] idx 2 similarity self.cosine_similarity(embedding1, embedding2) is_similar similarity threshold results.append({ text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似 }) return results # 使用示例 def main(): # 创建系统实例 system SemanticSimilaritySystem() # 测试用例 test_cases [ (我喜欢吃苹果, 苹果是一种水果), (今天天气真好, 外面阳光明媚), (学习编程很难, 晚餐吃什么), (如何学习Python, Python编程入门指南), (这个电影很好看, 这部电影很精彩) ] print(语义相似度判断结果) print( * 60) for text1, text2 in test_cases: result system.judge_similarity(text1, text2) print(f文本1: {text1}) print(f文本2: {text2}) print(f相似度: {result[similarity_score]:.4f}) print(f判断: {result[judgment]}) print(- * 40) if __name__ __main__: main()5.3 系统测试与优化运行上面的代码你会看到类似这样的输出语义相似度判断结果 文本1: 我喜欢吃苹果 文本2: 苹果是一种水果 相似度: 0.8567 判断: 语义相似 ---------------------------------------- 文本1: 今天天气真好 文本2: 外面阳光明媚 相似度: 0.9123 判断: 语义相似 ---------------------------------------- 文本1: 学习编程很难 文本2: 晚餐吃什么 相似度: 0.1234 判断: 语义不相似 ----------------------------------------这个系统现在已经可以工作了但我们可以进一步优化优化建议1调整阈值相似度阈值threshold是个很重要的参数。默认0.8可能不适合所有场景你可以根据实际需求调整严格场景如法律文档比对设为0.85-0.9一般场景如客服问答设为0.75-0.8宽松场景如话题聚类设为0.65-0.7优化建议2添加缓存机制如果经常处理相同的文本可以添加缓存来提升性能import hashlib from functools import lru_cache class OptimizedSemanticSimilaritySystem(SemanticSimilaritySystem): def __init__(self, base_url: str http://localhost:30000/v1): super().__init__(base_url) self.embedding_cache {} def _get_text_hash(self, text: str) - str: 生成文本的哈希值作为缓存键 return hashlib.md5(text.encode()).hexdigest() lru_cache(maxsize1000) def get_embedding_cached(self, text: str) - List[float]: 带缓存的嵌入获取 text_hash self._get_text_hash(text) if text_hash in self.embedding_cache: return self.embedding_cache[text_hash] embedding self.get_embedding(text) self.embedding_cache[text_hash] embedding return embedding def judge_similarity_optimized(self, text1: str, text2: str, threshold: float 0.8) - Dict: 优化版的相似度判断使用缓存 embedding1 self.get_embedding_cached(text1) embedding2 self.get_embedding_cached(text2) similarity self.cosine_similarity(embedding1, embedding2) is_similar similarity threshold return { text1: text1, text2: text2, similarity_score: similarity, threshold: threshold, is_similar: is_similar, judgment: 语义相似 if is_similar else 语义不相似, from_cache: True # 可以添加缓存命中信息 }6. 第五步实际应用案例现在我们的系统已经搭建好了来看看它能用在哪些实际场景中。6.1 智能客服问答匹配在客服系统中用户的问题可能有很多种表达方式但核心意思是一样的。我们的系统能帮助匹配用户问题与标准答案def customer_service_application(): 智能客服问答匹配应用 system SemanticSimilaritySystem() # 知识库中的标准问答对 knowledge_base { 如何重置密码: 请访问设置页面点击忘记密码按照提示操作即可重置。, 账户被锁定怎么办: 请联系客服解锁账户或等待24小时自动解锁。, 如何查看账单: 登录后进入我的账户-账单详情即可查看。, 支持哪些支付方式: 我们支持支付宝、微信支付、银行卡支付。 } # 用户可能的各种提问方式 user_questions [ 密码忘了怎么弄, 账号被锁了怎么处理, 我想查一下我的消费记录, 可以用微信付款吗, 今天天气怎么样 # 这个不在知识库中 ] print(智能客服问答匹配) print( * 60) for question in user_questions: best_match None best_score 0 for standard_question, answer in knowledge_base.items(): result system.judge_similarity(question, standard_question, threshold0.7) if result[similarity_score] best_score: best_score result[similarity_score] best_match (standard_question, answer) if best_match and best_score 0.7: print(f用户问题{question}) print(f匹配到{best_match[0]} (相似度{best_score:.4f})) print(f回答{best_match[1]}) else: print(f用户问题{question}) print(未找到匹配答案将转接人工客服) print(- * 40) # 运行客服应用 customer_service_application()6.2 文档去重与聚类在处理大量文档时经常需要去除重复内容或相似内容def document_deduplication(): 文档去重应用 system SemanticSimilaritySystem() # 假设我们有一批文档 documents [ 人工智能是计算机科学的一个分支, AI技术正在快速发展, 机器学习是人工智能的重要部分, 计算机科学包含人工智能领域, 今天的天气非常不错, 人工智能研究如何让机器智能, 外面阳光明媚适合外出 ] print(文档去重结果) print( * 60) # 用于存储唯一文档 unique_docs [] for i, doc in enumerate(documents): is_duplicate False # 与已有唯一文档比较 for unique_doc in unique_docs: result system.judge_similarity(doc, unique_doc, threshold0.8) if result[is_similar]: print(f文档{i1}: {doc[:30]}...) print(f与文档{unique_doc[:30]}...相似 (相似度{result[similarity_score]:.4f})) print(→ 判定为重复文档) is_duplicate True break if not is_duplicate: unique_docs.append(doc) print(f文档{i1}: {doc[:30]}...) print(→ 添加为唯一文档) print(- * 40) print(f\n去重后剩余 {len(unique_docs)} 个唯一文档) for doc in unique_docs: print(f • {doc}) # 运行文档去重 document_deduplication()6.3 内容推荐系统根据用户历史喜欢的内容推荐相似的新内容def content_recommendation(): 内容推荐系统 system SemanticSimilaritySystem() # 用户历史喜欢的内容 user_likes [ Python编程入门教程, 机器学习基础概念, 深度学习实战指南 ] # 可推荐的内容库 content_library [ Python数据分析实战, Java编程从入门到精通, 机器学习算法详解, 深度学习框架比较, 人工智能发展趋势, Web前端开发教程, 数据库设计原理 ] print(为您推荐的内容) print( * 60) # 计算每个内容的推荐分数 recommendations [] for content in content_library: total_similarity 0 for liked_content in user_likes: result system.judge_similarity(content, liked_content) total_similarity result[similarity_score] # 平均相似度作为推荐分数 avg_similarity total_similarity / len(user_likes) recommendations.append({ content: content, score: avg_similarity }) # 按推荐分数排序 recommendations.sort(keylambda x: x[score], reverseTrue) # 输出前3个推荐 for i, rec in enumerate(recommendations[:3]): print(f{i1}. {rec[content]}) print(f 推荐分数{rec[score]:.4f}) print() # 运行内容推荐 content_recommendation()7. 总结通过这5个步骤我们完成了一个完整的语义相似度判断系统的搭建环境准备获取并部署Qwen3-Embedding-0.6B模型服务启动用sglang启动嵌入服务基础调用验证服务正常运行获取文本嵌入向量相似度计算实现余弦相似度计算和阈值判断系统构建创建完整的语义相似度判断系统这个系统的优势很明显部署简单几个命令就能跑起来使用方便API接口标准化调用简单效果实用能满足大多数语义相似度判断需求扩展性强可以轻松集成到各种应用中在实际使用中你可能会遇到一些常见问题这里也给你一些建议如果相似度判断不准怎么办尝试调整阈值参数检查文本预处理是否充分考虑使用更大的Qwen3-Embedding模型4B或8B版本如果响应速度慢怎么办启用上面提到的缓存机制考虑批量处理请求确保服务器资源充足如果想处理更长的文本怎么办Qwen3-Embedding支持长文本但要注意token限制对于超长文本可以考虑分段处理再合并结果这个系统现在已经可以投入使用了。你可以把它集成到你的网站、APP或者内部系统中用来做智能客服、内容去重、文档检索等各种有趣的应用。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。