Telegram关键词搜索Bot 文本相似度计算：如何过滤Telegram机器人中的重复刷屏消息？

在 Telegram 群组中，机器人经常会遇到重复刷屏、改符号刷屏、批量复制广告等问题。传统的关键词黑名单只能拦截固定词汇，却很难识别“免费资源点击领取”和“免费资源，点击领取！”这类高度相似但并不完全相同的消息。

Telegram关键词搜索Bot 更稳定的做法，是在机器人内部加入文本相似度计算，先对消息进行标准化，再结合哈希、字符 n-gram、Jaccard 相似度或余弦相似度，判断消息是否属于重复内容。本文将从算法选择、Telegram 消息结构、阈值设计和误判控制四个方面，讲清楚如何搭建一套可落地的过滤方案。

🧭 一、为什么关键词过滤无法解决重复刷屏

关键词过滤依赖预先维护的词表，只能判断消息中是否出现某些词。当发送者不断加入空格、表情、标点或随机数字时，原有规则就容易失效。

文本相似度的核心不是寻找某一个词，而是比较两段文本的整体结构。即使消息被修改了标点、大小写或少量词语，只要主要内容仍然接近，系统就可以将其判定为疑似重复。

1. 先区分三种重复类型

完全重复是指文本内容完全一致，适合使用 SHA-256 等哈希快速判断；格式重复是指仅改变大小写、空格、表情或标点，适合在标准化后比较。

Telegram关键词搜索Bot 近似重复则是发送者替换少量词语、增加推广链接或调整句子顺序。此时需要使用字符 n-gram、Jaccard 或余弦相似度，而不能只依赖字符串相等。

🧹 二、第一步：对 Telegram 消息进行标准化

Telegram关键词搜索Bot 相似度计算之前，必须先清洗输入内容。Telegram 消息可能来自 message.text，也可能来自图片、文件或视频的 caption，因此机器人需要统一读取文本字段。

标准化通常包括转换大小写、统一空白字符、移除多余标点、替换链接和清除部分表情。链接建议替换成统一的占位符，而不是直接删除，否则两个本来不同的推广链接可能被错误视为相同。

import re
import hashlib

def normalize_text(text: str) -> str:
    text = (text or "").casefold()

    # 将不同链接统一为占位符，避免链接参数制造大量噪声
    text = re.sub(r"https?://\S+|t\.me/\S+", " URL ", text)

    # 保留中文、字母和数字，压缩空白字符
    text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", text, flags=re.UNICODE)
    text = re.sub(r"\s+", " ", text).strip()

    return text

def text_hash(text: str) -> str:
    normalized = normalize_text(text)
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

需要注意，清洗规则不能过度激进。例如完全移除数字，可能会让“课程 2024”和“课程 2025”被误判为相同；删除所有链接，也可能掩盖真正需要拦截的恶意域名。

2. 短文本要采用更谨慎的规则

“好的”“收到”“哈哈”等短消息天然具有高重复率，如果直接使用高相似度阈值，极易误删正常对话。建议对长度较短的消息只执行完全重复检测，不要贸然使用近似匹配。

📐 三、第二步：选择适合中文的相似度算法

1. 字符 n-gram 与 Jaccard 相似度

中文通常不使用空格分词，因此字符 n-gram 比简单按空格切词更稳。例如“Telegram 机器人过滤刷屏”可以拆成连续的二字或三字片段，再比较两条消息的片段集合。

Jaccard 相似度的计算方式，是用两个集合的交集数量除以并集数量。数值越接近 1，代表两段文本共享的片段越多，越可能属于重复消息。

def shingles(text: str, size: int = 2) -> set:
    compact = text.replace(" ", "")
    if len(compact) < size:
        return {compact} if compact else set()
    return {compact[i:i + size] for i in range(len(compact) - size + 1)}

def jaccard_similarity(a: str, b: str) -> float:
    set_a = shingles(normalize_text(a))
    set_b = shingles(normalize_text(b))

    if not set_a or not set_b:
        return 0.0

    return len(set_a & set_b) / len(set_a | set_b)

2. 余弦相似度与 SimHash

余弦相似度适合比较词频向量，消息数量较大时可以配合 TF-IDF 使用；SimHash 则能将相似文本映射为接近的指纹，适合在 Redis 或数据库中进行快速近邻查询。

如果群组规模不大，推荐使用“标准化哈希加字符 Jaccard”的组合方案，部署成本低、解释性强。只有当机器人每天处理数十万条以上消息，才有必要引入 SimHash、向量数据库或专门的语义模型。

3. 推荐的初始阈值

阈值不能直接照搬到所有群组，应当先收集一批真实消息进行测试。下面的参数可以作为起点，再根据误删率和漏放率持续调整。

MIN_SIMILAR_LENGTH = 12       # 少于此长度，仅判断完全重复
JACCARD_DELETE_THRESHOLD = 0.86
JACCARD_REVIEW_THRESHOLD = 0.72
DUPLICATE_WINDOW_SECONDS = 60
MAX_COMPARE_MESSAGES = 30
SAME_USER_REPEAT_LIMIT = 3

Telegram关键词搜索Bot 通常可以将 0.86 以上作为自动删除或折叠候选，将 0.72 至 0.86 作为限流、提醒或进入审核队列。对于管理员、认证账号和机器人发布的固定公告，应当设置白名单，避免正常通知被系统拦截。

