Práctica: Creación de potentes búsquedas multimodales con Voyager-3 y LangGraph

Tutoriales prácticos sobre IAPublicado hace 7 meses Círculo de intercambio de inteligencia artificial

1.6K 00

Voyage AI 的 Voyager 3 是一种新的最先进的模型，它允许您将文本和图像嵌入到同一空间中。在这篇文章中，我将解释如何从杂志中提取这些多模态嵌入，将它们存储在向量数据库（Weaviate）中，并使用相同的嵌入向量对文本和图像执行相似性搜索。

将图像和文本嵌入到同一空间中，将使我们能够对多模态内容（如网页、PDF 文件、杂志、书籍、宣传册和各种论文）执行高度精确的搜索。为什么这种技术如此有趣？将文本和图像嵌入到同一空间的主要令人兴奋之处在于，您可以搜索和检索与特定图像相关的文本，反之亦然。例如，如果您正在搜索猫，您将找到显示猫的图片，但您也会得到引用这些图像的文本，即使文本没有明确地说出“猫”这个词。

让我展示一下传统的文本嵌入相似性搜索和多模态嵌入空间之间的区别：

示例问题：杂志上关于猫说了什么？

一张来自摄影杂志的截图 —— OUTDOOR

常规相似性搜索答案

提供的搜索结果不包含关于猫的具体信息。它们提到了动物肖像和摄影技巧，但没有明确提及猫或与其相关的细节。

如上图所示，“猫”这个词没有被提及；只有一张图片和关于如何拍摄动物照片的解释。由于没有提到“猫”这个词，常规的相似性搜索没有产生任何结果。

多模态搜索答案

这本杂志刊登了一张猫的肖像，突出了其面部特征和性格的精细捕捉。文字强调了制作精良的动物肖像如何深入到主题的灵魂，并通过引人注目的眼神交流与观看者建立情感联系。

使用多模态搜索，我们将找到一张猫的图片，然后将相关的文字链接到它。将这些数据提供给模型将使其能够更好地回答和理解上下文。

如何构建多模态嵌入和检索管道

现在，我将分几个步骤描述这样一个管道的工作原理：

我们将使用 Unstructured（一个用于数据提取的强大 Python 库）从 PDF 文件中提取文本和图像。
我们将使用 Voyager Multimodal 3 模型为同一向量空间内的文本和图像创建多模态向量。
我们将把它插入到向量存储 (Weaviate) 中。
最后，我们将执行相似性搜索并比较文本和图像的结果。

第 1 步：设置向量存储并从文件 (PDF) 中提取图像和文本

在这里，我们必须做一些手动工作。通常，Weaviate 是一个非常易于使用的向量存储，它会在插入时自动转换数据并添加嵌入。但是，没有用于 Voyager Multimodal v3 的插件，因此我们必须手动计算嵌入。在这种情况下，我们必须创建一个集合而不定义向量化器模块。

import weaviate
from weaviate.classes.config import Configure
client = weaviate.connect_to_local()
collection_name = "multimodal_demo"
client.collections.delete(collection_name)
try:
client.collections.create(
name=collection_name,
vectorizer_config=Configure.Vectorizer.none() # 不为此集合设置向量化器
)
collection = client.collections.get(collection_name)
except Exception:
collection = client.collections.get(collection_name)pyt

在这里，我在 Docker 容器中运行一个本地 Weaviate 实例。

第 2 步：从 PDF 中提取文档和图像

这是流程工作的关键步骤。在这里，我们将获取一个包含文本和图片的 PDF。然后，我们将提取内容（图像和文本）并将其存储在相关的块中。因此，每个块将是一个包含字符串（实际文本）和 Python PIL 图像的元素列表。

我们将使用 Unstructured 库来完成一些繁重的工作，但我们仍然需要编写一些逻辑并配置库参数。

from unstructured.partition.auto import partition
from unstructured.chunking.title import chunk_by_title
elements = partition(
filename="./files/magazine_sample.pdf",
strategy="hi_res",
extract_image_block_types=["Image", "Table"],
extract_image_block_to_payload=True)
chunks = chunk_by_title(elements)

