StructBERT中文相似度模型实战教程：Python批量调用+Redis缓存+阈值动态配置完整示例-171主机测评

StructBERT中文相似度模型实战教程：Python批量调用+Redis缓存+阈值动态配置完整示例

1. 引言

你有没有遇到过这样的问题？面对海量的用户评论，想找出哪些是重复的；或者搭建一个智能客服系统，需要快速匹配用户问题和知识库答案；又或者在做内容推荐时，需要找到语义相近的文章。这些场景的核心，其实都是要判断两段文字的意思有多接近。

今天我要分享的，就是基于百度StructBERT大模型的中文句子相似度计算实战方案。这不是一个简单的API调用教程，而是一个完整的工程化解决方案。我会带你从零开始，搭建一个高可用的相似度计算服务，并实现批量处理、性能优化和智能阈值配置。

想象一下，你有一个电商平台，每天产生上万条商品评论。人工去重几乎不可能，但用这个方案，几行代码就能自动找出相似评论。或者你正在做一个问答系统，用户问"怎么改密码"，系统能自动匹配到"如何重置密码"这个标准问题。这就是文本相似度计算的魔力。

在接下来的内容里，我不会只讲理论，而是直接给你可运行的代码和实战经验。你会学到如何用Python批量处理数据，如何用Redis缓存加速计算，如何根据不同的业务场景动态调整判断阈值。这些都是我在实际项目中踩过坑、验证过的方案。

2. StructBERT相似度服务快速上手

2.1 服务已经准备好了

首先告诉你一个好消息：StructBERT相似度服务已经预装并配置好了开机自启。这意味着你不需要从头搭建环境，可以直接开始使用。

访问地址很简单：

http://gpu-pod698386bfe177c841fb0af650-5000.web.gpu.csdn.net/

打开这个链接，你会看到一个紫色渐变的Web界面。这个界面设计得很友好，支持电脑和手机访问，还能实时显示服务状态。如果你看到顶部的状态点是绿色的，就说明服务运行正常。

2.2 三种使用方式

这个服务提供了三种使用方式，适合不同需求的用户：

方式一：Web界面（最简单）如果你只是想偶尔计算几个句子的相似度，或者给非技术人员使用，Web界面是最佳选择。界面分为两个主要功能：

单句对比：输入两个句子，立即得到相似度分数
批量对比：输入一个源句子和多个目标句子，一次性计算所有相似度

方式二：API调用（最灵活）如果你是开发者，需要把相似度计算集成到自己的系统里，API接口是最佳选择。服务提供了两个核心接口：

/similarity：计算两个句子的相似度
/batch_similarity：批量计算相似度

方式三：命令行测试（最直接）如果你想快速验证服务是否正常，可以用curl命令测试：

curl -X POST http://127.0.0.1:5000/similarity \\
-H "Content-Type: application/json" \\
-d '{
"sentence1": "今天天气很好",
"sentence2": "今天阳光明媚"
}'

2.3 理解相似度分数

相似度计算的结果是一个0到1之间的数字。这个数字怎么理解呢？我给大家一个直观的参考：

0.9以上：意思几乎完全一样
- 例："我喜欢吃苹果" vs "我爱吃苹果" → 0.95
- 适用场景：严格查重、数据去重
0.7-0.9：意思很接近
- 例："怎么修改密码" vs "如何重置密码" → 0.85
- 适用场景：问答匹配、客服系统
0.4-0.7：有一定关联
- 例："手机没电了" vs "充电宝在哪借" → 0.65
- 适用场景：语义检索、内容推荐
0.4以下：基本没关系
- 例："今天天气很好" vs "我喜欢吃苹果" → 0.12
- 适用场景：不相关过滤

记住这些参考值，后面配置阈值时会很有用。

3. Python批量调用实战

现在进入实战环节。在实际项目中，我们很少只计算一对句子的相似度，更多时候需要批量处理。比如从数据库读取1000条评论，找出相似的；或者用户上传一个文件，里面有几百个问题需要匹配。

3.1 基础批量处理

先来看一个最简单的批量处理例子：

import requests
import time
from typing import List, Dict

class SimilarityClient:
"""相似度计算客户端"""

def __init__(self, base_url="http://127.0.0.1:5000"):
self.base_url = base_url
self.session = requests.Session() # 使用会话提高性能

def batch_compare(self, source: str, targets: List[str]) -> List[Dict]:
"""批量计算相似度"""
url = f"{self.base_url}/batch_similarity"

# 构建请求数据
data = {
"source": source,
"targets": targets
}

try:
# 发送请求
response = self.session.post(url, json=data, timeout=10)
response.raise_for_status() # 检查HTTP错误

result = response.json()
return result.get("results", [])

except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return []

def compare_multiple_pairs(self, pairs: List[tuple]) -> List[float]:
"""计算多对句子的相似度"""
results = []

for sentence1, sentence2 in pairs:
url = f"{self.base_url}/similarity"
data = {
"sentence1": sentence1,
"sentence2": sentence2
}

try:
response = self.session.post(url, json=data, timeout=5)
if response.status_code == 200:
similarity = response.json().get("similarity", 0)
results.append(similarity)
else:
results.append(0) # 出错时返回0

except Exception as e:
print(f"计算失败: {sentence1} vs {sentence2}, 错误: {e}")
results.append(0)

return results

