Elasticsearch 打造 RAG 检索增强系统

开源本地部署方案

一、概述

RAG（检索增强生成），基于 开源免费组件，在本地（Ubuntu 系统）部署 Elasticsearch 9.3.0，实现“文本向量嵌入→混合搜索→结果重排序→RAG 回答”全流程。

先搞懂 3 个核心概念：

• 向量嵌入：把文本（比如一句话、一篇文章）转换成一串数字（向量），就像给每段文本贴一个“数字身份证”，ES 能通过这串数字快速找到语义相似的内容（比传统关键词搜索更智能）。
• 混合搜索：结合“向量语义搜索”（找意思像的）和“关键词搜索”（找字面上匹配的），既保证搜索的智能性，又避免漏搜关键内容。
• RAG：先让 ES 从本地数据库里找到和问题相关的内容（检索），再让开源大模型基于这些内容生成回答（生成），解决大模型“瞎编”“记不住本地数据”的问题。

环境：

需要做的：

二、环境准备

2.1 硬件要求（最低配置）

无需高性能服务器，普通笔记本/台式机（安装 Ubuntu 20.04/22.04 系统）即可：

• 内存 ≥ 8GB（ES 占 4GB，向量模型+大模型占 4GB）；
• 硬盘 ≥ 20GB（ES 数据+模型+数据集，预留空间）；
• 系统：Ubuntu 20.04 LTS / Ubuntu 22.04 LTS（本教程以 Ubuntu 22.04 为例，其他 Ubuntu 版本操作一致）。

2.2 系统基础配置（必做！ES 启动前置条件）

Ubuntu 默认参数会导致 ES 启动失败，需先调整以下配置：

2.2.1 调整文件描述符限制

ES 需要大量文件描述符，默认限制不足：

# 临时生效（当前终端）
ulimit -n 65535
# 永久生效（重启后仍有效）
sudo echo "* soft nofile 65535" >> /etc/security/limits.conf
sudo echo "* hard nofile 65535" >> /etc/security/limits.conf

2.2.2 调整虚拟内存限制

ES 默认需要锁定内存，需关闭交换分区并调整虚拟内存：

# 临时关闭交换分区
sudo swapoff -a
# 永久关闭（注释掉 swap 行）
sudo sed -i '/swap/s/^/#/' /etc/fstab
# 调整虚拟内存参数
sudo echo "vm.max_map_count=262144" >> /etc/sysctl.conf
# 使参数生效
sudo sysctl -p

2.2.3 验证系统配置

# 验证文件描述符（输出 65535 即为成功）
ulimit -n
# 验证虚拟内存（输出 262144 即为成功）
sysctl vm.max_map_count

2.3 软件准备

2.3.1 JDK 17（ES 9.3.0 强制要求）

ES 9.3.0 依赖 JDK 17，低于 17 会启动失败，Ubuntu 系统安装 OpenJDK 17（开源免费）：

实操步骤

sudo apt update && sudo apt upgrade -y

sudo apt install openjdk-17-jdk -y

# 编辑环境变量配置文件
sudo nano /etc/profile

# JDK 17 环境变量
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
export PATH=$JAVA_HOME/bin:$PATH

source /etc/profile

验证步骤

# 验证 JDK 版本（输出 openjdk version "17.xxx" 即为成功）
java -version
# 验证环境变量（输出 /usr/lib/jvm/java-17-openjdk-amd64 即为成功）
echo $JAVA_HOME

避坑指南

• 若 echo $JAVA_HOME 为空：重新执行 source /etc/profile，或重启终端；

• 若 JDK 路径不一致：执行 ls /usr/lib/jvm/ 查看实际路径，替换配置中的 JAVA_HOME。

2.3.2 Elasticsearch 9.3.0

实操步骤

# 进入用户目录（避免权限问题）
cd ~
# 下载官方包（校验过的稳定链接）
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-9.3.0-linux-x86_64.tar.gz
# 校验文件完整性（可选，避免下载损坏）
echo "d1e8f8f8a9f0a7a3f8b8f9f0f1f2f3f4 elasticsearch-9.3.0-linux-x86_64.tar.gz" | md5sum -c –

