こんにちは、CCCMKホールディングス TECH LABの三浦です。
この前いつも乗り換えをするだけの駅で降りて駅の周りを歩いてみました。のんびり出来そうな公園や、色々なものが売っている商店街などがあって、こんないいところがあったんだと、なんだか得をした気持ちになりました。ゴールデンウィークも始まったので、他にもいつも通り過ぎるだけの駅で降りて、探索してみたいなと思います。
さて、今回MicrosoftのBot開発用のSDK"Bot Framework"のPythonのSDKを使ってAzure OpenAI ServiceのChatGPTと会話が出来るBotアプリケーションの作り方について調べて実際に動かしてみました。Bot Frameworkで作成したアプリケーションはAzure AI Bot Serviceで動かすことでMicrosoft Teamsからアクセス出来るようになるので活用できるシーンは多いと思います。
Bot Framework
Bot FrameworkはAzure AI Bot Serviceで動作するBotアプリケーションを開発するためのフレームワークでSDKが提供されています。実は以前Bot Frameworkを使ってBotアプリケーションを作ったことがあるのですが、その時にはBot Framework Composerというローコードのツールを使ってアプリケーションの開発をしていました。
Bot Framework ComposerでもBotアプリケーションの開発は出来るのですが、個人ではなく複数人でBotアプリケーションの開発に取り組む機会が最近出てきていて、そうなると普段慣れているプログラミング言語で開発する方が作業がはかどりそうだな・・・と感じるようになりました。
そこで今回はBot FrameworkのPython SDKを使ってAzure OpenAI ServiceのChatGPTとやり取りが出来るBotアプリケーションの作り方について調べた内容をまとめていきたいと思います。
Bot Frameworkを構成する要素
まずBot Framework SDKで開発されたBotアプリケーションのサンプルがGitHubに公開されていて、これを眺めつつ公式のドキュメントを見てBot Frameworkについて調べてみました。
- サンプル
- 公式のドキュメント
サンプルコードやドキュメントを読んでいて、Bot Frameworkを使う上でいくつか基本事項として抑えておいた方がいいな、と感じた構成要素があるので、まずそれらをまとめていきたいと思います。
また、ここで触れているBotアプリケーションとは、ユーザーとアプリケーションが交互にデータ(主にテキスト形式)をやり取りすることが出来るアプリケーションを指しています。
Activity
Botアプリケーションでユーザーとアプリケーションが交互にやり取りをするデータを、Bot FrameworkではActivityと呼びます。ユーザーがアプリケーションに入力したテキストデータ(Message activity)はもちろん、Teamsでユーザーが会話に参加した時にも参加したユーザーに関する情報がActivity(Conversation update activity)としてアプリケーションに送信されます。
Turn
Botアプリケーションはユーザーの入力に対してアプリケーション側で何らかの処理を行って応答を返します。ユーザーの入力に対するアプリケーションで行う処理をBot FrameworkではTurnと呼びます。Turnの中にはTurn Contextが含まれていて、Turn Contextの中にユーザーの入力した情報(Activity)が含まれています。
ユーザーとアプリケーションは複数回のTurnを繰り返すことも多いですが、基本的に前のTurnの状態は次のTurnには持ち越されません。しかし、たとえば前のTurnでユーザーの名前を聞いた後、次のTurnでそのユーザーの名前を使って会話をするなど、Turnの状態を保持できた方が良いことも多いです。その場合は後述するStateを使って実現することが出来ます。
またレストランの予約が出来るBotアプリケーションでは最初に人数を入力して予算を入力して場所を入力して・・・といったように、何度もTurnを重ねる必要もあります。この場合はDialogで実装すると良いそうですが、今回はDialogについては調べていません。
ActivityHandler
Turn Contextに含まれるActivityの内容を受け取り、それに対する処理を行い、アプリケーション側のActivityを作成し、ユーザーに送信する役割を担います。Botアプリケーションのコアの処理を担っていて、Python SDKではActivityHandlerというクラスのサブクラスを定義して処理を実装します。
Adapter
Azure AI Bot ServiceのBotはTeamsやLINEといった様々なアプリケーションからアクセスが出来ます。それらの接続先をBot FrameworkではChannelと呼んでいます。Channelから送られてきたActivityをActivityHandlerに渡す役割を担うのがAdapterです。AdapterはActivityを受け取り、処理を施してTurn Contextに含めてActivityHandlerに渡します。今回は調べられなかったのですが、その際に認証も行うことが出来るようです。
State
Turnの状態を次のTurnに持ち越す時はStateを使用します。Stateは会話におけるトピックや直近のやり取りを記憶したり、ユーザーの名前などを記憶するのに利用することが出来ます。これらの情報は様々なタイプのStorageに保存することが出来、In-Memory, Azure Blob Storage, Azure Cosmos DBが保存先として選択出来ます。
Bot FrameworkではStateにアクセスしてデータを取り出したりデータを保存するのにState property accessorsを使用します。State property accessorsを通じてStorageからアプリケーションのキャッシュにデータが保存され、必要なタイミングでデータにアクセスが出来るようになります。キャッシュは一時的な保存先なので、Turnの終了時にキャッシュからStorageに変更内容を書き込み、反映させる必要があります。
Azure OpenAI Serviceを利用するBotアプリケーション
以上のBot Frameworkの要素を踏まえ、Azure OpenAI Serviceを利用したBotアプリケーションをBot Framework Python SDKで作成してみました。
- Botアプリケーションの応答はgpt-35-turboで生成する
- 会話の履歴を保存出来るようにする。保存先のStorageはIn-Memoryにする
0から自力で作成するのはとても大変なのですが、基本的なBotアプリケーションは先ほど記載したリンク先にあるサンプルをベースに独自の処理を追加していくことで、比較的スムーズに開発を進めることが出来ました。
今回参考にしたサンプルアプリケーションは「02.echo-bot」「03.welcome-user」です。「02.echo-bot」ではBotアプリケーションの基本構成を、「03.welcome-user」ではStateの使い方を理解するのに参考になりました。
Bot Frameworkのサンプルのソースコードは最初、ファイルがいくつかに分かれていてどこから見たらいいんだろう、と少し戸惑いました。流れとしてはまずBotのコアのロジックであるActivityHandlerの実装(各サンプルフォルダのbotsフォルダに含まれている)を理解し、その後アプリケーションのメインとなるapp.pyの内容を見ていくのが良いのでは・・・と感じました。
ActivityHandlerの実装
Botアプリケーションのコアになる部分の実装です。まずAzure OpenAI Serviceに接続し、応答を生成する処理をazopai.pyというファイルに以下の様に書きました。
import os from typing import List, Dict from dotenv import load_dotenv from openai import AzureOpenAI def generate(messages: List[Dict[str,str]]): """Azure OpenAI ServiceのChatGPTを使用して応答を生成する Args: messages (List[Dict[str,str]]): "role"と"content"をキーに持つDictで表現されたメッセージList Return: ChatCompletion: openaiのレスポンス(ChatCompletion)。エラー時はNone """ client = AzureOpenAI( api_version="2023-12-01-preview", azure_endpoint=os.environ.get("AZURE_OPENAI_API_BASE") ) try: response = client.chat.completions.create( model="gpt-35-turbo", messages=messages ) return response except Exception as e: print(f"Error occurred: {e}") return None
それからStateの構造を定義するクラスをdata_models.pyというファイルに書きました。
from typing import Dict, List class ConversationData(): """会話の履歴を保持するState """ def __init__(self, messages: List[Dict[str, str]] = None): self.messages = messages
そしてBotのコアの処理をbot_logic.pyというファイルに以下の様に書きました。
from botbuilder.core import ( ActivityHandler, ConversationState, MessageFactory, TurnContext ) from azopai import generate from data_models import ConversationData class ChatBotWithState(ActivityHandler): def __init__(self, conversation_state: ConversationState): if conversation_state is None: raise TypeError( "[ChatBotWithState]: Missing parameter. conversation_state is \ required but None was given" ) self.conversation_state = conversation_state self.conversation_data_accessor = conversation_state.create_property("ConversationData") async def on_message_activity(self, turn_context: TurnContext): """ユーザーからのMessage Activityに対する応答を生成する Args: turn_context (TurnContext): ユーザーからのMessage Activityを含むTurnContext """ conversation_data = await self.conversation_data_accessor.get( turn_context, ConversationData ) messages = [] if conversation_data.messages is None: # Stateに会話の履歴が未登録の場合 messages = [ { "role": "user", "content": turn_context.activity.text } ] else: # Stateから会話の履歴が取れる場合は取得する messages = conversation_data.messages messages.append( { "role": "user", "content": turn_context.activity.text } ) # 応答の生成 response = generate(messages) if response: generate_text = response.choices[0].message.content messages.append( { "role": "assistant", "content": generate_text } ) # キャッシュ上のStateに変更を行う conversation_data.messages = messages return await turn_context.send_activity( MessageFactory.text(generate_text) ) async def on_turn(self, turn_context: TurnContext): await super().on_turn(turn_context) # 必須 # Turnに対する処理が完了後、StateをStorageに書き込む await self.conversation_state.save_changes(turn_context)
メイン処理の実装
アプリケーションのメインの処理を実装します。ここからはほぼサンプルの内容をそのまま使用しています。
まずアプリケーションの設定をConfigクラスで定義します。config.pyというファイルに以下の様に書きました。
import os class DefaultConfig: """ BOTのConfigを設定するClass""" PORT = 3978 APP_ID = os.environ.get("MicrosoftAppId", "") APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "") APP_TYPE = os.environ.get("MicrosoftAppType", "MultiTenant") APP_TENANTID = os.environ.get("MicrosoftAppTenantId", "")
そしてメインの処理をapp.pyに書きます。ActivityHandlerの箇所やAzure OpenAI Serviceへの接続情報の読み込み部分で手を加えていますが、それ以外はサンプルのコードの内容を使用しました。
import sys import traceback from datetime import datetime from http import HTTPStatus from aiohttp import web from aiohttp.web import Request, Response, json_response from botbuilder.core import ( ConversationState, MemoryStorage, TurnContext, UserState ) from botbuilder.core.integration import aiohttp_error_middleware from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication from botbuilder.schema import Activity, ActivityTypes from dotenv import load_dotenv from bot_logic import ChatBotWithState from config import DefaultConfig # Azure OpenAI Serviceへの接続情報を環境変数に読み込む load_dotenv() CONFIG = DefaultConfig() ADAPTER = CloudAdapter(ConfigurationBotFrameworkAuthentication(CONFIG)) MEMORY = MemoryStorage() CONVERSATION_STATE = ConversationState(MEMORY) # Botロジックを担うActivityHandler BOT = ChatBotWithState(CONVERSATION_STATE) async def on_error(context: TurnContext, error: Exception): # This check writes out errors to console log .vs. app insights. # NOTE: In production environment, you should consider logging this to Azure # application insights. print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr) traceback.print_exc() # Send a message to the user await context.send_activity("The bot encountered an error or bug.") await context.send_activity( "To continue to run this bot, please fix the bot source code." ) # Send a trace activity if we're talking to the Bot Framework Emulator if context.activity.channel_id == "emulator": # Create a trace activity that contains the error object trace_activity = Activity( label="TurnError", name="on_turn_error Trace", timestamp=datetime.utcnow(), type=ActivityTypes.trace, value=f"{error}", value_type="https://www.botframework.com/schemas/error", ) # Send a trace activity, which will be displayed in Bot Framework Emulator await context.send_activity(trace_activity) ADAPTER.on_turn_error = on_error async def messages(req: Request) -> Response: if "application/json" in req.headers["Content-Type"]: body = await req.json() else: return Response(status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE) activity = Activity().deserialize(body) auth_header = req.headers["Authorization"] if "Authorization" in req.headers else "" response = await ADAPTER.process_activity(auth_header, activity, BOT.on_turn) if response: return json_response(data=response.body, status=response.status) return Response(status=HTTPStatus.OK) APP = web.Application(middlewares=[aiohttp_error_middleware]) APP.router.add_post("/api/messages", messages) if __name__ == "__main__": try: web.run_app(APP, host="localhost", port=CONFIG.PORT) except Exception as error: raise error
Bot Framework Emulatorでのテスト
作成したアプリケーションは以下の手順でテストをすることが出来ます。
まずメイン処理が書かれたapp.pyを実行します。
python app.py
コンソールに以下のようなメッセージが表示されるとアプリケーションが起動していることが確認出来ます。
======== Running on http://localhost:3978 ======== (Press CTRL+C to quit)
このアプリケーションをテストするためにBotFramework Emulatorというデスクトップアプリを使用します。アプリケーションのインストール方法はこちらから確認することが出来ます。
BotFramework Emulatorを起動したら、"Open Bot"というボタンをクリックし、"Bot URL"にコンソールに表示されているアドレス＋"/api/messages"を入力するとBotアプリケーションにアクセスしテストをすることが出来ます。
まとめ
今回はBot Framework Python SDKを使ってAzure OpenAI Serviceを利用するBotアプリケーションを作ってみた話をまとめてみました。Bot Frameworkは最初覚えることが多く大変だったのですが、サンプルのコードを見たり、公式のドキュメントを読んだりして少しずつ理解することが出来てきたように思います。
今回は触れなかったのですが、テキストだけでなく、ファイルもやり取りすることが出来るので、ファイルの内容について答えてくれるBotアプリケーションもBot Frameworkで構築出来ると思います。今後Teamsから利用出来るようにしていきたいと考えています。
