こんにちは、CCCMKホールディングスTECH LABの三浦です。
いつの間にか桜が散って、街の中で緑が目立つようになってきました。外に出るのが心地よい時期なので、ベランダでのんびり出来るようにしようとこの前の休みにベランダの掃除をしました。半日くらいかけてきれいにし、なんだか部屋に入ってくる空気もきれいになったように感じました。ベランダの掃除はついつい後回しにしがちだったのですが、家全体の雰囲気が良くなるので、もっと定期的にベランダの掃除をしようと思いました。
前回LangGraphを使ってPDFファイルとBing Searchを使って質問に回答してくれるAgentアプリケーションを作った話をご紹介しました。
これまではNotebookで開発をし動作を確認していたのですが、GUIを通じてチャット形式で使えるようにしたいな・・・と考えました。そこで以前少し使ってみたChainlitというPythonのライブラリを使ってこれまで作ってきたAgentアプリケーションを組み込んだチャットアプリケーションを作ってみました。
Chainlitの基本的な使い方は以前の記事でまとめています。
今回はさらにChainlitでパスワード認証機能や使用したToolの情報を表示させる機能を実現してみましたので、その内容についてまとめていこうと思います。
アプリケーションの画面
まずはアプリケーションの画面をどんな感じにしたのかをご紹介します。
WebブラウザでURLにアクセスすると、以下のようなEmail/パスワードによる認証画面が表示されます。
認証に成功すると、アプリケーションで参照するPDFファイルのアップロードが可能になります。
アップロードが完了すると、チャット形式で対話をすることが出来るようになります。
Toolが呼び出された場合は、以下の様にその内容を確認することが出来ます。
認証機能もToolの詳細の表示も、どちらもChainlitの機能を活用することで実現することが出来ます。
パスワード認証機能
まずアプリケーションを利用するためのユーザー/パスワードによる認証を実装します。参考にしたドキュメントはChainlitのドキュメントの"Authentication"関連の項目です。
Chainlitで認証機能を使用するためにはCHAINLIT_AUTH_SECRETという環境変数を設定する必要があります。この値はトークンの生成に使用されるそうですが、パスワードによる認証でも環境変数の設定は必須なようです。CHAINLIT_AUTH_SECRETに指定するべき値は、以下のコマンドを実行することで得ることが出来ます。
chainlit create-secret
環境変数に設定する方法以外に、.envというファイルにCHAINLIT_AUTH_SECRET=xxxxx...といった記述で設定させる方法もあります。.envファイルはchainlit runコマンドを実行するディレクトリと同一階層に配置する必要があります。
あとは登録済みのユーザー情報を格納しておきます。データベースに格納する方法などが考えられますが、今回はconfigファイルにユーザー情報をYAMLで記述しました。Chainlit側ではYAMLの構造に特に指定はないので、一旦以下のような構成にしました。パスワードについてはハッシュ化済みの文字列を指定します。
users:
- {
name: test_user,
email: test_user@test.test,
password: ハッシュ化したパスワードを記載,
}
パスワード文字列のハッシュ化処理はPythonで行いました。hashlibというライブラリを使用した以下のような処理です。
import hashlib password = "xxxx" # パスワードの平文 hashed_password = hashlib.sha256(password.encode()).hexdigest()
あとはChainlitの機能を利用して認証画面を呼び出す処理を簡単に書くことが出来ます。
import hashlib import os import chainlit as cl from pydantic import BaseModel, Field import yaml from yaml.loader import SafeLoader AUTH_FILE_PATH = "登録ユーザー情報を格納したconfigファイルのパス" class UserInfo(BaseModel): """ログイン認証に使用するユーザー情報のスキーマ""" name: str = Field( description="登録されたユーザー名" ) email: str = Field( description="登録されたemail" ) password: str = Field( description="登録された、ハッシュ化済みパスワード" ) @cl.password_auth_callback def auth_callback(email: str, password: str): """パスワード認証の際に呼ばれる処理 Args: - email(str): Email - password(str): パスワード平文 Return: - cl.User: 認証成功 - None: 認証失敗 """ # configファイルの存在確認と読み込み if os.path.isfile(AUTH_FILE_PATH): with open(AUTH_FILE_PATH) as file: config = yaml.load(file, Loader=SafeLoader) else: return None # 入力されたEmailアドレスと一致するユーザー情報をconfigから取得 user_list = config.get("users",[]) registered_user = [UserInfo(**user) for user in user_list if email == user["email"]] # 一致するユーザー情報がなければ失敗 if len(registered_user) == 0: return None # ユーザー情報の取得 registered_user = registered_user[0] registered_name = registered_user.name registered_email = registered_user.email registered_password = registered_user.password # 認証画面で入力されたパスワードのハッシュ化 hashed_password= hashlib.sha256(password.encode()).hexdigest() # Emailとパスワードの比較 if (email, hashed_password) == (registered_email, registered_password): # 認証成功 return cl.User( identifier=registered_name, metadata={"role": "user", "provider": "credentials"} ) else: # 認証失敗 return None
Toolの詳細表示
前回LangGraphで作成したAgentは、ユーザーの質問によって必要に応じてPDFからの情報抽出、Bing Searchの2つのToolを実行します。どのようなToolが呼ばれ、どのような結果が得られたのかを確認する機能は、どうしてAgentがそのような回答をしたのかを把握するために必要です。
LLMをエンジンに持つAgentに複雑なタスクを解かせるため、タスクをいくつかのステップに分解してシーケンシャルに実行させる手段を取ることがありますが、各ステップの情報を画面上に表示させる機能がChainlitに備わっています。それがStepです。
Stepとして表示させたい内容を文字列で返す関数に@cl.stepを付与し、Stepを表示させたい箇所でこの関数を実行することで表示させることが出来ます。
前回LangGraphで作成したアプリは、Toolが実行された時に実行したToolの情報を"action_input"というキーにlanggraphのToolInvocation型のオブジェクトで返し、実行結果を"action_result"というキーにlangchain_coreのFunctionMessage型のオブジェクトで返します。Toolの実行情報と実行結果で型が異なることからそれぞれ別の関数を作成しました。
import chainlit as cl from langchain_core.messages import HumanMessage, AIMessage, FunctionMessage from langgraph.prebuilt import ToolInvocation @cl.step(name="実行ツール") async def action_input_display(action_input: ToolInvocation): """Toolを呼び出してActionを実行した時の実行情報をStepに表示する Args: action_input(ToolInvocation) Return: display_message(Str) """ display_message = f""" tool: {action_input.tool} tool_input: {action_input.tool_input} """ return display_message @cl.step(name="ツール結果") async def action_result_display(action_result: FunctionMessage): """Toolを呼び出してActionを実行した時の実行結果をStepに表示する Args: action_result(FunctionMessage) Return: display_message(Str) """ display_message = f""" tool_result: {action_result.content} """ return display_message
PDFファイルの読み込み
ログインに成功すると、このアプリケーションでは最初に参照させるPDFファイルのアップロードを求めます。アップロードされたPDFファイルはChainlitが実行されている環境に保存されます。ファイルのアップロードメッセージの表示やファイルの受け取りはcl.AskFileMessageで実行され、戻り値としてアップロードされたファイルの情報を渡します。
前回作成したLangGraphのStateGraphを作成する処理はbuild_agent_graphという関数にまとめました。この関数は読み込むPDFファイルのパスを受け取り、langgraphのStateGraph型のオブジェクトを返します。
@cl.on_chat_start async def chat_start(): """ログイン成功後、最初に実行される処理。 PDFのアップロードを求め、アップロードされたPDFを保存し、それを参照するToolを持った StageGraph型のオブジェクトを生成する初期化処理を行う。 """ await cl.Avatar( name="assistant", path="アバターの画像ファイルパス(png)" ).send() files = None while files is None: files = await cl.AskFileMessage( content="PDFファイルをアップロードしてください。", accept=["application/pdf"], max_size_mb=20, timeout=180 ).send() file = files[0] agent_graph = build_agent_graph(file.path) cl.user_session.set("history",[]) # 会話の履歴を保存 cl.user_session.set("agent_graph",agent_graph) # StageGraphを保存 msg = cl.Message(content=f"ファイルを読み込みました。ファイル名: {file.name}",author="assistant") await msg.send()
メッセージ受け取り時の処理
ユーザーからメッセージを受け取り、StageGraphに処理を実行させ、その結果を画面に表示する処理です。
@cl.on_message async def send_message(message): """ ユーザーからメッセージを受け取り、受け取ったメッセージから応答メッセージをStageGraphを通じて生成し、 画面に表示する。 """ # 会話の履歴を取得 history = cl.user_session.get("history",[]) # StageGraphを取得 agent_graph = cl.user_session.get("agent_graph") history.append(HumanMessage(content=message.content)) response = agent_graph.invoke( { "messages":history } ) # 実行Toolの情報を取得 action_input = response.get("action_input","") if action_input: # "実行ツール"のStep表示 await action_input_display(action_input) action_result = response.get("action_result","") if action_result: # "ツール結果"のStep表示 await action_result_display(action_result) # StageGraphの直近生成メッセージの取得と表示処理 ai_message = response["messages"][-1].content history.append(AIMessage(content=ai_message)) await cl.Message(content=f"{ai_message}",author="assistant").send()
これらの処理をPythonのスクリプトにまとめてchainlit runコマンドを実行すると、アプリケーションを起動することが出来ます。
まとめ
今回は前回作成したLangGraphのアプリケーションにChainlitでGUIを追加した内容をまとめました。特にパスワードによる認証機能や実行したToolの詳細情報を関連情報として表示させることが出来るようになったことが今回の主な成果です。
Chainlitはシンプルな実装できれいなデザインのGUIを構築することが出来るのが良いなぁと思います！