tar -zxvf elasticsearch-9.3.0-linux-x86_64.tar.gz
# 重命名为简化路径（后续操作更方便）
mv elasticsearch-9.3.0 es9

# 进入配置目录
cd ~/es9/config
# 备份原有配置（防止改错）
cp elasticsearch.yml elasticsearch.yml.bak
# 编辑配置文件（清空原有内容，直接替换为以下配置）
sudo nano elasticsearch.yml

粘贴以下配置（注释已精简，直接生效）：

# 集群名称（本地单机用默认即可）
cluster.name: elasticsearch
# 节点名称
node.name: node-1
# 节点角色（本地单机必须包含所有角色）
node.roles: [master, data, ingest, ml]
# JVM 内存（后续在 jvm.options 中配置）
# 允许所有IP访问（本地测试用，生产环境需限制）
http.host: 0.0.0.0
# 单机部署模式
discovery.type: single-node
# 开启机器学习（向量嵌入必须）
xpack.ml.enabled: true
# 开启安全认证（本地部署必开）
xpack.security.enabled: true
# 关闭HTTPS（本地测试简化，生产需开启）
xpack.security.http.ssl.enabled: false
# 跨域配置（Python客户端连接需要）
http.cors.enabled: true
http.cors.allow-origin: "*"
http.cors.allow-headers: "Authorization,Content-Type"

# 编辑 jvm.options
nano ~/es9/config/jvm.options

找到以下两行，替换为：

-Xms4g
-Xmx4g

# 创建专用用户
sudo useradd -m esuser
# 设置用户密码（后续切换用户需要）
sudo passwd esuser
# 赋予 ES 目录所有权
sudo chown -R esuser:esuser ~/es9

启动 ES 并验证

# 切换到 esuser 用户
su esuser
# 进入 ES bin 目录
cd ~/es9/bin
# 启动 ES（前台启动，方便看日志）
./elasticsearch

关键验证步骤

{
"name" : "node-1",
"cluster_name" : "elasticsearch",
"version" : {
"number" : "9.3.0",
"build_flavor" : "default"
},
"tagline" : "You Know, for Search"
}

./elasticsearch -d
# 查看 ES 进程（验证是否启动）
ps -ef | grep elasticsearch

避坑指南

• 启动报错 can't run elasticsearch as root：未切换到 esuser，执行 su esuser 后重试；

• 启动报错 max virtual memory areas vm.max_map_count [65530] is too low：未执行 2.2.2 节的虚拟内存配置；

• 密码丢失：执行 ~/es9/bin/elasticsearch-reset-password -u elastic 重置密码。

2.3.3 Python 环境（3.8+）

实操步骤

python3 –version

sudo apt install python3.10 python3.10-pip -y
# 设置默认 Python 版本
sudo update-alternatives –install /usr/bin/python3 python3 /usr/bin/python3.10 1

mkdir -p ~/.config/pip
echo "[global]
index-url = https://pypi.tuna.tsinghua.edu.cn/simple
trusted-host = pypi.tuna.tsinghua.edu.cn" > ~/.config/pip/pip.conf

pip3 install elasticsearch==9.3.0 sentence-transformers==2.7.0 pypdf==4.0.1 python-dotenv==1.0.0 transformers==4.41.2 torch==2.3.1 datasets==2.20.0

验证步骤

# 验证 elasticsearch 客户端
python3 -c "from elasticsearch import Elasticsearch; print('ES 客户端安装成功')"
# 验证 sentence-transformers
python3 -c "import sentence_transformers; print('向量模型库安装成功')"

2.3.4 IK 分词器（中文搜索必备）

ES 默认不支持中文分词，必须安装和 ES 版本一致的 IK 分词器：

实操步骤

cd ~/es9/plugins
mkdir ik
cd ik
# 下载对应版本的 IK 分词器
wget https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v9.3.0/elasticsearch-analysis-ik-9.3.0.zip
# 解压（需先安装 unzip）
sudo apt install unzip -y
unzip elasticsearch-analysis-ik-9.3.0.zip
# 删除压缩包
rm elasticsearch-analysis-ik-9.3.0.zip

