OpenAIでPDFファイルを要約する方法（サンプルプログラムをご紹介）

＜目次＞

OpenAIでPDFファイルを要約する方法（サンプルプログラムをご紹介）
　やりたいこと／概要
　サンプルプログラム
　動かし方

OpenAIでPDFファイルを要約する方法（サンプルプログラムをご紹介）

やりたいこと／概要

●やりたいこと

・PDFファイルを読み込んで、OpenAIを使い内容を要約する

●概要

①PDFのテキスト化

・Azureの「Form Recognizer」機能を使ってPDFを読込み、文字列を抽出。

（参考）

https://learn.microsoft.com/ja-jp/azure/applied-ai-services/form-recognizer/quickstarts/get-started-sdks-rest-api?view=form-recog-3.0.0&pivots=programming-language-python

↓

②要約処理

・指定の文字数（len(text) < order_wcount * 2）になるまで要約を続ける。

∟・要約する際、まずは一定の規模に分割（例：order_wcount * 5）する。

∟・分割したものを、それぞれ要約して短くなったもの同士を連結する

（図１１１）

＞目次にもどる

サンプルプログラム

●メイン処理：openai_pdf_summarize.py

import openai
import re
import time
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
from log_utils import prepare_logger
import env

FR_ENDPOINT = env.get_env_variable("FR_ENDPOINT")
FR_KEY = env.get_env_variable("FR_KEY")
openai.api_key = env.get_env_variable("OPEN_AI_KEY")
openai.api_base = env.get_env_variable("OPEN_AI_BASE")
openai.api_type = env.get_env_variable("OPEN_AI_TYPE")
openai.api_version = env.get_env_variable("OPEN_AI_VERSION")
model_deploy_name = env.get_env_variable("OPEN_AI_DEPLOY_NAME")

def analyze_general_documents(binary_file,logger):
    # -----------------------------------------
    # binaryファイルを解析し、ファイル中の文字列を抽出する
    # -----------------------------------------  

    # DocumentAnalysisClientクラスのインスタンスを作成
    # →Form Recognizerのサービスを使える様にする
    document_analysis_client = DocumentAnalysisClient(endpoint=FR_ENDPOINT, credential=AzureKeyCredential(FR_KEY))

    # ドキュメントの解析
    # →ドキュメントの取得方法は「ファイルから」と「URLから」の2通りがある
    # (1)ファイル形式のドキュメント
    poller = document_analysis_client.begin_analyze_document("prebuilt-document", binary_file)
    # (2)URLからドキュメントを取得
    # poller = document_analysis_client.begin_analyze_document_from_url("prebuilt-document", docUrl)
    
    # 戻り値：AsyncLROPoller のインスタンス
    # →poller オブジェクトで result() を呼び出して、 を AnalyzeResult返します。
    result = poller.result()
    # 結果格納用の配列
    text_of_doc = []
    # 結果は「ページ単位」に分かれている
    for page in result.pages:
        # 更にその中で「行単位」に分かれる
        for line_idx, line in enumerate(page.lines):
            # 行単位に「単語」の情報を抽出
            words = line.get_words()
            for word in words:
                # リストに追加
                text_of_doc.append(word.content)

    # 配列の中身を、1つの変数に格納しなおす       
    final_text = ""
    for text in text_of_doc:
        final_text += text

    # logger.debug(f"Azure Form Recognizerの文字認識結果: {final_text}")
    return final_text

def get_summary(text, order_wcount, logger):

    # 始めから、要約文字数×2 以下の場合
    if len(text) < order_wcount * 2:
        return text
    # 要約文字数×2 以上の場合
    while len(text) > order_wcount * 2:
        # 分割して要約した結果を連結した結果
        final_result = __get_summary_internal(text, order_wcount, logger)
        # while文の判定対象であるtextを更新
        text = final_result      
    return final_result 

# 分割して要約
def __get_summary_internal(text, order_wcount, logger):
    # TODO:要約の単位は何文字ずつに区切るのが最適か？
    split_size = order_wcount * 5
    # 上限に抵触しない範囲のカタマリに分割した配列を取得
    splited_text = split_text(text, split_size)
    entire_size = len(splited_text)   
    final_result = ""

    # 上限に抵触しない範囲のカタマリに分割した配列をループし、その単位で要約
    logger.info(f"全体サイズ： {entire_size}")
    for index, text in enumerate(splited_text):
        
        logger.info(f"内部要約回数： {index + 1} 、内部要約サイズ： {len(text)}...")        
        prompt = __get_system_prompt(order_wcount)
        prompt += text
        # max_tokensの設定方法
        # 　ひらがな／カタカナ／英数字 = 約2Token、漢字 = 約3Token
        # 　とすると、最大3*X Tokenが来る想定。
        # 　X=300の場合　3*300*10=6000
        response = ""
        try:
            response = openai.ChatCompletion.create(
                engine=model_deploy_name,
                messages=[{"role": "system", "content": prompt}],
                temperature=0,
                max_tokens=5000,
                top_p=0.95,
                frequency_penalty=0,
                presence_penalty=0,
                stop=None)
        except openai.error.InvalidRequestError:
            logger.error(
                f"openai.ChatCompletion.create() でエラーが発生しました。入力プロンプト : {prompt}")
            raise
        except openai.error.RateLimitError as e:
            logger.error(e)
            raise
        final_result += response['choices'][0]['message']['content']
        # time.sleep(60)
    logger.info(f"最終要約結果: {final_result}")
    return final_result