# 使用示例
if __name__ == "__main__":
client = SimilarityClient()

# 场景1：客服问题匹配
user_question = "我的密码忘记了怎么办"
faq_questions = [
"如何修改登录密码",
"密码忘记了怎么找回",
"怎样注册新账号",
"账号被锁定了怎么办",
"如何联系客服"
]

print("=== 客服问题匹配 ===")
results = client.batch_compare(user_question, faq_questions)

# 按相似度排序
sorted_results = sorted(results, key=lambda x: x["similarity"], reverse=True)

for item in sorted_results:
similarity = item["similarity"]
sentence = item["sentence"]

# 根据相似度添加标签
if similarity >= 0.7:
tag = "✅ 高度匹配"
elif similarity >= 0.4:
tag = "⚠️ 部分匹配"
else:
tag = "❌ 不匹配"

print(f"{similarity:.4f} – {tag} – {sentence}")

# 场景2：文本去重
print("\\n=== 文本去重示例 ===")
comments = [
"这个产品非常好用，推荐购买",
"产品很好用，强烈推荐",
"质量不错，性价比高",
"物流速度很快，包装完好",
"这个产品非常好用，推荐购买" # 重复内容
]

# 计算所有两两之间的相似度
pairs = []
for i in range(len(comments)):
for j in range(i + 1, len(comments)):
pairs.append((comments[i], comments[j]))

similarities = client.compare_multiple_pairs(pairs)

# 找出相似度高的对
duplicate_threshold = 0.85
for idx, similarity in enumerate(similarities):
if similarity >= duplicate_threshold:
i, j = pairs[idx]
print(f"发现重复内容 (相似度: {similarity:.4f}):")
print(f" 原文: {i}")
print(f" 重复: {j}")

这个基础版本已经能处理很多场景了，但它有个问题：每次计算都要调用API，如果数据量大，速度会很慢。而且相同的句子对会被重复计算，浪费资源。

3.2 添加Redis缓存优化

为了解决性能问题，我们引入Redis缓存。Redis是一个内存数据库，读写速度极快，适合做缓存。

import redis
import json
import hashlib
from functools import lru_cache

class CachedSimilarityClient(SimilarityClient):
"""带缓存的相似度客户端"""

def __init__(self, base_url="http://127.0.0.1:5000", redis_host="localhost", redis_port=6379):
super().__init__(base_url)

# 连接Redis
try:
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
db=0,
decode_responses=True # 自动解码为字符串
)
self.redis_client.ping() # 测试连接
print("Redis连接成功")
except Exception as e:
print(f"Redis连接失败: {e}")
self.redis_client = None

def _get_cache_key(self, sentence1: str, sentence2: str) -> str:
"""生成缓存键"""
# 对句子进行排序，确保 (A,B) 和 (B,A) 使用相同的缓存键
sorted_sentences = tuple(sorted([sentence1, sentence2]))
key_str = f"{sorted_sentences[0]}|{sorted_sentences[1]}"

# 使用MD5生成固定长度的键
return f"similarity:{hashlib.md5(key_str.encode()).hexdigest()}"

def get_similarity_with_cache(self, sentence1: str, sentence2: str, expire_seconds=3600) -> float:
"""带缓存的相似度计算"""

# 如果Redis不可用，直接调用API
if not self.redis_client:
return self._call_api_directly(sentence1, sentence2)

cache_key = self._get_cache_key(sentence1, sentence2)

# 尝试从缓存读取
cached_value = self.redis_client.get(cache_key)
if cached_value is not None:
print(f"缓存命中: {sentence1[:20]}… vs {sentence2[:20]}…")
return float(cached_value)

# 缓存未命中，调用API
print(f"缓存未命中，调用API: {sentence1[:20]}… vs {sentence2[:20]}…")
similarity = self._call_api_directly(sentence1, sentence2)

# 存入缓存
if similarity is not None:
self.redis_client.setex(cache_key, expire_seconds, str(similarity))

return similarity

def _call_api_directly(self, sentence1: str, sentence2: str) -> float:
"""直接调用API（无缓存）"""
url = f"{self.base_url}/similarity"
data = {
"sentence1": sentence1,
"sentence2": sentence2
}

try:
response = self.session.post(url, json=data, timeout=5)
if response.status_code == 200:
return response.json().get("similarity", 0)
except Exception as e:
print(f"API调用失败: {e}")

return 0

def batch_compare_with_cache(self, source: str, targets: List[str]) -> List[Dict]:
"""带缓存的批量计算"""
results = []

for target in targets:
similarity = self.get_similarity_with_cache(source, target)
results.append({
"sentence": target,
"similarity": similarity
})

# 按相似度排序
return sorted(results, key=lambda x: x["similarity"], reverse=True)

def clear_cache(self):
"""清空相似度缓存"""
if self.redis_client:
# 删除所有以similarity:开头的键
keys = self.redis_client.keys("similarity:*")
if keys:
self.redis_client.delete(*keys)
print(f"已清除 {len(keys)} 个缓存项")

def get_cache_stats(self) -> Dict:
"""获取缓存统计信息"""
if not self.redis_client:
return {"error": "Redis未连接"}

stats = {
"total_keys": len(self.redis_client.keys("*")),
"similarity_keys": len(self.redis_client.keys("similarity:*")),
"memory_usage": self.redis_client.info("memory")["used_memory_human"]
}

return stats

