書類の整合性検証をAIで効率化してみた

こんにちは、AILabの神﨑睦人です。

AILabでは、AIを使用して文書管理における業務効率化を目指したツールを開発しています。今回は、書類における矛盾をAIを用いて指摘するツールの開発を行いました。

各自治体が管理するような申請書や契約書の構造は統一されておらず、手動での記入情報は期間や生年月日などの記入段階でのミスをすべてのフォーマットに対応する形で検知するのは容易ではありません。

書類の記入段階でのミスを検知するための本ツールを紹介します。

目次

開発環境

Python

PDFのPNGへの変換やCSV形式での保存、AzureのAPI使用を含め、豊富なライブラリを持つPythonを使用して開発工程を効率化しました。

Azure AI Document Intelligence

AzureのDocument Intelligenceを使用して、PDFの文字起こしと、オブジェクトの構造の抽出を行います。テキストに加えて選択肢の情報も抽出することで、どの選択肢に丸がついているかも確認します。

Azure Open AI Service

同様に、Azure Open AIでは、PNGに変換した記入前後の書類と、Document Intelligenceの文字起こしや選択肢の情報をAIに渡します。必要箇所が記入されているか、日付に矛盾がないかなどを確認して出力します。

実装フロー

実装におけるポイントを解説します。

ディレクトリ構成は以下となります。

project_root/
├── src/
│   ├── main.py
│   ├── encoder/
│   │   └── encode.py
│   ├── extracter/
│   │   └── extract_problem.py
│   └── converter/
│       ├── pdf_to_png.py
│       └── txt_to_json.py
├── output_csv/
├── img/
├── json/
└── pdf/
    ├── 書類サンプル_before.pdf # 記入前のPDF
    └── 書類サンプル_after.pdf  # 記入後のPDF

① PDFをPNGへ変換

顧客から受領したPDF書類をページごとにPNG画像へ変換します。 pdfフォルダ内のPDFファイルが全てPNG画像に変換される仕様となっています。

② Document Intelligenceによる文字起こし・選択肢抽出

ここではまだPNG画像は使用せず、PDF形式のままAzure AI Document Intelligenceに送信し、ページ内の文字列や選択肢情報をJSON形式で取得します。 取得したJSONはjsonフォルダ内に保存されます。

これにより、どの選択肢が選ばれているかを取得することができます。

③ Open AIを用いた内容整合性チェック

①でPNG画像に変換した書類と、②でDocument Intelligenceによって取得した選択肢情報をOpen AIに送信し、整合性のチェック結果を取得します。

JSON形式を含む文字列を返すようプロンプトで指定することで、後から扱いやすい形で取得することができます。

ただし、Open AIが返すのは文字列なので、返却された文字列からJSON形式の箇所を抽出してパースする必要があります。次の手順で詳しく紹介します。

④ 文字列として得られた結果からJSON形式の箇所を抽出

Open AIに対して、応答としてJSON形式で返すようプロンプトを調整しても、応答に余計な文章や文字が含まれることがあるので、必要なJSONの箇所のみを抽出してパースします。

Open AIによる出力例

以下が指摘内容のJSONになります。
'
[
  {
    "項目名" : "",
    "記載内容": "",
    "指摘内容": "",
    "修正案": ""
  },
...
]
'

本来であれば

[
  {
    "項目名" : "",
    "記載内容": "",
    "指摘内容": "",
    "修正案": ""
  },
...
]

の部分のみが欲しいので、大括弧の中身を抽出して、JSONとして扱えるようにパースします。

⑤ 指摘内容をCSV形式で保存

晴れて指摘内容をJSON形式で扱えるようになったので、このJSONをCSVに保存します。

使用方法

ここからは、本ツールで使用するコードを紹介します。

環境変数の設定

まず、環境変数を以下のように設定します。

# Azure Open AI
AZURE_API_KEY=      #APIキー 
AZURE_ENDPOINT=     #エンドポイント
AZURE_DEPLOYMENT=   #デプロイ名
API_VERSION=        #APIバージョン

# Document Intelligence
AZURE_DOC_ENDPOINT=  #エンドポイント
AZURE_DOC_KEY=       #APIキー

PDFからPNGへの変換

はじめに、PDFをPNGに変換します。

import os
import pypdfium2 as pdfium

def convert_pdf_to_png(pdf_file, output_folder):
    os.makedirs(output_folder, exist_ok=True)
    pdf_name = os.path.splitext(os.path.basename(pdf_file))[0]

    # PDFをオープン
    pdf = pdfium.PdfDocument(pdf_file)
    output_paths = []

    for page_num in range(len(pdf)):
        # dpi=300相当, scale=4.0/72*300 ≒ 16.66
        page = pdf[page_num]
        # scale: 4.0倍相当
        image = page.render(scale=4.0).to_pil()
        output_path = os.path.join(output_folder, f"{pdf_name}_page_{page_num + 1:03}.png")
        image.save(output_path)
        output_paths.append(output_path)

    print(f'Converted {pdf_file} to PNG files.')
    return output_paths

