制定藍圖：什么樣的API策略能夠確保未來的成功？
本著開放的精神，我們引入兩種開源技術來輔助 Elasticsearch：Hugging Face 轉換器庫和新穎有趣的 Python 庫 LangChain，它們可以加快作為矢量數據庫的 Elasticsearch 的工作速度。此外，LangChain 還有一個優勢，我們的 LLM 一旦設置好，就可以通過編程實現互換，讓我們可以自由地試驗各種模型。
LangChain 是什么？LangChain 是一個 Python 和 JavaScript 框架，用于開發由大型語言模型提供支持的應用程序。LangChain 將與 OpenAI 的 API 配合使用，但它也擅長于抽象化數據庫和 AI 工具之間的差異。
ChatGPT 本身在星球大戰問答方面表現不錯。不過，它的訓練數據集已經是幾年前的數據，而我們需要的是最新電視節目和活動中有關星球大戰宇宙的答案。還有一點要注意，我們是假裝這些數據過于私密，不能與云中的大型 LLM 共享。我們可以自己用較新的數據調整大型語言模型，但有一種簡單得多的方法可以實現這一點，而且還能讓我們始終使用最新的數據。
現在介紹一種規模較小、易于自托管的 LLM。我用 Google 的 flan-t5-large 模型得到了不錯的結果，它可以很好地從注入的上下文中解析出答案，彌補了訓練不足的缺陷。我們將使用語義搜索對私有知識進行檢索，然后將上下文和一個問題注入到我們的私有 LLM 中。
此外這里也有一些流行的大語言模型API可供使用：
1.從 Wookieepedia 中抓取所有規則文章，并將數據放入暫存的 Python Pickle 文件中。
2A.使用 LangChain 內置的 Vectorstore 庫將這些文章的每個段落加載到 Elasticsearch 中。
2B.或者，我們可以將 LangChain 與在 Elasticsearch 中托管 pytorch 轉換器的新方法進行比較。我們將把文本嵌入模型部署到 Elasticsearch 中，以利用分布式計算并加快這個過程。
3.問題傳入后，我們將使用 Elasticsearch 的矢量搜索找到與問題在語義上最相似的段落。然后，選取該段落并將其添加到一個本地小型 LLM 的提示中，作為這個問題的上下文，然后讓神奇的生成式 AI 針對我們的問答題目給出簡短回答。
確保您的計算機上安裝了 Python 3.9 或類似版本。我使用 3.9 版本是為了更容易與 GPU 加速庫兼容，但這個項目不需要這樣做。任何最近的 3.X 版本的 Python 都可以。
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install beautifulsoup4 eland elasticsearch huggingface-hub langchain tqdm torch requests sentence_transformers如果您已經下載了示例代碼，可以使用下面的 pip install 命令拉取我使用的代碼的確切版本。
pip install -r requirements.txt您可以按照此處的說明設置 Elasticsearch 集群。最簡單的入門方法是使用免費的云試用版。
在文件夾中創建一個 .env 文件，然后為 Elasticsearch 加載連接詳細信息。
export ES_SERVER="YOURDESSERVERNAME.es.us-central1.gcp.cloud.es.io"
export ES_USERNAME="YOUR READ WRITE AND INDEX CREATING USER"
export ES_PASSWORD="YOUR PASSWORD"代碼存儲庫中有一個小型數據集，位于 Dataset/starwars_small_sample_data.pickle。如果您能接受小規模的任務，可以跳過這一步。
抓取代碼改編自?Dennis Bakhuis 的優秀數據科學博客和項目，去看看吧！他只拉取了每篇文章的第一段，而我修改了代碼，拉取了全部內容。他可能需要將數據保持在適合主內存的大小，但我們沒有這方面的問題，因為我們有 Elasticsearch，它可以將內存擴展到 PB 級別。
您也可以在這里輕松地插入自己的私有數據源。LangChain 有一些優秀的實用程序庫，可以將文本數據拆分成更短的塊。
抓取不是本文的重點，如果您想自己小規模運行，可以看看 Python 記事本，或者下載源代碼并運行以下內容：
source .env
python3 step-1A-scrape-urls.py
python3 step-1B-scrape-content.py完成后，您應該能夠查看如下所示的已保存的 Pickle 文件，確保它正常工作。
from pathlib import Path
import pickle
bookFilePath = "starwars_*_data*.pickle"
files = sorted(Path('./Dataset').glob(bookFilePath))
for fn in files:
with open(fn,'rb') as f:
part = pickle.load(f)
for key, value in part.items():
title = value['title'].strip()
print(title)如果您跳過了網頁抓取，只需將 bookFilePath 更改為“starwars_small_sample_data.pickle”，即可使用我在 GitHub 存儲庫中提供的示例。
完整的代碼展示了我是如何只使用 LangChain 來實現這一操作的。代碼的關鍵部分是循環遍歷已保存的 Pickle 文件（如上面的示例），提取出作為段落的字符串列表，然后將它們傳遞給 LangChain Vectorstore 的 from_texts() 函數。
from langchain.vectorstores import ElasticVectorSearch
from langchain.embeddings import HuggingFaceEmbeddings
from pathlib import Path
import pickle
import os
from tqdm import tqdm
model_name = "sentence-transformers/all-mpnet-base-v2"
hf = HuggingFaceEmbeddings(model_name=model_name)
index_name = "book_wookieepedia_mpnet"
endpoint = os.getenv('ES_SERVER', 'ERROR')
username = os.getenv('ES_USERNAME', 'ERROR')
password = os.getenv('ES_PASSWORD', 'ERROR')
url = f"https://{username}:{password}@{endpoint}:443"
db = ElasticVectorSearch(embedding=hf, elasticsearch_url=url, index_name=index_name)
batchtext = []
bookFilePath = "starwars_*_data*.pickle"
files = sorted(Path('./Dataset').glob(bookFilePath))
for fn in files:
with open(fn,'rb') as f:
part = pickle.load(f)
for ix, (key, value) in tqdm(enumerate(part.items()), total=len(part)):
paragraphs = value['paragraph']
for p in paragraphs:
batchtext.append(p)
db.from_texts(batchtext,
embedding=hf,
elasticsearch_url=url,
index_name=index_name)利用 Elasticsearch 向量存儲進行檢索增強生成 (RAG)
檢索增強生成利用外部知識來增強語言模型生成的響應。以下是有關如何使用 Elasticsearch 向量存儲進行 RAG 的指南。
循序漸進教程：整合外部知識
RAG 問答
理解檢索概念
其他資源
通過遵循這些準則，您可以有效地利用 Elasticsearch 的向量存儲來改進檢索增強生成，從而獲得更準確、更具上下文感知的語言模型輸出。
為了增強您對 Elasticsearch 向量存儲進行檢索增強生成 (RAG) 的理解和實踐技能，請探索這些綜合指南：
了解如何將來自各種數據源的信息整合到您的 Elasticsearch 向量存儲中，以實現有效的檢索增強生成。這些分步教程將幫助您掌握集成過程。
深入了解這些詳細的操作指南，提升您的問答能力。它們涵蓋了您需要了解的所有內容，從設置環境到微調系統以獲得最佳問答性能。
通過探索這些文檔加深您對概念的理解。它們提供了對檢索理論方面的見解，幫助您掌握基本原理并改進您的實際實施。
通過遵循這些指南，您將能夠利用 Elasticsearch 向量存儲來實現各種高級檢索增強生成應用程序。
要將 Elasticsearch 向量存儲轉換為檢索器，請遵循以下簡明步驟：
as_retriever在您的矢量存儲中利用該方法。search_type和search_kwargs。retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.2}
)invoke與您的查詢相關的方法來獲取相關文檔。results = retriever.invoke("Stealing from the bank is a crime")檢索器將返回滿足指定相似度分數閾值的相關文檔。
[
Document(metadata={'source': 'news'}, page_content='Robbers broke into the city bank and stole $1 million in cash.'),
Document(metadata={'source': 'news'}, page_content='The stock market is down 500 points today due to fears of a recession.'),
Document(metadata={'source': 'website'}, page_content='Is the new iPhone worth the price? Read this review to find out.'),
Document(metadata={'source': 'tweet'}, page_content='Building an exciting new project with LangChain - come check it out!')
]通過遵循這些步驟，您可以將 Elasticsearch 向量存儲轉變為高效的檢索器，能夠根據與輸入查詢的相似性獲取相關文檔。
我發現，在我的舊版 Intel Macbook 上，創建嵌入模型需要處理好幾個小時。我還是把時間說短了，好像需要好幾天才能完成。我覺得使用 Elastic 托管服務中可動態擴展的 Machine Learning (ML) 節點，可以更快、更經濟地完成這項工作。免費試用版中的集群不支持擴展該層，因此，這一步對某些用戶可能更有意義。
最后的結果是：這種方法在 Elastic Cloud 中運行的節點上耗時 40 分鐘，成本為 5 美元/小時，這比我在本地完成的速度要快得多，而且與 OpenAI（目前按 token 計費）處理嵌入模型的成本不相上下。如何高效地完成是一個更大的話題，但有一點我非常滿意，那就是我可以快速得到一個在 Elastic Cloud 中運行的并行推理管道，而不需要學習新技能或將數據交給非私有 API。
在這一步中，我們將把嵌入生成的任務轉移給 Elasticsearch 集群，它可以托管嵌入模型，并以分布式的方式嵌入文本段落。為此，我們必須加載數據并使用采集管道，確保最終的形式與 LangChain 所用的索引映射一致。在 Kibana 的開發工具中運行以下 REST 命令：
PUT /book_wookieepedia_mpnet
{
"settings": {
"number_of_shards": 4
},
"mappings": {
"properties": {
"metadata": {
"type": "object"
},
"text": {
"type": "text"
},
"vector": {
"type": "dense_vector",
"dims": 768
}
}
}
}接下來，使用 eland Python 庫將嵌入模型上傳到 Elasticsearch。
source .env
python3 step-3A-upload-model.py下面，我們進入 Elastic Cloud 控制臺，將我們的 ML 層擴展到總共 64 個 vCPU（是我目前筆記本電腦性能的 8 倍）。
現在，我們將在 Kibana 中部署訓練好的 ML 模型。性能測試表明，在大規模部署時，用戶應該從每個模型分配 1 個線程開始，然后逐步增加分配數量以提高吞吐量。可在此處找到相關文檔和指南。我進行了試驗，對于這個較小的數據集，我用 32 個實例（每個實例 2 個線程）得到了最佳結果。要設置這一項，請前往“堆棧管理”>“Machine Learning”。使用同步保存的對象功能，讓 Kibana 看到我們用 Python 代碼推送到 Elasticsearch 中的模型。然后，在單擊模型后出現的菜單中進行部署。
現在，我們再次使用開發工具創建一個新的索引和采集管道，該管道用于處理文檔中的文本段落，將結果放入名為“vector”的密集矢量字段中，并將該段落復制到預期的“text”字段中。
PUT /book_wookieepedia_mpnet
{
"settings": {
"number_of_shards": 4
},
"mappings": {
"properties": {
"metadata": {
"type": "object"
},
"text": {
"type": "text"
},
"vector": {
"type": "dense_vector",
"dims": 768
}
}
}
}
PUT _ingest/pipeline/sw-embeddings
{
"description": "Text embedding pipeline",
"processors": [
{
"inference": {
"model_id": "sentence-transformers__all-mpnet-base-v2",
"target_field": "text_embedding",
"field_map": {
"text": "text_field"
}
}
},
{
"set":{
"field": "vector",
"copy_from": "text_embedding.predicted_value"
}
},
{
"remove": {
"field": "text_embedding"
}
}
],
"on_failure": [
{
"set": {
"description": "Index document to 'failed-<index>'",
"field": "_index",
"value": "failed-{{{_index}}}"
}
},
{
"set": {
"description": "Set error message",
"field": "ingest.failure",
"value": "{{_ingest.on_failure_message}}"
}
}
]
}測試管道，確保它能正常運行。
POST _ingest/pipeline/sw-embeddings/_simulate
{
"docs": [
{
"_source": {
"text": "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.",
"metadata": {
"a": "b"
}
}
}
]
}現在，我們已經準備好使用 Elasticsearch 的普通 Python 庫批量加載數據，目標是我們的采集管道正確地創建矢量嵌入并對數據進行轉換，以符合 LangChain 的預期。
source .env
python3 step-3B-batch-hosted-vectorize.py成功！按照 OpenAI 的標準，這些數據約為 1300 萬個 token，因此在 OpenAI 或類似云服務中生成這些矢量的費用約為 5.40 美元。使用 Elastic Cloud，處理這些數據需要 40 分鐘，成本為 5 美元/小時。
加載完數據后，請記得使用云控制臺將 Cloud ML 的規模縮減到零或更合理的值。
接下來我們來試一試 LLM 和 LangChain。我創建了一個 lib_llm.py 庫文件來保存這些代碼。
from langchain import PromptTemplate, HuggingFaceHub, LLMChain
from langchain.llms import HuggingFacePipeline
from transformers import AutoTokenizer, pipeline, AutoModelForSeq2SeqLM
from langchain.vectorstores import ElasticVectorSearch
from langchain.embeddings import HuggingFaceEmbeddings
import os
cache_dir = "./cache"
def getFlanLarge():
model_id = 'google/flan-t5-large'
print(f">> Prep. Get {model_id} ready to go")
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForSeq2SeqLM.from_pretrained(model_id, cache_dir=cache_dir)
pipe = pipeline(
"text2text-generation",
model=model,
tokenizer=tokenizer,
max_length=100
)
llm = HuggingFacePipeline(pipeline=pipe)
return llm
local_llm = getFlanLarge()
def make_the_llm():
template_informed = """
I am a helpful AI that answers questions.
When I don't know the answer I say I don't know.
I know context: {context}
when asked: {question}
my response using only information in the context is: """
prompt_informed = PromptTemplate(
template=template_informed,
input_variables=["context", "question"])
return LLMChain(prompt=prompt_informed, llm=local_llm)
## continued belowtemplate_informed 是其中非常關鍵但也很容易理解的部分。我們要為提示模板設置格式，它將接受兩個參數：上下文和用戶提出的問題。
最后的主代碼（接上文）如下所示：
## continued from above
topic = "Star Wars"
index_name = "book_wookieepedia_mpnet"
# Create the HuggingFace Transformer like before
model_name = "sentence-transformers/all-mpnet-base-v2"
hf = HuggingFaceEmbeddings(model_name=model_name)
## Elasticsearch as a vector db, just like before
endpoint = os.getenv('ES_SERVER', 'ERROR')
username = os.getenv('ES_USERNAME', 'ERROR')
password = os.getenv('ES_PASSWORD', 'ERROR')
url = f"https://{username}:{password}@{endpoint}:443"
db = ElasticVectorSearch(embedding=hf, elasticsearch_url=url, index_name=index_name)
## set up the conversational LLM
llm_chain_informed= make_the_llm()
def ask_a_question(question):
## get the relevant chunk from Elasticsearch for a question
similar_docs = db.similarity_search(question)
print(f'The most relevant passage: \n\t{similar_docs[0].page_content}')
informed_context= similar_docs[0].page_content
informed_response = llm_chain_informed.run(
context=informed_context,
question=question)
return informed_response
# The conversational loop
print(f'I am a trivia chat bot, ask me any question about {topic}')
while True:
command = input("User Question >> ")
response= ask_a_question(command)
print(f"\tAnswer : {response}")通過執行一些數據整理，我們現在已經使用了 AI，不會將我們的數據向第三方托管式 LLM 公開。AI 的世界瞬息萬變，但對于私有數據的安全性和控制權的保護，我們所有人都應該嚴肅對待，因為數據泄露會對監管、財務和人員等方面造成不良后果。這種情況不太可能改變。我們與客戶合作，通過搜索來調查欺詐行為、保衛國家安全并改善弱勢患者群體的治療效果。隱私是非常重要的。
您是否和我一樣愛上了 LangChain？一位睿智的老絕地武士曾說過：“很好。你已經邁出了通往更廣闊世界的第一步。” 從這里出發，有很多條路可以走。LangChain 讓 AI 提示工程變得不再復雜。我知道 Elasticsearch 在這里還能發揮許多其他作用，比如充當生成式 AI 的長期記憶，我非常期待看到這個瞬息萬變的領域中層出不窮的新事物。
在本博文中，我們可能使用了第三方生成式 AI 工具，這些工具由其各自所有者擁有和運營。Elastic 對第三方工具沒有任何控制權，對其內容、操作或使用不承擔任何責任或義務，對您使用此類工具可能造成的任何損失或損害也不承擔任何責任或義務。在 AI 工具中使用個人、敏感或機密信息時，請務必謹慎。您提交的任何數據都可能用于 AI 訓練或其他目的。Elastic 不保證您所提供信息的安全性或保密性。在使用任何生成式 AI 工具之前，您應自行熟悉隱私慣例和使用條款。
原文鏈接：使用 LangChain 和 Elasticsearch 實現隱私至上的 AI 搜索 | Elastic Blog