# 使用示例
if __name__ == "__main__":
# 初始化带缓存的客户端
client = CachedSimilarityClient(
base_url="http://127.0.0.1:5000",
redis_host="localhost", # 根据实际情况修改
redis_port=6379
)

# 测试数据
test_pairs = [
("今天天气很好", "今天阳光明媚"),
("我喜欢吃苹果", "我爱吃苹果"),
("如何修改密码", "怎么重置密码"),
("今天天气很好", "我喜欢吃苹果") # 不相关的
]

print("=== 第一次计算（会调用API）===")
for s1, s2 in test_pairs:
similarity = client.get_similarity_with_cache(s1, s2)
print(f"{s1} vs {s2}: {similarity:.4f}")

print("\\n=== 第二次计算（从缓存读取）===")
for s1, s2 in test_pairs:
similarity = client.get_similarity_with_cache(s1, s2)
print(f"{s1} vs {s2}: {similarity:.4f}")

# 查看缓存统计
stats = client.get_cache_stats()
print(f"\\n缓存统计: {stats}")

# 批量计算示例
print("\\n=== 批量计算示例 ===")
source = "快递什么时候能到"
targets = [
"物流什么时候送达",
"包裹什么时候能到",
"快递延误了怎么办",
"如何查询快递状态",
"我要退货"
]

results = client.batch_compare_with_cache(source, targets)

for item in results:
sim = item["similarity"]
sent = item["sentence"]
if sim >= 0.7:
print(f"✅ {sim:.4f} – {sent}")
elif sim >= 0.4:
print(f"⚠️ {sim:.4f} – {sent}")
else:
print(f"❌ {sim:.4f} – {sent}")

这个带缓存的版本有什么好处呢？

速度大幅提升：相同的计算只做一次，后续直接从缓存读取

减少API调用：降低服务压力，特别是高并发场景

成本降低：如果使用付费API，能节省大量费用

稳定性增强：即使API暂时不可用，缓存中的数据还能用

缓存时间我设置了3600秒（1小时），你可以根据业务需求调整。对于变化不大的内容，可以设置更长的缓存时间；对于实时性要求高的，可以设置短一些。

3.3 处理大规模数据

在实际项目中，我们经常要处理成千上万条数据。这时候需要更高效的批量处理策略：

import concurrent.futures
from queue import Queue
import threading
import time

class BatchProcessor:
"""批量处理器 – 支持多线程和进度显示"""

def __init__(self, client, max_workers=5):
self.client = client
self.max_workers = max_workers

def process_large_dataset(self, source_sentences: List[str], target_sentences: List[str]) -> List[Dict]:
"""处理大规模数据集"""
total_pairs = len(source_sentences) * len(target_sentences)
print(f"需要计算 {total_pairs} 个句子对")

results = []
completed = 0
start_time = time.time()

# 使用线程池并行处理
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# 提交所有任务
future_to_pair = {}
for source in source_sentences:
for target in target_sentences:
future = executor.submit(self.client.get_similarity_with_cache, source, target)
future_to_pair[future] = (source, target)

# 收集结果
for future in concurrent.futures.as_completed(future_to_pair):
source, target = future_to_pair[future]
try:
similarity = future.result()
results.append({
"source": source,
"target": target,
"similarity": similarity
})

completed += 1
if completed % 100 == 0: # 每100个显示一次进度
elapsed = time.time() – start_time
speed = completed / elapsed if elapsed > 0 else 0
print(f"进度: {completed}/{total_pairs} ({completed/total_pairs*100:.1f}%), "
f"速度: {speed:.1f} 对/秒")

except Exception as e:
print(f"计算失败: {source} vs {target}, 错误: {e}")

elapsed = time.time() – start_time
print(f"\\n计算完成! 总计: {total_pairs} 对, 耗时: {elapsed:.2f} 秒, "
f"平均速度: {total_pairs/elapsed:.1f} 对/秒")

return results

def find_similar_groups(self, sentences: List[str], threshold: float = 0.8) -> List[List[str]]:
"""找出相似的句子组（聚类）"""
print(f"开始聚类 {len(sentences)} 个句子，阈值: {threshold}")

groups = []
processed = set()

for i, sentence in enumerate(sentences):
if sentence in processed:
continue

# 为新句子创建一个组
current_group = [sentence]
processed.add(sentence)

# 找出所有相似的句子
for j in range(i + 1, len(sentences)):
other = sentences[j]
if other in processed:
continue

# 计算相似度
similarity = self.client.get_similarity_with_cache(sentence, other)

if similarity >= threshold:
current_group.append(other)
processed.add(other)

if len(current_group) > 1:
groups.append(current_group)

print(f"找到 {len(groups)} 个相似组")
return groups

# 使用示例
if __name__ == "__main__":
client = CachedSimilarityClient()
processor = BatchProcessor(client, max_workers=10)

# 示例：评论去重
comments = [
"产品质量很好，非常满意",
"商品质量不错，很满意",
"物流速度很快，包装完好",
"快递很快，包装很好",
"客服态度很差，不推荐",
"服务态度不好，不满意",
"价格实惠，性价比高",
"价格便宜，物超所值",
"功能齐全，使用方便",
"功能很多，操作简单"
]

print("=== 评论聚类分析 ===")
groups = processor.find_similar_groups(comments, threshold=0.7)

