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

本文将深入探讨流程中第一个节点的实现方法，即如何从自然语言查询中高效抽取关键词。

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

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

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

目标

本节点的核心目标是从用户输入的自然语言查询中，精准提取出对后续任务有价值的关键词。这些关键词将作为向量检索召回字段、指标以及枚举值的输入依据。例如，当用户输入“各性别销售额分布”时，系统应能提取出“性别”、“销售额”、“分布”以及完整的“各性别销售额分布”等关键词。

思路

本节点采用的核心策略是使用 jieba 分词工具并结合词性过滤，而非直接调用大型语言模型（LLM）。具体原因如下：

1. 为什么不用 LLM？ 获取关键词属于高频操作，LLM 调用存在延迟和成本问题。相比之下，jieba 分词的算法复杂度为 O(n)，能够在毫秒级完成处理，更为高效。
2. 为什么按词性过滤？ 用户查询中真正具有语义价值的部分通常是名词（如实体名、指标名）、动词和英文单词。而像“的”、“了”、“在”等助词、代词以及标点符号，则对下游任务贡献不大，需要剔除。
3. 为什么保留原始 query？ 部分查询本身即为复合词，例如“各性别销售额分布”。将完整的原始查询作为兜底关键词传递给向量检索，可以确保信息不丢失，提高召回率。

实现步骤

首先，需要安装 jieba 库：

uv add jieba

整个节点的逻辑可以清晰拆解为三个步骤：获取查询并判空、定义词性白名单并调用 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 权重排序：根据 TF-IDF 算法计算每个词的权重，并保留权重最高的前 N 个词（默认 20 个），按重要性从高到低排列。

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

2c. 去重 + 追加原始 query

1. 如果 extract_tags 返回的词与原始 query 完全相同（即单关键词场景），则先将其移除，以避免冗余。
2. 随后，将原始 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”，因此无需分词。而中文没有空格，分词因此成为中文自然语言处理的基础环节。一旦分词错误，后续的检索、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：深度学习方案，精度天花板

参考资料

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

综上所述，通过结合 jieba 分词与词性过滤的策略，我们能够高效、低成本地从自然语言查询中提取出关键信息，为后续的向量检索和数据分析任务奠定坚实基础。