PNG画像ファイルをbase64エンコード

PNG画像ファイルをbase64エンコードし、データURI形式の文字列に変換して返します。

import base64

def encode_image_base64(file_path):
    with open(file_path, "rb") as image_file:
        encoded = base64.b64encode(image_file.read()).decode("utf-8")
    return f"data:image/png;base64,{encoded}"

Open AIとDocument Intelligenceの呼び出し

Open AIとDocument Intelligenceを使用するコードを記述します。

ファイル名はextract_problem.pyとしています。

from converter.txt_to_json import convert_to_json
import csv
import os
import math
import json

def center(polygon):
    # polygon=[x1,y1,x2,y2,x3,y3,x4,y4]
    x = sum(polygon[::2]) / 4
    y = sum(polygon[1::2]) / 4
    return (x, y)

def distance(p1, p2):
    return math.hypot(p1[0] - p2[0], p1[1] - p2[1])

def extract_selection(client, file_path):
    base_dir = os.path.dirname(__file__)
    json_dir = os.path.join(base_dir, "../..", "json")
    os.makedirs(json_dir, exist_ok=True)

    output_path = os.path.join(json_dir, "extracted_layout.json")

    result_messages = []

    # レイアウト解析&JSON保存(なければ解析、あればスキップしてそのまま読み込み)
    if not os.path.exists(output_path):
        with open(file_path, "rb") as f:
            poller = client.begin_analyze_document(
                model_id="prebuilt-layout",
                body=f
            )
            result = poller.result()

        result_dict = result.as_dict()
        output = {
            "status": "succeeded",
            "analyzeResult": result_dict
        }

        with open(output_path, "w", encoding="utf-8") as f:
            json.dump(output, f, ensure_ascii=False, indent=2)

        print(f"解析結果をJSONに保存しました。")
    else:
        print(f"{output_path} は既に存在します。再解析をスキップします。")


    with open(output_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    selection_marks = []
    pages = data["analyzeResult"].get("pages", [])
    for page in pages:
        selection_marks.extend(page.get("selectionMarks", []))

    if not selection_marks:
        print("selectionMarks が見つかりません。")

        mark_idx = 1  # 選択マークの連番
        for page in pages:
            words = page.get("words", [])

            for mark in selection_marks:
                if mark.get("state") != "selected":
                    continue

                mark_center = center(mark["polygon"])

                closest_word = None
                min_dist = float("inf")

                for word in words:
                    text = word.get("content", "").strip()
                    polygon = word.get("polygon", [])
                    if not text or not polygon:
                        continue

                    word_center = center(polygon)
                    dist = distance(mark_center, word_center)

                    if dist < min_dist:
                        min_dist = dist
                        closest_word = text

                result_messages.append(
                    f"[選択マーク {mark_idx}] → 最も近い文字: 「{closest_word}」 距離: {round(min_dist, 2)} px"
                )
                mark_idx += 1

    return "\n".join(result_messages)


def extract_problem(client, deployment, image_before_url, image_after_url, user_input, selection):

    # プロンプト例、ここでJSON形式での出力を指定
    prompt = f"""
            画像を読み取り、記入内容を確認して誤りを指摘してください。

            渡す画像:

            1枚目:原本(記入前)

            2枚目:記入後の書類 

            画像内の選択肢情報です:
            {selection}

            [
                {{
                    "項目名" : "[項目名]",
                    "記載内容": "指摘箇所(修正前)",
                    "指摘内容": "指摘内容",
                    "修正案": "修正案"
                }},
                ...
            ]
            """

    response = client.chat.completions.create(
        model=deployment,
        messages=[
            {"role": "system", "content": prompt},
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": user_input},
                    {"type": "image_url", "image_url": {"url":image_before_url}},
                    {"type": "image_url", "image_url": {"url":image_after_url}}
                ]
            }
        ],
        max_tokens=4096,
        temperature=1.0,
        top_p=1.0
    )
    return response

def save_as_csv(response, output_csv, write_header=False):
    os.makedirs(os.path.dirname(output_csv), exist_ok=True)
    json_data = convert_to_json(response.choices[0].message.content)
    mode = "w" if write_header else "a"
    with open(output_csv, mode, newline="", encoding="utf-8") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=["項目名", "記載内容", "指摘内容", "修正案"])
        if write_header:
            writer.writeheader()
        writer.writerows(json_data)

JSON形式への変換

Open AIからのレスポンスに対して、大括弧内を抽出します。

import re
import json