def split_text(text, split_size):
    # 文章を「。」「．」「！」「？」で分割して配列に格納
    # FIXME:「.」による分割はYYYY.MM.DD形式の日付が途中で分割されてしまう
    sentence_array = re.split("。|．|！|？|\.", text)
    result,chunk = [],""
    # OpenAIに渡せる「最大文字数」になるまで文章を連結し、その単位で配列に格納
    #   ・enumerate() 関数を使用することで、要素のインデックスと値の両方を取得できます。
    for index, sentence in enumerate(sentence_array):
        chunk += sentence
        if __check_if_next_iteration_exeeds_limit(chunk, index, sentence_array, split_size):
            result.append(chunk)
            chunk = ""
    return result

def __get_system_prompt(order_wcount):
    instructions =  "以下の【文章】を要約してください。" \
                    "要約する際のルールは以下です。" \
                    f"【ルール１】{order_wcount}文字以内に要約すること。" \
                    "【ルール２】要約時に内容と関連が無い内容が含まれている場合、結果からは除外してください。" \
                    "【ルール３】要約結果は箇条書きではなく、連続した文章として返却すること。" \
                    "【文章】\n"
    # logger.debug(f"システムプロンプト: {instructions}")
    return instructions

# 文字列がOpenAIに渡せる「最大文字数」に抵触しないか？のチェック
#   ・特殊メソッドは、Pythonの内部で特別な役割を果たすため、直接的に呼び出すことは一般的ではありません。
#   ・代わりに、特定の操作や構文が使用されるときに自動的に呼び出されます。
def __check_if_next_iteration_exeeds_limit(chunk, idx, sentence_array, split_size):
    # 次の要素がsentence_arrayの最後の要素の場合はTrue
    if idx + 1 == len(sentence_array):
        return True
    # 次の要素を連結し、split_sizeの値を超す場合はTrue、その他の場合はFalse
    return (len(chunk) + len(sentence_array[idx + 1])) >= split_size

def get_pdf_summary(logger):

    # PDFをテキスト化
    with open("./leaf18.pdf", "rb") as f:
        prompt_pdf_text = analyze_general_documents(f,logger)
    
    # 要約
    summary = get_summary(prompt_pdf_text,300,logger)

if __name__ == "__main__":
    logger = prepare_logger()
    get_pdf_summary(logger)

●ロギング：log_utils.py

import logging
def prepare_logger() -> logging.Logger:
    # ロガーの作成
    logger = logging.getLogger(__name__)
    # ログハンドラの作成
    handler = logging.StreamHandler()
    # ログフォーマットの設定
    formatter = logging.Formatter('xxxxxxxxxxxxx : %(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    # ログハンドラをロガーに追加
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    return logger

●環境変数：env_sample.py

def get_env_variable(key):

    env_variable_dict = {
        # ------------------------------------
        # Azure Form Recognizer
        # ------------------------------------
        "FR_ENDPOINT" : "",
        "FR_KEY" : "",

        # ------------------------------------
        # OpenAI
        # ------------------------------------
        "OPEN_AI_KEY" : "",
        "OPEN_AI_BASE" : "",
        "OPEN_AI_TYPE" : "",
        "OPEN_AI_VERSION" : "",
        "OPEN_AI_DEPLOY_NAME" : "",

        # ------------------------------------
        # App名：Slack_Python_Flask
        # ------------------------------------
        # BotユーザーID
        "BOT_USER_ID" : "",
        # Botトークン（Flask）
        "WEBAPPS_SLACK_TOKEN" : "",
        "WEBAPPS_SIGNING_SECRET" : "",

        # Botトークン（ソケットモード）
        "SOCK_SLACK_BOT_TOKEN" : "",
        "SOCK_SLACK_APP_TOKEN" : ""
    }
    ret_val = env_variable_dict.get(key, None)
    return ret_val

＞目次にもどる