GW中に開発した、ボケをおすすめしてくれるLINEアプリをエンジニア向けに解説する。
今回つくったのはボケてをAI検索するLINEアプリ 〜AIボケテンダー「エイジ」〜
LINEで入力された文章(お題)について、意味的に近い回答(ボケ)をピックアップしてレコメンドする仕様だ。この記事ではOpenAIのAPIでこのようなアプリを作りたい人向けの解説をソース付きで書きたいと思う。
開発の手順は下記の通り
①OpenAIのAPIライセンスを取得
②OpenAI API Embeddingsでボケデータをベクター化
③GCPのCloud Functionsのアカウント発行
④LINE Developersのアカウント発行
⑤CGP上でLINE BOT APIの開発
①、③、④についてはすでにネットにたくさんの記事があるので省略する
②OpenAI API Embeddingsでボケデータをベクター化
OpenAIはたくさんの機能をAPIとして提供している。
今回はたくさんのボケの中から、「入力された文章」と「ボケてのボケの文章」とマッチ度が高い文章検索が求められている。
更には大量の処理済みボケデータをOpenAIのサーバ側にはおかず、自社サーバ(GCP)に「組み込む」ことによって、高速性や低コストでの実現を図っている。
それを実現するOpenAIのAPIがEmbeddingsだ。
EmbeddingsのAPIドキュメント
https://platform.openai.com/docs/guides/embeddings
簡単に説明すると
・事前に文章DBをAPIでベクター処理化してローカルに保存できる
(この際の言語処理力がOpenAIの真価)
・そのデータ(CSV)をGCP上にアップしておく
・検索時に入力された文章をAPIでベクター化し、保存済みベクター化データに対して検索マッチ度が高い順にピックアップして表示させる
ベクター化(数値化)して保存・比較するから数十万のデータに対する検索も一瞬でOK
処理前のボケてデータ【bokeData.csv】
id,text 103973108,を乗せずに海外 103971848,殿、毒見が終わりました。 103971447,バターとチーズ
ボケidとボケの内容のみのシンプルな構成
ベクター処理済みボケてデータ【outputVec.csv】
id,text 103973108,を乗せずに海外,[-0.025080611929297447, -0.012434509582817554,..(省略)] 103971848,殿、毒見が終わりました。,[-0.005088195204734802, -0.018677828833460808, ..(省略)] 103971447,バターとチーズ,[-0.010187446139752865, -0.009211978875100613, ..(省略)]
後半の数値の羅列がいわゆるベクター化された数値データ。↑では2項目ずつだけに省略してるが、その後このような数値が一行あたり1500項目も並ぶ。これらの数値がボケの数文字のテキストをいろいろな方向(ベクトル)に強弱を表現したデータということだ。
データ処理用のソースコード【dataset.py】
pythonでコンソールからこれを実行すれば、同フォルダにあるbokeData.csvを読み込み、OpenAIのAPIに接続しベクター変換処理をして、OutputVec.csvに保存する。
import csv import openai import pandas as pd import tiktoken import math from openai.embeddings_utils import get_embedding # OpenAI APIキーを設定(本番運用時は環境変数にセットすること) openai.api_key = "ここにOpenAIのAPIキーをいれます" # 入力データCSVファイル名 input_csv_path = "./bokeData.csv" #input_csv_path = "./data/sample100.csv" # Embeddedベクトルデータとして出力するファイル名 output_csv_path = "./outputVec.csv" # APIにわたすモデル名など embedding_model = "text-embedding-ada-002" embedding_encoding = "cl100k_base" max_tokens = 8000 # 入力CSVファイルを読み込む df = pd.read_csv(input_csv_path) df = df[["id", "text"]] df = df.dropna() df["combined"] = ( "id: " + df.id.astype(str).str.strip() + "; text: " + df.text.str.strip() ) df.head(2) encoding = tiktoken.get_encoding(embedding_encoding) # 分割数の設定(1000件ずつ) chunk_size = 1000 total_rows = len(df) num_chunks = math.ceil(total_rows / chunk_size) #APIを叩いてベクトルデータにしてCSVへ保管 for i in range(num_chunks): start_index = i * chunk_size end_index = start_index + chunk_size chunk_df = df.iloc[start_index:end_index] chunk_df["embedding"] = chunk_df.combined.apply(lambda x: get_embedding(x, engine=embedding_model)) # 結果をCSVに付け足して保存 if i == 0: chunk_df.to_csv(output_csv_path, index=False) else: chunk_df.to_csv(output_csv_path, mode='a', header=False, index=False) # 進捗状況を表示 print(f"書き込みました: {end_index if end_index <= total_rows else total_rows} / {total_rows} 行") print("埋め込みが保存されました:", output_csv_path)
例えば数万件のデータなら数時間で実行が完了する。
出力されるCSV中身は小数点だらけのベクトルデータ。データ容量はだいぶ増える。例えば元データが10万件、10MBだとしたら、出力データは10万件のままで、容量は50倍の500MB程度の大きさにもになる。
⑤CGP上でLINE BOT APIの開発
ベクトルデータが作成できたら、それをGCPのCloud Storageにアップして、Cloud Functionsから参照できるようにしよう。
そして、Cloud Functionsにデプロイするpythonスクリプトは下記の通り。
LINEへのチャット投稿をトリガーとしてLINEにwebhook登録してあるこのpythonのAPIで、入力された内容とマッチ度の高いボケをcsvからピックアップしてLINEにチャットで回答する。
LINE BOTアプリとしてGCP上で待機するAPI【main.py】
import os import sys import openai import json import pandas as pd import numpy as np from io import BytesIO from google.cloud import storage #Gクラストレージからcsvを読み取る from flask import make_response, abort, jsonify from linebot import ( LineBotApi, WebhookHandler ) from linebot.exceptions import ( InvalidSignatureError ) from linebot.models import ( MessageEvent, TextMessage, TextSendMessage ) #LINE関連 LINE_CHANNEL_SECRET = "ここにLINEのキーをいれます" LINE_CHANNEL_ACCESS_TOKEN = "ここにLINEのアクセストークンをいれます" # OpenAI APIキーを設定 openai.api_key = "ここにOpenAIのAPIキーをいれます" # OpenAI APIのモデル名 embedding_model = "text-embedding-ada-002" #Gクラバケット名とcsvファイルパス GC_BUCKET = "bokedata" GC_CSV = "outputVec.csv" #LINEへの接続設定 line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN) handler = WebhookHandler(LINE_CHANNEL_SECRET) # Cloudストレージのデータファイルのパス storage_client = storage.Client() # Cloud Storageのバケット名とcsvファイル名 bucket = storage_client.get_bucket(GC_BUCKET) # ベクトル変換済みcsvデータをバイナリから取得 blob = bucket.blob(GC_CSV) content = blob.download_as_bytes() df = pd.read_csv(BytesIO(content)) ##ローカルcsvでのデバッグ用 # CSVファイルを読み込む #df = pd.read_csv(datafile_path) # データファイルのパス #datafile_path = "../data/outputVec.csv" # embeddingカラムの文字列を評価し、NumPy配列に変換する df["embedding"] = df.embedding.apply(eval).apply(np.array) from openai.embeddings_utils import get_embedding, cosine_similarity # ボケを検索する関数 def search_boke(df, search_query, n=3, pprint=True): # ボケデータからembeddedベクトルを取得する boke_embedding = get_embedding( search_query, engine=embedding_model ) # データフレームの各埋め込みベクトルとのコサイン類似度を計算する df["similarity"] = df.embedding.apply(lambda x: cosine_similarity(x, boke_embedding)) # コサイン類似度の高い順に並べ替え、上位n件の結果を取得する results = ( df.sort_values("similarity", ascending=False) .head(n) .combined.str.replace("Title: ", "") .str.replace("; Content:", ": ") ) # マッチ度の高いボケのidからURLを生成して表示する if pprint: for r in results: print(f"https://bokete.jp/boke/{r.split(';')[0].split(':')[1].strip()}") return results # サーバAPIへのリクエストを受け付けて検索の場合 def main(request): signature = request.headers['X-Line-Signature'] body = request.get_data(as_text=True) try: handler.handle(body, signature) except InvalidSignatureError: abort(400) return 'OK' #LINEからのメッセージをうけて実行 @handler.add(MessageEvent, message=TextMessage) def handle_message(event): results = search_boke(df, event.message.text, n=5) #top5を出力 # ボケのリンク一覧を生成 links = [] for r in results: boke_id = r.split(";")[0].split(":")[1].strip() link = f"https://bokete.jp/boke/{boke_id}" links.append(link) # 配列を改行で区切られた文字列に変換 return_text = "お待たせしました!\n" return_text += "\n".join(links) return_text += "\n次の注文もどうぞ!" #LINEに返信されるメッセージ line_bot_api.reply_message( event.reply_token, TextSendMessage(text=return_text)) #デバッグ用 # サーバAPIへのリクエストを受け付けて検索の場合 """ def main(request): # パラメータqに検索ワードが入力されて、search_queryに代入 search_query = request.args.get("q", "デフォルト検索ワード") results = search_boke(df, search_query, n=5) #top5を出力 # リンクをHTML形式で作成 links = [] for r in results: boke_id = r.split(";")[0].split(":")[1].strip() link = f"https://bokete.jp/boke/{boke_id}" links.append(f'{link}') # リンクを一覧にして文字列に変換 links_html = "
".join(links) # HTMLレスポンスを作成 response = make_response(links_html) response.headers["Content-Type"] = "text/html" return response """ #デバッグ用 #ターミナルから引数を受け取って検索の場合 if __name__ == "__main__": search_query = sys.argv[1] if len(sys.argv) > 1 else "デフォルト検索ワード" results = search_boke(df, search_query, n=5) #top5を出力
コストを抑えたCloud Functionsなため、最初の実行でのメモリへのロードに1分程度時間がかかるが、一度ロードしたら2回目以降の回答は数秒で完了する。
(ある程度コストを掛けてちゃんとしたサーバを借りれば常時数秒でレスポンスできる)
OpenAIのAPIコストは比較的安いと感じた
GW中、何十万回、何十MB分のAPIを叩いたが、APIに送るデータを最小限の文字列にしたのもあってか、1日数百円程度しかかからなかった。
送る文字データが大きくなればなるほど課金が大きくなるらしいが、そもそも安く感じる
開発はますます簡単に!
一つ前の記事にも書いたが、そもそも今回ChatGPTを開発ツールとしても使うことで3倍速を実現している。
また最先端のAIアルゴリズムをAPI経由で活用することが出来るようになった、APIを使用するための労力も課金コストも思ったよりも少ない。
AI開発がこんなに簡単になったのだ。これを機にアイディアを実現すべく挑戦しよう!
あわせて読みたい
①今回開発したLINEアプリで笑いたいならこちら!
今回つくったボケてをAI検索するLINEアプリ 〜AIボケテンダー「エイジ」〜
②今回開発にChatGPTをフル活用することで、「開発が体感3倍速」になった。そこではどんな指示をGPTにしていたのか? その過程を詳細解説!
ChatGPTを使ったら開発が3倍速になった件&GPTへの生々しい指示一覧 #33