for i, group in enumerate(groups, 1):
print(f"\\n第{i}组 ({len(group)} 条相似评论):")
for comment in group:
print(f" – {comment}")

# 示例：大规模批量计算
print("\\n=== 大规模批量计算示例 ===")

# 生成测试数据
sources = [f"测试句子{i}" for i in range(10)]
targets = [f"对比句子{j}" for j in range(10)]

# 小规模测试
test_results = processor.process_large_dataset(sources[:3], targets[:3])

print(f"\\n计算结果示例（前5个）:")
for result in test_results[:5]:
print(f"{result['source']} vs {result['target']}: {result['similarity']:.4f}")

这个批量处理器有几个关键特性：

多线程并行：同时计算多个句子对，大幅提升速度

进度显示：实时显示计算进度和速度

自动聚类：自动找出相似的句子组

错误处理：单个计算失败不影响整体进度

对于真正的大规模数据（比如百万级别），你可能还需要考虑：

分批处理，避免内存溢出
使用数据库存储中间结果
考虑分布式计算

4. 阈值动态配置策略

相似度计算出来后，怎么判断两个句子是否"相似"？这就是阈值配置的问题。不同的业务场景需要不同的阈值，而且这个阈值可能还需要动态调整。

4.1 静态阈值配置

先来看基础的阈值配置：

class ThresholdConfig:
"""阈值配置管理器"""

# 不同场景的默认阈值
DEFAULT_THRESHOLDS = {
"strict_deduplication": 0.9, # 严格去重
"qa_matching": 0.7, # 问答匹配
"content_recommendation": 0.5, # 内容推荐
"spam_detection": 0.8, # 垃圾检测
"plagiarism_check": 0.85, # 抄袭检测
}

def __init__(self):
self.thresholds = self.DEFAULT_THRESHOLDS.copy()
self.history = [] # 记录阈值调整历史

def get_threshold(self, scenario: str) -> float:
"""获取场景对应的阈值"""
return self.thresholds.get(scenario, 0.7)

def set_threshold(self, scenario: str, value: float):
"""设置阈值"""
if not 0 <= value <= 1:
raise ValueError("阈值必须在0到1之间")

old_value = self.thresholds.get(scenario)
self.thresholds[scenario] = value

# 记录调整历史
self.history.append({
"timestamp": time.time(),
"scenario": scenario,
"old_value": old_value,
"new_value": value,
"reason": "手动调整"
})

print(f"阈值更新: {scenario} = {value}")

def judge_similarity(self, similarity: float, scenario: str) -> Dict:
"""根据阈值判断相似度"""
threshold = self.get_threshold(scenario)

# 多级判断
if similarity >= threshold:
level = "high"
match = True
confidence = min(1.0, (similarity – threshold) / (1 – threshold))
elif similarity >= threshold * 0.7:
level = "medium"
match = False
confidence = (similarity – threshold * 0.7) / (threshold – threshold * 0.7)
else:
level = "low"
match = False
confidence = similarity / (threshold * 0.7)

return {
"similarity": similarity,
"threshold": threshold,
"match": match,
"level": level,
"confidence": confidence,
"scenario": scenario
}

def auto_adjust_threshold(self, scenario: str, feedback_data: List[Dict]):
"""根据反馈数据自动调整阈值"""
if not feedback_data:
return

# 分析反馈数据
correct_matches = []
incorrect_matches = []

for item in feedback_data:
similarity = item["similarity"]
is_correct = item["is_correct"]

if is_correct:
correct_matches.append(similarity)
else:
incorrect_matches.append(similarity)

if not correct_matches or not incorrect_matches:
print(f"数据不足，无法自动调整 {scenario} 的阈值")
return

# 计算最佳阈值（最大化F1分数）
best_threshold = 0.7
best_f1 = 0

for test_threshold in [i/100 for i in range(30, 96, 5)]: # 0.3到0.95，步长0.05
# 计算在这个阈值下的表现
tp = len([s for s in correct_matches if s >= test_threshold])
fp = len([s for s in incorrect_matches if s >= test_threshold])
fn = len([s for s in correct_matches if s < test_threshold])

# 计算精确率、召回率、F1分数
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0

if precision + recall > 0:
f1 = 2 * precision * recall / (precision + recall)
else:
f1 = 0

if f1 > best_f1:
best_f1 = f1
best_threshold = test_threshold

# 更新阈值
old_threshold = self.get_threshold(scenario)
self.set_threshold(scenario, best_threshold)

# 记录自动调整
self.history.append({
"timestamp": time.time(),
"scenario": scenario,
"old_value": old_threshold,
"new_value": best_threshold,
"reason": f"自动调整 (F1={best_f1:.3f})",
"stats": {
"correct_samples": len(correct_matches),
"incorrect_samples": len(incorrect_matches),
"best_f1": best_f1
}
})

print(f"自动调整完成: {scenario} 阈值从 {old_threshold:.3f} 调整为 {best_threshold:.3f}, F1={best_f1:.3f}")

# 使用示例
if __name__ == "__main__":
config = ThresholdConfig()

# 测试不同场景的阈值
test_cases = [
(0.92, "strict_deduplication", "论文查重"),
(0.75, "qa_matching", "客服问答"),
(0.60, "content_recommendation", "内容推荐"),
(0.35, "spam_detection", "垃圾检测"),
]

print("=== 阈值判断示例 ===")
for similarity, scenario, desc in test_cases:
result = config.judge_similarity(similarity, scenario)

