大模型 RAG 中关键字检索的实现与实战

针对中文文本的处理实现

中文处理主要依赖分词工具（如 jieba），NLTK 对中文支持有限，建议结合 jieba 使用。

import re
import jieba

# 定义简单的中文停用词表，实际项目中可加载更完整的列表
STOP_WORDS_CN = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'}

def to_keywords(input_string):
    """将句子转成检索关键词序列"""
    # 按搜索引擎模式分词
    word_tokens = list(jieba.cut_for_search(input_string))
    # 去除停用词
    filtered_sentence = [w for w in word_tokens if w not in STOP_WORDS_CN and len(w) > 1]
    return ' '.join(filtered_sentence)

def sent_tokenize(input_string):
    """按标点断句"""
    # 按标点切分
    sentences = re.split(r'(?<=[。！？；?!])', input_string)
    # 去掉空字符串
    return [sentence for sentence in sentences if sentence.strip()]

将文本灌入检索引擎

连接 Elasticsearch 并创建索引，将处理后的文本写入。

import os, time

# 引入配置文件
ELASTICSEARCH_BASE_URL = os.getenv('ELASTICSEARCH_BASE_URL', 'http://localhost:9200')
ELASTICSEARCH_PASSWORD = os.getenv('ELASTICSEARCH_PASSWORD', '')
ELASTICSEARCH_NAME= os.getenv('ELASTICSEARCH_NAME', 'elastic')

# 1. 创建 Elasticsearch 连接
es = Elasticsearch(
    hosts=[ELASTICSEARCH_BASE_URL],  # 服务地址与端口
    http_auth=(ELASTICSEARCH_NAME, ELASTICSEARCH_PASSWORD),  # 用户名，密码
    verify_certs=False  # 开发环境可关闭证书验证
)

# 2. 定义索引名称
index_name = "teacher_demo_index"

# 3. 如果索引已存在，删除它（仅供演示，实际应用时不需要这步）
if es.indices.exists(index=index_name):
    es.indices.delete(index=index_name)

# 4. 创建索引（包含自定义映射以优化搜索）
es.indices.create(
    index=index_name,
    body={
        "mappings": {
            "properties": {
                "keywords": {"type": "text", "analyzer": "ik_max_word"},
                "text": {"type": "text"}
            }
        }
    }
)

# 模拟段落数据
paragraphs = [
    "Llama 2 is a collection of pretrained and fine-tuned large language models.",
    "It ranges in scale from 7 billion to 70 billion parameters.",
    "The model supports grouped-query attention and has increased context length."
]

# 5. 灌库指令
actions = [
    {
        "_index": index_name,
        "_source": {
            "keywords": to_keywords(para),
            "text": para
        }
    }
    for para in paragraphs
]

# 6. 文本灌库
helpers.bulk(es, actions)

# 灌库是异步的，等待刷新
time.sleep(2)
es.indices.refresh(index=index_name)

实现关键字检索

使用 ES 的 match 查询语言进行搜索。

def search(query_string, top_n=3):
    # ES 的查询语言
    search_query = {
        "query": {
            "multi_match": {
                "query": to_keywords(query_string),
                "fields": ["keywords"]
            }
        },
        "size": top_n
    }
    res = es.search(index=index_name, query=search_query)
    return [hit["_source"]["text"] for hit in res["hits"]["hits"]]
    
results = search("how many parameters does llama 2 have?", 2)
for r in results:
    print(r + "\n")

LLM 接口封装

封装 OpenAI 接口用于生成最终回复。

from openai import OpenAI
import os
from dotenv import load_dotenv, find_dotenv

load_dotenv(find_dotenv())  # 读取本地 .env 文件
client = OpenAI()

def get_completion(prompt, model="gpt-3.5-turbo"):
    '''封装 openai 接口'''
    messages = [{"role": "user", "content": prompt}]
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,  # 模型输出的随机性，0 表示随机性最小
    )
    return response.choices[0].message.content

Prompt 模板

设计清晰的 Prompt 模板以规范 LLM 的输出行为。

def build_prompt(prompt_template, **kwargs):
    '''将 Prompt 模板赋值'''
    inputs = {}
    for k, v in kwargs.items():
        if isinstance(v, list) and all(isinstance(elem, str) for elem in v):
            val = '\n\n'.join(v)
        else:
            val = v
        inputs[k] = val
    return prompt_template.format(**inputs)

