PythonからOllama APIを使って、文章を要約する

PythonからOllama APIを使って、文章を要約する
Photo by Douglas Lopes / Unsplash

前回「Ollama APIを使用してみる」で、Ollama APIを試してみました。
今回は、PythonとOllama APIを使って、PDFファイルの要約をGemma3にしてもらおうと思います。

春期の情報処理試験も近いですので、それに関連するPDFの要約を作成してみましょう。

必要なライブラリのインストール

pip install aiohttp PyMuPDF

import fitzとあるのでpip install fitzとしがちですが、pip install PyMuPDFです。

Pythonコード(ファイル名: summarize_pdf_ollama.py)

import asyncio
import aiohttp
import fitz  # pip install fitzではなく、pip install PyMuPDF
import sys
import os
import time
import json
from datetime import datetime

OLLAMA_API_URL = "http://localhost:11434/api/generate"
MODEL_NAME = "gemma3:4b"

PROMPT_TEMPLATE = """
以下のテキストから、情報処理安全確保支援士試験に出題されると予想される情報のみを抽出して、簡単に解説してください。
出題される情報がなければ、何も出力しないでください。
また、要約文が出力された場合は、4つの選択肢のある選択問題を作成してください。
その選択問題の解答と解説を、選択問題の後につけてください。

テキスト:
{content}
"""

# 非同期処理の同時実行数を2に制限します。
SEM = asyncio.Semaphore(2)
# 処理に時間がかかるので、クライアントのタイムアウトを180秒に設定。
timeout = aiohttp.ClientTimeout(total=180)

async def query_ollama(session, chunk, retries=3):
    # 同時実行数を2にすると、たまにエラーになるので、エラーになったらリトライします。
    for attempt in range(retries):
        try:        
            async with SEM:
                payload = {
                    "model": MODEL_NAME,
                    "prompt": PROMPT_TEMPLATE.format(content=chunk),
                    "stream": False,
                    "format": {
                        "type": "object",
                        "properties": {
                            "要約": {
                                "type": "string"
                            },
                            "選択問題": {
                                "type": "string"
                            },
                            "解答と解説": {
                                "type": "string"
                            }
                        },
                        "required": [
                            "要約", "選択問題", "解答と解説"
                        ]
                    }
                }
                async with session.post(OLLAMA_API_URL, json=payload) as resp:
                    data = await resp.json()
                    print(data)
                    return data.get("response", "")
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            print(f"[{attempt + 1}/{retries}] retry ... error: {e}")
            await asyncio.sleep(5)
    print("Result: Error.")
    return ""
    
def extract_text_chunks(pdf_path, pages_per_chunk=5): #5ページ単位で要約します。
    doc = fitz.open(pdf_path)
    chunks = []
    for i in range(0, len(doc), pages_per_chunk):
        chunk_text = ""
        for page in doc[i:i+pages_per_chunk]:
            chunk_text += page.get_text()
        chunks.append(chunk_text)
    return chunks

async def main(pdf_path):
    chunks = extract_text_chunks(pdf_path)
    # print(chunks)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [query_ollama(session, chunk) for chunk in chunks]
        raw_results = await asyncio.gather(*tasks)
    
        summaries = []
        for result in raw_results:
            try:
                summary_obj = json.loads(result)
                summaries.append(summary_obj.get("要約", "") + "\n")
                summaries.append("選択問題:\n" + summary_obj.get("選択問題", "") + "\n")
                summaries.append("解答と解説:\n" + summary_obj.get("解答と解説", "") + "\n -------------------------")
            except json.JSONDecodeError:
                summaries.append(result)
                
        combined = "\n\n".join(summaries)

        script_dir = os.path.dirname(os.path.abspath(__file__))
        timestamp = datetime.now().strftime("%Y%m%d%H%M")
        filename = f"summary{timestamp}.txt"
        output_path = os.path.join(script_dir, filename)
        with open(output_path, "w") as file:
            file.write(f"{combined}\n")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 summarize_pdf_ollama.py <PDF File Path>")
        sys.exit(1)
    
    input_path = sys.argv[1]
    if not os.path.isfile(input_path):
        print(f"Can not find the file. {input_path}")
        sys.exit(1)

    start_time = time.time()
    asyncio.run(main(input_path))
    end_time = time.time()
    
    elapsed_time = end_time - start_time
    print(f"\n実行時間: {elapsed_time:.2f} 秒")

