使用 Docling 和 Granite 构建人工智能驱动的多模式 RAG 系统

作者

Open Source Developer, STSM

Data Scientist

IBM

在本教程中，您将使用 IBM 的 Docling 和开源 IBM Granite Vision、基于文本的嵌入和生成式 AI 模型，创建一个 RAG 系统。这些模型可通过各种开源框架获得。在本教程中，我们将使用 Replicate 连接 IBM Granite Vision 和生成式 AI 模型，并使用 HuggingFace 连接嵌入模型。

多模态检索增强生成

检索增强生成 (RAG) 是一种用于将与大型语言模型 (LLM) 与模型训练数据之外的信息知识库相连接的技术，无需进行微调。传统的 RAG 仅限于基于文本的用例，例如文本摘要和聊天机器人。

多模态 RAG 可以使用多模态 LLM (MLLM) 处理来自多种类型数据的信息，将其作为 RAG 使用的外部知识库的一部分。多模态数据可以包括文本、图片、音频、视频或其他形式。常用的多模态 LLM 包括 Google 的 Gemini、Meta 的 Llama 3.2 和 OpenAI 的 GPT-4 和 GPT-4o。

在本方案中，您将使用能够处理不同模态的 IBM Granite 模型。您将创建一个 AI 系统，用于根据 PDF 中的非结构化数据回答用户的实时查询。

教程概述

欢迎进入学习本 Granite 教程。在本教程中，您将了解如何利用高级工具的强大功能来构建人工智能驱动的多模式 RAG 管道。本教程将指导您完成以下步骤：

文档预处理：了解如何处理来自各种来源的文档，如何使用 Docling 进行解析并将其转换为可用格式，然后将其存储在矢量数据库中。您将使用 Granite MLLM 为文档中的图像生成图像描述。
RAG： 了解如何将 Granite 等 LLM 与外部知识库连接，以增强查询回复并生成有价值的洞察分析。
用于工作流整合的 LangChain：探索如何使用 LangChain 简化和编排文档处理和检索工作流，实现系统不同组件之间的无缝交互。

本教程使用三种尖端科技：

Docling：一个用于解析和转换文档的开源工具包。
Granite：一款最先进的 LLM，提供强大的自然语言能力，以及提供图片到文本生成的视觉语言模型。
LangChain：一个强大的框架，用于构建由语言模型驱动的应用程序，旨在简化复杂的工作流并无缝集成外部工具。

完成本教程后，您将实现以下目标：

熟练掌握文档预处理、区块分析和图片理解。
集成矢量数据库以增强检索能力。
使用 RAG 对实际应用程序执行高效、准确的数据检索。

本教程专为希望增强文档管理和高级自然语言处理 (NLP) 技术知识的 AI 开发人员、研究人员和爱好者而设计。该教程还可以在 IBM Granite Community 的 Granite Snack Cookbook GitHub 中以 Jupyter Notebook 的形式找到。

先决条件

熟悉 Python 编程。
基本了解 LLM、NLP 概念和计算机视觉。

步骤

第 1 步：安装依赖项

! echo "::group::Install Dependencies"
%pip install uv
! uv pip install git+https://github.com/ibm-granite-community/utils.git \
    transformers \
    pillow \
    langchain_classic \
    langchain_core \
    langchain_huggingface sentence_transformers \
    langchain_milvus 'pymilvus[milvus_lite]' \
    docling \
    'langchain_replicate @ git+https://github.com/ibm-granite-community/langchain-replicate.git'
! echo "::endgroup::"

第 2 步：选择 AI 模型

日志记录

为了查看一些日志信息，我们可以将日志级别配置为 INFO。

注意：可以跳过运行此单元。

import logging

logging.basicConfig(level=logging.INFO)

加载 Granite 模型

指定用于生成文本嵌入矢量的嵌入模型。这里我们将使用 Granite 嵌入模型中的一个

要使用不同的嵌入模型，请将此代码单元替换为此嵌入模型方案中的一个。

from langchain_huggingface import HuggingFaceEmbeddings
from transformers import AutoTokenizer

embeddings_model_path = “ibm-granite/granite-embedding-30m-english”
embeddings_model = HuggingFaceEmbeddings(
model_name=embeddings_model_path,
)
embeddings_tokenizer = AutoTokenizer.from_pretrained(embeddings_model_path)

指定用于图像理解的 MLLM。我们将使用 Granite Vision 模型。

from ibm_granite_community.notebook_utils import get_env_var
from langchain_community.llms import Replicate
from transformers import AutoProcessor

vision_model_path = “ibm-granite/granite-vision-3.2-2b”
vision_model = Replicate(
    model=vision_model_path,
    replicate_api_token=get_env_var(“REPLICATE_API_TOKEN”),
    model_kwargs={
        “max_tokens”: embeddings_tokenizer.max_len_single_sentence, # Set the maximum number of tokens to generate as output.
        “min_tokens”: 100, # Set the minimum number of tokens to generate as output.
    },
)
vision_processor = AutoProcessor.from_pretrained(vision_model_path)

