譯者 | 晶顏
審校 | 重樓
本文提供了關于如何使用Milvus構建多模態RAG系統以及如何為AI系統開辟各種可能性的深入指南。
局限于單一的數據格式已經逐漸落伍。隨著企業越來越依賴信息來做出關鍵決策，他們需要能夠比較不同格式的數據。幸運的是，傳統的限于單一數據類型的人工智能系統已經讓位于能夠理解和處理復雜信息的多模態（Multimodal）系統。
多模態搜索和多模態檢索增強生成（RAG）系統近年來在這一領域取得了很大進展。這些系統能夠處理多種類型的數據，包括文本、圖像和音頻，以提供上下文感知的響應。
在這篇文章中，我們將討論開發人員如何使用Milvus構建他們自己的多模態RAG系統。我們還將引導你構建這樣一個系統，該系統可以處理文本和圖像數據，特別是執行相似性搜索，并利用語言模型來優化輸出。
Milvus是什么？
向量數據庫是一種特殊類型的數據庫，用于存儲、索引和檢索向量嵌入，向量嵌入是數據的數學表示（如圖像、文本和音頻），不僅可以比較數據的等價性，還可以比較數據的語義相似性。Milvus就是一個開源、高性能的向量數據庫。你可以在GitHub上找到它，它有Apache-2.0許可證并已獲得超過3萬顆星星。
Milvus幫助開發人員提供靈活的解決方案來管理和查詢大規模向量數據。Milvus的效率使其成為開發人員使用深度學習模型構建應用程序的理想選擇，例如檢索增強生成（RAG）、多模態搜索、推薦引擎和異常檢測。
Milvus提供多種部署選項來滿足開發人員的需求。Milvus Lite是一個輕量級版本，可以在Python應用程序中運行，非常適合在本地環境中創建應用程序原型。Milvus Standalone和Milvus Distributed是可擴展和“生產就緒”（即產品已經過充分測試和優化，可在生產環境中使用）的選項。
多模態RAG：擴展至文本之外
在構建系統之前，了解傳統的基于文本的RAG及其向多模態RAG的演變是很重要的。
檢索增強生成（RAG）是從外部源檢索上下文信息并從大型語言模型（LLM）生成更準確輸出的一種方法。傳統的RAG是提高LLM輸出的一種非常有效的策略，但是它仍然局限于文本數據。而在許多現實世界的應用程序中，數據已經擴展到文本之外，結合圖像、圖表和其他形式提供了關鍵的上下文。
多模態RAG通過支持使用不同的數據類型解決了上述限制，為LLM提供了更好的上下文。
簡單地說，在多模態RAG系統中，檢索組件能夠跨不同的數據模態搜索相關信息，生成組件根據檢索到的信息生成更準確的結果。
理解向量嵌入和相似性搜索
向量嵌入和相似性搜索是多模態RAG的兩個基本概念。讓我們先來理解它們。
向量嵌入
如前所述，向量嵌入是數據的數學/數值表示。機器使用這種表示來理解不同數據類型（如文本、圖像和音頻）的語義含義。
在使用自然語言處理（NLP）時，將文檔塊轉換為向量，并將語義相似的單詞映射到向量空間中的附近點。圖像也是如此，其中嵌入表示語義特征。這使我們能夠以數字格式理解顏色、紋理和物體形狀等指標。
使用向量嵌入的主要目的是幫助保持不同數據塊之間的關系和相似性。
相似性搜索
相似性搜索用于查找和定位給定數據集中的數據。在向量嵌入的背景下，相似性搜索在給定的數據集中找到最接近查詢向量的向量。
以下是幾種常用的度量向量之間相似性的方法：
- 歐幾里得距離：測量向量空間中兩點之間的直線距離。
- 余弦相似度：測量兩個向量之間夾角的余弦值（關注它們的方向而不是大小）。
- 點積：對應元素相加的簡單乘法。
相似性度量的選擇通常取決于特定于應用程序的數據以及開發人員處理問題的方式。
在大規模數據集上進行相似性搜索時，需要很強大的計算能力和資源。這就是近似最近鄰（ANN）算法發揮作用的地方。人工神經網絡算法用于交換小百分比或數量的準確性，以獲得顯著的速度提升。這使得它們成為大規模應用程序的合適選擇。
Milvus還使用先進的人工神經網絡算法（包括HNSW和DiskANN），在大型向量嵌入數據集上執行高效的相似性搜索，使開發人員能夠快速找到相關數據點。此外，Milvus支持其他索引算法，如HSNW， IVF， CAGRA等，使其成為一個更有效的向量搜索解決方案。
用Milvus構建多模態RAG
現在我們已經學習了這些概念，是時候使用Milvus構建一個多模態RAG系統了。在下述示例中，我們將使用Milvus Lite（Milvus的輕量級版本，非常適合實驗和原型設計）進行向量存儲和檢索，BGE用于精確的圖像處理和嵌入，GPT用于高級結果重新排序。
先決條件
首先，你需要一個Milvus實例來存儲你的數據。你可以使用pip設置Milvus Lite，使用Docker運行本地實例，或者通過Zilliz Cloud注冊一個免費托管的Milvus帳戶。
其次，你需要為你的RAG管道提供LLM，因此建議前往OpenAI并獲取API密鑰。免費層足以使此代碼運行。
接下來，創建一個新目錄和一個Python虛擬環境（或者采取你用來管理Python的任何步驟）。
對于本教程，你還需要安裝pymilvus庫（它是Milvus的官方Python SDK）和一些常用工具。
設置Milvus Lite
pip install -U pymilvus安裝依賴項
pip install --upgrade pymilvus openai datasets opencv-python timm einops ftfy peft tqdm
git clone https://github.com/FlagOpen/FlagEmbedding.git
pip install -e FlagEmbedding下載數據
下面的命令將下載示例數據并將其解壓縮到本地文件夾“./images_folder”，其中包括：
- 圖片：Amazon Reviews 2023的一個子集，包含大約900張來自“Appliance”、 “Cell_Phones_and_Accessories”和“Electronics”類別的圖片。
- 查詢圖片示例：leopard.jpg
wget https://github.com/milvus-io/bootcamp/releases/download/data/amazon_reviews_2023_subset.tar.gztar -xzf amazon_reviews_2023_subset.tar.gz
加載嵌入模型
我們將使用可視化BGE模型“big - visualizing -base-en-v1.5”來生成圖像和文本的嵌入。
現在從HuggingFace下載權重。
wget https://huggingface.co/BAAI/bge-visualized/resolve/main/Visualized_base_en_v1.
然后，讓我們構建一個編碼器。
import torchfrom visual_bge.modeling import Visualized_BGE
class Encoder:
def __init__(self, model_name: str, model_path: str):
self.model = Visualized_BGE(model_name_bge=model_name, model_weight=model_path)
self.model.eval()
def encode_query(self, image_path: str, text: str) -> list[float]:
with torch.no_grad():
query_emb = self.model.encode(image=image_path, text=text)
return query_emb.tolist()[0]
def encode_image(self, image_path: str) -> list[float]:
with torch.no_grad():
query_emb = self.model.encode(image=image_path)
return query_emb.tolist()[0]
model_name = "BAAI/bge-base-en-v1.5"
model_path = "./Visualized_base_en_v1.5.pth" # Change to your own value if using a different model path
encoder = Encoder(model_name, model_path)生成嵌入和加載數據到Milvus
本節將指導你如何將示例圖像與其相應的嵌入加載到數據庫中。
生成嵌入
首先，我們需要為數據集中的所有圖像創建嵌入。
從data目錄加載所有圖像并將它們轉換為嵌入。
import os
from tqdm import tqdm
from glob import glob
data_dir = (
"./images_folder" # Change to your own value if using a different data directory
)
image_list = glob(
os.path.join(data_dir, "images", "*.jpg")
) # We will only use images ending with ".jpg"
image_dict = {}
for image_path in tqdm(image_list, desc="Generating image embeddings: "):
try:
image_dict[image_path] = encoder.encode_image(image_path)
except Exception as e:
print(f"Failed to generate embedding for {image_path}. Skipped.")
continue
print("Number of encoded images:", len(image_dict))執行多模態搜索和重新排序結果
在本節中，我們將首先使用多模態查詢搜索相關圖像，然后使用LLM服務對檢索結果進行重新排序，并找到帶有解釋的最佳圖像。
運行多模態搜索
現在，我們準備使用由圖像和文本指令組成的查詢來執行高級多模態搜索。
query_image = os.path.join(
data_dir, "leopard.jpg"
) # Change to your own query image path
query_text = "phone case with this image theme"
query_vec = encoder.encode_query(image_path=query_image, text=query_text)
search_results = milvus_client.search(
collection_name=collection_name,
data=[query_vec],
output_fields=["image_path"],
limit=9, # Max number of search results to return
search_params={"metric_type": "COSINE", "params": {}}, # Search parameters
)[0]
retrieved_images = [hit.get("entity").get("image_path") for hit in search_results]
print(retrieved_images)結果如下：
['./images_folder/images/518Gj1WQ-RL._AC_.jpg',
'./images_folder/images/41n00AOfWhL._AC_.jpg'用GPT-40重新排序結果
現在，我們將使用GPT-40對檢索到的圖像進行排序，并找到最匹配的結果。最后，LLM還將解釋排名原因。
1. 創建全景視圖。
import numpy as np
import cv2
img_height = 300
img_width = 300
row_count = 3
def create_panoramic_view(query_image_path: str, retrieved_images: list) -> np.ndarray:
"""
creates a 5x5 panoramic view image from a list of images
args:
images: list of images to be combined
returns:
np.ndarray: the panoramic view image
"""
panoramic_width = img_width * row_count
panoramic_height = img_height * row_count
panoramic_image = np.full(
(panoramic_height, panoramic_width, 3), 255, dtype=np.uint8
)
# create and resize the query image with a blue border
query_image_null = np.full((panoramic_height, img_width, 3), 255, dtype=np.uint8)
query_image = Image.open(query_image_path).convert("RGB")
query_array = np.array(query_image)[:, :, ::-1]
resized_image = cv2.resize(query_array, (img_width, img_height))
border_size = 10
blue = (255, 0, 0) # blue color in BGR
bordered_query_image = cv2.copyMakeBorder(
resized_image,
border_size,
border_size,
border_size,
border_size,
cv2.BORDER_CONSTANT,
value=blue,
)
query_image_null[img_height * 2 : img_height * 3, 0:img_width] = cv2.resize(
bordered_query_image, (img_width, img_height)
)
# add text "query" below the query image
text = "query"
font_scale = 1
font_thickness = 2
text_org = (10, img_height * 3 + 30)
cv2.putText(
query_image_null,
text,
text_org,
cv2.FONT_HERSHEY_SIMPLEX,
font_scale,
blue,
font_thickness,
cv2.LINE_AA,
)
# combine the rest of the images into the panoramic view
retrieved_imgs = [
np.array(Image.open(img).convert("RGB"))[:, :, ::-1] for img in retrieved_images
]
for i, image in enumerate(retrieved_imgs):
image = cv2.resize(image, (img_width - 4, img_height - 4))
row = i // row_count
col = i % row_count
start_row = row * img_height
start_col = col * img_width
border_size = 2
bordered_image = cv2.copyMakeBorder(
image,
border_size,
border_size,
border_size,
border_size,
cv2.BORDER_CONSTANT,
value=(0, 0, 0),
)
panoramic_image[
start_row : start_row + img_height, start_col : start_col + img_width
] = bordered_image
# add red index numbers to each image
text = str(i)
org = (start_col + 50, start_row + 30)
(font_width, font_height), baseline = cv2.getTextSize(
text, cv2.FONT_HERSHEY_SIMPLEX, 1, 2
)
top_left = (org[0] - 48, start_row + 2)
bottom_right = (org[0] - 48 + font_width + 5, org[1] + baseline + 5)
cv2.rectangle(
panoramic_image, top_left, bottom_right, (255, 255, 255), cv2.FILLED
)
cv2.putText(
panoramic_image,
text,
(start_col + 10, start_row + 30),
cv2.FONT_HERSHEY_SIMPLEX,
1,
(0, 0, 255),
2,
cv2.LINE_AA,
)
# combine the query image with the panoramic view
panoramic_image = np.hstack([query_image_null, panoramic_image])
return panoramic_image2. 將查詢圖像和檢索圖像與索引結合在一個全景視圖中。
from PIL import Image
combined_image_path = os.path.join(data_dir, "combined_image.jpg")
panoramic_image = create_panoramic_view(query_image, retrieved_images)
cv2.imwrite(combined_image_path, panoramic_image)
combined_image = Image .open(combined_image_path )
show_combined_image = combined_image.resize((300, 300))
show_combined_image.show()
多模態搜索結果
3. 對結果重新排序并給出解釋
我們將把所有組合的圖像發送到多模態LLM服務，并提供適當的提示，對檢索結果進行排序并給出解釋。注意：要啟用GPT- 40作為LLM，你需要提前準備好你的OpenAI API Key。
import requests
import base64
openai_api_key = "sk-***" # Change to your OpenAI API Key
def generate_ranking_explanation(
combined_image_path: str, caption: str, infos: dict = None
) -> tuple[list[int], str]:
with open(combined_image_path, "rb") as image_file:
base64_image = base64.b64encode(image_file.read()).decode("utf-8")
information = (
"You are responsible for ranking results for a Composed Image Retrieval. "
"The user retrieves an image with an 'instruction' indicating their retrieval intent. "
"For example, if the user queries a red car with the instruction 'change this car to blue,' a similar type of car in blue would be ranked higher in the results. "
"Now you would receive instruction and query image with blue border. Every item has its red index number in its top left. Do not misunderstand it. "
f"User instruction: {caption} \n\n"
)
# add additional information for each image
if infos:
for i, info in enumerate(infos["product"]):
information += f"{i}. {info}\n"
information += (
"Provide a new ranked list of indices from most suitable to least suitable, followed by an explanation for the top 1 most suitable item only. "
"The format of the response has to be 'Ranked list: []' with the indices in brackets as integers, followed by 'Reasons:' plus the explanation why this most fit user's query intent."
)
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {openai_api_key}",
}
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": information},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{base64_image}"},
},
],
}
],
"max_tokens": 300,
}
response = requests.post(
"https://api.openai.com/v1/chat/completions", headers=headers, json=payload
)
result = response.json()["choices"][0]["message"]["content"]
# parse the ranked indices from the response
start_idx = result.find("[")
end_idx = result.find("]")
ranked_indices_str = result[start_idx + 1 : end_idx].split(",")
ranked_indices = [int(index.strip()) for index in ranked_indices_str]
# extract explanation
explanation = result[end_idx + 1 :].strip()
return ranked_indices, explanation得到排名后的圖像指標和最佳結果的原因：
ranked_indices, explanation = generate_ranking_explanation(
combined_image_path, query_text
)4. 顯示最佳結果并附有說明
print(explanation)
best_index = ranked_indices[0]
best_img = Image.open(retrieved_images[best_index])
best_img = best_img.resize((150, 150))
best_img.show()結果：
“原因：最適合用戶查詢意圖的項是索引6，因為指令指定了一個以圖片為主題的手機殼，是一只豹子。索引為6的手機殼采用了類似豹紋的主題設計，最符合用戶對圖像主題手機殼的需求。”
豹紋手機殼-最佳效果
結語
在這篇文章中，我們討論了使用Milvus（一個開源向量數據庫）構建一個多模態RAG系統的具體操作指南，介紹了開發人員如何設置Milvus、加載圖像數據、執行相似性搜索以及使用LLM對檢索結果進行重新排序以獲得更準確的響應。
可以說，多模態RAG解決方案為人工智能系統提供了多種可能性，可以輕松理解和處理多種形式的數據。一些常見的可能性包括改進圖像搜索引擎、更好的上下文驅動結果等等，其他更多可能性等你來探索！
原文標題：Want to Search for Something With an Image and a Text Description? Try a Multimodal RAG，作者：Jiang Chen
