在 Elasticsearch 中尝试使用 RAG（检索增强生成）
首先
我相信很多人都在ChatGPT这个生成人工智能上投入了很多关注，并尝试了各种使用方式。生成人工智能的用途广泛，既有非常有效的方面，也存在一些不太理想的地方。
例如，对于文章摘要、程序编写和翻译等已经具备可以在工作中使用的水平的事物，有很多。与此相反，它对于最新信息和尚未在互联网上出现的机密信息相对较薄弱。
在这里，我们将解释如何使用Elasticsearch实现用于生成人工智能的最新信息和机密信息的RAG（检索增强生成）方法。
这篇内容我们也在我们公司的网络研讨会上公开了。
https://www.elastic.co/jp/virtual-events/delivering-generative-ai
如果您已经熟悉RAG和Elasticsearch，您只需查看下面的存储库即可理解。
-
- Elasticsearchの公式リポジトリ
https://github.com/elastic/elasticsearch-labs
Jupyter Notebookのシンプルな実装
https://github.com/elastic/elasticsearch-labs/blob/main/notebooks/integrations/openai/openai-KNN-RAG.ipynb
Elasticsearch社員のリポジトリ
https://github.com/jeffvestal/ElasticDocs_GPT/
こちらはこのブログで紹介されているものです
https://www.elastic.co/search-labs/chatgpt-elasticsearch-openai-meets-private-data
https://github.com/legacyworld/esre/
今回解説に使うリポジトリ
RAG和Elasticsearch
以下是对RAG（Risk Assessment Grid）的流程的大致概述：
首先，假设公司内部的文本数据（存储在文件服务器或云存储中）已被索引到Elasticsearch中。
-
- 用户输入搜索文本
-
- 在Elasticsearch内进行搜索
-
- 可以进行普通搜索（关键词搜索）或者向量搜索，甚至可以进行混合搜索
-
- Elasticsearch将按相关性返回结果
-
- 从前面的结果中提取一些，将其作为上下文附加到问题文本中一起发送给生成AI（此次为OpenAI）
- 生成AI根据上下文和问题文本返回结果
在上述的例子中，加入確定拠出年金的方法因每家公司而異（如證券公司或開始時間），用戶公司的步驟已被索引到Elasticsearch中。從那裡，我們可以提取包括證券公司和帳戶創建方法在內的文件並請AI生成摘要。
如果在搜索部分得不到正確的結果，那麼生成的AI回答也就不會正確。此外，用戶的權限會決定哪些信息可以公開，哪些不能（例如一般員工不能查看管理相關信息等）。這也是RAG的困難之處，也是搜索領域老牌工具Elasticsearch的用武之地。
构成
简图
运动环境
只要Docker能夠運行，環境就不是問題。
操作系统：Linux（ubuntu 20.04）
虛擬機：Azure Standard DS1 v2
Docker 版本：24.0.5
Docker Compose 版本：v2.20.2
Elasticsearch（Elastic Cloud）版本：8.10.3
文档
正在索引NewsAPI的文章。
https://newsapi.org/
查询语句也按照这个结构进行设置。
操作步骤
请参考以下链接：https://github.com/legacyworld/esre/blob/main/README.md，其中有关于README的详细说明。
在Elastic Cloud上创建集群。
根据这篇博客，我们将在Elastic Cloud上部署一个集群。
链接：https://qiita.com/tomo_s_el/items/3584d0b1fabb0bafa4fa
转换结构
（注意）所截屏幕为2023年10月的时间点。
扩展进行添加
增加机器学习节点
设定
一旦完成弹性云的准备工作后，将进行模型的上传、映射设置以及文档的投入等操作。
将在运行RAG的环境中进行工作。
克隆 Git
git clone https://github.com/legacyworld/esre/
.env文件
进入esre文件夹，使用以下内容创建.env文件。newsapi_key是可选的。
openai_api_key=<openapi key>
openai_api_type=azure
openai_api_base=<openapi base url>
openai_api_version=<openapi version>
openai_api_engine=<openapi engine>
cloud_id=<cloud id of Elasticsearch Cluster>
cloud_pass=<Cloud pass of Elasticsearch Cluster>
cloud_user=<Cloud User. Normally it is elastic>
search_index=<your index name>
newsapi_key=<newsapi key>
每个人的确认方法
构建和启动Docker镜像
docker compose up -d
请使用docker compose ps和docker compose logs来确认您已经启动。
$ docker compose ps
NAME IMAGE COMMAND SERVICE CREATED STATUS PORTS
esre_flask esre-esre_flask "python3 -u app.py" esre_flask 4 hours ago Up 3 hours 0.0.0.0:4000->4000/tcp
如果在docker compose logs中显示以下内容，则表示成功连接到Elastic Cloud。
{'name': 'instance-0000000000', 'cluster_name': 'e48072429f3d44beb9286fecc64f4529', 'cluster_uuid': 'u0PuN1DCSN-yVJZhowZ_mw', 'version': {'number': '8.10.3', 'build_flavor': 'default', 'build_type': 'docker', 'build_hash': 'c63272efed16b5a1c25f3ce500715b7fddf9a9fb', 'build_date': '2023-10-05T10:15:55.152563867Z', 'build_snapshot': False, 'lucene_version': '9.7.0', 'minimum_wire_compatibility_version': '7.17.0', 'minimum_index_compatibility_version': '7.0.0'}, 'tagline': 'You Know, for Search'}
将模型上传并初始化索引。
如果执行initialize.sh脚本，则所有操作将自动完成。但是，请注意，执行环境已在启动的Docker容器中构建，因此请在进入容器后执行。
通过查看Dockerfile的WORKDIR指令，您可以了解到初始位置在/src目录下。
docker exec -it esre_flask /bin/bash
python ./initialize.sh
进行中的事项如下所示。
导入机器学习模型
请下载在Hugging Face上的东北大学模型，并上传到机器学习节点。
我们正在使用这个模型。
您可以在以下链接找到该模型：
https://huggingface.co/cl-tohoku/bert-base-japanese-char-v2
在initialize.sh脚本的下面部分，进行上传操作。
这一步需要几分钟时间，请您喝杯茶稍作等待。
eland_import_hub_model \
--cloud-id $cloud_id \
-u $cloud_user \
-p $cloud_pass \
--hub-model-id cl-tohoku/bert-base-japanese-v2 \
--task-type text_embedding \
--start
eland是用于在Elasticsearch中使用机器学习模型的工具。官方文档链接如下：https://www.elastic.co/guide/en/elasticsearch/client/eland/current/index.html
有关如何在eland上上传机器学习模型的方法请参考以下链接：https://www.elastic.co/guide/en/machine-learning/current/ml-nlp-import-model.html
创建摄取管道
在Elasticsearch中，Ingest Pipeline是用于定义在将文档输入到Elasticsearch之前进行的预处理的部分。
官方文档链接如下：
https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html
Elasticsearch的一个优点是不需要在外部创建Dense Vector，而是可以在文档投入的同时进行嵌入（这是ESRE的一部分）。
本次设定将使用东北大学的模型进行Dense Vector Embedding。在该模型中，将给定的文章转换为768维向量并写入索引中。
在initialize.sh中调用了create_index.py，其中的以下部分是用于创建流水线的代码。
pipeline_id = "japanese-text-embeddings"
body = {
"description": "Text embedding pipeline",
"processors": [
{
"inference": {
"model_id": "cl-tohoku__bert-base-japanese-v2",
"target_field": "text_embedding",
"field_map": {
"title": "text_field"
}
}
}
]
}
print(es.ingest.put_pipeline(id=pipeline_id,body=body))
目标字段和字段映射是固定值。
创建Mapping。
我們將進行設置，以便進行kuromoji的日本語形態素解析和密集向量場的向量檢索。
词法分析
中文全文搜索需要进行特殊的词法分析处理。详细说明可在此网址找到：
https://www.elastic.co/jp/blog/how-to-implement-japanese-full-text-search-in-elasticsearch
除了使用kuromoji之外，有时也会使用另一种名为sudachi的词法分析工具。
分析器/分詞器設定
以下是create_index.py中用于设置形态分析器/分词器的部分。在实际运行中，我们可能需要添加N-Gram字段，或创建同义词字典。
'settings': {
'index': {
'analysis': {
'char_filter': {
'normalize': {
'type': 'icu_normalizer',
'name': 'nfkc',
'mode': 'compose'
}
},
'tokenizer': {
'ja_kuromoji_tokenizer': {
'mode': 'search',
'type': 'kuromoji_tokenizer'
}
},
'analyzer': {
'kuromoji_analyzer': {
'tokenizer': 'ja_kuromoji_tokenizer',
'filter': [
'kuromoji_baseform',
'kuromoji_part_of_speech',
'cjk_width',
'ja_stop',
'kuromoji_stemmer',
'lowercase'
]
}
}
}
}
},
在各个领域中的配置设定(映射)
在create_index.py文件中的下面部分是Mapping设置。
'mappings': {
'properties': {
'author': {
'type': 'text',
'analyzer': 'kuromoji_analyzer'
},
...
'text_embedding': {
'properties': {
'model_id': {
'type': 'text',
'fields': {
'keyword': {
'type': 'keyword',
'ignore_above': 256
}
}
},
'predicted_value': {
'type': 'dense_vector',
'dims': 768,
'index': True,
'similarity': 'cosine'
}
}
},
...
文件投放
esre/data文件夹下放置了从NewsAPI获取的一些JSON文件。暂时先按原样进行索引。
docker exec -it esre_flask /bin/bash
cd data
./load_all.sh
确认数据
让我们从Kibana中检查投入的数据。
在登录进入云页面后，点击创建的群集，即可打开Kibana的界面。
创建Dataview
让我们尝试玩RAG吧
-
- 最初の3つの検索は非常に早い
-
- OpenAI検索は時間がかかる
- 最新情報はわからない
在中国，常规搜索和向量搜索都非常快速（混合搜索也一样快）。因此，在华为智能问答系统（RAG）中，设计这一部分非常重要。
现在我们来点击OpenAI中的“Hybrid Search”进行摘要。
显示出“目前普通汽油的零售价格为每升186.5日元。”
可以看出我们获得了所需的信息。
在这个例子中，通常搜索是有优势的，但是向量搜索也可以产生良好的结果。
我认为互相补充的使用方式可能会获得良好的结果。
代碼解說
app.py是主要的部分。由于部分使用了Flask（以及默认的jinja2渲染功能），所以在index.html中包含了一些渲染部分，但这并非其核心部分。
只讲重点概括说明。
以下解析是在假设读者已经有关于Web框架Flask的知识的前提下编写的（也就是说没有对该部分进行解释）。
弹性云连接
cloud_id = os.environ['cloud_id']
cloud_pass = os.environ['cloud_pass']
cloud_user = os.environ['cloud_user']
es = Elasticsearch(cloud_id=cloud_id, basic_auth=(cloud_user, cloud_pass),request_timeout=10)
print(es.info())
首先，读取.env文件中连接到Elastic Cloud所需的信息。如果在docker的日志中输出了print语句显示的内容，则表示连接成功。
搜索准备
search_index = os.environ['search_index']
bm25_search_fields = ["title", "description"]
bm25_result_fields = ["description", "url", "category", "title"]
在搜尋時，search_field是用來指定要在哪個欄位進行搜索的。這裡是根據NewsAPI的結構進行設定的。
返回的值是根據result_field指定的欄位。通常可以利用這個來創建Facet。
一般在內部文檔搜索中，如果存在文檔到url的連結，我認為也會包含在其中。
def get_rrf_search_request_body(query, search_fields, result_fields, size=10):
return {
'_source': False,
'fields': result_fields,
'size': size,
"query": {
"multi_match": {
"query": query,
"fields": search_fields
}
},
"knn": {
"field": "text_embedding.predicted_value",
"k": 10,
"num_candidates": 100,
"query_vector_builder": {
"text_embedding": {
"model_id": "cl-tohoku__bert-base-japanese-v2",
"model_text": query
}
}
},
"rank": {
"rrf": {
"window_size": 50,
"rank_constant": 20
}
}
}
这个函数用于创建执行混合搜索的JSON。
其中的query部分是用于常规搜索的，knn是用于向量搜索，rrf是用于排名创建。
还有用于常规搜索和向量搜索的JSON也是通过get_text_search_request_body和get_vector_search_request_body来创建的。
由于设置了size=10，所以结果只显示10个。
执行搜索
def get_es_result():
query = request.args['var1']
bm25_body = get_text_search_request_body(query,bm25_search_fields,bm25_result_fields)
bm25_result = es.search(index=search_index, query=bm25_body["query"], fields=bm25_body["fields"], size=bm25_body["size"], source=bm25_body["_source"])
bm25_documents = bm25_result['hits']['hits'][:10]
查询文来自 base.html 和 index.html 的这部分。这是使用旧的 JQuery 方法。
在 Elasticsearch-labs 中，有一个很酷的实现是使用 React。
<script>
$(function () {
$('#btn').click(function () {
console.log($('#text').val());
let url = "/api/search_results?var1=" + $('#text').val()
console.log(url)
window.location.href = url;
})
});
</script>
当接收到查询文时，我们使用es.search进行搜索，并通过JSON格式接收结果。由于返回结果有10个，因此将其存储在bm25_documents中。
bm25_all = []
for hit in bm25_documents[:3]:
temp_contents = {'url': hit['fields']['url'][0],'title': hit['fields']['title'][0],'description': hit['fields']['description'][0]}
bm25_all.append(temp_contents)
正在从返回结果中提取前三个最佳结果。
将一个结果存储为字典格式，并将其存储在一个数组中的三个位置。
同样的操作也适用于向量搜索和混合搜索，以确保所有搜索结果中的前三个内容一致。
最后，
return {‘bm25’: bm25_all, ‘vector’: vector_all, ‘rrf’: rrf_all, ‘openai_answer’: “”}
将每个内容都以字典格式存储起来。在这里，由于尚未向OpenAI提出问题，openai_answer中没有任何内容。
展示搜索结果
在index.html文件中，以下部分被显示出来。
<div class="col text-center" style="border:solid 2px">
<p class="text-center">通常検索</p>
<button type="button" class="btn btn-primary" id="bm25" style="margin: 5px;">OpenAIで要約</button>
{% for item in all.bm25: %}
<div style="border:solid 1px; margin: 10px">
<p>{{ item.title }}</p>
<p>{{ item.description }}</p>
<a href="{{ item.url }}">View Documents</a>
</div>
{% endfor %}
</div>
这是根据Jinja2模板的基本方法。通过这种方式，将显示出每个部分的前三名。
给OpenAI发送查询
這是一個將問題直接提交給OpenAI而不是RAG的部分。
def get_all_results():
query = request.args['var1']
response = get_es_result()
openai_answer = ""
if request.args['var2'] == "openai":
prompt = query
messages = {"message": [{"role": "system", "content": prompt}]}
completion = completion_with_backoff(engine=engine,temperature=0.2,messages=messages["message"])
openai_answer = completion["choices"][0]["message"]["content"]
只需要一种选择，用汉语来解释以下内容：
提示：我认为您已经看到了这个请求。
在这里启动的是一个名为UI with OpenAI Search的按钮，它对应于index.html文件中的以下部分。
（console文已被省略）。
我仍然在强行使用JQuery来实现。
$('#openai').click(function () {
let url = "/api/all_results?var1=" + $('#text').val() + "&var2=openai"
window.location.href = url;
});
为了避免保留过去的结果变得繁琐，现在我们重新对Elasticsearch进行了搜索（如果查询语句没有改变，结果将保持不变）。
垃圾
好的，现在就是实际考验的时候了。我将Top3的内容（仅描述部分）与上下文一起提交给OpenAI。
def route_api_stream():
all = get_es_result()
all['openai_answer'] = request.args['var3']
query = request.args['var1']
...
documents = all[request.args['var2']]
for item in documents:
prompt += f"Description: {item['description']}"
prompt += f"\nQuestion: {query}"
truncated_prompt = truncate_text(prompt, max_context_tokens - max_tokens - safety_margin)
messages = {"message": [{"role": "system", "content": "Given the following extracted parts of a long document and a question, create a final answer. If you don't know the answer, just say '検索対象からは回答となる情報が見つかりませんでした'. Don't try to make up an answer."},{"role": "system", "content": truncated_prompt}]}
completion = completion_with_backoff(engine=engine,temperature=0.2,messages=messages["message"],stream=True)
每次页面转换都会获取搜索结果（保持这些结果很麻烦）。
根据 request.args[‘var2’] 决定要对常规搜索、向量搜索还是混合搜索进行总结。
在第一个 for 循环中将前三个描述放入提示中，最后再添加查询文本。
如果描述过长，truncated_prompt 会将其截断为不超过 max_context_token 的长度（在此源代码中为4000）。
然后只需要构建 messages 并调用 completion_with_backoff。
注意温度等已被固定，请根据需要进行更改。
混乱的部分是用来实时显示OpenAI的回答的设计。大家可能已经注意到了，在ChatGPT中它是逐渐显示出来的（尽管之间有jinja2的处理，没有那么流畅）。
在中文中进行重述：这里启动的部分是一个名为“OpenAI要点概括”的按钮。由于向OpenAI再次普通地提问会花费时间，因此我们直接从以下的HTML中提取信息。
$('#bm25').click(function () {
let url = "/api/completions?var1=" + $('#text').val() + "&var2=bm25&var3=" + $('#openai_answer').text()
window.location.href = url;
});
太容易了。
如果将界面简化并仅限于常规搜索，这些内容只需几十行就能撰写完成。
总结
如果您查看源代码，就可以理解到为了将上下文加入内容（即搜索结果），实际上将其分成了两个完全独立的部分，一个是准备部分，一个是实际提交给OpenAI的部分。
试用RAG时需要考虑到这两个方面。如果您想先尝试一下简单的通常搜索，可以尝试按以下步骤逐渐升级，看看怎么样呢？
-
- 通常検索だけでとりあえずTopNの結果でRAGを行う
-
- 検索結果のチューニングをしてより正確な結果が得られるようにする
形態素解析を工夫したり、同義語辞書を作ったり、文章は短いけど重要なフィールドであるタイトルだけ重みを増やす、などいろいろあります
ベクトル検索で通常検索では拾いきれない部分を補完する
ベクトル検索のチューニングもしてみる
モデルを変えてみたり、ファインチューニングしたり
Prompt Engineering頑張る
コンテキストに良い内容が入れられるようになれば、より良い回答を生成AIから引き出す工夫をする
如果这篇文章能帮助你欣然利用生成AI，将会使我感到非常愉快。