status = "✅ 匹配" if result["match"] else "❌ 不匹配"
print(f"{desc}: 相似度={similarity:.3f}, 阈值={result['threshold']:.3f}, "
f"等级={result['level']}, {status}")

# 模拟反馈数据，用于自动调整阈值
print("\\n=== 自动阈值调整示例 ===")

# 模拟一些反馈数据
feedback_data = []

# 正确匹配的例子（相似度高，应该匹配）
for _ in range(50):
similarity = 0.7 + random.random() * 0.3 # 0.7-1.0
feedback_data.append({
"similarity": similarity,
"is_correct": True
})

# 错误匹配的例子（相似度低，不应该匹配）
for _ in range(50):
similarity = random.random() * 0.5 # 0-0.5
feedback_data.append({
"similarity": similarity,
"is_correct": False
})

# 自动调整阈值
config.auto_adjust_threshold("qa_matching", feedback_data)

# 查看调整历史
print("\\n=== 阈值调整历史 ===")
for record in config.history[-3:]: # 显示最后3条
time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record["timestamp"]))
print(f"{time_str} – {record['scenario']}: {record['old_value']:.3f} -> {record['new_value']:.3f} "
f"({record['reason']})")

4.2 动态阈值策略

静态阈值有时候不够用，特别是当数据分布变化时。我们需要能动态调整的阈值策略：

class DynamicThresholdManager:
"""动态阈值管理器"""

def __init__(self, initial_threshold=0.7, min_threshold=0.3, max_threshold=0.95):
self.current_threshold = initial_threshold
self.min_threshold = min_threshold
self.max_threshold = max_threshold

# 历史数据
self.similarity_history = []
self.feedback_history = [] # (similarity, is_correct)

# 滑动窗口
self.window_size = 100
self.adaptation_rate = 0.1 # 调整速率

def add_feedback(self, similarity: float, is_correct: bool):
"""添加反馈数据"""
self.feedback_history.append((similarity, is_correct))

# 保持窗口大小
if len(self.feedback_history) > self.window_size * 2:
self.feedback_history = self.feedback_history[-self.window_size*2:]

def update_threshold(self):
"""根据反馈更新阈值"""
if len(self.feedback_history) < self.window_size:
return # 数据不足

# 分析最近的数据
recent_data = self.feedback_history[-self.window_size:]

# 计算当前阈值下的表现
tp = fp = fn = tn = 0

for similarity, is_correct in recent_data:
predicted_match = similarity >= self.current_threshold

if predicted_match and is_correct:
tp += 1
elif predicted_match and not is_correct:
fp += 1
elif not predicted_match and is_correct:
fn += 1
else:
tn += 1

# 计算指标
precision = tp / (tp + fp) if (tp + fp) > 0 else 0
recall = tp / (tp + fn) if (tp + fn) > 0 else 0

# 根据业务需求调整
# 如果精确率太低（太多误报），提高阈值
# 如果召回率太低（太多漏报），降低阈值

adjustment = 0

if precision < 0.8: # 精确率太低
adjustment = self.adaptation_rate # 提高阈值
elif recall < 0.8: # 召回率太低
adjustment = -self.adaptation_rate # 降低阈值

# 应用调整
new_threshold = self.current_threshold + adjustment
new_threshold = max(self.min_threshold, min(self.max_threshold, new_threshold))

if new_threshold != self.current_threshold:
print(f"阈值调整: {self.current_threshold:.3f} -> {new_threshold:.3f} "
f"(精确率={precision:.3f}, 召回率={recall:.3f})")
self.current_threshold = new_threshold

def get_adaptive_threshold(self, context: Dict = None) -> float:
"""获取自适应阈值"""
if context:
# 可以根据上下文动态调整
# 例如：根据文本长度、领域、时间等因素
text_length = context.get("text_length", 0)

# 长文本通常需要更高的阈值
if text_length > 100:
return min(self.current_threshold + 0.05, self.max_threshold)
# 短文本可以放宽阈值
elif text_length < 20:
return max(self.current_threshold – 0.05, self.min_threshold)

return self.current_threshold

def judge_with_context(self, similarity: float, context: Dict = None) -> Dict:
"""带上下文的判断"""
threshold = self.get_adaptive_threshold(context)

return {
"similarity": similarity,
"threshold": threshold,
"match": similarity >= threshold,
"context": context
}

# 使用示例：智能客服系统
class SmartQASystem:
"""智能问答系统"""

def __init__(self):
self.client = CachedSimilarityClient()
self.threshold_manager = DynamicThresholdManager(initial_threshold=0.7)
self.qa_pairs = {
"如何修改密码": "请登录后进入个人中心，在安全设置中修改密码。",
"密码忘记了怎么办": "可以通过注册邮箱或手机号找回密码。",
"如何注册账号": "点击首页的注册按钮，填写相关信息即可。",
"账号被锁定了怎么办": "请联系客服解锁账号。",
}

def find_answer(self, question: str) -> Dict:
"""查找最匹配的答案"""
# 获取所有问题
candidate_questions = list(self.qa_pairs.keys())

# 批量计算相似度
results = self.client.batch_compare_with_cache(question, candidate_questions)

if not results:
return {"answer": "抱歉，没有找到相关答案。", "confidence": 0}

