欢迎，

本实践教程将教你如何使用 Elasticsearch 建立一个完整的搜索方法。其实，在本教程中，你将学习：

• 如何对数据集执行全文关键词搜索。并可选择使用过滤条件，
• 如何使用机器学习模型生成、存储和搜索密集向量嵌入。不过，
• 如何使用 ELSER 模型生成和搜索稀疏向量。
• 如何使用 Elastic 的 Reciprocal Rank Fusion算法，将上述方法的搜索结果进行融合。

使用者痛点：在实际项目中。往往缺少从基础开始搭建完整搜索程序的实战案例，导致“想实现却找不到完整步骤”。本教程通过一步步来的小步骤，帮助你在本地项目中一次性实现所有功能。

本教程最关键的一点是它将向你展示如何在一个运行于你自己计算机上的项目中实现所有这些功能，而且整个过程通过一步步来的小步骤完成。

你将学习的示例使用 Python 编写。但其中的概念是通用的，可以应用于你喜欢的任何编程语言或技术栈。

为了利用本教程。我们建议你跟着教程一起操作，并运行所有示例。怎么说呢，

前提条件

要完成本教程。你需要安装以下组件：

• 一个 Elasticsearch 安装运行环境，可以使用我们托管的 Elastic Cloud 服务，或者在你自己的计算机上运行一个自托管服务。安装说明请参阅上面的 “安装 Elasticsearch” 部分。如果你想使用 Elastic Cloud，请参考我之前录制的视频。在本展示中，我讲使用本地部署的 Elastic Stack 来进行展示。
• 一个 Python 解释器。请确保使用较新的版本，例如 Python 3.9 或更高版本。

使用者痛点：很多开发者对 Elasticsearch 的部署方式感到困惑——不确定是选云端还是自建，甚至不知道如何快速启动一个实例。老实说，本节提供了两种常见方案，让你快速决定并落地。

技术基础

本教程假定你此前没有 Elasticsearch 或搜索相关主题的知识，但希望你至少具备以下技术的基础了解：

• Python 开发
• Python 的 Flask Web 框架
• 操作程序中的命令提示符或终端应用程序

项目设置

下载入门应用程序

点击下面的链接下载搜索入门应用程序。

$ pwd
/Users/liuxg/python/tutorial
$ ls
search-tutorial-starter.zip

为你的项目选择一个合适的父目录。例如你的“文档”目录，并将压缩包内容解压到该目录中。解压后会生成一个名为 search-tutorial 的目录。其中包含多个子目录和文件：

$ pwd
/Users/liuxg/python/tutorial
$ tree -L 1
.
├── search-tutorial
│ ├── LICENSE
│ ├── README.md
│ ├── app.py
│ ├── data.json
│ ├── requirements.txt
│ ├── static
│ │ └── elastic-logo.svg
│ └── templates
│ ├── base.html
│ ├── document.html
│ └── index.html
└── search-tutorial-starter.zip

安装 Python 依赖项

1. 创建并激活虚拟环境：

cd search-tutorial # 切换到项目根目录
python3 -m venv .venv # 创建虚拟环境
# 激活虚拟环境
source .venv/bin/activate
# 若在 Windows PowerShell 中：
# .venv\Scripts\activate

*注意*: 虚拟环境仅在当前终端会话有效，关闭窗口后需重新激活。

2. 安装依赖包：

pip install -r requirements.txt
pip freeze> requirements.txt # 保持 requirements 同步

运行应用程序

flask run # 启动 Flask 开发服务器

A 浏览器访问  确认页面加载。此时页面仅显示空壳，暂无搜索结果，后续章节会逐步填充功能。

全文搜索实现步骤

设置 Elasticsearch Python 客户端

安装客户端库

pip install elasticsearch
pip list | grep elasticsearch # 验证已安装版本与 ES stack 匹配

连接到 Elasticsearch 服务

Create a new file /search-tutorial/search.py,n add following code:

import json
from pprint import pprint
import os
import time
from dotenv import load_dotenv
from elasticsearch import Elasticsearch
load_dotenv
class Search:
def __init__:
# 以下两段任选其一，# ---- Elastic Cloud 示例 -----
# self.es = Elasticsearch(
# cloud_id=os.getenv,# api_key=os.getenv
# )
# ---- 本地自托管示例 -----
self.es = Elasticsearch(
os.getenv,api_key=os.getenv,verify_certs=False,ssl_show_warn=False。)
client_info = self.es.info
print
pprint
# 基础 CRUD 方法将在后续章节补全...