指定用于 RAG 生成操作的语言模型。在这里，我们使用 Replicate LangChain 客户端连接 Replicate 上 ibm-granite org 的 Granite 模型。

要使用 Replicate 进行设置，请参阅 Replicate 入门指南。要连接到除 Replicate 之外的提供商上的模型，请将些代码单元替换为 LLM 组件方案中的一个。

要连接到 Replicate 之外的提供商上的模型，请将此代码单元替换为 LLM 组件方案中的一个。

from langchain_replicate import ChatReplicate

model_path = "ibm-granite/granite-4.0-h-small"
model = ChatReplicate(
    model=model_path,
    replicate_api_token=get_env_var("REPLICATE_API_TOKEN"),
    model_kwargs={
        "max_tokens": 1000, # Set the maximum number of tokens to generate as output.
        "min_tokens": 100, # Set the minimum number of tokens to generate as output.
    },
)

第 3 步：为矢量数据库准备文档

在本示例中，我们使用 Docling 将一组源文档转换为文本和图片。然后将文本分割成若干区块。MLLM 处理这些图片以生成图片摘要。

使用 Docling 下载文档并将其转换为文本和图片。

Docling 将下载 PDF 文档并进行处理，以便获取文件中包含的文本和图片。PDF 中有各种数据类型，包括文本、表格、图形和图像。

from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import PdfPipelineOptions

pdf_pipeline_options = PdfPipelineOptions(
  do_ocr=False,
    generate_picture_images=True,
)
format_options = {
    InputFormat.PDF: PdfFormatOption(pipeline_options=pdf_pipeline_options),
}
converter = DocumentConverter(format_options=format_options)

sources = [
    “https://midwestfoodbank.org/images/AR_2020_WEB2.pdf”,
]
conversions = { source: converter.convert(source=source).document for source in sources }

文档处理后，我们会进一步处理文档中的文本元素。我们将其区块成适合我们使用的嵌入模型的大小。由文本区块创建 LangChain 文档列表。

from docling_core.transforms.chunker.hybrid_chunker import HybridChunker
from docling_core.types.doc.document import TableItem
from langchain_core.documents import Document

doc_id = 0
texts: list[Document] = []
for source, docling_document in conversions.items():
    for chunk in HybridChunker(tokenizer=embeddings_tokenizer).chunk(docling_document):
        items = chunk.meta.doc_items
        if len(items) == 1 and isinstance(items[0], TableItem):
            continue # we will process tables later
        refs = “ “.join(map(lambda item: item.get_ref().cref, items))
        print(refs)
        text = chunk.text
        document = Document(
            page_content=text,
            metadata={
                “doc_id”: (doc_id:=doc_id+1),
                “source”: source,
                “ref”: refs,
            },
        )
        texts.append(document)

print(f”{len(texts)} text document chunks created”)

接下来，我们将处理文档中的所有表格。我们将表格数据转换为 markdown 格式，以便传递给语言模型。从表格的 markdown 渲染创建 LangChain 文档列表。

from docling_core.types.doc.labels import DocItemLabel

doc_id = len(texts)
tables: list[Document] = []
for source, docling_document in conversions.items():
    for table in docling_document.tables:
        if table.label in [DocItemLabel.TABLE]:
            ref = table.get_ref().cref
            print(ref)
            text = table.export_to_markdown()
            document = Document(
                page_content=text,
                metadata={
                    “doc_id”: (doc_id:=doc_id+1),
                    “source”: source,
                    “ref”: ref
                },
            )
            tables.append(document)

print(f”{len(tables)} table documents created”)

最后，我们处理文档中的所有图片。在这里，我们使用视觉语言模型来理解图片的内容。在此示例中，我们对图片中的所有文本信息感兴趣。您可能需要尝试使用不同的提示文本，看看它如何改善结果。

注意：处理图片中有需要很长时间，具体取决于图片数量和运行视觉语言模型的服务。

import base64
import io
import PIL.Image
import PIL.ImageOps
from IPython.display import display

def encode_image(image: PIL.Image.Image, format: str = “png”) -> str:
    image = PIL.ImageOps.exif_transpose(image) or image
    image = image.convert(“RGB”)

    buffer = io.BytesIO()
    image.save(buffer, format)
    encoding = base64.b64encode(buffer.getvalue()).decode(“utf-8”)
    uri = f”data:image/{format};base64,{encoding}”
    return uri

