langchainとDatabricksで(私が)学ぶRAG : 半構造化データに対するRAG①

# 導入

私が学ぶRAGの実質7回目です。シリーズ一覧は[こちら](https://qiita.com/isanakamishiro2/items/66411e1443cf78a2c1e8)。
今回は半構造化データに対するRAGの実践です。

# これは何？

Langchainの下記Blog内 Semi-Structured Data のセクションで解説されています。

https://blog.langchain.dev/semi-structured-multi-modal-rag/

ざっくり邦訳＋図を掲載。

> 非構造化ファイルの構文解析とマルチベクトル検索を組み合わせることで、半構造化データのRAGをサポートすることができます。テーブル要素の要約を生成し、自然言語検索により適しています。
> もしテーブルの要約がユーザーの質問に対して意味的類似性によって検索された場合、生のテーブルは上記のように回答合成のためにLLMに渡されます。

![semi-structure RAG](https://blog.langchain.dev/content/images/size/w1600/2023/10/image-16.png)

図中にあるように、例えばPDFデータを対象にRAGを実現しようとする場合、表なども含まれています。
表はそのままに扱わなければ十分意味をもった情報となりませんが、単純にテキスト化してチャンクにすると意味を失ってしまいます。
このようなケースにおいて、表は表のままに検索できるように、LLMに渡す必要があります。

LangchainのMulti-Vector Retrieverは、これを実現するために、「表に対しては、サマリした検索用の情報を生成し、それに対してベクトル検索させる。LLMに与える情報はサマリ元である表情報を渡す」ということを実現しています。

<br>

というわけで実践してみましょう。

**今回は非常に長くなってしまったので2回に分割します。**

この記事では要約含めた特徴量データを保管するところまで実施します。

なお、今回の内容は、以下のLangchain Templatesの内容を参考に実践します。
https://python.langchain.com/docs/templates/rag-semi-structured


DatabricksのDBRは14.1 ML、GPUクラスタで動作を確認しています。

# Step0. Package Install

今回は[Unstructured](https://github.com/Unstructured-IO)というpythonパッケージを使用します。
このパッケージを使うために必要な外部ライブラリを最初にインストールします。

tesseractを使えるようにするために、aptのリポジトリに追加。

```sh
%sh add-apt-repository ppa:alex-p/tesseract-ocr5 && apt update
```

tesseract等必要なライブラリをaptからインストール。

```sh
%sh apt install -y libmagic-dev tesseract-ocr libtesseract-dev tesseract-ocr-jpn tesseract-ocr-jpn-vert tesseract-ocr-script-jpan tesseract-ocr-script-jpan-vert poppler-data
```

popplerがうまくubuntuのapt上でインストールできなかったので、Debianのパッケージファイルをダウンロードして直接インストール。

```sh
%sh wget http://security.ubuntu.com/ubuntu/pool/main/p/poppler/libpoppler118_22.02.0-2ubuntu0.3_amd64.deb -P /tmp/pkg/ && \
  wget http://security.ubuntu.com/ubuntu/pool/main/p/poppler/poppler-utils_22.02.0-2ubuntu0.3_amd64.deb -P /tmp/pkg/ && \
  wget http://security.ubuntu.com/ubuntu/pool/main/p/poppler/libpoppler-dev_22.02.0-2ubuntu0.3_amd64.deb -P /tmp/pkg/ && \
  dpkg -i /tmp/pkg/libpoppler118_22.02.0-2ubuntu0.3_amd64.deb /tmp/pkg/poppler-utils_22.02.0-2ubuntu0.3_amd64.deb /tmp/pkg/libpoppler-dev_22.02.0-2ubuntu0.3_amd64.deb
```

tesseractで日本語データを扱えるようにするためのファイルをダウンロード。

```sh
%sh wget -P /usr/share/tesseract-ocr/4.00/tessdata https://github.com/tesseract-ocr/tessdata/raw/4.00/jpn.traineddata
```


最後に、pipから必要なパッケージをインストール。

```python
%pip install -U -qq transformers accelerate langchain langchainhub faiss-cpu unstructured[all-docs]
# CUDA 11.8用
%pip install https://github.com/casper-hansen/AutoAWQ/releases/download/v0.1.7/autoawq-0.1.7+cu118-cp310-cp310-linux_x86_64.whl
%pip install "databricks-feature-engineering"

dbutils.library.restartPython()
```

LLMはこのシリーズで共通のAWQ量子化フォーマットのファイルをTransformersで読み込みます。
そのためにAutoAWQをインストールしていますが、AutoAWQはCUDA 11.8版をWheelを指定してインストールしています。
(DatabricksのDBR ML 14台はCUDAのバージョンが11.8のため)

# Step1. Document Loading

今回は[準備編](https://qiita.com/isanakamishiro2/items/66411e1443cf78a2c1e8)で保管したデータは使わず、PDFファイルを最初から取り込みなおします。

まず、以下のサイトに掲載されている「①令和５年度概算契約書」のPDFファイルをダウンロード。

https://www.meti.go.jp/information_2/publicoffer/gaisan/r5keiyaku_format-3.html

```python
# PDFファイルをローカルにダウンロード
import os
import requests

pdf_url = "https://www.meti.go.jp/information_2/downloadfiles/r5gaisan-2_format.pdf"
local_dir = "/tmp/unstructured/pdf/"
local_path = local_dir + "r5gaisan-2_format.pdf"

resp = requests.get(pdf_url)

if not os.path.exists(local_dir):
    os.makedirs(local_dir)

with open(local_path, "wb") as f:
    f.write(resp.content)
```

```Unstructured```を使ってPDFファイルを取り込みます。

```python
import os
from unstructured.partition.pdf import partition_pdf

# tesseractの設定
os.environ["TESSDATA_PREFIX"] = "/usr/share/tesseract-ocr/4.00/tessdata"

raw_pdf_elements = partition_pdf(
    filename=local_path,
    extract_images_in_pdf=False,
    infer_table_structure=True,
    chunking_strategy="by_title",
    max_characters=4000,
    new_after_n_chars=3800,
    combine_text_under_n_chars=2000,
    image_output_dir_path="/tmp/unstructured/images/",
    languages=["jpn"],
)
```

これで、PDFファイルが読み込まれました。
取り込まれたデータはElementというチャンク化したデータの情報を含んだリストになります。

# Step2. Conver to DataFrame

このままでは取り扱いづらいので、DataFrame形式に変換します。

```python
import pyspark.sql.functions as F

from unstructured.documents.elements import Title, NarrativeText
from unstructured.staging.base import convert_to_dataframe

# idをUUID化
for r in raw_pdf_elements:
    r.id_to_uuid()

# Pandasデータフレームへ変換
pdf = convert_to_dataframe(raw_pdf_elements)

# 対象を最低限に絞る
pdf = pdf[["element_id", "type", "text", "page_number", "filename"]]

# Spark DataFrameへ変換
df = spark.createDataFrame(pdf)
# ノイズ文字を簡易除去
df = df.withColumn("text", F.regexp_replace(F.col("text"), "R5-G-2\n", ""))

display(df)

df.createOrReplaceTempView("doc")
```

出力結果は次のようになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/4153324b-4ff0-d623-9b94-8b69617d9b53.png)


typeがTableのデータが、PDFに含まれる表データを取り込んだデータです。

取込処理自体が結構時間がかかるため、一旦特徴量テーブルに保管しておきます。

```python
from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
feature_name = "training.llm.sample_doc_features2"

df = spark.table("doc")

doc_table = fe.create_table(
    name=feature_name,
    primary_keys="element_id",
    schema=df.schema,
    description="Unstructured PDF Data",
)

fe.write_table(name=feature_name, df=df, mode="merge")
```


# Step3. Chunking

保管したデータを読み直し、まずは通常のテキストデータをチャンキングします。
Unstructuredで読み込んだ時点でチャンキングはしているのですが、検索性能が上がるかなと思い、もう少し割ってみました。
チャンキング自体はこれまでと同様、シンプルなやり方で分割しています。

```python
from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
feature_name = "training.llm.sample_doc_features2"

df = fe.read_table(name=feature_name)
df.createOrReplaceTempView("feature_doc")
```

```python
from typing import Any
import pandas as pd
import pyspark.sql.functions as F
from pyspark.sql.functions import pandas_udf
from langchain.text_splitter import RecursiveCharacterTextSplitter


class JapaneseCharacterTextSplitter(RecursiveCharacterTextSplitter):
    """句読点も句切り文字に含めるようにするためのスプリッタ"""

    def __init__(self, **kwargs: Any):
        separators = ["\n\n", "\n", "。", "、", " ", ""]
        super().__init__(separators=separators, **kwargs)


@pandas_udf("array<string>")
def split_text(texts: pd.Series) -> pd.Series:

    # 適当なサイズとオーバーラップでチャンク分割する
    text_splitter = JapaneseCharacterTextSplitter(chunk_size=200, chunk_overlap=40)
    return texts.map(lambda x: text_splitter.split_text(x))


# type=CompositeElementのみチャンキング
df = spark.table("feature_doc")
df = df.withColumn("chunk", F.when(F.col("type") == F.lit('CompositeElement'), split_text("text")))

display(df)

df = df.cache()
df.createOrReplaceTempView("chunked_doc")
```


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/0ee4b32d-02b9-c782-5a8e-b90c9ce14c7c.png)

# Step4. Summarize Table Data

```type=="Table"```の表データからサマリテキストを生成します。
サマリの生成はLLMに実施させます。モデルは前回同様OpenChat 3.5を利用しました。
また、今回はSpark上で処理させるためにpandas_udfで実装しましたが、今回のボリュームでは一度Pandas Dataframe化して処理する方が速いと思います。


```python
import pyspark.sql.functions as F
import pandas as pd

@pandas_udf("string")
def summarize_texts(texts: pd.Series) -> pd.Series:

    from transformers import AutoModelForCausalLM, AutoTokenizer
    from transformers_chat import ChatHuggingFaceModel
    from langchain.schema.output_parser import StrOutputParser
    from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
    from langchain.prompts import ChatPromptTemplate
    from langchain.prompts.chat import (
        AIMessagePromptTemplate,
        HumanMessagePromptTemplate,
    )

    model_path = "/Volumes/training/llm/model_snapshots/models--TheBloke--openchat_3.5-AWQ"

    generator = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto")
    tokenizer = AutoTokenizer.from_pretrained(model_path)

    prompt_text = """You are an assistant tasked with summarizing tables and text. \ 
    Give a concise summary of the table or text. Table or text chunk: {element} """
    summarize_prompt = ChatPromptTemplate.from_messages(
        [
            HumanMessagePromptTemplate.from_template(prompt_text),
            AIMessagePromptTemplate.from_template(""),
        ]
    )

    summarize_model = ChatHuggingFaceModel(
        generator=generator,
        tokenizer=tokenizer,
        human_message_template="GPT4 Correct User: {}
",
        ai_message_template="GPT4 Correct Assistant: {}",
        repetition_penalty=1.2,
        temperature=0.1,
        max_new_tokens=2048,
    )
    summarize_chain = {"element": lambda x: x} | summarize_prompt | summarize_model | StrOutputParser()

    texts = list(texts)
    results = summarize_chain.batch(texts, {"max_concurrency": 2})

    return pd.Series(results)

df = spark.table("chunked_doc")

df = df.repartition(1)
df = df.withColumn("table_summary", F.when(F.col("type") == "Table", summarize_texts("text")))
df = df.cache()

display(df)
df.createOrReplaceTempView("summarized_doc")
```

以下のように、table_summaryという表データのサマリデータ列を追加しました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/168727/7bf6fd61-aa08-acca-ae38-5e78e1678b2a.png)

# Step5. Save as Features

加工データをDatabricksの特徴量テーブルとして保管。

```python
from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
feature_name = "training.llm.sample_doc_features3"

df = spark.table("summarized_doc")

doc_table = fe.create_table(
    name=feature_name,
    primary_keys="element_id",
    schema=df.schema,
    description="Unstructured PDF Data2",
)

fe.write_table(name=feature_name, df=df, mode="merge")
```

これでRAGに使えるデータ生成が完了しました。

# まとめ

次回に続きます。