03 | 完成节点1 — 抽取关键词

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

这是一篇系列文，请按照顺序阅读。

本文实现流程第一个节点，抽取关键词。

目标

从用户输入的自然语言查询中提取出有价值的关键词，作为后续向量检索召回字段、指标、枚举值的输入。

例如输入 "各性别销售额分布"，应提取出：性别、销售额、分布、各性别销售额分布。

思路

核心思路是 使用 jieba 分词 + 词性过滤，而不是直接调用 LLM：

1. 为什么不用 LLM？ 获取关键词的需求是高频操作，LLM 调用有延迟和成本，而 jieba 分词基本是 O(n) 级别，毫秒级完成
2. 为什么按词性过滤？ 用户查询中真正有意义的是名词（实体名、指标名）、动词、英文等，助词（"的""了""在"）、代词、标点等要剔除
3. 为什么保留原始 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: strkeywords: list[str]error: str | None# 1. 从用户自然语言中提取关键词async def extract_keywords(state: State, runtime: Runtime[RuntimeContext]) -> State:import jiebaimport jieba.analysepush_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"},)

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

科普：中文分词

什么是分词？

计算机处理中文的第一步，就是把连续的汉字序列切成有意义的词语。

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

英文天然以空格分隔单词（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 jiebaimport jieba.posseg as pseg# 基础分词print(list(jieba.cut("统计华北地区销售额")))# → ['统计', '华北', '地区', '销售额']# 带词性分词for word, flag in pseg.cut("统计华北地区销售额"):print(f"{word}({flag})")# → 统计(v) 华北(ns) 地区(n) 销售额(n)

核心概念

三种分词模式

import jiebatext = "我来到北京清华大学"# 精确模式（默认）：最精确的切分，适合文本分析jieba.cut(text, cut_all=False)# → ['我', '来到', '北京', '清华大学']# 全模式：把所有可能的词都扫出来，速度快但有冗余jieba.cut(text, cut_all=True)# → ['我', '来到', '北京', '清华', '清华大学', '华大', '大学']# 搜索引擎模式：在精确模式基础上对长词再切分，提高召回率jieba.cut_for_search(text)# → ['我', '来到', '北京', '清华', '华大', '大学', '清华大学']

我们项目用的是精确模式，因为你只需要最准确的词，不需要冗余。

我们用的三个关键 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 psegfor 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 jiebaimport jieba.analyse# ❌ 错误：没有先 import jieba 就直接 import jieba.analyse

因为 jieba.analyse 是在 jieba 模块初始化后动态挂载的子模块。

进阶路线

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