Elasticsearch 9.x 本地RAG个人知识库实操

一、概述

核心原理：3个角色组成你的离线私人知识库

我们要搭的系统，就像一个完全装在你电脑里、不联网的私人智库，3个核心角色分工明确，没有任何一个环节需要连外网：

全流程闭环

二、第一步：环境准备

关键提醒：物理隔离环境无外网，必须先在有外网的设备上，把所有安装包、镜像、模型、代码全部下载完成，再通过U盘/合规介质拷贝到离线环境，缺一不可！

【必下物料清单】（固定版本，避免兼容性问题，全为官方正版地址）

【有网环境必做】Docker镜像导出命令

在有网设备上拉取完镜像后，执行以下命令导出镜像文件，拷贝到离线环境：

# 导出ES 9.1.3镜像
docker save -o es-9.1.3.tar docker.elastic.co/elasticsearch/elasticsearch:9.1.3
# 导出LocalAI CPU镜像
docker save -o localai-cpu.tar localai/localai:latest-cpu

执行完成后，当前目录会生成es-9.1.3.tar和localai-cpu.tar两个文件，必须和其他物料一起拷贝到离线环境。

三、第二步：离线环境基础环境部署

所有操作均在离线环境的服务器/电脑上执行，全程无需联网

3.1 安装Docker（容器运行环境）

原理大白话

Docker就像一个“隔离的盒子”，我们把ES和LocalAI分别装在两个独立的盒子里，不会和设备上的其他软件冲突，离线环境无需解决复杂的依赖问题，导入镜像即可直接运行。

Linux离线安装步骤