Pythonを実行

$ python summarize_pdf_ollama.py /your_path/your_pdf_file.pdf

your_pathyour_pdf_file.pdf の部分は、自分の環境に合わせて変更してください。

完了すると、Pythonのファイルと同じディレクトリに、summary.txtファイルが作成されます。

実行結果

このテキストは、セキュリティ技術、特に認証技術とデジタル署名について解説しています。主な内容は以下の通りです。

* **認証の3要素:** 記憶、所持、生体という3つの要素による認証の概念と、2要素認証の重要性が説明されています。
* **PKI (Public Key Infrastructure):** 公開鍵基盤の仕組み、認証局(CA)の役割、デジタル証明書の発行と利用について解説されています。
* **デジタル署名:** 真正性(本人証明)と改ざん検出の両方を実現するデジタル署名の仕組みと、その利用方法が説明されています。
* **PKIの仕組み:** CAが発行するデジタル証明書、プライベートCA、政府認証基盤(GPKI)といった用語も紹介されています。
* **デジタル証明書:** Webブラウザで確認できるデジタル証明書の例も示されています。


選択問題:
以下のうち、デジタル署名によって実現される主な機能はどれですか?

A. 暗号化されたデータの送受信
B. 真正性(本人証明)と改ざん検出
C. 生体認証によるアクセス制御
D. 秘密鍵によるデータの保護

正解: B


解答と解説:
正解はBです。デジタル署名は、送信者の秘密鍵でハッシュ値を暗号化することで、データの改ざんを検出し、送信者の真正性を証明する機能を実現します。Aは暗号化、Cは生体認証、Dは秘密鍵の保護です。
 -------------------------

このテキストは、セキュリティ技術に関する情報を提供しています。主な内容は以下の通りです。

* **ワンタイムパスワード (OTP)**: ユーザー認証を強化するための技術で、通常、SMSや認証アプリを通じて送信される一回しか使えないコードを使用します。
* **SAML (Security Assertion Markup Language)**: 異なるWebサイト間での認証を実現するための標準化されたフレームワークです。
* **OAuth**: 認可のプロトコルで、ユーザーが第三者アプリケーションにリソースへのアクセスを許可する際に使用されます。
* **OpenID Connect**: OAuth 2.0プロトコルの上にアイデンティティレイヤを追加したプロトコルで、認証機能を提供します。
* **DNSSEC (DNS Security Extensions)**: DNS でのセキュリティに関する拡張仕様で、DNS 応答レコードの偽造や改ざんを検出します。


選択問題:
以下のうち、ユーザー認証を強化するための技術として最も適切なものはどれですか?

A. DNSSEC
B. SAML
C. ワンタイムパスワード (OTP)
D. OAuth


解答と解説:
正解: C. ワンタイムパスワード (OTP)

解説: ワンタイムパスワード (OTP) は、ユーザー認証を強化するための技術で、通常、SMSや認証アプリを通じて送信される一回しか使えないコードを使用します。他の選択肢は、認証技術ではなく、セキュリティ拡張仕様または認証プロトコルです。
 -------------------------

変なところもありましたが、PDFを読む前の概要把握には、ちょうど良いと思います。

実行速度について

Semaphore(2) のとき、実行時間: 888.95秒

はじめは調子が良かったのですが、最後の方はリソース不足なのか、エラーが多発していました…

Semaphore(1) のとき、実行時間: 1172.61 秒

Ollama は 内部でリクエストをキュー(待ち行列)に積んで順番に処理するみたいなので、非同期処理の効果はあまりなさそうですが、結果を見ると一応効果はあったようです。