下面介绍一种简单的关键字检索,主要用于PostgreSQL数据库。主要利用的是应用层切词+PostgreSQL存储的方式。
假设表articles有1个tsvector列search_vec,此时可以这样操作:
CREATE INDEX idx_articles_search ON articles USING GIN (search_vector);
SELECT * FROM articles
WHERE search_vec @@ to_tsquery('simple', '南京市 & 人口');
在此之前需要理由jieba对文档内容进行停用词处理后分词,之后将这些分词的结果存储在search_vec列中.
# Python
import jieba
text = "南京市人口多少呢?...."
# 先进行停用词处理后再分词
words = jieba.lcut(text)
# 构造带位置的 tsvector 字符串: '南京市':1 '人口':2 '多少':3
tsv_str = " ".join([f"'{w}':{i+1}" for i, w in enumerate(words)])
# 存入数据库时，需要强制转换类型
# SQL: UPDATE articles SET search_vec = %s::tsvector WHERE id = 1;
我们可以jieba的TF-IDF算法提供关键字:
import jieba.analyse
query = "南京市人口多少"
# 提取前3个关键词
keywords = jieba.analyse.extract_tags(query, topK=3, withWeight=False)
# 结果: ['南京市', '人口', '多少']
# 关键步骤：将关键词用 '&' (AND) 连接，构造 tsquery 语法
# 注意：如果关键词包含特殊字符，需要进行转义处理
tsquery_str = " & ".join(keywords)
# 结果: "南京市 & 人口 & 多少"
如果使用的是pg_jieba等插件,则直接使用PG中to_tsquery函数即可,类似如下:
select to_tsvector('jiebacfg', document_text);
为了简化tsquery字符串的编写,我们可以利用LLM进行编写:
import jieba.analyse
import json
import re
import psycopg2
from openai import OpenAI
# =================配置区域=================
# 1. 数据库配置
DB_CONFIG = {
"dbname": "your_db_name",
"user": "your_user",
"password": "your_password",
"host": "localhost",
"port": 5432
}
# 2. LLM 配置 (以 OpenAI 为例，也可替换为国内大模型接口)
client = OpenAI(
api_key="YOUR_API_KEY",
base_url="https://api.openai.com/v1" # 如果是国内模型，替换为对应的 base_url
)
# 3. PostgreSQL 使用的分词配置名称 (需与建表时一致，如 'jiebacfg', 'zhparser' 等)
PG_TS_CONFIG = "jiebacfg"
# =========================================
def extract_keywords(query: str, top_k: int = 5) -> list:
"""
使用 jieba 提取核心关键词，过滤停用词和虚词
"""
# extract_tags 基于 TF-IDF，能较好识别核心名词/动词
keywords = jieba.analyse.extract_tags(query, topK=top_k, withWeight=False)
# 可选：进一步过滤单个字符或无意义词（根据业务需求调整）
filtered_keywords = [k for k in keywords if len(k) > 1]
return filtered_keywords if filtered_keywords else keywords
def generate_tsquery_via_llm(original_query: str, keywords: list) -> str:
"""
调用大模型，根据用户意图生成合法的 PostgreSQL tsquery 字符串
"""
if not keywords:
return ""
system_prompt = """
你是一个 PostgreSQL 全文检索专家。
你的任务是将用户的自然语言查询转换为合法的 PostgreSQL tsquery 格式字符串。
规则：
1. 输出必须是纯文本字符串，不要包含 Markdown 格式（如 ```json），不要包含任何解释。
2. 关键词必须用单引号包裹，例如 'keyword'。
3. 逻辑运算符使用：
- AND: &
- OR: |
- NOT: !
- FOLLOWED BY (短语紧邻): <->
4. 默认策略：
- 如果用户只是普通陈述或疑问，关键词之间默认用 & (AND) 连接。
- 如果用户明确表达“或者”、“任一”，使用 | (OR)。
- 如果用户表达“排除”、“不含”，使用 ! (NOT)。
- 如果关键词构成一个固定的专有名词或短语（如“北京大学”），使用 <-> 连接。
5. 如果关键词列表为空，返回空字符串。
示例 1:
输入: Query="南京市人口多少", Keywords=['南京市', '人口']
输出: '南京市' & '人口'
示例 2:
输入: Query="苹果或者华为的手机评测", Keywords=['苹果', '华为', '手机', '评测']
输出: ('苹果' | '华为') & '手机' & '评测'
示例 3:
输入: Query="北京到上海的高铁", Keywords=['北京', '上海', '高铁']
输出: '北京' <-> '上海' & '高铁'
"""
user_prompt = f"""
原始查询: "{original_query}"
提取关键词: {json.dumps(keywords, ensure_ascii=False)}
请生成 tsquery 字符串:
"""
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo", # 或 gpt-4, qwen-plus 等
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.1 # 低温度以保证输出格式稳定
)
tsquery_str = response.choices[0].message.content.strip()
# 清理可能存在的 Markdown 标记 (虽然 prompt 禁止了，但防万一)
tsquery_str = re.sub(r'```.*?```', '', tsquery_str, flags=re.DOTALL).strip()
return tsquery_str
except Exception as e:
print(f"LLM 调用失败: {e}")
# 降级策略：如果 LLM 失败，默认使用 AND 连接
return " & ".join([f"'{k}'" for k in keywords])
def validate_tsquery(tsquery_str: str) -> bool:
"""
简单的合法性校验，防止非法字符导致 SQL 错误
注意：最严格的校验应该在 DB 端通过 try-except 完成
"""
if not tsquery_str:
return False
# 基本检查：只允许包含字母、数字、中文、空格、单引号、& | ! <-> ( ) : *
# 这是一个宽松的正则，具体可根据需求调整
pattern = r"\w\u4e00-\u9fa5\s'&|!<>\-\(\):*]+$"
return bool(re.match(pattern, tsquery_str))
def search_in_pg(tsquery_str: str):
"""
在 PostgreSQL 中执行搜索
"""
if not tsquery_str:
return []
conn = None
try:
conn = psycopg2.connect(**DB_CONFIG)
cur = conn.cursor()
# 使用 parameterized query 防止 SQL 注入
# 注意：to_tsquery 的第二个参数是 query 字符串
sql = """
SELECT id, title, snippet
FROM articles
WHERE search_vector @@ to_tsquery(%s, %s)
ORDER BY ts_rank(search_vector, to_tsquery(%s, %s)) DESC
LIMIT 10;
"""
# 传入两次 config 和 query，因为 select 和 order by 都用到了
params = (PG_TS_CONFIG, tsquery_str, PG_TS_CONFIG, tsquery_str)
cur.execute(sql, params)
results = cur.fetchall()
return results
except Exception as e:
print(f"数据库查询错误: {e}")
# 如果是因为 tsquery 语法错误，可以尝试降级为 plainto_tsquery
if "syntax error" in str(e).lower():
print("尝试降级为 plainto_tsquery...")
return fallback_search(tsquery_str) # 需自行实现降级逻辑
return []
finally:
if conn:
conn.close()
def main():
# 模拟用户输入
user_query = "我想找关于南京或者苏州的人口统计数据，不要北京的"
print(f"用户输入: {user_query}")
# 1. 应用层分词
keywords = extract_keywords(user_query)
print(f"提取关键词: {keywords}")
# 2. LLM 生成 tsquery
tsquery_str = generate_tsquery_via_llm(user_query, keywords)
print(f"LLM 生成的 tsquery: {tsquery_str}")
# 3. 校验与执行
if validate_tsquery(tsquery_str):
results = search_in_pg(tsquery_str)
print(f"搜索结果数量: {len(results)}")
for row in results:
print(row)
else:
print("生成的查询字符串无效，跳过搜索。")
if __name__ == "__main__":
main()
但是这种效果往往不如BM25的。
如果喜欢这篇文章或对您有帮助，可以：[☕] 请我喝杯咖啡 | [💓] 小额赞助