prompt_template = """
你是一个问答机器人。
你的任务是根据下述给定的已知信息回答用户问题。

已知信息:
{context}

用户问：
{query}

如果已知信息不包含用户问题的答案，或者已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。
请不要输出已知信息中不包含的信息或答案。
请用中文回答用户问题。
"""

一个简单的 RAG Pipeline 就产生了

什么是 Pipeline

首先，RAG 不是一个单项技术，它是一个流水线（Pipeline）。只有对流水线上的每一步骤都进行精细打磨，最后才能出来效果，我们的目标就是努力使每个环节都达到尽可能准确。

Demo 运行流程

user_query = "how many parameters does llama 2 have?"

# 1. 检索
search_results = search(user_query, 2)

# 2. 构建 Prompt
prompt = build_prompt(prompt_template, context=search_results, query=user_query)
print("===Prompt===")
print(prompt)

# 3. 调用 LLM
response = get_completion(prompt)

print("===回复===")
print(response)

效果展示

===Prompt===
你是一个问答机器人。
你的任务是根据下述给定的已知信息回答用户问题。

已知信息:
1. Llama 2, an updated version of Llama 1, trained on a new mix of publicly available data...
2. In this work, we develop and release Llama 2, a collection of pretrained and fine-tuned large language models (LLMs) ranging in scale from 7 billion to 70 billion parameters...

用户问：
how many parameters does llama 2 have?

如果已知信息不包含用户问题的答案，或者已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。
请不要输出已知信息中不包含的信息或答案。
请用中文回答用户问题。

===回复===
Llama 2 有 7B, 13B 和 70B 参数。

高级配置与优化建议

为了提升关键字检索的效果，除了基础的分词和索引外，还可以考虑以下优化策略：

1. 自定义 Analyzer

Elasticsearch 默认的分词器可能不适合特定业务场景。可以配置自定义分析器，例如针对中文使用 ik_max_word 或 smartcn，针对英文使用 standard 配合同义词过滤器。

{
  "settings": {
    "analysis": {
      "analyzer": {
        "my_analyzer": {
          "tokenizer": "ik_max_word",
          "filter": ["lowercase", "synonym_filter"]
        }
      },
      "filter": {
        "synonym_filter": {
          "type": "synonym",
          "synonyms": ["llama, llama2, l2"]
        }
      }
    }
  }
}

2. 查询优化

使用 multi_match 查询可以同时在多个字段中搜索，提高召回率。同时，可以使用 fuzziness 参数来容忍拼写错误。

search_query = {
    "query": {
        "multi_match": {
            "query": to_keywords(query_string),
            "fields": ["keywords^2", "text"],
            "fuzziness": "AUTO"
        }
    }
}

3. 性能考量

• 批量写入：使用 helpers.bulk 进行批量写入，避免频繁网络请求。
• 刷新间隔：生产环境中调整 refresh_interval 以减少写入时的 IO 压力。
• 缓存：利用 ES 的 Query Cache 加速重复查询。

总结

本文我们对关键字检索做了讲解以及从实战的角度做了详细解析。

首先我们应该认识到 RAG 的核心在于外部知识库，这个外部知识库甚至可以是基于关系型数据库的 SQL 查询，也可以是 Elasticsearch 这样的搜索引擎。

而本文基于 ES 的关键字检索也是一个比较传统和基础的检索方式，它比较简单，也有它适用的场景，特别是在精确匹配和术语搜索方面表现优异。

关键字检索的局限性

最关键的就在于：关键字检索可能会受到一些问题的影响，例如同义词、拼写错误等，这可能会导致一些相关的文档被漏掉或者一些不相关的文档被检索到。语义理解能力较弱是其核心短板。

关键字检索通常在简单的检索场景下使用，例如在搜索引擎中用户输入关键字进行网页检索。

这样我们就引出了向量检索，在下一篇文章中我们再进行分享。随着 RAG 的发展，单纯的关键字检索或者向量检索都很难满足我们面临的复杂场景，了解关键字检索是为了后面更好的了解混合检索（Hybrid Search），即结合两者的优势以获得最佳检索效果。