# 进入Docker安装包目录
cd /opt/offline-packages/docker/
# 解压安装包
tar -zxvf docker-20.10.24.tgz
# 将Docker二进制文件复制到系统PATH，实现全局调用
cp docker/* /usr/bin/

# 创建Docker服务配置文件
cat > /etc/systemd/system/docker.service <<EOF
[Unit]
Description=Docker Application Container Engine
Documentation=https://docs.docker.com
After=network-online.target firewalld.service
Wants=network-online.target

[Service]
Type=notify
ExecStart=/usr/bin/dockerd
ExecReload=/bin/kill -s HUP $MAINPID
TimeoutSec=0
RestartSec=2
Restart=always

LimitNOFILE=infinity
LimitNPROC=infinity
LimitCORE=infinity

TasksMax=infinity
Delegate=yes
KillMode=process

[Install]
WantedBy=multi-user.target
EOF

# 重载系统服务配置
systemctl daemon-reload
# 启动Docker服务
systemctl start docker
# 设置开机自启
systemctl enable docker

Windows离线安装步骤

直接双击Docker Desktop离线安装包，跟随向导完成安装，安装完成后启动Docker Desktop，右下角Docker图标显示running，说明安装成功。

离线环境常见报错处理

• 报错permission denied：使用root用户执行，或给当前用户配置sudo权限

• 报错docker服务启动失败：检查防火墙是否关闭，系统内核版本是否符合Docker要求（3.10+）

3.2 导入离线Docker镜像

原理大白话

我们在有网环境把ES和LocalAI的“完整安装包”导出成了tar文件，现在把这些文件导入到离线环境的Docker中，就像手机安装APK文件一样，无需联网即可完成安装。

操作步骤

# 进入镜像tar文件存放目录
cd /opt/offline-packages/images/
# 导入ES 9.1.3镜像
docker load -i es-9.1.3.tar
# 导入LocalAI CPU镜像
docker load -i localai-cpu.tar

【成功校验标准】

执行docker images，能看到docker.elastic.co/elasticsearch/elasticsearch:9.1.3和localai/localai:latest-cpu两个镜像，说明导入成功。

3.3 安装Python与离线依赖包

原理大白话

RAG核心脚本基于Python开发，需要安装Python环境和对应的依赖库。离线环境无法使用pip install在线安装，因此我们提前把所有依赖包的离线安装文件（whl格式）下载完成，直接本地安装即可。

操作步骤

# 进入whl包存放目录
cd /opt/offline-packages/whl/
# 本地安装所有依赖包，完全离线，无外网调用
pip install –no-index –find-links=./ elasticsearch requests openai

四、第三步：Elasticsearch 9.x 离线部署

4.1 提前规划：ES离线环境目录

所有数据、配置、模型均存储在宿主机目录，容器重启不会丢失数据，离线环境必须做持久化配置

# 创建ES工作目录，所有数据均存储在此处
mkdir -p /opt/rag-offline/es/{data,logs,config,models}
# 给目录赋读写权限，ES容器内部用户为1000，必须开放权限否则启动失败
chmod -R 777 /opt/rag-offline/es/

4.2 启动ES 9.x 离线单节点容器

原理大白话

我们通过Docker启动ES，所有配置均适配离线单节点环境，关闭了所有需要联网的功能，开启了安全认证，同时把数据、配置持久化到宿主机，即使容器删除，数据也不会丢失。

启动命令

docker run -d \\
–name es-rag-offline \\
–restart=always \\
-p 9200:9200 \\
-p 9300:9300 \\
-e "discovery.type=single-node" \\ # 单节点模式，离线环境无需集群配置
-e "ES_JAVA_OPTS=-Xms2g -Xmx2g" \\ # 堆内存设置，建议为设备内存的1/4，最低2G
-e "xpack.security.enabled=true" \\ # 开启安全认证，符合内网合规要求
-e "xpack.security.authc.api_key.enabled=true" \\ # 开启API密钥认证，脚本调用使用
-e "xpack.ml.enabled=true" \\ # 开启机器学习功能，用于加载嵌入模型
-e "xpack.ml.model_repository.file.enabled=true" \\ # 开启本地模型导入，离线环境核心配置
-v /opt/rag-offline/es/data:/usr/share/elasticsearch/data \\
-v /opt/rag-offline/es/logs:/usr/share/elasticsearch/logs \\
-v /opt/rag-offline/es/config:/usr/share/elasticsearch/config \\
-v /opt/rag-offline/es/models:/usr/share/elasticsearch/models \\ # 挂载离线模型目录
docker.elastic.co/elasticsearch/elasticsearch:9.1.3

Windows环境适配

将目录替换为Windows路径，比如D:\\rag-offline\\es\\data，命令修改为：

docker run -d ^
–name es-rag-offline ^
–restart=always ^
-p 9200:9200 ^
-p 9300:9300 ^
-e "discovery.type=single-node" ^
-e "ES_JAVA_OPTS=-Xms2g -Xmx2g" ^
-e "xpack.security.enabled=true" ^
-e "xpack.security.authc.api_key.enabled=true" ^
-e "xpack.ml.enabled=true" ^
-e "xpack.ml.model_repository.file.enabled=true" ^
-v D:\\rag-offline\\es\\data:/usr/share/elasticsearch/data ^
-v D:\\rag-offline\\es\\logs:/usr/share/elasticsearch/logs ^
-v D:\\rag-offline\\es\\config:/usr/share/elasticsearch/config ^
-v D:\\rag-offline\\es\\models:/usr/share/elasticsearch/models ^
docker.elastic.co/elasticsearch/elasticsearch:9.1.3

【成功校验标准】

curl -X GET "http://localhost:9200/" -u "elastic:你的elastic密码"

补充说明：ES首次启动会自动生成elastic用户的密码，执行docker logs es-rag-offline | grep "Password for the elastic user is"即可获取，务必妥善保存。

4.3 离线导入multilingual-e5-small嵌入模型到ES

原理大白话

在线环境中，ES会自动从官网下载该嵌入模型，但离线环境无法联网，因此我们需要提前把模型文件下载好，放到ES的模型目录中，让ES离线加载该模型，才能在无外网的环境下生成语义向量。

操作步骤

curl -X GET "http://localhost:9200/_ml/trained_models/.multilingual-e5-small" -u "elastic:你的elastic密码"

【成功校验标准】

命令返回模型的详细信息，无报错，说明模型离线加载成功，ES可在离线环境使用该模型生成语义向量。

4.4 生成ES API密钥

操作步骤

curl -X POST "http://localhost:9200/_security/api_key" \\
-u "elastic:你的elastic密码" \\
-H "Content-Type: application/json" \\
-d '{
"name": "rag-offline-api-key",
"expiration": "10y",
"role_descriptors": {
"rag-admin": {
"cluster": ["all"],
"index": [
{
"names": ["*"],
"privileges": ["all"]
}
]
}
}
}'

【成功校验标准】

返回结果中的encoded字段，就是你的API密钥，务必复制保存，后续脚本需要使用，格式示例：MTk0NVRwa0IxRGRjVmVKSGl2ZFc6dUFaNkZTUHVXRzEzVjdrejQzSUNxZw==

五、第四步：LocalAI+大模型 完全离线部署

5.1 提前规划：LocalAI离线环境目录

# 创建LocalAI工作目录，存放大模型文件
mkdir -p /opt/rag-offline/localai/models
# 给目录赋读写权限
chmod -R 777 /opt/rag-offline/localai/

5.2 放置大模型文件

把提前下载好的dolphin3.0-qwen2.5-0.5b-instruct.Q4_K_M.gguf文件，拷贝到/opt/rag-offline/localai/models目录中，LocalAI启动时会自动加载该模型，完全离线，无需通过API下载。

5.3 启动LocalAI离线容器

原理大白话

LocalAI就像一个本地的OpenAI服务，我们把大模型文件挂载到容器中，它启动时会自动加载，对外提供和OpenAI完全兼容的API接口，RAG脚本无需修改任何代码，即可调用本地大模型，全程无外网调用。

启动命令

docker run -d \\
–name localai-rag-offline \\
–restart=always \\
-p 8080:8080 \\
-v /opt/rag-offline/localai/models:/models \\ # 挂载本地模型目录
-e DEBUG=false \\
-e MODELS_PATH=/models \\
-e THREADS=4 \\ # 改为你的CPU物理核心数，核心越多生成速度越快
-e CONTEXT_SIZE=2048 \\ # 上下文窗口，平衡内存占用和长文本支持
-e PRELOAD_MODELS=dolphin3.0-qwen2.5-0.5b-instruct.Q4_K_M.gguf \\ # 启动时预加载模型，无需等待调用再加载
localai/localai:latest-cpu

Windows环境适配

将目录替换为Windows路径，比如D:\\rag-offline\\localai\\models，修改为对应Windows格式即可。

【成功校验标准】

curl -X POST http://localhost:8080/v1/chat/completions \\
-H "Content-Type: application/json" \\
-d '{
"model": "dolphin3.0-qwen2.5-0.5b-instruct.Q4_K_M.gguf",
"messages": [{"role": "user", "content": "你好"}],
"stream": false
}'

六、第五步：RAG核心配置与离线脚本部署

6.1 原理大白话

这个脚本是整个系统的“总指挥”，全程无外网调用，自动完成以下工作：

6.2 准备知识库文档

在离线环境中，创建/opt/rag-offline/dataset目录，把所有txt格式的文档（会议纪要、报告、制度文件等）放入该目录，脚本会自动读取所有txt文件。

6.3 完整离线RAG脚本

把以下脚本保存为/opt/rag-offline/rag_offline.py，仅需修改开头的3个配置项（ES_API_KEY、AI_MODEL_NAME、DATASET_FOLDER），其余内容无需修改。

import os
import time
from elasticsearch import Elasticsearch, helpers
from openai import OpenAI

# ===================== 【仅需修改这里的配置，其他内容无需改动】=====================
# 替换为之前生成的ES API密钥
ES_API_KEY = "你的ES API密钥"
# 替换为放入models目录的大模型文件名，必须和文件名完全一致
AI_MODEL_NAME = "dolphin3.0-qwen2.5-0.5b-instruct.Q4_K_M.gguf"
# 知识库文档存放目录
DATASET_FOLDER = "/opt/rag-offline/dataset"
# ==============================================================================

# 固定配置，离线环境无需修改
ES_URL = "http://localhost:9200"
INDEX_NAME = "offline-rag-knowledge-base"
LOCAL_AI_URL = "http://localhost:8080/v1"

# 初始化客户端，全程无外网调用
es_client = Elasticsearch(ES_URL, api_key=ES_API_KEY)
ai_client = OpenAI(base_url=LOCAL_AI_URL, api_key="sk-offline-rag") # API key填写任意非空字符串即可，无需真实密钥

def check_es_connect():
"""检查ES连接是否正常，离线环境第一步必做"""
try:
if not es_client.ping():
print("❌ 无法连接到Elasticsearch，请检查服务是否启动、API密钥是否正确")
exit(1)
es_info = es_client.info()
print(f"✅ 成功连接Elasticsearch，版本：{es_info['version']['number']}")
return True
except Exception as e:
print(f"❌ ES连接失败，错误信息：{str(e)}")
exit(1)

def setup_embedding_inference():
"""创建离线嵌入推理端点，使用提前导入的e5-small模型，无外网调用"""
inference_id = "offline-e5-small-embedding"
# 先检查推理端点是否已存在，存在则无需重复创建
try:
es_client.inference.get(inference_id=inference_id)
print(f"✅ 嵌入推理端点 {inference_id} 已存在，无需重复创建")
return
except Exception:
pass

# 创建推理端点，完全适配离线环境，无外部调用
try:
es_client.inference.put(
inference_id=inference_id,
task_type="text_embedding",
body={
"service": "elasticsearch",
"service_settings": {
"num_allocations": 1,
"num_threads": 1,
"model_id": ".multilingual-e5-small", # 使用提前离线导入的模型
},
"chunking_settings": {
"strategy": "sentence",
"max_chunk_size": 250,
"sentence_overlap": 2
}
}
)
print(f"✅ 嵌入推理端点 {inference_id} 创建成功")
except Exception as e:
print(f"❌ 嵌入推理端点创建失败，错误信息：{str(e)}")
exit(1)

def setup_knowledge_index():
"""创建知识库索引，使用ES9.x原生semantic_text字段，自动处理向量，无需手动编码"""
try:
if es_client.indices.exists(index=INDEX_NAME):
print(f"✅ 知识库索引 {INDEX_NAME} 已存在，无需重复创建")
return False

print(f"📦 正在创建知识库索引 {INDEX_NAME}…")
es_client.indices.create(
index=INDEX_NAME,
body={
# 离线单节点环境专属配置，避免索引黄标，提升性能
"settings": {
"number_of_shards": 1,
"number_of_replicas": 0,
"refresh_interval": "1s"
},
# 索引映射，核心为semantic_text语义字段，自动处理向量生成和检索
"mappings": {
"properties": {
"file_name": {
"type": "text",
"copy_to": "semantic_content" # 把文件名复制到语义字段，支持按文件名检索
},
"file_content": {
"type": "text",
"copy_to": "semantic_content" # 把文件内容复制到语义字段，支持按内容检索
},
"semantic_content": {
"type": "semantic_text",
"inference_id": "offline-e5-small-embedding" # 绑定创建的离线推理端点
}
}
}
}
)
print(f"✅ 知识库索引 {INDEX_NAME} 创建成功")
return True
except Exception as e:
print(f"❌ 知识库索引创建失败，错误信息：{str(e)}")
exit(1)

def load_documents_from_folder():
"""从本地文件夹加载所有txt文档，生成ES批量写入数据，完全离线"""
if not os.path.exists(DATASET_FOLDER):
print(f"❌ 知识库目录 {DATASET_FOLDER} 不存在，请先创建目录并放入txt文档")
exit(1)

file_count = 0
for filename in os.listdir(DATASET_FOLDER):
if filename.endswith(".txt"):
file_path = os.path.join(DATASET_FOLDER, filename)
# 用UTF-8编码读取，避免中文乱码
with open(file_path, "r", encoding="utf-8") as f:
file_content = f.read()

yield {
"_index": INDEX_NAME,
"_source": {
"file_name": filename,
"file_content": file_content
}
}
file_count += 1

if file_count == 0:
print(f"❌ 知识库目录 {DATASET_FOLDER} 中没有找到任何txt文档，请先放入文档")
exit(1)
print(f"📄 找到 {file_count} 个文档，准备导入ES")

def bulk_index_documents():
"""批量把文档导入ES知识库，使用bulk API提升导入速度"""
try:
success_count, failed_count = helpers.bulk(
es_client,
load_documents_from_folder()
)
if failed_count:
print(f"⚠️ {failed_count} 个文档导入失败")
print(f"✅ 成功导入 {success_count} 个文档到知识库")
return success_count
except Exception as e:
print(f"❌ 文档导入失败，错误信息：{str(e)}")
exit(1)

def semantic_search(user_query, top_n=3):
"""ES原生语义检索，自动把用户问题转成向量，匹配最相关的文档，全程离线"""
start_time = time.time()
# 语义检索查询，一行代码完成，无需手动生成向量
search_query = {
"query": {
"semantic": {
"field": "semantic_content",
"query": user_query
}
},
"size": top_n,
"_source": ["file_name", "file_content"]
}

response = es_client.search(index=INDEX_NAME, body=search_query)
search_latency = (time.time() – start_time) * 1000 # 转换为毫秒
return response["hits"]["hits"], search_latency

def call_local_llm(prompt):
"""调用本地离线大模型生成答案，全程无外网调用"""
start_time = time.time()
try:
response = ai_client.chat.completions.create(
model=AI_MODEL_NAME,
messages=[{"role": "user", "content": prompt}],
temperature=0.3, # RAG场景使用低温度，减少幻觉，确保答案仅来自文档
stream=False
)
# 计算耗时和生成速度
llm_latency = (time.time() – start_time) * 1000
answer = response.choices[0].message.content
tokens_per_second = 0

# 计算token生成速度
if hasattr(response, "usage") and response.usage:
completion_tokens = response.usage.completion_tokens
if llm_latency > 0:
tokens_per_second = (completion_tokens / llm_latency) * 1000

return answer, llm_latency, tokens_per_second
except Exception as e:
llm_latency = (time.time() – start_time) * 1000
return f"大模型调用失败，错误信息：{str(e)}", llm_latency, 0

def build_rag_prompt(user_query, search_results):
"""构建RAG提示词，强制大模型仅使用检索到的文档内容，减少幻觉，带引文标注"""
context = ""
citations = []
for idx, hit in enumerate(search_results, 1):
source = hit["_source"]
context += f"[{idx}] 文档名称：{source['file_name']}\\n文档内容：{source['file_content']}\\n\\n"
citations.append(f"[{idx}] {source['file_name']}")

# 中文提示词，适配中文大模型，明确规则减少幻觉
prompt = f"""
请严格基于下面提供的参考文档，回答用户的问题，必须遵守以下规则：
1. 只使用参考文档里的内容回答问题，绝对不能编造参考文档里没有的信息
2. 每引用一句参考文档的内容，必须在后面标注来源，格式为 [1][2] 这样的编号
3. 回答条理清晰，分点说明核心内容，语言简洁准确
4. 如果参考文档里没有和问题相关的内容，直接回答“未在知识库中找到相关信息”

参考文档：
{context}

用户问题：{user_query}
"""
return prompt, citations

if __name__ == "__main__":
print("="*60)
print("🚀 【完全离线版】RAG个人知识库助手 启动")
print("="*60)

# 1. 检查ES连接
print("\\n===== 1. 检查Elasticsearch连接 =====")
check_es_connect()

# 2. 创建离线嵌入推理端点
print("\\n===== 2. 初始化离线嵌入模型 =====")
setup_embedding_inference()

# 3. 创建知识库索引
print("\\n===== 3. 初始化知识库索引 =====")
is_new_index = setup_knowledge_index()

# 4. 新创建的索引自动导入文档
if is_new_index:
print("\\n===== 4. 导入知识库文档 =====")
import_count = bulk_index_documents()
if import_count == 0:
exit(1)
time.sleep(1) # 等待索引刷新，确保文档可被检索

# 5. 执行RAG问答
print("\\n===== 5. 执行RAG智能问答 =====")
# 此处可修改为你要提问的问题，支持中文
USER_QUESTION = "请总结API存在的性能问题"
print(f"🔍 用户问题：{USER_QUESTION}")

# 语义检索
search_results, search_latency = semantic_search(USER_QUESTION, top_n=3)
if not search_results:
print("❌ 未在知识库中找到相关文档")
exit(0)
print(f"✅ 找到 {len(search_results)} 个相关文档，检索耗时：{search_latency:.0f}ms")

# 构建RAG提示词
rag_prompt, citations = build_rag_prompt(USER_QUESTION, search_results)

# 调用本地大模型生成答案
print(f"\\n🤖 正在调用本地大模型 {AI_MODEL_NAME} 生成答案…")
answer, llm_latency, tokens_per_second = call_local_llm(rag_prompt)

# 输出最终结果
print("\\n" + "="*60)
print(f"💡 问题：{USER_QUESTION}")
print(f"📝 答案：\\n{answer}")
print("\\n📚 引用来源：")
for cite in citations:
print(f" {cite}")
print("\\n" + "="*60)
print(f"🔍 语义检索耗时：{search_latency:.0f}ms")
print(f"🤖 大模型生成耗时：{llm_latency:.0f}ms | 生成速度：{tokens_per_second:.1f} tokens/s")
print("✅ 离线RAG问答执行完成")

6.4 运行脚本

# 进入脚本所在目录
cd /opt/rag-offline/
# 运行脚本
python3 rag_offline.py

【成功校验标准】

脚本正常执行，输出最终的答案、引用来源和性能数据，无报错，说明整个离线RAG系统部署成功！

七、离线环境常见报错与终极解决方案

八、离线环境进阶优化