昨晚十一点多,我在公司楼下抽烟(别学),组里的小李突然在群里问:“哥,产品让我‘从一段文本里自动抠关键字’,要能解释的那种……” 我当时也卡了一下,因为真要讲清楚,不能就一句“用个库完事”。
后来我边走边想,顺手把常用的四种路子捋了一遍。这几种方法都能在“单段文本”这个场景里跑起来,而且各有各的脾气。
先说个前提,不管你用啥算法,第一步都逃不掉:清洗 + 分词。英文就切词,中文就得想办法切。下面的代码写成“有 jieba 就用,没有就退化到正则切”,确保你粘过去能跑起来。
import re
from collections import Counter, defaultdict
from math import log
STOPWORDS_ZH = set("""
的 了 和 是 就 都 而 及 与 着 或 一个 没有 我 你 他 她 它 我们 你们 他们
以及 这个 那个 这些 那些 因为 所以 然后 如果 但是 还是 可能 可以 进行
""".split())
def tokenize(text: str):
text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", text.lower())
# 尝试中文分词
try:
import jieba
words = [w.strip() for w in jieba.lcut(text) if w.strip()]
except Exception:
# 没装jieba就凑合:按空格/下划线切(更偏英文)
words = [w for w in re.split(r"[\s_]+", text) if w]
# 通用过滤:停用词、太短、纯数字
cleaned = []
for w in words:
if w in STOPWORDS_ZH:
continue
if len(w) <= 1:
continue
if w.isdigit():
continue
cleaned.append(w)
return cleaned
1. 词频统计法
第一种最朴素,就是词频法。你别笑,线上很多需求就吃这套,尤其是“文本比较短、关键词本来就重复出现”的情况。缺点也明显:容易把“常见但没意义的高频词”顶上来,所以停用词列表得认真准备。
def keywords_by_freq(text: str, topk: int = 10):
words = tokenize(text)
cnt = Counter(words)
return [w for w, _ in cnt.most_common(topk)]
我当时就跟小李说,这玩意像你去便利店买水,最便宜但也最不挑人。如果你是运营同学,随手贴一段活动文案,让你“给我十个词”,词频法往往已经够交差了。
2. TF-IDF 法
有人会问:单篇文本哪来的 IDF?对,这就是坑点。我一般的土办法是把一段文本拆成很多“伪文档”,比如按句子、按段落,甚至按滑动窗口。这样 IDF 至少能体现“哪些词只在少数句子里出现”,就不会被通篇反复出现的口水词压死。
def split_sentences(text: str):
# 够用了,别太较真
parts = re.split(r"[。!?!?;;\n]+", text)
return [p.strip() for p in parts if p.strip()]
def keywords_by_tfidf_single(text: str, topk: int = 10):
sents = split_sentences(text)
docs = [tokenize(s) for s in sents]
N = len(docs) if docs else 1
df = Counter()
for d in docs:
df.update(set(d))
tfidf = defaultdict(float)
for d in docs:
tf = Counter(d)
for w, c in tf.items():
# tf用对数平滑一下,不然长句子太吃亏
tf_w = 1.0 + log(1 + c)
idf_w = log((N + 1) / (df[w] + 1)) + 1.0
tfidf[w] += tf_w * idf_w
return [w for w, _ in sorted(tfidf.items(), key=lambda x: x[1], reverse=True)[:topk]]
这个有点像你在会议里听人扯半天,突然有一句“关键决策点”只出现一次,但特别刺眼——TF-IDF 就是在抓这种“稀有但重要”的感觉。缺点呢,也有:如果文本特别短,拆出来句子也没几条,IDF 就不太稳。
这个我真用得多,原因很现实:不用训练、不用外部词库,效果还挺“像那么回事”。它的思路粗暴点说,就是把词当点,词跟词在窗口里共现就连边,谁跟谁关系多、关系强,谁就更重要。
def keywords_by_textrank(text: str, topk: int = 10, window: int = 4, iters: int = 20, d: float = 0.85):
words = tokenize(text)
if not words:
return []
# 建图:共现窗口
graph = defaultdict(set)
for i, w in enumerate(words):
for j in range(i + 1, min(i + window, len(words))):
u = words[j]
if w == u:
continue
graph[w].add(u)
graph[u].add(w)
# 初始化分数
score = {w: 1.0 for w in graph}
for _ in range(iters):
new_score = {}
for w in graph:
s = 1 - d
for v in graph[w]:
s += d * (score[v] / max(len(graph[v]), 1))
new_score[w] = s
score = new_score
return [w for w, _ in sorted(score.items(), key=lambda x: x[1], reverse=True)[:topk]]
TextRank 特别适合那种“技术排障复盘、一段比较长的叙述”,它能把“核心对象、核心动作”捞上来。但它也会犯轴:如果分词效果差,构建的图就烂,最后出来的词可能就不准确。
4. RAKE 算法
第四个是 RAKE(Rapid Automatic Keyword Extraction),这玩意很像“把停用词当分隔符,把剩下的词拼成短语,然后用词的共现度给短语打分”。它的优势是:不仅能出单词,还能出“词组/短语”,比如“服务端超时”“连接池配置”这种,比单个词更像人写的关键字。
def keywords_by_rake(text: str, topk: int = 10):
words = tokenize(text)
if not words:
return []
# 用停用词切短语:这里简单点,把停用词当断点(tokenize里已经过滤了大部分)
# 所以我们用“原始粗切”再断一次会更像RAKE,这里做个轻量版:按标点/换行切片再分词
chunks = re.split(r"[。!?!?;;\n,,]+", text)
phrases = []
for ch in chunks:
toks = tokenize(ch)
if toks:
phrases.append(toks)
# 词频 & 共现度(词所在短语长度的累计)
freq = Counter()
degree = Counter()
for p in phrases:
unique = p # 不去重也行,短语里重复词本来就少
L = len(unique)
for w in unique:
freq[w] += 1
degree[w] += (L - 1)
# word score = (degree + freq) / freq
wscore = {}
for w in freq:
wscore[w] = (degree[w] + freq[w]) / freq[w]
# phrase score = sum(word score)
pscore = []
for p in phrases:
s = sum(wscore.get(w, 0.0) for w in p)
pscore.append(("".join(p), s)) # 中文直接拼一起更像词组
seen = set()
out = []
for ph, _ in sorted(pscore, key=lambda x: x[1], reverse=True):
if ph in seen:
continue
seen.add(ph)
out.append(ph)
if len(out) >= topk:
break
return out
方法对比与实战选择
这四种方法,感觉就像你做饭:词频是泡面,TF-IDF 是加了点料的泡面,TextRank 是你开始研究火候了,RAKE 是你都开始摆盘了。
真到项目里怎么选?我自己的习惯是这样:
- 文本很短:优先词频法或 RAKE。
- 文本中等偏长:用 TextRank 或 TF-IDF。
- 想要“短语”:优先 RAKE。
- 想要“全局核心词”:优先 TextRank。
最后,我给小李写了个“统一入口”函数,他复制过去就能跑,省得一个个函数试。
def extract_keywords(text: str, topk: int = 10):
return {
"freq": keywords_by_freq(text, topk),
"tfidf": keywords_by_tfidf_single(text, topk),
"textrank": keywords_by_textrank(text, topk),
"rake": keywords_by_rake(text, topk),
}
if __name__ == "__main__":
demo = "昨晚线上接口超时,查到是连接池默认配置太小,线程堆积,最后还触发了重试风暴。"
print(extract_keywords(demo, topk=6))
运行上面的示例,你可以直观地看到四种算法从同一段故障描述中提取出的不同关键词,这对于理解它们的侧重点很有帮助。
归根结底,没有一种方法能通吃所有场景。关键是根据你的文本特点(长度、领域、格式)和业务需求(要单词还是短语,要全面还是精准)来灵活选择和组合。在云栈社区里,也有很多开发者分享过他们在不同业务中使用这些Python文本处理技巧的心得,多看看实战案例能帮你更快做出选择。
行了,思路和代码都在这儿了。如果你手头有实际的文本,不妨把这四种方法都跑一遍,看看哪个出来的结果更贴近你的“人话”直觉。实践永远是检验算法的唯一标准。