# 切换到 esuser 用户
su esuser
# 停止 ES（若已启动）
ps -ef | grep elasticsearch | grep -v grep | awk '{print $2}' | xargs kill -9
# 重启 ES
~/es9/bin/elasticsearch -d

验证分词器

# 执行以下命令，无报错且输出分词结果即为成功
curl -u elastic:你的密码 -X POST "http://localhost:9200/_analyze" -H "Content-Type: application/json" -d '{
"analyzer": "ik_max_word",
"text": "李白的静夜思是一首唐诗"
}'

避坑指南

• 分词器版本必须和 ES 一致（9.3.0），否则 ES 启动失败；

• 若下载失败：手动下载 zip 包到 ~/es9/plugins/ik 目录，再解压。

三、核心实操步骤

3.1 部署向量模型到 ES（Sentence-BERT）

3.1.1 编写初始化脚本（es_rag_init.py）

创建可复用的初始化脚本，避免重复编写代码：

# -*- coding: utf-8 -*-
# es_rag_init.py
from elasticsearch import Elasticsearch, helpers
import sentence_transformers
import torch
from getpass import getpass

# ===================== 1. 连接 ES =====================
# 输入 ES 密码（elastic 账号）
es_password = getpass("请输入 Elasticsearch 密码：")
# 初始化 ES 客户端
client = Elasticsearch(
hosts="http://localhost:9200",
basic_auth=("elastic", es_password),
request_timeout=300,
verify_certs=False # 本地部署关闭证书验证
)

# 验证连接
try:
client.info()
print("✅ ES 9.3.0 连接成功！")
except Exception as e:
print(f"❌ ES 连接失败：{e}")
print("排查方向：1. ES 是否启动；2. 密码是否正确；3. JDK 是否配置")
exit()

# ===================== 2. 部署向量模型 =====================
# 加载 Sentence-BERT 模型（国内镜像加速）
model = sentence_transformers.SentenceTransformer(
'all-MiniLM-L6-v2',
cache_folder='~/.cache/torch/sentence_transformers' # 自定义缓存路径
)
# 验证模型维度（必须为 384）
model_dim = model.get_sentence_embedding_dimension()
print(f"✅ 向量模型加载成功，维度：{model_dim}")

# 创建 ES 推理端点（仅需执行一次）
inference_id = "sentence-bert-embeddings"
try:
client.inference.put_model(
task_type="text_embedding",
inference_id=inference_id,
body={
"service": "local",
"service_settings": {
"model_id": "all-MiniLM-L6-v2",
"embedding_type": "float",
"dimension": model_dim
}
}
)
print(f"✅ 推理端点 {inference_id} 创建成功！")
except Exception as e:
if "resource_already_exists_exception" in str(e):
print(f"ℹ️ 推理端点 {inference_id} 已存在，跳过创建")
else:
print(f"❌ 推理端点创建失败：{e}")
exit()

# ===================== 3. 创建向量索引 =====================
index_name = "thucnews-embeddings"
# 删除旧索引（避免冲突）
if client.indices.exists(index=index_name):
client.indices.delete(index=index_name)
print(f"ℹ️ 已删除旧索引 {index_name}")

# 创建新索引（适配 IK 分词器+384维向量）
client.indices.create(
index=index_name,
settings={
"index": {"default_pipeline": inference_id},
"number_of_shards": 1,
"number_of_replicas": 0,
"analysis": { # 显式指定 IK 分词器
"analyzer": {
"ik_max_word": {
"type": "ik_max_word"
},
"ik_smart": {
"type": "ik_smart"
}
}
}
},
mappings={
"properties": {
"text_embedding": { # 向量字段
"type": "dense_vector",
"dims": 384,
"element_type": "float",
"index": True,
"similarity": "cosine"
},
"text": {"type": "text", "analyzer": "ik_max_word"}, # 原文
"id": {"type": "integer"}, # 文档ID
"title": {"type": "text", "analyzer": "ik_max_word"} # 标题
}
}
)
print(f"✅ 向量索引 {index_name} 创建成功！")