# Feel free to experiment with this prompt
image_prompt = “If the image contains text, explain the text in the image.”
conversation = [
    {
        “role”: “user”,
        “content”: [
            {“type”: “image”},
            {“type”: “text”, “text”: image_prompt},
        ],
    },
]
vision_prompt = vision_processor.apply_chat_template(
    conversation=conversation,
    add_generation_prompt=True,
)
pictures: list[Document] = []
doc_id = len(texts) + len(tables)
for source, docling_document in conversions.items():
    for picture in docling_document.pictures:
        ref = picture.get_ref().cref
        print(ref)
        image = picture.get_image(docling_document)
        if image:
            text = vision_model.invoke(vision_prompt, image=encode_image(image))
            document = Document(
                page_content=text,
                metadata={
                    “doc_id”: (doc_id:=doc_id+1),
                    “source”: source,
                    “ref”: ref,
                },
            )
            pictures.append(document)

print(f”{len(pictures)} image descriptions created”)

接着，我们可以显示从输入文档生成的 LangChain 文档。

import itertools
from docling_core.types.doc.document import RefItem

# Print all created documents
for document in itertools.chain(texts, tables):
    print(f”Document ID: {document.metadata[‘doc_id’]}”)
    print(f”Source: {document.metadata[‘source’]}”)
    print(f”Content:\n{document.page_content}”)
    print(“=” * 80) # Separator for clarity

for document in pictures:
    print(f”Document ID: {document.metadata[‘doc_id’]}”)
    source = document.metadata[‘source’]
    print(f”Source: {source}”)
    print(f”Content:\n{document.page_content}”)
    docling_document = conversions[source]
    ref = document.metadata[‘ref’]
    picture = RefItem(cref=ref).resolve(docling_document)
    image = picture.get_image(docling_document)
    print(“Image:”)
    display(image)
    print(“=” * 80) # Separator for clarity

填充矢量数据库

使用嵌入模型，我们将文本区块并将生成的图片说明加载到矢量数据库中。通过创建这个矢量数据库，我们可以轻松地在文档中进行语义相似性搜索。

注意：矢量数据库填充可能需要一些时间，具体取决于您的嵌入模型和服务。

选择矢量数据库

指定用于存储和检索嵌入矢量的数据库。

要连接到 Milvus 以外的矢量数据库，请将此代码单元替换为此矢量存储方案中的一个。

import tempfile
from langchain_core.vectorstores import VectorStore
from langchain_milvus import Milvus

db_file = tempfile.NamedTemporaryFile(prefix=”vectorstore_”, suffix=”.db”, delete=False).name
print(f”The vector database will be saved to {db_file}”)

vector_db: VectorStore = Milvus(
    embedding_function=embeddings_model,
    connection_args={“uri”: db_file},
    auto_id=True,
    enable_dynamic_field=True,
    index_params={“index_type”: “AUTOINDEX”},
)

现在，我们将所有用于文本、表格和图片描述的所有 LangChain 文档添加到矢量数据库中。

import itertools
documents = list(itertools.chain(texts, tables, pictures))
ids = vector_db.add_documents(documents)
print(f”{len(ids)} documents added to the vector database”)

第 4 步：使用 Granite 进行 RAG

现在，我们已成功转换文档并将其矢量化，接下来可以设置 RAG 管道。

检索相关区块

这里，我们通过在矢量空间中搜索与查询相关的区块来测试矢量数据库。我们显示与检索到的图片描述相关的文档。

欢迎尝试不同的查询方式。

query = "How much was spent on food distribution relative to the amount of food distributed?"
for doc in vector_db.as_retriever().invoke(query):
print(doc)
print("=" * 80) # Separator for clarity

返回的文档应该对查询做出响应。让我们继续构建我们的 RAG 管道。

为 Granite 创建 RAG 管道

首先，我们为 Granite 执行 RAG 查询创建提示。我们使用 Granite 聊天模板，并提供 LangChain RAG 管道将替换的占位符值。

接下来，我们利用之前创建的 Granite 提示模板构建 RAG 管道。

from ibm_granite_community.langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_classic.chains.retrieval import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate

# Create a Granite prompt for question-answering with the retrieved context
prompt_template = ChatPromptTemplate.from_template("{input}")

# Assemble the retrieval-augmented generation chain
combine_docs_chain = create_stuff_documents_chain(
    llm=model,
    prompt=prompt_template,
)
rag_chain = create_retrieval_chain(
    retriever=vector_db.as_retriever(),
    combine_docs_chain=combine_docs_chain,
)

生成对问题的检索增强响应

管道使用查询从矢量数据库中查找文档，并将其用作查询的上下文。

from ibm_granite_community.notebook_utils import wrap_text

output = rag_chain.invoke({"input": query})

print(wrap_text(output['answer']))

太棒了！我们已经创建了一个 AI 应用程序，它能够利用源文档的文本和图片知识。

后续步骤

探索适用于其他行业的高级 RAG 工作流。
尝试使用其他文档类型和更大的数据集。
优化提示工程，以获得更好的 Granite 回复。

解锁生成式 AI + ML 的强大功能

了解如何自信地将生成式 AI 和机器学习融入您的业务中。