CloudFunctions(Python)でChatGPTを使った要約LINEBotを作ってみる

@tamina_ryosukeさんの記事

に影響を受け、流行っているChatGPTを使った要約LINEBotをPythonで作ってみました。

グループ内の特定ユーザーの連投を要約できるようにします。

Botの流れ

1. LINEからメッセージをFunctionsで受け取ってFirestoreに溜める
2. LINEから要約を指示したメッセージを受け取ったら
   1. 溜めたメッセージにURLが含まれる場合、スクレイピングする
      1. スクレイピング結果を何回かChatGPTに送る(要約)
   2. ChatGPTのスクレイピング要約結果と元のメッセージ(URL削除)を混ぜてもう一度ChatGPTに送る
   3. 要約結果をLINEに送る
   4. 溜めたメッセージを削除する

事前準備

説明記事も多いため具体的な手順は割愛します。

• Firebaseプロジェクトの作成
  • Functions、Firestore、Secret Manager を使用します。
• LINE Developersアカウントの作成

• OpenAI APIキーの取得
  • 連続でAPIを利用する場合は、請求情報を登録する必要があります。

• Python 3.10以降の環境構築
  • Python 3.9だとエミュレータが上手く動作しませんでした。

プロジェクトセットアップ

1. 最新のfirebase-toolsをインストールするとfirebase initの選択肢にPythonが追加されます。Functions、Firestore、Emulatorを選択してください。
2. functions/venv/Scripts/activate.batをcmd等で起動
3. functionsフォルダに移動し、pip install -r requirements.txtでインストール
   プロジェクトで使用するライブラリはrequirements.txtに記入
4. functionsフォルダに.secret.localをして以下を記入

LINE_KEY=ここにLINEトークン
OPENAI_KEY=ここにOPENAI APIトークン
LINE_SEND_ID=ここに送信先LINEユーザーID
LINE_TARGET_ID=ここに受信対象LINEユーザーID(LINE_SEND_IDと同じでOK)

.secret.localはエミュレータで使用します。

5.Secret Managerに↑のOpenAI APIトークンなどを登録
以下が参考になります。

6.コードを追加します。

from firebase_functions import https_fn
from firebase_admin import initialize_app
import sys, os
import time
import json
import openai
sys.path.append(os.path.dirname(os.path.abspath("__file__"))) #これ無いとエミュレータで下層ファイルを読み込めなかった。
from libs import LineApiResponseData,LineApiRequest,firestore,scraping_and_summarize_with_gpt
initialize_app()

@https_fn.on_request(secrets=["LINE_KEY","LINE_SEND_ID","LINE_TARGET_ID","OPENAI_KEY"])
def line_webhook(req: https_fn.Request) -> https_fn.Response:
    data = json.loads(req.data.decode("utf-8"))
    openai.api_key =   os.environ["OPENAI_KEY"]
    line_key = os.environ["LINE_KEY"]
    line_send_id = os.environ["LINE_SEND_ID"]
    line_target_id = os.environ["LINE_TARGET_ID"]
    print(data)
    line_res = LineApiResponseData(data)
    event = line_res.events[0] if len(line_res.events) > 0 else None
    #メッセージがあるならそれに対応する処理を行う
    if event is not None:
        messages = firestore.get_message_list()
        if event.message.text == "要約して":
            res_text = "要約できるメッセージがありません。"
            if len(messages) > 0:
                now = time.time()
                pre_text =  "\n".join([m["text"] for m in messages if now - m["timestamp"] / 1000 < 86400]) #24時間(86400)経過していないメッセージのみ抽出
                res_text = scraping_and_summarize_with_gpt(pre_text)
            line_req = LineApiRequest(line_key)
            line_req.push_message(line_send_id,text=res_text)
            firestore.reset_message_list()# 発言の記録をリセット
        elif event.source.user_id == line_target_id:
            messages.append(event.to_dict())
            firestore.set_message_list(messages)
    return https_fn.Response(f"Recieved")

libs内に必要な処理を置いています(後述)。
@https_fn.on_requestにあるsecretsは指定しなくてもエミュレータは動作しますが、指定しないと本番環境でSecret Managerに登録した値が環境変数へ反映されません。

eventsが複数届くケースを捨てて、最初のeventのみで処理しています。
messages = firestore.get_message_list()で過去のメッセージをfirestoreから取得し、今回届いたメッセージが要約指示の場合は要約処理を、要約指示ではないならmessagesに今回のメッセージを追加してfirestoreに再度保存しています。