The connection parameters are read from a hidden .env. Create it in project root:

# 对于 Elastic Cloud：
ELASTIC_CLOUD_ID=your_cloud_id_here
ELASTIC_API_KEY=your_api_key_here
# 对于本地部署：
ES_URL=https://localhost:9200 # 根据实际端口调整
ES_API_KEY=your_local_api_key_here # 如未启用，可留空但保持键名存在

Add file name to .gitignore,防止凭据泄露：

echo ".env">> .gitignore

测试连接是否成功

1. Python REPL 中导入并实例化：

>> from search import Search # 导入类
>> es = Search # 实例化，会打印连接信息
Connected to Elasticsearch!{'cluster_name': 'elasticsearch','name': 'my-node',...}
>>

If you encounter errors,double‑check your credentials in .env,ensure ES service is running and reachable.

把客户端整合进 Flask 应用

• Edit /search-tutorial/app.py 。add:

from search import Search
app = Flask
es = Search # 全局实例，可供所有路由调用
...
Now every request can use global `es` object.

创建索引 

Create Index 方法 
python
class Search:
...
def create_index:
self.es.indices.delete
self.es.indices.create


The index name is hard‑coded for simplicity;in real projects you may pass it as an argument.


Add Documents 方法 
python
def insert_document:
return self.es.index
def insert_documents:
ops =
for doc in documents:
ops.append
ops.append
return self.es.bulk


Bulk API dramatically reduces indexing time compared with single‑document calls.


python
def reindex:
self.create_index
with open as f:
docs = json.load
return self.insert_documents


Add a Flask CLI command for easy re‑indexing:

python
@app.cli.command
def reindex:
"""Regenerate Elasticsearch index."""
resp = es.reindex
print} docs in {resp} ms')


User Pain Point: 手动删除旧索引、重新创建、批量导入很繁琐；通过 CLI 一键搞定，


基本全文检索实现 

Create a simple search wrapper 

python
class Search:
...
def search:
return self.es.search



Match Query

Edit app.py → handle_search:

python
@app.post
def handle_search:
query = request.form.get
results = es.search(
query={'match': {'name': {'query': query}}}
)
hits = results
total = results
return render_template('index.html'。results=hits,query=query,from_=0,total=total)


Now you can search by document title . 

Pain point: 初始实现只能匹配标题，很多业务需求需要同时匹配正文与摘要，这里仅演示了最基础的单字段匹配。


Multi‑Match

Upgrade to multi‑match so title、summary、content 都参与检索：

python
@app.post
def handle_search:
query = request.form.get
results = es.search(
query={'multi_match':
{'query': query,'fields': }}
)
...



Pagination

Add size & from_ parameters :

python
@app.post
def handlesearch:
query = request.form.get
from = request.form.get
results = es.search(
query={...},# 上面的 multi_match 块保持不变
size=10,from_=from_
)


...


Update templates/index.html to display pagination controls:

+jinja2

