03 | 实现节点1 — 抽取关键词

本文属于系列教程，建议按顺序阅读。但本小节内容相对独立，将详细讲解一个关键步骤——如何从用户的自然语言查询中高效提取关键词，这是构建中文搜索引擎的重要环节。

本文实现流程的第一个节点：抽取关键词。这步看似简单，却直接影响后续向量检索能否准确命中相关字段与指标。

目标

本节点的目标是从用户输入的自然语言查询中提取出有价值的关键词，为后续向量检索提供字段、指标、枚举值的召回依据。例如，输入查询 "各性别销售额分布"，应提取出 性别、销售额、分布 等关键词，并将完整查询 各性别销售额分布 作为兜底输入。

思路

核心思路非常明确：采用 jieba 分词 + 词性过滤 方案，而非直接调用 LLM。为什么选择这种方式？

• 为什么不用 LLM？ 关键词提取是高频操作，LLM 调用存在延迟和成本问题，而 jieba 分词通常为 O(n) 级别，毫秒级完成，性价比极高。
• 为什么按词性过滤？ 用户查询中真正有价值的是名词（实体名、指标名）、动词、英文等，而助词（"的""了""在"）、代词、标点等应剔除，否则会干扰向量检索结果。
• 为什么保留原始 query？ 部分查询本身就是复合词（如"各性别销售额分布"），将其作为兜底关键词传递给向量检索，可确保信息不丢失。即使分词效果不佳，完整 query 也能起到兜底作用。

实现步骤

首先安装 jieba 分词库：

uv add jieba

整个节点逻辑可拆解为三步：获取 query → 定义词性白名单并调用 jieba 提取关键词 → 清洗后返回结果。下面逐一详解。

第一步：取 query + 判空

从 State 中取出用户输入。若为空，则直接返回空列表，不进入后续处理。这是常见的防御性编程技巧，但至关重要——许多线上问题都源于未处理空输入，进而导致后续流程崩溃。

第二步：定义词性白名单 + 调 jieba 提取 + 清洗

这三件事本质上是一条流水线，拆开讲太碎，合在一起说明白。

2a. 定义允许的词性集合

jieba 的 analyse.extract_tags() 支持 allowPOS 参数按词性过滤。我们保留以下词性：

剔除的词性：uj（"的"）、ul（"了"）、p（介词）、r（代词）、w（标点）、x（非语素）等——这些对下游检索无贡献，反而会引入噪声。

2b. 调用 jieba 提取关键词

jieba.analyse.extract_tags() 内部做了两件事：

1. 分词——将连续文本切分为词语序列
2. TF-IDF 权重排序——保留权重高的前 N 个词（默认 20 个），按重要性从高到低排列

以 "各性别销售额分布" 为例，分词结果为：各 / 性别 / 销售额 / 分布。剔除助词"各"（不在 allow_pos 中）后保留 性别、销售额、分布。

2c. 去重 + 追加原始 query

• 如果 extract_tags 返回的词与 query 完全相同（单关键词场景），先移除避免冗余
• 再把原始 query 追加到列表末尾，作为兜底词——即使分词效果不好，完整 query 也能作为检索输入

第三步：返回结果

返回的 {"keywords": keywords} 只更新 State 中的 keywords 字段，query、error 等其他字段保持不变。后续节点通过 state["keywords"] 即可拿到关键词列表。

以下是 app/agent/graph.py 中 extract_keywords 节点的完整代码。

# graph.py 修改的部分
# State 定义图中各节点间流转的共享状态
class State(TypedDict):
    query: str
    keywords: list[str]
    error: str | None

# 1. 从用户自然语言中提取关键词
async def extract_keywords(state: State, runtime: Runtime[RuntimeContext]) -> State:
    import jieba
    import jieba.analyse

    push_progress(runtime, STEP_NAMES["extract_keywords"], "running")

    # 第一步：取 query，判空
    query = state["query"]
    if not query:
        push_progress(runtime, STEP_NAMES["extract_keywords"], "error")
        return {"keywords": []}

    # 第二步：定义词性白名单
    allow_pos = (
        "n",  # 名词: 数据、服务器、表格
        "nr", # 人名: 张三、李四
        "ns", # 地名: 北京、上海
        "nt", # 机构团体名: 政府、学校、某公司
        "nz", # 其他专有名词: Unicode、哈希算法、诺贝尔奖
        "v",  # 动词: 运行、开发
        "vn", # 名动词: 工作、研究
        "a",  # 形容词: 美丽、快速
        "an", # 名形词: 难度、合法性、复杂度
        "eng", # 英文
        "i",  # 成语
        "l",  # 常用固定短语
    )

    # 调 jieba 提取 + 去重追加
    keywords = jieba.analyse.extract_tags(query, allowPOS=allow_pos)
    keywords = [k for k in keywords if k != query]
    keywords.append(query)

    # TODO 仅仅为了测试，后期删掉
    push_progress(runtime, f"关键词：{keywords}", "running")
    push_progress(runtime, STEP_NAMES["extract_keywords"], "success")

    # 第三步：返回结果
    return {"keywords": keywords}

接入接口：让 query 从用户输入流入 graph

节点本身写完了，但 graph 的 initial_state 里 query 还是硬编码的。需要让用户发来的问题真正流入流程图。

修改 main.py 的 /api/query 接口，从 payload 中取出 query 并传入 graph：

@app.post("/api/query")
async def query(payload: dict):
    """自然语言查询入口，以 SSE 流式返回处理进度和最终结果"""
    query_text = payload.get("query", "")
    return StreamingResponse(
        sse_stream(query_text),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
    )