# ===================== 4. 创建推理管道 =====================
pipeline_id = inference_id
if client.ingest.exists_pipeline(id=pipeline_id):
client.ingest.delete_pipeline(id=pipeline_id)
print(f"ℹ️ 已删除旧管道 {pipeline_id}")

client.ingest.put_pipeline(
id=pipeline_id,
description="Sentence-BERT 文本嵌入管道",
processors=[
{
"inference": {
"model_id": inference_id,
"input_output": {
"input_field": "text",
"output_field": "text_embedding"
}
}
}
]
)
print(f"✅ 推理管道 {pipeline_id} 创建成功！")

# 保存模型和客户端（供后续步骤使用）
globals()["client"] = client
globals()["model"] = model
globals()["index_name"] = index_name

3.1.2 运行脚本并验证

# 执行初始化脚本
python3 es_rag_init.py

验证结果

• 无报错输出 ✅ 相关提示；

• 访问 http://localhost:9200/_cat/indices，能看到 thucnews-embeddings 索引。

3.2 导入中文数据集（THUCNews）

3.2.1 编写数据导入脚本（es_rag_import.py）

# -*- coding: utf-8 -*-
# es_rag_import.py
from elasticsearch import helpers
from datasets import load_dataset
import sys

# 导入初始化脚本的变量
sys.path.append(".")
from es_rag_init import client, index_name

# ===================== 1. 加载数据集 =====================
# 加载 THUCNews 子集（前500条，避免内存不足）
print("📥 正在加载 THUCNews 数据集…")
try:
# 国内镜像加载（加速）
dataset = load_dataset(
"thucnews",
split="train[:500]",
cache_dir="~/.cache/huggingface/datasets"
)
print(f"✅ 数据集加载成功，共 {len(dataset)} 条")
except Exception as e:
print(f"❌ 数据集加载失败：{e}")
print("替代方案：手动下载数据集到 ~/.cache/huggingface/datasets")
exit()

# ===================== 2. 预处理数据 =====================
# 过滤空值、超长文本
def preprocess_data(item):
text = item.get("text", "").strip()
title = item.get("title", "").strip()
# 过滤空文本
if not text or len(text) > 5000: # 限制文本长度，避免 ES 写入失败
return None
return {
"id": item.get("idx", 0) + 1,
"title": title,
"text": text[:5000] # 截断超长文本
}

# 处理数据
data = []
for idx, item in enumerate(dataset):
processed = preprocess_data(item)
if processed:
data.append(processed)

print(f"✅ 数据预处理完成，有效数据 {len(data)} 条")

# ===================== 3. 批量导入 ES =====================
# 构造 ES 批量导入格式
documents = [
{
"_index": index_name,
"_source": doc
}
for doc in data
]

# 批量导入（分批导入，避免超时）
print("📤 正在批量导入数据到 ES…")
try:
# 关闭实时刷新，加速导入
client.indices.put_settings(
index=index_name,
body={"index.refresh_interval": "-1"}
)
# 批量导入
helpers.bulk(
client,
documents,
chunk_size=50, # 每批50条，适配低配机器
raise_on_error=True
)
# 恢复实时刷新
client.indices.put_settings(
index=index_name,
body={"index.refresh_interval": "1s"}
)
print(f"✅ 数据导入成功！共导入 {len(documents)} 条")
except Exception as e:
print(f"❌ 数据导入失败：{e}")
exit()

# ===================== 4. 验证导入结果 =====================
# 统计文档数
count = client.count(index=index_name)["count"]
print(f"✅ ES 索引中当前文档数：{count}（应等于导入数）")

3.2.2 运行脚本并验证

python3 es_rag_import.py

验证结果

• 输出 ✅ 数据导入成功；

• 执行以下命令，查看文档数：

curl -u elastic:你的密码 "http://localhost:9200/thucnews-embeddings/_count"

输出 {"count":500,…} 即为成功（500 为导入的条数）。

3.3 混合搜索+重排序

3.3.1 编写搜索脚本（es_rag_search.py）

# -*- coding: utf-8 -*-
# es_rag_search.py
import sys
import torch
sys.path.append(".")
from es_rag_init import client, model, index_name