def convert_to_json(text):
    text = re.sub(r'^json\s*', '', text)
    text = re.sub(r'\s*$', '', text)
    start = text.find("[")
    end = text.rfind("]")
    if start == -1 or end == -1 or end < start:
        return None
    array_str = text[start:end+1]
    array_str = re.sub(r'\\(?!["\\/bfnrtu])', '', array_str)
    return json.loads(array_str)

main.pyでの処理の呼び出し

ここまでで記述したコードはmain.pyで呼び出します。 今回は、記入前の書類を書類サンプル_before.pdf、記入後の書類を書類サンプル_after.pdfであるものとしています。

from encoder.encode import encode_image_base64
from extracter.extract_problem import extract_selection, extract_problem, save_as_csv
from converter.pdf_to_png import convert_pdf_to_png
from dotenv import load_dotenv
from openai import AzureOpenAI
from azure.core.credentials import AzureKeyCredential
from azure.ai.documentintelligence import DocumentIntelligenceClient
import os
import glob
import unicodedata

# .envファイルを読み込む
load_dotenv()

# 環境変数の取得
openai_endpoint = os.getenv("AZURE_ENDPOINT")
deployment = os.getenv("AZURE_DEPLOYMENT")
subscription_key = os.getenv("AZURE_API_KEY")
api_version = os.getenv("API_VERSION")
doc_intelligence_endpoint = os.getenv("AZURE_DOC_ENDPOINT")
doc_intelligence_key = os.getenv("AZURE_DOC_KEY")

# Azure OpenAIクライアント作成
openai_client = AzureOpenAI(
    api_version=api_version,
    azure_endpoint=openai_endpoint,
    api_key=subscription_key,
)

# Azure Document Intelligenceクライアント作成
doc_intelligence_client = DocumentIntelligenceClient(
    endpoint=doc_intelligence_endpoint,
    credential=AzureKeyCredential(doc_intelligence_key)
)

def normalize_filename(filename):
    return unicodedata.normalize("NFD", filename)

base_filename = normalize_filename("書類サンプル")

# ここで質問内容を定義
user_input = "記入内容で記入形式のミスや整合性に矛盾が見つかる部分をリストアップして下さい。"


def find_page_pairs(img_folder, base_filename):
    before_glob = os.path.join(img_folder, f"{base_filename}_before_page_*.png")
    after_glob  = os.path.join(img_folder, f"{base_filename}_after_page_*.png")
    before_imgs = sorted(glob.glob(before_glob))
    after_imgs  = sorted(glob.glob(after_glob))

    before_map = {page_from_filename(p): p for p in before_imgs}
    after_map  = {page_from_filename(p): p for p in after_imgs}
    pages = sorted(set(before_map.keys()) & set(after_map.keys()))
    return [(before_map[p], after_map[p], p) for p in pages]

def page_from_filename(fname):
    return os.path.splitext(fname)[0].split("_")[-1]

if __name__ == "__main__":
    pdf_folder = "../pdf"
    img_folder = "../img"
    output_dir = "../output_csv"
    selection = ""

    output_csv = os.path.join(output_dir, "check.csv")

    pdf_files = glob.glob(os.path.join(pdf_folder, "*.pdf"))

    for pdf_file in pdf_files:
        output_path = convert_pdf_to_png(pdf_file, img_folder)
        print(output_path)

    page_pairs = find_page_pairs(img_folder, base_filename)

    first_page = True
    for before_path, after_path, page in page_pairs:
        print(f"Processing: {before_path}, {after_path}")

        image_before_url = encode_image_base64(before_path)
        image_after_url = encode_image_base64(after_path)

        selection = extract_selection(doc_intelligence_client, after_path)
        response = extract_problem(openai_client, deployment, image_before_url, image_after_url, user_input, selection)

        save_as_csv(response, output_csv, write_header=first_page)
        first_page = False

    print("csvへの記述が完了しました。")

スクリプトの実行

スクリプトは以下のコマンドで実行します。

python main.py

実行結果

python main.py
Converted ../pdf/書類サンプル_after.pdf to PNG files.
['../img/書類サンプル_after_page_001.png']
Converted ../pdf/書類サンプル_before.pdf to PNG files.
['../img/書類サンプル_before_page_001.png']
Processing: ../img/書類サンプル_before_page_001.png, ../img/書類サンプル_after_page_001.png
解析結果をJSONに保存しました。
csvへの記述が完了しました。

output_csvフォルダ内に作成されたCSVファイルに処理結果が保存されます。

今後の展望

現在はオブジェクトをOpen AIでも認識させるために画像化していますが、今後はOpen AIにDocument Intelligenceで取得したオブジェクト情報を高精度で読み込ませることで処理結果の精度向上とPNG処理の省略を図る予定です。