7.firebase emulators:startでエミュレータを起動し、正常に動作するならデプロイします。
エミュレータ用にLINE Messaging APIのダミーデータ作っておくと良いです。ダミーは次の関数をデプロイし、デプロイ後に生成されるURLをwebhookに登録して適当にLINE Botとお話してればGoogle Cloudのログに出力されます。

@https_fn.on_request()
def test(req: https_fn.Request) -> https_fn.Response:
    print(req.data)
    return https_fn.Response(f"ok")

Firestoreの読み書きについて

あんまり深く考えてません。1つのドキュメントでなんとかしてます。

from firebase_admin import firestore
from typing import List

doc_path = "line/talk"
def get_message_list()->List[str]:
    db = firestore.client()
    talk = db.document(doc_path).get().to_dict()
    if talk is None:
        return []
    else:
        return talk["message_list"]

def set_message_list(lst:list):
    db = firestore.client()
    db.document(doc_path).set({
        "message_list":lst
    })

def reset_message_list():
    db = firestore.client()
    db.document(doc_path).set({
        "message_list":[]
    })

LineApiRequestやLineApiResponseDataについて

ただ辞書形式のjsonが扱いづらかったのでクラスにしただけです。

• line.py - GitHub

ChatGPTとスクレイピングについて

ChatGPTは要約したいテキストに含まれるURLをクロールしてくれないので自分でスクレイピングします。

スクレイピング

import requests
import re
from bs4 import BeautifulSoup

#再帰的にテキスト部分を抜き取る(改行調整のため親ノードでget_textを使用しない)
def get_node_text(node):
    text = ""
    children = node.findChildren(recursive=False)
    if (node.name == "body" or node.name == "div" or node.name == "p") and len(children) > 0:
        text = "".join([get_node_text(child) for child in children]) + "\n"
    elif node.name == "br":
        text = "\n"
    else:
        text = node.get_text().strip()
    return text

def get_text(url:str):
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')
    for script in soup(["script", "style","header","footer","img","nav"]):
        script.decompose()
    #lines = [line for line in lines if line != ""] #余計な改行を消すけど可読性下がる
    text = get_node_text(soup.find("body"))
    text = re.sub("\n+", "\n", text)
    text = re.sub('[ 　]+', ' ', text)
    return text

とりあえず本文のみ取れます。要約にインデントは重要ではないと判断したので文字数の節約のため消しています。

ChatGPT

ちょっと雑なプロンプトですがChatGPTに要約をお願いしています。
スクレイピング記事数がnとしてn+1回要約を行っています。wikiなど文字数が多すぎるページは2000文字の制限をかけています。

import openai
def summarize_with_gpt(text:str,limit_text_size = 2000,system_text="要約してください。"):
    text =  text[:limit_text_size] #過剰な問い合わせ防止
    response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system", "content": system_text},
        {"role": "user", "content": text[:2000]}
    ]   
    )
    res_text = response.choices[0].message.content
    return res_text

# %%
def scraping_and_summarize_with_gpt(text:str):
    limit_scraping_list = 5
    limit_tmp_text_size = 3000
    text_url_lst = split_url(text) #テキストとURLが分離されたリスト
    tmp_text = ""
    scraping_list = []
    is_contained_url = False
    for text in text_url_lst:
        if text.startswith("http"):
            tmp_text += "[URL削除済]"
            scraping_list.append(get_text(text))
            is_contained_url = True
        else:
            tmp_text += text
    if is_contained_url:
        tmp_text += "\n ■URL削除済は以下の内容のURLでした。\n"
    
    scraping_list = scraping_list[:limit_scraping_list]
    res_gpt_list = []
    for sc_text in scraping_list:
         res_gpt_list.append(summarize_with_gpt(sc_text))
    tmp_text += "\n\n".join(res_gpt_list)
    tmp_text = tmp_text[:limit_tmp_text_size] #結合後のテキスト最大サイズ
    result_text = summarize_with_gpt(tmp_text,system_text="要約してください。内容は、複数の話題から構成されている場合があります。")
    return result_text

def split_url(text:str):
    pattern = "(https?://[A-Za-z0-9_/:%#$&?()~.=+-]+?(?=https?:|[^A-Za-z0-9_/:%#$&?()~.=+-]|$))"
    return [text for text in re.split(pattern, text) if text != ""]

おわりに

Firebaseで要約LINEBotができました。Python対応のCloud Functionsを触れたので満足です。エミュレータはまだPythonをサポートしていないと思ってましたが使えました。

他参考