零基础玩转bge-large-zh-v1.5：中文文本相似度匹配实战教程

1. 引言：从零开始构建中文语义匹配系统

在自然语言处理的实际应用中，如何准确衡量两段中文文本的语义相似度一直是一个核心挑战。传统的关键词匹配方法难以捕捉深层语义关系，而基于深度学习的嵌入模型为我们提供了全新的解决方案。

本文将带你从零开始，使用已部署的bge-large-zh-v1.5模型服务，完成一个完整的中文文本相似度匹配实践项目。无论你是NLP新手还是希望快速验证模型能力的开发者，都能通过本教程快速上手。

阅读本文后，你将掌握：

如何验证和调用本地部署的embedding模型服务
基于真实场景的中文句子相似度计算全流程
文本向量化与余弦相似度匹配的核心实现
可直接复用的完整代码模板与工程化建议

2. 环境准备与模型验证

2.1 进入工作目录

首先确保你已经进入指定的工作空间目录：

cd /root/workspace

该路径是模型服务默认的工作目录，日志文件和服务端点均在此上下文中运行。

2.2 验证模型服务状态

通过查看启动日志确认模型是否成功加载：

cat sglang.log

当输出中包含类似以下信息时，表示bge-large-zh-v1.5模型已成功启动并监听请求：

INFO: Started server process [1] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:30000 (Press CTRL+C to quit)

同时，日志应显示模型加载完成的信息，如：

Loading BAAI/bge-large-zh-v1.5... Model loaded successfully with max length 512

2.3 测试模型基本调用

使用Python脚本测试模型的基本embedding功能：

import openai # 初始化客户端，连接本地SGLang服务 client = openai.Client( base_url="http://localhost:30000/v1", api_key="EMPTY" ) # 发起文本嵌入请求 response = client.embeddings.create( model="bge-large-zh-v1.5", input="今天天气真好" ) # 输出结果结构 print(response)

预期输出为包含嵌入向量的对象，其主要字段包括：

object: 类型标识（"list"）
data: 包含向量数据的列表
model: 使用的模型名称
usage: token使用统计
向量维度：1024维浮点数数组

这表明模型服务已正常运行，可以接收后续的批量请求。

3. 核心实现：中文文本相似度匹配

3.1 构建文本嵌入函数

封装一个通用的文本向量化函数，便于后续重复调用：

def get_embedding(text): """ 获取单个文本的embedding向量 """ response = client.embeddings.create( model="bge-large-zh-v1.5", input=text ) return response.data[0].embedding def get_embeddings(texts): """ 批量获取多个文本的embedding向量 """ response = client.embeddings.create( model="bge-large-zh-v1.5", input=texts ) return [d.embedding for d in response.data]

注意：对于长文本（超过512个token），模型会自动截断。建议提前进行文本预处理以保证效果。

3.2 计算余弦相似度

定义向量间相似度计算方法，采用最常用的余弦相似度：

import numpy as np from sklearn.metrics.pairwise import cosine_similarity def calculate_similarity(vec1, vec2): """ 计算两个向量的余弦相似度 """ v1 = np.array(vec1).reshape(1, -1) v2 = np.array(vec2).reshape(1, -1) return cosine_similarity(v1, v2)[0][0] def batch_similarity(query_vec, corpus_vecs): """ 批量计算查询向量与语料库向量的相似度 """ query_vec = np.array(query_vec).reshape(1, -1) corpus_vecs = np.array(corpus_vecs) similarities = cosine_similarity(query_vec, corpus_vecs)[0] return similarities

3.3 实现相似句子检索器

构建完整的检索类，整合向量化与匹配逻辑：

class SimilarSentenceRetriever: def __init__(self): self.client = openai.Client(base_url="http://localhost:30000/v1", api_key="EMPTY") self.corpus = [] self.corpus_embeddings = [] def add_corpus(self, sentences): """添加语料库并生成嵌入""" self.corpus = sentences self.corpus_embeddings = get_embeddings(sentences) print(f"已加载 {len(sentences)} 条语料到索引") def search(self, query, top_k=5): """搜索最相似的句子""" if not self.corpus_embeddings: raise ValueError("请先调用add_corpus()加载语料库") # 获取查询向量 query_embedding = get_embedding(query) # 计算相似度 similarities = batch_similarity(query_embedding, self.corpus_embeddings) # 排序并返回top-k结果 top_indices = np.argsort(similarities)[::-1][:top_k] results = [] for idx in top_indices: results.append({ 'sentence': self.corpus[idx], 'similarity': float(similarities[idx]) }) return results

4. 实战演练：构建问答匹配系统

4.1 准备语料数据

定义一个常见问题库作为匹配目标：

faq_corpus = [ "如何重置我的密码", "忘记登录账号怎么办", "会员服务有哪些权益", "订单支付失败怎么处理", "发票申请流程是什么", "商品退换货政策说明", "客服工作时间是几点", "物流配送需要多久" ]

4.2 初始化检索系统

# 创建检索器实例 retriever = SimilarSentenceRetriever() # 加载FAQ语料库 retriever.add_corpus(faq_corpus)

4.3 执行相似度查询

模拟用户提问并查找最匹配的标准问题：

# 用户输入的各种表述方式 user_queries = [ "我忘了密码该怎么找回", "登录不了账户怎么办", "你们的会员有什么好处", "付款的时候出错了", "怎么开购货发票" ] # 对每个查询执行匹配 for query in user_queries: print(f"\n用户提问: {query}") results = retriever.search(query, top_k=3) for i, res in enumerate(results): print(f" Top{i+1}: '{res['sentence']}' (相似度: {res['similarity']:.3f})")

输出示例：

用户提问: 我忘了密码该怎么找回 Top1: '如何重置我的密码' (相似度: 0.876) Top2: '忘记登录账号怎么办' (相似度: 0.792) Top3: '客服工作时间是几点' (相似度: 0.413)

可以看到，尽管用户使用了不同的表达方式，系统仍能准确识别其意图并匹配到标准问题。

5. 性能优化与工程建议

5.1 批量处理提升效率

避免逐条请求，尽量使用批量接口减少网络开销：

# ✅ 推荐：批量处理 batch_texts = ["文本A", "文本B", "文本C"] embeddings = get_embeddings(batch_texts) # ❌ 不推荐：循环单条请求 embeddings = [] for text in texts: emb = get_embedding(text) # 多次HTTP请求 embeddings.append(emb)

5.2 添加结果过滤机制

设置相似度阈值，防止低质量匹配：

def search_with_threshold(self, query, top_k=5, threshold=0.6): results = self.search(query, top_k * 2) # 先取更多候选 filtered = [r for r in results if r['similarity'] >= threshold] return filtered[:top_k]

5.3 缓存高频查询结果

对常见查询进行缓存，减少重复计算：

from functools import lru_cache @lru_cache(maxsize=1000) def cached_get_embedding(text): return get_embedding(text)

5.4 监控与日志记录

添加基本的性能监控：

import time def timed_search(retriever, query): start = time.time() results = retriever.search(query) duration = time.time() - start print(f"查询耗时: {duration*1000:.2f}ms") return results