在这里，我们必须使用 hi_res 策略，并使用 extract_image_block_to_payload 将图像导出到有效载荷，因为我们稍后需要此信息用于实际的嵌入。一旦我们提取了所有元素，我们将根据文档中的标题将它们分组到块中。

有关更多信息，请查看关于分块的 Unstructured 文档.

在下面的脚本中，我们将使用这些块输出两个列表：

一个包含我们将发送到 Voyager 3 以创建向量的对象的列表
一个包含 Unstructured 提取的元数据的列表。此元数据是必需的，因为我们必须将其添加到向量存储中。它将为我们提供额外的属性进行过滤，并告诉我们一些关于检索到的数据的信息。

from unstructured.staging.base import elements_from_base64_gzipped_json
import PIL.Image
import io
import base64
embedding_objects = []
embedding_metadatas = []
for chunk in chunks:
embedding_object = []
metedata_dict = {
"text": chunk.to_dict()["text"],
"filename": chunk.to_dict()["metadata"]["filename"],
"page_number": chunk.to_dict()["metadata"]["page_number"],
"last_modified": chunk.to_dict()["metadata"]["last_modified"],
"languages": chunk.to_dict()["metadata"]["languages"],
"filetype": chunk.to_dict()["metadata"]["filetype"]
}
embedding_object.append(chunk.to_dict()["text"])
# 将图像添加到嵌入对象
if "orig_elements" in chunk.to_dict()["metadata"]:
base64_elements_str = chunk.to_dict()["metadata"]["orig_elements"]
eles = elements_from_base64_gzipped_json(base64_elements_str)
image_data = []
for ele in eles:
if ele.to_dict()["type"] == "Image":
base64_image = ele.to_dict()["metadata"]["image_base64"]
image_data.append(base64_image)
pil_image = PIL.Image.open(io.BytesIO(base64.b64decode(base64_image)))
# 如果图像大于 1000x1000，则在保持纵横比的同时调整图像大小
if pil_image.size[0] > 1000 or pil_image.size[1] > 1000:
ratio = min(1000/pil_image.size[0], 1000/pil_image.size[1])
new_size = (int(pil_image.size[0] * ratio), int(pil_image.size[1] * ratio))
pil_image = pil_image.resize(new_size, PIL.Image.Resampling.LANCZOS)
embedding_object.append(pil_image)
metedata_dict["image_data"] = image_data
embedding_objects.append(embedding_object)
embedding_metadatas.append(metedata_dict)

此脚本的结果将是一个列表的列表，其内容如下所示：

[['来自\n\n冰岛 KIRKJUFELL 的位置',
<PIL.Image.Image image mode=RGB size=1000x381>,
<PIL.Image.Image image mode=RGB size=526x1000>],
['这座标志性的山峰是我们冰岛拍摄地点的首选，而且在我们去那里之前，我们就看过许多从附近瀑布拍摄的照片。因此，这是我们在日出时前往的第一个地方 - 我们没有失望。这些瀑布为这张照片（顶部）提供了完美的近景趣味，而从这个角度来看，Kirkjufell 是一座完美的尖山。我们花了一两个小时简单地探索这些瀑布，找到了几个不同的角度。']]

第 3 步：向量化提取的数据

在这一步中，我们将使用上一步中创建的块，并使用 Voyager Python 包将它们发送到 Voyager。它将返回给我们所有嵌入对象的列表。然后，我们可以使用此结果，并最终将其存储在 Weaviate 中。

from dotenv import load_dotenv
import voyageai
load_dotenv()
vo = voyageai.Client()
# 这将自动使用环境变量 VOYAGE_API_KEY。
# 或者，您可以使用 vo = voyageai.Client(api_key="<您的密钥>")
# 包含文本字符串和 PIL 图像对象的示例输入
inputs = embedding_objects
# 向量化输入
result = vo.multimodal_embed(
inputs,
model="voyage-multimodal-3",
truncation=False
)

如果我们访问 result.embeddings，我们将获得一个包含所有计算出的嵌入向量的列表的列表：

[[-0.052734375, -0.0164794921875, 0.050048828125, 0.01348876953125, -0.048095703125, …]]

我们现在可以使用 batch.add_object 方法将此嵌入数据以单个批次存储在 Weaviate 中。请注意，我们还在 properties 参数中添加了元数据。