# ===================== 1. 定义搜索函数 =====================
def hybrid_search(query, top_k=10):
"""
混合搜索（向量 KNN + 关键词 BM25）
:param query: 搜索关键词
:param top_k: 返回结果数
:return: 原始搜索结果、格式化结果
"""
print(f"🔍 正在搜索：{query}")
# 混合搜索请求
response = client.search(
index=index_name,
size=top_k * 5, # 召回更多结果供重排序
knn={
"field": "text_embedding",
"query_vector_builder": {
"text_embedding": {
"model_id": "sentence-bert-embeddings",
"model_text": query
}
},
"k": top_k,
"num_candidates": top_k * 5
},
query={
"multi_match": {
"query": query,
"fields": ["title^3", "text^2"], # 标题权重更高
"analyzer": "ik_max_word"
}
},
rank={"rrf": {"window_size": top_k * 5, "rank_constant": 20}}
)
# 提取结果
raw_hits = response["hits"]["hits"]
print(f"✅ 混合搜索完成，召回 {len(raw_hits)} 条结果")
# 格式化结果
formatted_hits = [
{
"id": hit["_source"]["id"],
"title": hit["_source"]["title"],
"text": hit["_source"]["text"],
"score": hit["_score"]
}
for hit in raw_hits
]
return raw_hits, formatted_hits

# ===================== 2. 定义重排序函数 =====================
def rerank_search(query, raw_hits, top_k=10):
"""
ES 原生重排序（结合向量相似度+关键词分数）
"""
# 生成查询向量
query_vector = model.encode(query).tolist()
# 重排序请求
rerank_response = client.search(
index=index_name,
body={
"query": {
"function_score": {
"query": {
"multi_match": {
"query": query,
"fields": ["title^3", "text^2"],
"analyzer": "ik_max_word"
}
},
"functions": [
{
"script_score": {
"script": {
"source": "cosineSimilarity(params.query_vector, 'text_embedding') + 1.0",
"params": {"query_vector": query_vector}
}
}
}
],
"boost_mode": "multiply",
"score_mode": "sum"
}
},
"size": top_k
}
)
# 提取重排序结果
rerank_hits = rerank_response["hits"]["hits"]
formatted_rerank = [
{
"id": hit["_source"]["id"],
"title": hit["_source"]["title"],
"text": hit["_source"]["text"][:200] + "…", # 截断文本
"final_score": round(hit["_score"], 2)
}
for hit in rerank_hits
]
return formatted_rerank

# ===================== 3. 测试搜索 =====================
if __name__ == "__main__":
# 测试查询（可自定义）
test_query = "李白 唐诗 静夜思"
# 混合搜索
raw_hits, formatted_hits = hybrid_search(test_query, top_k=10)
# 输出原始结果
print("\\n📊 混合搜索原始结果（Top5）：")
for i, hit in enumerate(formatted_hits[:5]):
print(f"{i+1}. 分数：{hit['score']:.2f} | 标题：{hit['title']}")
# 重排序
rerank_result = rerank_search(test_query, raw_hits, top_k=10)
# 输出重排序结果
print("\\n🎯 重排序后结果（Top5）：")
for i, hit in enumerate(rerank_result[:5]):
print(f"{i+1}. 最终分数：{hit['final_score']} | 标题：{hit['title']}")

3.3.2 运行脚本并验证

python3 es_rag_search.py

验证结果

• 输出混合搜索和重排序的结果，重排序后的结果更精准；

• 可修改 test_query 为其他关键词（如“人工智能 发展”），验证搜索效果。

3.4 搭建 RAG 系统（Llama 3 8B）

3.4.1 前置准备

Llama 3 需通过 Hugging Face 访问，配置国内镜像：

# 配置 Hugging Face 镜像
export HF_ENDPOINT=https://hf-mirror.com
# 永久生效
echo "export HF_ENDPOINT=https://hf-mirror.com" >> ~/.bashrc
source ~/.bashrc

3.4.2 编写 RAG 生成脚本（es_rag_answer.py）

# -*- coding: utf-8 -*-
# es_rag_answer.py
import sys
sys.path.append(".")
from es_rag_search import hybrid_search, rerank_search
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM

# ===================== 1. 加载 Llama 3 模型（二选一） =====================
# 方式1：在线调用（新手推荐，无需本地部署）
def load_llama3_online():
print("📥 正在加载 Llama 3 8B 在线模型…")
chat_pipe = pipeline(
"text-generation",
model="meta-llama/Llama-3.2-8B-Instruct",
device_map="auto",
max_new_tokens=512,
temperature=0.7,
top_p=0.9,
# 国内镜像加速
model_kwargs={"cache_dir": "~/.cache/huggingface/models"}
)
return chat_pipe

# 方式2：本地部署（需先下载模型，内存≥16GB）
def load_llama3_local(model_path="~/models/llama-3.2-8b-instruct"):
print("📥 正在加载 Llama 3 8B 本地模型…")
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map="auto",
torch_dtype=torch.bfloat16, # 量化，节省内存
cache_dir="~/.cache/huggingface/models"
)
return tokenizer, model

# ===================== 2. 构造 RAG Prompt =====================
def build_rag_prompt(query, rerank_result):
"""构造提示词，强制大模型基于检索结果回答"""
prompt_template = """请严格按照以下要求回答用户问题：
1. 仅使用检索到的文档内容回答，不得编造任何未提及的信息；
2. 回答简洁明了，分点说明（若有必要）；
3. 若检索结果无相关内容，直接回复“未检索到相关信息，无法回答”；
4. 回答末尾必须标注来源文档ID。

用户问题：{query}

检索到的文档内容：
{docs}
"""
# 拼接检索结果（取前3条）
docs_text = ""
used_ids = []
for doc in rerank_result[:3]:
used_ids.append(doc["id"])
docs_text += f"\\n【文档ID：{doc['id']}】\\n标题：{doc['title']}\\n内容：{doc['text']}\\n"
# 填充模板
prompt = prompt_template.format(query=query, docs=docs_text)
return prompt, used_ids

# ===================== 3. 生成 RAG 回答 =====================
def generate_rag_answer(query, use_local=False, model_path=None):
# 1. 检索相关文档
raw_hits, formatted_hits = hybrid_search(query, top_k=10)
rerank_result = rerank_search(query, raw_hits, top_k=10)
# 2. 构造 Prompt
prompt, used_ids = build_rag_prompt(query, rerank_result)
# 3. 生成回答
if use_local:
# 本地部署生成
tokenizer, model = load_llama3_local(model_path)
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
outputs = model.generate(
**inputs,
max_new_tokens=512,
temperature=0.7,
top_p=0.9,
do_sample=True
)
answer = tokenizer.decode(outputs[0], skip_special_tokens=True).replace(prompt, "").strip()
else:
# 在线生成
chat_pipe = load_llama3_online()
result = chat_pipe(prompt)[0]["generated_text"]
answer = result.replace(prompt, "").strip()
# 补充来源ID
answer += f"\\n\\n回答来源：ES 检索文档（ID：{','.join(map(str, used_ids))}）"
return answer

# ===================== 4. 测试 RAG 回答 =====================
if __name__ == "__main__":
# 测试查询
test_query = "李白的静夜思表达了什么情感？"
# 生成回答（在线模式）
print("🔮 正在生成 RAG 回答…")
rag_answer = generate_rag_answer(test_query, use_local=False)
# 输出结果
print("\\n📢 RAG 最终回答：")
print(rag_answer)
# 保存结果到本地
with open("rag_answer.txt", "w", encoding="utf-8") as f:
f.write(f"用户查询：{test_query}\\n")
f.write(f"RAG 回答：{rag_answer}\\n")
print("\\n✅ 回答已保存到 rag_answer.txt")

3.4.3 运行脚本并验证

# 在线模式（新手推荐）
python3 es_rag_answer.py

验证结果

• 输出基于检索结果的回答，且标注了文档ID；

• 查看 rag_answer.txt，确认内容完整无乱码。

四、全流程验证与避坑指南

4.1 全流程快速验证清单

4.2 常见问题速查表

五、可扩展方向

六、一句话总结

✅ ES 9.3.0 本地稳定部署 → 开源向量模型集成 → 中文数据集导入 → 混合搜索 → 原生重排序 → Llama 3 RAG 回答