# 获取最佳匹配
best_match = results[0]
similarity = best_match["similarity"]
matched_question = best_match["sentence"]

# 获取上下文信息
context = {
"text_length": len(question),
"question_type": self._classify_question(question),
"time_of_day": time.localtime().tm_hour
}

# 使用动态阈值判断
judgment = self.threshold_manager.judge_with_context(similarity, context)

if judgment["match"]:
answer = self.qa_pairs[matched_question]
confidence = similarity

# 收集反馈（假设用户点击了"有帮助"）
self._collect_feedback(question, matched_question, similarity, is_correct=True)
else:
answer = "抱歉，没有找到相关答案，请尝试其他问法或联系人工客服。"
confidence = similarity

# 收集反馈（假设用户点击了"无帮助"）
self._collect_feedback(question, matched_question, similarity, is_correct=False)

# 更新阈值
self.threshold_manager.update_threshold()

return {
"answer": answer,
"matched_question": matched_question,
"similarity": similarity,
"threshold": judgment["threshold"],
"confidence": confidence,
"match": judgment["match"]
}

def _classify_question(self, question: str) -> str:
"""简单的问题分类"""
question_lower = question.lower()

if any(word in question_lower for word in ["密码", "登录", "账号"]):
return "account"
elif any(word in question_lower for word in ["支付", "退款", "订单"]):
return "payment"
elif any(word in question_lower for word in ["物流", "快递", "发货"]):
return "delivery"
else:
return "general"

def _collect_feedback(self, user_question: str, matched_question: str,
similarity: float, is_correct: bool):
"""收集用户反馈"""
# 在实际系统中，这里应该记录到数据库
# 这里简单模拟
self.threshold_manager.add_feedback(similarity, is_correct)

if is_correct:
print(f"✅ 反馈: 用户问题'{user_question[:20]}…' "
f"匹配到'{matched_question[:20]}…' (相似度={similarity:.3f}) – 正确")
else:
print(f"❌ 反馈: 用户问题'{user_question[:20]}…' "
f"匹配到'{matched_question[:20]}…' (相似度={similarity:.3f}) – 错误")

# 测试智能问答系统
if __name__ == "__main__":
qa_system = SmartQASystem()

test_questions = [
"怎么改密码",
"密码忘了咋办",
"如何注册",
"账号锁了",
"今天天气怎么样" # 不在知识库中的问题
]

print("=== 智能问答系统测试 ===")
for question in test_questions:
print(f"\\n用户问题: {question}")
result = qa_system.find_answer(question)

if result["match"]:
print(f"✅ 匹配成功 (相似度: {result['similarity']:.3f}, 阈值: {result['threshold']:.3f})")
print(f"匹配问题: {result['matched_question']}")
print(f"答案: {result['answer']}")
else:
print(f"❌ 未匹配 (相似度: {result['similarity']:.3f}, 阈值: {result['threshold']:.3f})")
print(f"系统回复: {result['answer']}")

# 显示当前阈值
print(f"\\n当前动态阈值: {qa_system.threshold_manager.current_threshold:.3f}")

这个动态阈值系统有几个优点：

自适应调整：根据用户反馈自动优化阈值

上下文感知：考虑文本长度、问题类型等因素

持续学习：随着数据积累，判断越来越准

业务定制：不同场景可以有不同的调整策略

5. 完整实战项目示例

现在我们把所有组件组合起来，构建一个完整的文本去重系统：

import json
import sqlite3
from datetime import datetime
from typing import List, Dict, Set
import hashlib

class TextDeduplicationSystem:
"""文本去重系统 – 完整示例"""

def __init__(self, db_path="text_deduplication.db"):
# 初始化组件
self.client = CachedSimilarityClient()
self.config = ThresholdConfig()
self.db_path = db_path

# 初始化数据库
self._init_database()

# 缓存已处理的文本哈希
self.text_cache = {}