with collection.batch.dynamic() as batch:
for i, data_row in enumerate(embedding_objects):
batch.add_object(
properties=embedding_metadatas[i],
vector=result.embeddings[i]
)

第 4 步：查询数据

我们现在可以执行相似性搜索并查询数据。这很容易，因为此流程类似于对文本嵌入执行的常规相似性搜索。由于 Weaviate 没有用于 Voyager 多模态的模块，因此我们必须在将搜索向量传递给 Weaviate 以执行相似性搜索之前，自己计算搜索查询的向量。

from weaviate.classes.query import MetadataQuery
question = "杂志上关于瀑布说了什么？"
vector = vo.multimodal_embed([[question]], model="voyage-multimodal-3")
vector.embeddings[0]
response = collection.query.near_vector(
near_vector=vector.embeddings[0], # 您的查询向量在此处
limit=2,
return_metadata=MetadataQuery(distance=True)
)
# 显示结果
for o in response.objects:
print(o.properties['text'])
for image_data in o.properties['image_data']:
# 使用 PIL 显示图像
img = PIL.Image.open(io.BytesIO(base64.b64decode(image_data)))
width, height = img.size
if width > 500 or height > 500:
ratio = min(500/width, 500/height)
new_size = (int(width * ratio), int(height * ratio))
img = img.resize(new_size)
display(img)
print(o.metadata.distance)

下图显示，搜索瀑布将返回与此搜索查询相关的文本和图像。如您所见，这些照片反映了瀑布，但文本本身并没有提及它们。这段文字是关于一张里面有瀑布的图片，这就是它也被检索的原因。这对于常规文本嵌入搜索是不可能的。

一张显示相似性搜索结果的图片

第 5 步：将其添加到整个检索管道中

现在我们已经从杂志中提取了文本和图像，为它们创建了嵌入，将它们添加到 Weaviate 中，并设置了我们的相似性搜索，我将把它添加到整个检索管道中。在此示例中，我将使用 LangGraph。用户将提出关于这本杂志的问题，管道将回答这个问题。既然所有工作都已完成，这部分就像使用常规文本设置典型的检索管道一样简单。

我已经将我们在前面部分讨论的一些逻辑抽象到其他模块中。这是一个我如何将其集成到 LangGraph 管道中的示例。

class MultiModalRetrievalState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
results: List[Document]
base_64_images: List[str]
class RAGNodes(BaseNodes):
def __init__(self, logger, mode="online", document_handler=None):
super().__init__(logger, mode)
self.weaviate = Weaviate()
self.mode = mode
async def multi_modal_retrieval(self, state: MultiModalRetrievalState, config):
collection_name = config.get("configurable", {}).get("collection_name")
self.weaviate.set_collection(collection_name)
print("正在运行多模态检索")
print(f"正在搜索 {state['messages'][-1].content}")
results = self.weaviate.similarity_search(
query=state["messages"][-1].content, k=3, type="multimodal"
)
return {"results": results}
async def answer_question(self, state: MultiModalRetrievalState, config):
print("正在回答问题")
llm = self.llm_factory.create_llm(mode=self.mode, model_type="default")
include_images = config.get("configurable", {}).get("include_images", False)
chain = self.chain_factory.create_multi_modal_chain(
llm,
state["messages"][-1].content,
state["results"],
include_images=include_images,
)
response = await chain.ainvoke({})
message = AIMessage(content=response)
return {"messages": message}
# 定义配置
class GraphConfig(TypedDict):
mode: str = "online"
collection_name: str
include_images: bool = False
graph_nodes = RAGNodes(logger)
graph = StateGraph(MultiModalRetrievalState, config_schema=GraphConfig)
graph.add_node("multi_modal_retrieval", graph_nodes.multi_modal_retrieval)
graph.add_node("answer_question", graph_nodes.answer_question)
graph.add_edge(START, "multi_modal_retrieval")
graph.add_edge("multi_modal_retrieval", "answer_question")
graph.add_edge("answer_question", END)
multi_modal_graph = graph.compile()
__all__ = ["multi_modal_graph"]

上面的代码将生成以下图表

创建的图表的直观表示

在此跟踪中，您可以看到内容和图像都被发送到 OpenAI 以回答问题。