🤖 四、第三步：把相似度检测接入 Telegram Bot

机器人收到消息后，建议按照“解析字段、去重、标准化、快速哈希、窗口比对、执行动作”的顺序处理。文本应该按群组 ID、发送者 ID 和时间窗口进行隔离，不能把所有群组的内容放在同一个比较池中。

消息去重还要区分 message_id 和文本指纹。message_id 用于避免重复处理同一条 Telegram 更新，文本指纹用于发现不同 message_id 之间的重复刷屏。

async def handle_message(message, redis):
    chat_id = message.chat.id
    user_id = message.from_user.id if message.from_user else 0
    raw_text = message.text or message.caption or ""

    if not raw_text.strip():
        return

    normalized = normalize_text(raw_text)
    if len(normalized.replace(" ", "")) < MIN_SIMILAR_LENGTH:
        return

    digest = text_hash(raw_text)
    exact_key = f"tg:exact:{chat_id}:{digest}"

    # 完全重复：快速处理，不必再次计算相似度
    if await redis.exists(exact_key):
        await safe_delete(message)
        return

    await redis.setex(exact_key, DUPLICATE_WINDOW_SECONDS, "1")

    recent_key = f"tg:recent:{chat_id}"
    recent_messages = await redis.lrange(recent_key, 0, MAX_COMPARE_MESSAGES - 1)

    for old_text in recent_messages:
        score = jaccard_similarity(normalized, old_text.decode())
        if score >= JACCARD_DELETE_THRESHOLD:
            await safe_delete(message)
            return

    await redis.lpush(recent_key, normalized)
    await redis.ltrim(recent_key, 0, MAX_COMPARE_MESSAGES - 1)
    await redis.expire(recent_key, DUPLICATE_WINDOW_SECONDS)

生产环境中还应当处理 edited_message，因为用户可能先发送正常内容，再编辑成广告。对于图片或文件消息，则需要读取 caption；如果只读取 message.text，机器人会漏掉大量带媒体的刷屏内容。

权限与 Telegram 机制

机器人只有在获得相应的群组管理权限后，才能删除其他用户的消息。实际接入前，应确认机器人拥有删除消息权限，并遵守 Telegram Bot API 对消息类型、删除时限和群组权限的限制。

如果群组启用了 Privacy Mode，机器人可能无法接收全部普通消息。因此，想要实现全量刷屏检测，需要根据使用场景调整隐私模式，并在上线前验证机器人实际收到的 Update 类型。

🛡️ 五、如何降低误判并提升长期效果

1. 不要只看相似度分数

相似度高不一定代表恶意刷屏。例如管理员可能连续发布相同格式的活动公告，客服也可能重复发送标准指引。因此，系统最好同时结合发送次数、发送间隔、账号角色和消息类型。

可以将连续重复三次以上、间隔少于一分钟、且来自普通成员的消息判定为高风险。对于只出现一次的相似内容，优先选择静默限流或提醒，而不是立即删除。

2. 建立可回溯的审核日志

每次执行删除、警告或放行，都应该记录群组 ID、用户 ID、相似度、规则版本和处理结果。日志不必长期保存完整原文，可以保存截断文本或不可逆哈希，以减少隐私风险。

运营者还应定期抽查误删样本，计算精确率、召回率和误删率。如果误删率明显上升，应优先收紧自动删除范围，而不是盲目降低阈值。

Telegram关键词搜索Bot 3. 语义模型不是越复杂越好

向量嵌入可以识别“领取福利”和“免费拿资源”这类词语不同但含义接近的消息，不过它需要更多计算资源，也可能把主题相近但用途不同的正常对话误判为重复。

较稳妥的架构是先用哈希和 n-gram 完成低成本初筛，只有进入灰度区间的消息才调用语义模型。这样既控制成本，也方便解释机器人为什么采取某个动作。

❓ 常见问题解答（FAQ）

Q1：文本相似度越高，就一定应该删除吗？

不一定。相似度只能说明内容接近，不能直接证明消息具有恶意，建议叠加发送频率、用户权限、时间窗口和群组白名单，再决定删除、限流还是提醒。

Q2：中文消息应该使用分词还是字符切片？

处理中短中文广告时，字符二元或三元切片通常更容易部署，也不依赖分词词典。对于长篇内容或需要语义判断的场景，可以再加入分词、TF-IDF 或向量模型。

Q3：Redis 在这个方案中主要负责什么？

Redis 适合保存短时间窗口内的消息指纹、最近文本和计数器，并通过 TTL 自动过期。这样既能快速判断重复，也不会让历史数据无限增长。

Q4：机器人收不到群里的普通消息怎么办？

首先检查 Privacy Mode、机器人是否真正加入目标群组，以及 Update 处理逻辑是否覆盖 message、edited_message 和带 caption 的媒体消息。确认能够稳定接收数据后，再排查相似度算法本身。

总体而言，Telegram 重复刷屏过滤不应依赖单一规则。通过消息标准化、哈希去重、相似度分层、时间窗口控制和人工复核组成多层防线，才能在拦截广告的同时保护正常交流，并让机器人系统具备更好的稳定性、可解释性与维护效率。