Showing {{ from_ + 1 }}–{{ from_ + results|length }} out of {{ total }}
{% if from> 0 %}
{% endif %}
{% if from + results|length 



Next page →
---
### Filter Support
#### Boolean Query Skeleton
json5
{
再看"bool"。{
"must"这方面，"filter":
}
}
#### Extract filter expressions from user input
Add helper in **app.py**:
python
def extract_filters:
filters=
# category filter e.g. "category:github"
cat_match=re.search',query)
if cat_match:
filters.append({
"term"的观点是，{"category.keyword":{"value":cat_match.group}}
})
query=re.sub.strip
# year filter e.g. "year:2022"
yr_match=re.search',query)
if yr_match:
year=yr_match.group
filters.append({
"range":{"updated_at":{
"gte"这方面，f"{year}||/y","lte":f"{year}||/y"
}}
})
query=re.sub.strip
return {"filter":filters},query
#### Integrate filters into search endpoint
python
@app.post
def handle_search:
raw_query=request.form.get
filters,clean_query=extract_filters
from_=request.form.get
if clean_query:
must_clause={"multi_match":
{"query":clean_query,"fields":}}
else的观点是，must_clause={"match_all":{}}
body={"bool":{"must":must_clause,"filter":filters}}
resp=es.search
...
**Pain point:** 使用者常抱怨“只能全局搜索，无法按分类或时间筛选”。通过正则解析过滤语法，实现了轻量级、可
的过滤机制。---
### Match‑All fallback
When only filters are supplied we replace `multi_match` with `match_all` so that filter alone returns results.
The updated endpoint already handles this case .
---
## Faceted Search
Facets give users quick access to common filter options toger with result counts.
### Terms Aggregation for Category
Add aggregations to same search call:
python
resp=es.search(
query=body,aggs={
"category-agg":{"terms":{"field":"category.keyword"}},"year-agg":{
"date_histogram":{
"field":"updated_at","calendar_interval":"year","format":"yyyy"
}
}
},size=10,from_=from_
)
### Transform aggregation response for template rendering
python
aggs={
"Category":{
bucket的观点是。bucket
for bucket in resp
},"Year":{
bucket这方面，bucket
for bucket in resp
if bucket> 0 # skip empty years
}
}
return render_template("index.html",results=resp,query=raw_query,from_=from_,total=resp,aggs=aggs)
### Template snippet – renders facets as clickable forms
+jinja2
{% for agg_name,buckets in aggs.items %}

{{ agg_name }}
{% for key,count in buckets.items %}

{% set new_query=query ~ ' ' ~ agg_name|lower ~ ':' ~ key %}
{{ new_query }}
{{ new_query }}
{{ key }}


{% endfor %}
{% endfor %}


Pain point: 手动统计每个分类/年份数量非常耗时；其实，聚合一次请求即可返回全部统计信息，明显提高交互体验。说起来，


完整代码概览

Below is a condensed view of all files after completing tutorial steps.

app.py

python
import re
from flask import Flask,render_template,request
from search import Search 

app = Flask
es = Search 

def extract_filters:
filters= 

------ 

@app.get
def index:
return render_template 

@app.post
def handlesearch:
rawquery=request.form.get
filters,cleanquery=extractfilters 

--- 构造 bool 查询体 --- 

results=es.search
aggs={...}
return render_template 

@app.get
def getdocument:
doc=es.retrievedocument
title=doc
paragraphs=doc.split
return render_template("document.html",title=title。paragraphs=paragraphs) 

if name=='main':
app.run


search.py

python
import json
import os
import time
from dotenv import load_dotenv
from elasticsearch import Elasticsearch 

load_dotenv 

class Search:
def init:
self.es = Elasticsearch(
os.getenv,apikey=os.getenv,verifycerts=False,sslshowwarn=False
) 
 info=self.es.info
print
print
def create_index:
self.es.indices.delete(index='my_documents',ignore_unavailable=True)
self.es.indices.create
def insert_document:
return self.es.index
def insert_documents:
ops=
for d in docs:
ops.append
ops.append
return self.es.bulk
def reindex:
self.create_index
with open as f:
docs=json.load
return self.insert_documents
def retrieve_document:
return self.es.get
def search:
return self.es.search


templates/index.html

+jinja2
…输入框 ，
Search​︎​︎​︎​︎​︎​︎​︎​︎​︎​︎​︎​​‍​​️​​️​​️​​​‍​​️​​​​​​​‍​​​​️​​‏‎​​​‏‏‎‏‍‏‌‍‌‎‮⁤‪‬‌‫‏‌⁦‪‪⁠‌‌‌‌​​​​‫‎‬ ‬⠀  ‌
{% endif %} 


{% for result in results %} 

\
{{ result._source.summary }}\
<\/span>\
…

{% endfor %} 

{% if from_>0 %} …{% endif %} 

{% if from_ + results|length 
{% endif %} 

…


小结 & 常见痛点回顾



痛点
本教程对应方法




不知道怎么快速启动 ES 环境
提供 Cloud 与自托管两套详细步骤


手动批量导入慢且易出错
使用 Bulk API 一键导入全部文档


只支持单字段检索，业务需求不满足
Multi‑Match 跨字段检索


结果太多。没有分页
Size + From 参数 + 前后页按钮


缺少筛选能力，只能全局搜索
正则解析 “category:” 与 “year:” 并转成 Bool Filter


只靠关键词检索，不知道哪些筛选项受欢迎
Terms & Date Histogram 聚合实现 Faceted UI


部署后每次改动都要手动重建索引
Flask CLI reindex 命令一键刷新




通过以上步骤，你已经拥有了：


可运行于本地或云端的完整 Elastic 搜索服务；

基础 CRUD、批量导入、全文检索、多字段查询；

分页、过滤、还有动态聚合；

与 Flask 前端完美结合，可直接打开浏览器交互。不过，


现在可以继续探索更高级特性。如向量相似度检索、ELSER 稀疏模型、RRF 融合等。祝开发顺利，

。

---