用 TF-IDF 提取文档关键词

功能说明

这段代码为一个小型文档集合计算 TF-IDF 分数，输出最具区分度的关键词。它适合给文章自动打标签、生成摘要提示或在站内搜索中做轻量关键词提取，不依赖任何第三方 NLP 库。

词频衡量词语在单篇文档中的重要性，逆文档频率则压低“每篇都出现”的常见词。对多篇文档的分数求和后，既能突出跨文档反复出现的主题词，也能保留只在单篇中出现的特征词。

关键代码

tokenize 用正则提取以小写字母开头、长度至少为 3 的 token，并过滤停用词。正则允许 +、#、.、-，因此 c++、c#、node.js 这类技术词不会被拆碎。

keyword_scores 先用集合统计每个词出现在多少篇文档中，再在每篇文档内计算归一化词频 count / length，套用平滑公式 log((1 + N) / (1 + df)) + 1，最后按分数降序、词名升序排序并截取前 N 个。

使用注意

IDF 在小语料上波动很大：只有两三篇文档时，一个词的分数可能被单篇内容主导。正则目前只处理 ASCII 英文，中文或带重音的文本需要先分词或改写正则。代码没有做词形还原，cache 和 caching 会被视为不同词；也没有处理短语。

分数是跨文档累加值，不是单篇文档内的 TF-IDF；如果要做“每篇文档各自的关键词”，应分别返回每篇的排序结果。生产环境还应限制文档长度和词表大小，避免内存膨胀。

验证方式

运行脚本会打印前四个词的分数，database 与 latency 分数相同，按字母序 database 排在前面，因此断言 keyword_scores(documents, 1)[0][0] == "database" 成立。可以再加入一篇完全不相关的文档，观察专有词分数上升、通用词分数下降。

代码示例

from __future__ import annotations

import math
import re
from collections import Counter

TOKEN_RE = re.compile(r"[a-z][a-z0-9+#.-]{1,}")
STOPWORDS = {
    "the", "and", "for", "with", "that", "this",
    "from", "into", "your", "are", "was", "have",
}


def tokenize(text: str) -> list[str]:
    return [
        token
        for token in TOKEN_RE.findall(text.lower())
        if token not in STOPWORDS and len(token) > 2
    ]


def keyword_scores(documents: list[str], top_n: int = 5) -> list[tuple[str, float]]:
    tokenized = [tokenize(document) for document in documents]
    document_frequency: Counter[str] = Counter()
    for tokens in tokenized:
        document_frequency.update(set(tokens))
    total_documents = len(tokenized)
    scores: Counter[str] = Counter()
    for tokens in tokenized:
        counts = Counter(tokens)
        length = sum(counts.values()) or 1
        for term, count in counts.items():
            tf = count / length
            idf = math.log((1 + total_documents) / (1 + document_frequency[term])) + 1
            scores[term] += tf * idf
    return sorted(scores.items(), key=lambda item: (-item[1], item[0]))[:top_n]


def main() -> None:
    documents = [
        "Use connection pooling to reduce database latency and reuse open sockets.",
        "A connection pool keeps sockets warm so database queries avoid handshake latency.",
        "Caching responses can hide database latency but risks stale data.",
    ]
    for term, score in keyword_scores(documents, 4):
        print(f"{term:12s} {score:.4f}")
    assert keyword_scores(documents, 1)[0][0] == "database"


if __name__ == "__main__":
    main()