使用PostgreSQL进行全文检索

一、概述

• 全文检索（Full-Text Search, FTS）是一种基于语义而非字面匹配的文本搜索技术。它能够对文档中的词语进行分词、标准化（如转为小写、词干提取）、去除停用词等处理，从而支持更智能、更高效的关键词查询。
• 常用的全文检索方案是使用ES（Elastic Search），但因为个人服务器资源有限 + 业务很小所以直接用PostgreSQL一把梭了，最后实现效果也挺不错。
• 在 PostgreSQL 中，全文搜索（Full-Text Search, FTS）是通过内置的文本搜索功能实现的，并不需要传统意义上的“全文索引”（如 MySQL 的 FULLTEXT INDEX），而是使用GIN（Generalized Inverted Index）或GiST（Generalized Search Tree）索引来加速全文搜索查询。

二、分词器选择

常用的中文分词器有zhparser和py_jieba，GPT推荐的zh_parser所以就以zhparser为准，反正大差不差

三、安装pg的分词器插件

因为我是用docker来安装的数据库（，所以下列步骤基于PostgreSQL17:latest docker镜像，具体版本号为：

docker run --rm postgres:latest postgres --version
# postgres (PostgreSQL) 17.6 (Debian 17.6-1.pgdg13+1)

1. 更新依赖

apt update && apt install -y git postgresql-server-dev-17 wget gcc make

2. 获取scws后解压并编译

wget http://www.xunsearch.com/scws/down/scws-1.2.3.tar.bz2
tar -jxvf scws-1.2.3.tar.bz2
cd scws-1.2.3
./configure && make && make install

3. 获取zhparser

git clone https://github.com/amutu/zhparser.git
cd zhparser
make && make install

4. 切换到数据库客户端添加依赖

-- 0. 添加依赖及相关配置
CREATE EXTENSION IF NOT EXISTS zhparser;
CREATE TEXT SEARCH CONFIGURATION zh_cfg (PARSER = zhparser);
ALTER TEXT SEARCH CONFIGURATION zh_cfg ADD MAPPING FOR n,v,a,i,e,l WITH simple; 
-- 1. 表中添加分词列（加快检索速度
ALTER TABLE blog_detail ADD COLUMN tsv TSVECTOR;
-- 2. 添加触发器，在数据更新时自动更新tsv（添加权重信息）
CREATE OR REPLACE FUNCTION blog_tsv_update() RETURNS trigger AS $$
BEGIN
  NEW.tsv := setweight(to_tsvector('zh_cfg', coalesce(NEW.title, '')), 'A') ||
      setweight(to_tsvector('zh_cfg', coalesce(NEW.blog_content, '')), 'B');
  RETURN NEW;
END $$ LANGUAGE plpgsql;

CREATE TRIGGER trg_blog_tsv BEFORE INSERT OR UPDATE OF title, blog_content
ON blog_detail FOR EACH ROW EXECUTE FUNCTION blog_tsv_update();
-- 3. 创建简单 GIN 索引
CREATE INDEX idx_blog_tsv ON blog_detail USING GIN(tsv);
-- 4. 更新历史数据的tsv
UPDATE blog_detail SET tsv =
    setweight(to_tsvector('zh_cfg', title), 'A') ||
    setweight(to_tsvector('zh_cfg', blog_content), 'B');
-- 5. 返回数据时添加html标签用于高亮关键词
SELECT ts_headline(
    'zh_cfg',
    blog_content,
    to_tsquery('zh_cfg', '开发 | 缓存'),
    'MaxWords=40, MinWords=20, StartSel=<mark>, StopSel=</mark>'
  ) AS snippet, t.*
FROM blog_detail t
WHERE tsv @@ to_tsquery('zh_cfg', '开发 | 缓存');

综上，使用PostgreSQL实现全文检索功能开发完成

四、补充

补充一段使用SQLAlchemy实现的带分页的全文检索代码：

	@classmethod
    async def search_blogs(cls,
                           session: AsyncSession,
                           keyword_str: str,
                           page: int,
                           size: int) -> Tuple[list[BlogSearchVO], int]:
        """全文检索博客"""
        # 总数（用于分页）
        count_sql = text(f"""
            SELECT COUNT(*)
            FROM blog_detail
            WHERE is_deleted = FALSE
                AND draft = FALSE
                AND tsv @@ to_tsquery('zh_cfg', '{keyword_str}')
        """)
        total_result = await session.execute(count_sql)
        total = total_result.scalar()
        if total == 0:
            # 如果无匹配结果，返回空结果
            return [], 0

        # 查询
        sql = text(f"""
            SELECT 
                id,
                title,
                ts_headline(
                    'zh_cfg',
                    title,
                    to_tsquery('zh_cfg', '{keyword_str}'),
                    'MaxWords=40, MinWords=20, StartSel=<mark>, StopSel=</mark>'
                ) AS title_highlight,
                blog_content,
                ts_headline(
                    'zh_cfg',
                    blog_content,
                    to_tsquery('zh_cfg', '{keyword_str}'),
                    'MaxWords=40, MinWords=20, StartSel=<mark>, StopSel=</mark>'
                ) AS blog_content_highlight,
                published,
                category,
                tags
            FROM blog_detail
            WHERE is_deleted = FALSE
                AND draft = FALSE
                AND tsv @@ to_tsquery('zh_cfg', '{keyword_str}')
            ORDER BY ts_rank(tsv, to_tsquery('zh_cfg', '{keyword_str}')) DESC
            LIMIT :size OFFSET :offset
        """)
        offset = (page - 1) * size
        result = await session.execute(sql, {"size": size, "offset": offset})
        rows = result.mappings().fetchall()

        return [BlogSearchVO.model_validate(row) for row in rows], total