现在刷新下页面，比如输入 "各性别销售额分布"，就能在控制台看到输出 ['性别', '销售额', '分布', '各性别销售额分布']。效果立竿见影。

科普：中文分词

什么是分词？

中文分词是计算机处理中文的第一步，即将连续的汉字序列切分为有意义的词语。这是自然语言处理（NLP）的基础环节。

输入：北京市海淀区中关村大街
输出：北京 / 市 / 海淀 / 区 / 中关村 / 大街

英文天然以空格分隔单词（如 Beijing Haidian District），无需分词。中文没有空格，因此分词成为中文 NLP 的地基——分错了，后续所有步骤（检索、SQL 生成）都会出错。这一点怎么强调都不过分。

核心难点

三大方法流派

进阶路线

1. 理解核心概念：分词、词性标注（POS）、未登录词（OOV）识别
2. 选一个工具上手：
   • 入门：jieba（几行代码搞定，本项目选择它）
   • 进阶：pkuseg（北大出品，领域自适应更好）
   • 深度：LAC（百度词法分析，精度高）
3. 调优方向：加载自定义词典 → 调整权重 → 切换模型
4. 参考资料：
   • jieba 官方：github.com/fxsjy/jieba
   • pkuseg：github.com/lancopku/pk…
   • 百度 LAC：github.com/baidu/lac

科普：jieba 分词

jieba 是什么？

jieba 是目前 Python 中文分词领域使用最广的开源库（31k+ Star），由百度工程师 fxsjy 开发。名字取自"结巴"的拼音——因为分词就是把"结结巴巴"的一句话切开。这个命名很有意思，也体现了开发者的幽默感。

一句话理解：你扔进去一段中文，它吐出来一串词语，附带每个词的词性。

import jieba
import jieba.posseg as pseg

# 基础分词
print(list(jieba.cut("统计华北地区销售额")))
# → ['统计', '华北', '地区', '销售额']

# 带词性分词
for word, flag in pseg.cut("统计华北地区销售额"):
    print(f"{word}({flag})")
# → 统计(v) 华北(ns) 地区(n) 销售额(n)

核心概念

三种分词模式

import jieba
text = "我来到北京清华大学"

# 精确模式（默认）：最精确的切分，适合文本分析
jieba.cut(text, cut_all=False)
# → ['我', '来到', '北京', '清华大学']

# 全模式：把所有可能的词都扫出来，速度快但有冗余
jieba.cut(text, cut_all=True)
# → ['我', '来到', '北京', '清华', '清华大学', '华大', '大学']

# 搜索引擎模式：在精确模式基础上对长词再切分，提高召回率
jieba.cut_for_search(text)
# → ['我', '来到', '北京', '清华', '华大', '大学', '清华大学']

我们项目使用的是精确模式，因为只需要最准确的词，不需要冗余。全模式可能会引入大量噪声，搜索引擎模式适合检索场景，但我们的关键词提取已经有 TF-IDF 排序，不需要额外切分。

我们用的三个关键 API

import jieba.analyse

# 1. extract_tags：提取关键词（TF-IDF 权重排序）
keywords = jieba.analyse.extract_tags("统计华北地区的销售额", topK=5)
# → ['销售额', '华北地区', '统计']

# 2. allowPOS 参数：按词性过滤
keywords = jieba.analyse.extract_tags("统计华北地区的销售额",
                                       allowPOS=('n', 'ns', 'vn'))
# → ['销售额', '华北地区']

# 3. posseg：获取每个词的词性
import jieba.posseg as pseg
for w, flag in pseg.cut("销售额环比增长20%"):
    print(f"{w}/{flag}")
# 销售额/n, 环比/d, 增长/v, 20%/x

关于 jieba 的一个大坑：import 顺序

这个坑我踩过不止一次，值得单独提出来：import jieba.analyse 必须写在 import jieba 之后，否则 jieba.analyse.extract_tags 会报 AttributeError。

# ✅ 正确
import jieba
import jieba.analyse

# ❌ 错误：没有先 import jieba 就直接 import jieba.analyse

原因很简单：jieba.analyse 是在 jieba 模块初始化后动态挂载的子模块。如果没先 import 主模块， Python 找不到这个子模块。很多新手会在这里卡住，记住就行了。

进阶路线

Level 1: 基础分词
└─ jieba.cut() / jieba.lcut() 了解精确模式、全模式、搜索引擎模式的区别
Level 2: 词性标注
└─ jieba.posseg.cut() 认识 n(名词)、v(动词)、ns(地名) 等词性标记
Level 3: 关键词提取（本项目级别）
├─ jieba.analyse.extract_tags() — TF-IDF 算法
└─ jieba.analyse.textrank() — TextRank 算法 用 allowPOS 按词性过滤
Level 4: 自定义优化
├─ jieba.add_word() / jieba.load_userdict()
│  如把 "转化率" 加入词典，避免被切成 "转化/率"
├─ jieba.suggest_freq()
│  调整词频让某些切分更优先
└─ jieba.set_dictionary() 更换更大的词典文件
Level 5: 进阶换装
├─ jieba_fast：C++ 重写，分词速度 2-3 倍提升
├─ pkuseg：北大出品，领域自适应更强
└─ LAC / HanLP：深度学习方案，精度天花板

参考资料

• jieba 官方文档：github.com/fxsjy/jieba
• jieba 词性标记表：github.com/fxsjy/jieba…
• pkuseg（进阶替代）：github.com/lancopku/pk…
• 百度 LAC（深度学习方案）：github.com/baidu/lac