def _init_database(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

# 创建文本表
cursor.execute('''
CREATE TABLE IF NOT EXISTS texts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
content_hash TEXT NOT NULL UNIQUE,
category TEXT,
source TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')

# 创建相似度记录表
cursor.execute('''
CREATE TABLE IF NOT EXISTS similarities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
text1_id INTEGER,
text2_id INTEGER,
similarity REAL NOT NULL,
is_duplicate BOOLEAN,
threshold REAL,
calculated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (text1_id) REFERENCES texts (id),
FOREIGN KEY (text2_id) REFERENCES texts (id),
UNIQUE(text1_id, text2_id)
)
''')

# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_texts_hash ON texts(content_hash)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_similarities_text1 ON similarities(text1_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_similarities_text2 ON similarities(text2_id)')

conn.commit()
conn.close()

def _get_text_hash(self, text: str) -> str:
"""计算文本哈希"""
# 清理文本：去除空格、转小写
cleaned = ' '.join(text.strip().lower().split())
return hashlib.md5(cleaned.encode()).hexdigest()

def add_text(self, text: str, category: str = None, source: str = None) -> int:
"""添加文本到数据库"""
content_hash = self._get_text_hash(text)

# 检查是否已存在
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

cursor.execute(
"SELECT id FROM texts WHERE content_hash = ?",
(content_hash,)
)
existing = cursor.fetchone()

if existing:
conn.close()
return existing[0] # 返回已有ID

# 插入新文本
cursor.execute(
"INSERT INTO texts (content, content_hash, category, source) VALUES (?, ?, ?, ?)",
(text, content_hash, category, source)
)

text_id = cursor.lastrowid
conn.commit()
conn.close()

# 更新缓存
self.text_cache[content_hash] = text_id

return text_id

def find_duplicates(self, new_texts: List[str], threshold: float = None) -> Dict:
"""查找重复文本"""
if threshold is None:
threshold = self.config.get_threshold("strict_deduplication")

print(f"开始查找重复文本，阈值: {threshold}")
print(f"新文本数量: {len(new_texts)}")

# 1. 添加新文本到数据库
new_text_ids = []
for text in new_texts:
text_id = self.add_text(text, category="new_batch")
new_text_ids.append(text_id)

# 2. 获取所有已有文本
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

cursor.execute("SELECT id, content FROM texts WHERE id NOT IN ({})".format(
','.join('?' for _ in new_text_ids)
), new_text_ids)

existing_texts = cursor.fetchall()
conn.close()

print(f"已有文本数量: {len(existing_texts)}")

# 3. 批量计算相似度
duplicates = []
processed_pairs = set()

for new_id in new_text_ids:
# 获取新文本内容
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT content FROM texts WHERE id = ?", (new_id,))
new_content = cursor.fetchone()[0]
conn.close()

# 与已有文本比较
existing_contents = [text for _, text in existing_texts]
existing_ids = [id for id, _ in existing_texts]

# 批量计算相似度
results = self.client.batch_compare_with_cache(new_content, existing_contents)

# 找出重复的
for result in results:
if result["similarity"] >= threshold:
# 找到对应的文本ID
idx = existing_contents.index(result["sentence"])
existing_id = existing_ids[idx]

# 避免重复记录
pair_key = tuple(sorted([new_id, existing_id]))
if pair_key in processed_pairs:
continue

duplicates.append({
"new_text_id": new_id,
"new_text": new_content,
"existing_text_id": existing_id,
"existing_text": result["sentence"],
"similarity": result["similarity"],
"is_duplicate": True
})

processed_pairs.add(pair_key)

# 保存到数据库
self._save_similarity(
new_id, existing_id,
result["similarity"],
True, threshold
)

# 4. 新文本之间的比较
print("检查新文本之间的重复…")
for i in range(len(new_texts)):
for j in range(i + 1, len(new_texts)):
text1 = new_texts[i]
text2 = new_texts[j]

similarity = self.client.get_similarity_with_cache(text1, text2)

if similarity >= threshold:
duplicates.append({
"new_text_id": new_text_ids[i],
"new_text": text1,
"existing_text_id": new_text_ids[j],
"existing_text": text2,
"similarity": similarity,
"is_duplicate": True
})

# 保存到数据库
self._save_similarity(
new_text_ids[i], new_text_ids[j],
similarity, True, threshold
)

return {
"total_new_texts": len(new_texts),
"total_existing_texts": len(existing_texts),
"threshold": threshold,
"duplicates_found": len(duplicates),
"duplicates": duplicates
}

def _save_similarity(self, text1_id: int, text2_id: int,
similarity: float, is_duplicate: bool, threshold: float):
"""保存相似度记录到数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

# 确保text1_id < text2_id，避免重复存储
id1, id2 = sorted([text1_id, text2_id])

cursor.execute('''
INSERT OR REPLACE INTO similarities
(text1_id, text2_id, similarity, is_duplicate, threshold)
VALUES (?, ?, ?, ?, ?)
''', (id1, id2, similarity, is_duplicate, threshold))

conn.commit()
conn.close()

def get_statistics(self) -> Dict:
"""获取统计信息"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

# 文本统计
cursor.execute("SELECT COUNT(*) FROM texts")
total_texts = cursor.fetchone()[0]

cursor.execute("SELECT COUNT(DISTINCT category) FROM texts WHERE category IS NOT NULL")
total_categories = cursor.fetchone()[0]

# 相似度统计
cursor.execute("SELECT COUNT(*) FROM similarities")
total_comparisons = cursor.fetchone()[0]

cursor.execute("SELECT COUNT(*) FROM similarities WHERE is_duplicate = 1")
total_duplicates = cursor.fetchone()[0]

cursor.execute("SELECT AVG(similarity) FROM similarities")
avg_similarity = cursor.fetchone()[0] or 0

conn.close()

return {
"total_texts": total_texts,
"total_categories": total_categories,
"total_comparisons": total_comparisons,
"total_duplicates": total_duplicates,
"duplicate_rate": total_duplicates / total_comparisons if total_comparisons > 0 else 0,
"average_similarity": round(avg_similarity, 4)
}

def export_duplicates(self, output_file: str = "duplicates.json"):
"""导出重复文本"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()

cursor.execute('''
SELECT
t1.content as text1,
t2.content as text2,
s.similarity,
s.calculated_at
FROM similarities s
JOIN texts t1 ON s.text1_id = t1.id
JOIN texts t2 ON s.text2_id = t2.id
WHERE s.is_duplicate = 1
ORDER BY s.similarity DESC
''')

duplicates = []
for row in cursor.fetchall():
duplicates.append({
"text1": row[0],
"text2": row[1],
"similarity": row[2],
"detected_at": row[3]
})

conn.close()

# 保存到文件
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(duplicates, f, ensure_ascii=False, indent=2)

print(f"已导出 {len(duplicates)} 条重复记录到 {output_file}")
return duplicates

# 使用示例
if __name__ == "__main__":
# 初始化系统
dedup_system = TextDeduplicationSystem()

# 示例数据
new_comments = [
"这个产品质量真的很好，非常满意！",
"商品质量不错，很满意的一次购物",
"物流速度很快，包装也很完好",
"快递很快，包装很好，点赞",
"客服态度很差，非常不满意",
"服务态度不好，体验很差",
"价格实惠，性价比很高",
"价格便宜，物超所值",
"功能齐全，使用起来很方便",
"功能很多，操作简单易懂",
# 一些明显的重复
"这个产品质量真的很好，非常满意！", # 完全重复
"商品质量不错，很满意", # 部分重复
"物流速度快，包装完好", # 语义重复
]

print("=== 文本去重系统演示 ===")

# 查找重复
result = dedup_system.find_duplicates(new_comments, threshold=0.85)

print(f"\\n分析结果:")
print(f"处理新文本: {result['total_new_texts']} 条")
print(f"对比已有文本: {result['total_existing_texts']} 条")
print(f"发现重复: {result['duplicates_found']} 处")

# 显示发现的重复
if result['duplicates']:
print("\\n发现的重复文本:")
for i, dup in enumerate(result['duplicates'][:5], 1): # 显示前5个
print(f"\\n{i}. 相似度: {dup['similarity']:.4f}")
print(f" 新文本: {dup['new_text'][:50]}…")
print(f" 重复文本: {dup['existing_text'][:50]}…")

# 获取统计信息
stats = dedup_system.get_statistics()
print(f"\\n系统统计:")
print(f"总文本数: {stats['total_texts']}")
print(f"总比较次数: {stats['total_comparisons']}")
print(f"重复数量: {stats['total_duplicates']}")
print(f"重复率: {stats['duplicate_rate']:.2%}")
print(f"平均相似度: {stats['average_similarity']:.4f}")

# 导出重复记录
dedup_system.export_duplicates("detected_duplicates.json")

# 测试不同阈值的效果
print("\\n=== 不同阈值效果测试 ===")
test_texts = [
"今天天气很好",
"今天天气真好",
"今天阳光明媚",
"我喜欢吃苹果"
]

for threshold in [0.7, 0.8, 0.9]:
print(f"\\n阈值: {threshold}")
test_result = dedup_system.find_duplicates(test_texts, threshold=threshold)

# 简单去重（只考虑新文本之间）
unique_texts = []
for dup in test_result['duplicates']:
if dup['new_text'] not in unique_texts:
unique_texts.append(dup['new_text'])

actual_unique = list(set(test_texts)) # 实际唯一文本

print(f" 去重前: {len(test_texts)} 条")
print(f" 去重后: {len(unique_texts)} 条")
print(f" 实际唯一: {len(actual_unique)} 条")
print(f" 准确率: {len(unique_texts)/len(actual_unique):.2%}")

这个完整的文本去重系统包含了：

数据库存储：保存所有文本和相似度记录

哈希去重：快速排除完全相同的文本

批量处理：高效处理大量数据

结果持久化：所有计算记录都保存到数据库

统计功能：提供各种统计信息

数据导出：方便结果分析

6. 总结

通过这个完整的实战教程，你应该已经掌握了StructBERT中文相似度计算的核心技术。我们来回顾一下重点：

6.1 核心要点总结

服务部署简单：StructBERT服务已经预装好，开箱即用，支持Web界面和API两种方式

批量处理高效：通过Redis缓存和多线程技术，大幅提升处理速度

阈值配置灵活：支持静态阈值、动态调整、上下文感知等多种策略

实战方案完整：从基础调用到完整系统，提供了可落地的解决方案

6.2 不同场景的应用建议

根据我的经验，不同场景可以这样配置：

严格查重（论文、代码）：阈值0.9+，配合文本预处理
问答匹配（客服、助手）：阈值0.7-0.8，使用动态调整
内容推荐（文章、视频）：阈值0.5-0.7，考虑用户兴趣
垃圾检测（评论、留言）：阈值0.8+，结合规则过滤

6.3 性能优化建议

如果你要处理海量数据，还可以考虑这些优化：

分布式计算：使用多台机器并行处理

向量化存储：将文本向量存入向量数据库，加速检索

分层过滤：先用简单规则（如关键词）过滤，再用模型计算

增量更新：只计算新增数据与已有数据的相似度

6.4 常见问题解决

在实际使用中，你可能会遇到这些问题：

服务响应慢：检查Redis是否正常，考虑增加缓存时间

内存占用高：批量处理时控制批次大小，及时清理缓存

准确率不够：尝试调整阈值，或者使用更复杂的文本预处理

并发问题：使用连接池，限制最大并发数

6.5 下一步学习方向

如果你想深入了解更多：

模型原理：了解StructBERT的架构和训练方式

微调模型：在自己的数据上微调，提升领域效果

多模态相似度：结合图像、语音等多模态信息

实时计算：构建流式处理系统，支持实时相似度计算

文本相似度计算是一个很有用的技术，在很多场景都能发挥作用。希望这个教程能帮你快速上手，在实际项目中用起来。记住，最好的学习方式就是动手实践，遇到问题就查文档、看日志，不断调整优化。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

StructBERT中文相似度模型实战教程：Python批量调用+Redis缓存+阈值动态配置完整示例