こんにちは、CCCMKホールディングスTECH LABの三浦です。
あっという間に11月も終わりですね！クリスマスまであと1か月なので、ブログのサムネイルもクリスマス仕様になりました！
- はじめに
- Chainlitのデータ永続化の仕組み
- 今回試した構成
- プロジェクトの構成
- docker-compose.yml
- PostgreSQLの設定
- Chainlitアプリの設定
- 使ってみて
- まとめ
はじめに
LLMを使ったチャット形式のアプリのフロントの実装に、私はよく"Chainlit"というPythonのライブラリを利用しています。スッキリとしていながら必要な機能がそろっているデザインがいいなと思っています。
さて、チャットアプリを使っていると、会話の記録を残しておきたいと思うことがあります。デフォルトではオフになっていますが、Chainlitにはデータを永続化する機能が実装されています。
便利な機能だな、と思ったので自分で試してみることにしました。
Chainlitのデータ永続化の仕組み
Chainlitでデータを永続化する場合は"Literal AI"というChainlitが運営する監視/評価プラットフォームにデータを送るか、自分でDBを用意してChainlitのCustom Data Layerを通じてDBにデータを送るかの2つの選択肢を選ぶことが出来ます。ChainlitのCustom Data Layerは"SQLAlchemy"というオブジェクト指向言語でRDBを操作できるライブラリで実装されています。さらにChainlitではアップロードされたファイルを"Element"として取り扱いますが、これらをData Layerを通じて"Azure Blob Storage"や"AWS S3"といったクラウドストレージに格納することも出来ます。
今回試した構成
いったんクラウドストレージは使わず、DBに会話の記録を残し、前回の会話を再開できるかどうかを今回試しました。使用するSQLデータベースは、最初Pythonに最初から搭載されているSQLiteを試していたのですが、途中でどうやっても解消できないエラーが発生しました。
エラーメッセージは"AttributeError: 'str' object has no attribute 'copy'"といった内容のもので、前回の会話を再開(on_chat_resume)しようとすると発生していることが分かりました。この現象に関するドキュメントが見つからなかったので憶測になりますが、恐らく私が使っていたPython 3.12に搭載されているSQLiteでは、json形式のデータをPythonの辞書型ではなく文字列型としてテーブルから抜き出していて、そのデータに対してcopyを呼び出そうとしてこのエラーが発生しているように考えらえます。
ためしにSQL DatabaseをSQLiteからPostgreSQLに変えてみたところ、正常に動作するようになりました。なので今回は正常に動かすことが出来たPostgreSQLを使った構成をご紹介します。
プロジェクトの構成
ChainlitもPostgreSQLも、Dockerのコンテナで動かす構成を取ってみました。プロジェクトの中の構成は次の通りです。
chainlit-postgres/
├── app.env
├── chainlit-app/
│ ├── Dockerfile
│ ├── requirements.txt
│ └── src/
│ ├── add-initial-user.py
│ ├── app.py
│ └── start.sh
├── db.env
├── docker-compose.yml
└── postgres/
├── data/
└── init/
└── init.sql
docker-compose.yml
コンテナを起動する、docker-compose.ymlを次のように書きました。chainlitのホスト側のポートやdockerのnetworkのip設定などは、必要に応じた値に設定してください。
services:
app:
container_name: chainlit-app
build:
context: "./chainlit-app"
ports:
- "your_port":8000
networks:
- chainlit-app-network
env_file: ./app.env
depends_on:
- db
db:
container_name: chainlit-db
image: postgres:17.2-alpine3.20
networks:
- chainlit-app-network
ports:
- 5432:5432
env_file: ./db.env
volumes:
- ./postgres/data:/var/lib/postgresql/data
- ./postgres/init:/docker-entrypoint-initdb.d
networks:
chainlit-app-network:
ipam:
config:
- subnet: "your_ip_range"
PostgreSQLの設定
PostgreSQLのイメージは公式のものを使用しました。また、環境変数POSTGRES_PASSWORDの設定が必要ですが、こちらは"db.env"というファイルで定義しています。volumesでは2つの項目を指定しています。
- ./postgres/data: DBファイルの格納先
- ./postgres/init: DB起動時の初期設定SQLファイルの格納先
コンテナ内の"/docker-entrypoint-initdb.d"に置かれたSQLファイルは、起動時に自動的に実行してくれるようです。この中に格納した"init.sql"はChainlitのドキュメントを参考に、一部変更を加えました。
初期設定スクリプトの内容
こちらのChainlit公式ドキュメントの"SQL alchemy data layer"に、必要なスキーマを作成するSQLが掲載されています。
こちらを参考にしながら、CREATE DATABASEの実行と、ユーザーのログインパスワード格納用のテーブルloginsのCREATE TABLEの処理を追加した次のようなSQLファイルを作成しました。
CREATE DATABASE chainlit_db; \c chainlit_db; CREATE TABLE IF NOT EXISTS logins ( "id" UUID PRIMARY KEY, "createdAt" TEXT, "passwordHash" TEXT ); CREATE TABLE IF NOT EXISTS users ( "id" UUID PRIMARY KEY, "identifier" TEXT NOT NULL UNIQUE, "metadata" JSONB NOT NULL, "createdAt" TEXT ); CREATE TABLE IF NOT EXISTS threads ( "id" UUID PRIMARY KEY, "createdAt" TEXT, "name" TEXT, "userId" UUID, "userIdentifier" TEXT, "tags" TEXT[], "metadata" JSONB, FOREIGN KEY ("userId") REFERENCES users("id") ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS steps ( "id" UUID PRIMARY KEY, "name" TEXT NOT NULL, "type" TEXT NOT NULL, "threadId" UUID NOT NULL, "parentId" UUID, "streaming" BOOLEAN NOT NULL, "waitForAnswer" BOOLEAN, "isError" BOOLEAN, "metadata" JSONB, "tags" TEXT[], "input" TEXT, "output" TEXT, "createdAt" TEXT, "start" TEXT, "end" TEXT, "generation" JSONB, "showInput" TEXT, "language" TEXT, "indent" INT ); CREATE TABLE IF NOT EXISTS elements ( "id" UUID PRIMARY KEY, "threadId" UUID, "type" TEXT, "url" TEXT, "chainlitKey" TEXT, "name" TEXT NOT NULL, "display" TEXT, "objectKey" TEXT, "size" TEXT, "page" INT, "language" TEXT, "forId" UUID, "mime" TEXT ); CREATE TABLE IF NOT EXISTS feedbacks ( "id" UUID PRIMARY KEY, "forId" UUID NOT NULL, "threadId" UUID NOT NULL, "value" INT NOT NULL, "comment" TEXT );
Chainlitアプリの設定
requirements.txt
以下のライブラリを"requirements.txt"に記入しインストールしました。
aiohttp==3.11.7 asyncpg==0.30.0 chainlit==1.3.2 bcrypt==4.2.1 sqlalchemy[asyncio]==2.0.36 psycopg2-binary==2.9.10
app.env
Chainlitアプリ側で必要になる環境変数を指定します。特にChainlitでデータ永続化をするためには認証機能が必要になるため、そのための設定も含めています。
| 変数名 | 説明 |
|---|---|
| DB_HOST | PostgreSQLのサーバホスト名 |
| DB_NAME | PostgreSQLのデータベース名 |
| DB_USER | PostgreSQLのユーザー名 |
| DB_PASSWORD | PostgreSQLのパスワード |
| DB_HOST | PostgreSQLのホスト名 |
| CHAINLIT_USER | Chainlitアプリにログインするためのユーザー名 |
| CHAINLIT_PASSWORD | Chainlitアプリにログインするためのユーザー名 |
| CHAINLIT_AUTH_SECRET | Chainlitで認証機能をオンにするために必要。Chainlitインストール済みの環境でchainlit create-secretを実行し、事前に生成しておく |
Chainlitアプリのログインユーザを追加する(add-initial-user.py)
アプリ起動時に実行する、アプリログインユーザを追加する処理です。
import datetime import json import os import uuid import bcrypt import psycopg2 def register_user(username, password, dbhost, dbuser, dbpassword, dbname): """ChainlitにログインするためのユーザをDBに追加する""" conn = psycopg2.connect( f"host={dbhost} dbname={dbname} user={dbuser} password={dbpassword}" ) id = str(uuid.uuid4()) # ユーザIDの生成 identifier = username metadata = {"role":"admin"} # いったんロールは"admin"にする created_date = datetime.datetime.now().isoformat() # パスワードはハッシュ化してDBに保存する password_hash = bcrypt.hashpw( password.encode("utf-8"), bcrypt.gensalt() ).decode('utf8') with conn: # 最初に指定されたユーザ名が既に登録済みかどうかを判定する。 with conn.cursor() as cur: cur.execute( "SELECT id FROM users WHERE identifier=%s", (identifier, ) ) check_exist = cur.fetchall() if len(check_exist) > 0: # すでに登録済みの場合は終了 print("already exist.") return 1 # 登録されていない場合は追加する with conn.cursor() as cur: # パスワード用のテーブル cur.execute( 'INSERT INTO logins (id, "createdAt", "passwordHash") VALUES (%s, %s, %s)', (id, created_date, password_hash) ) with conn.cursor() as cur: # ユーザ情報のテーブル cur.execute( 'INSERT INTO users (id, identifier, metadata, "createdAt") VALUES (%s, %s, %s, %s)', (id, identifier, json.dumps(metadata), created_date) ) conn.commit() return 0 if __name__=="__main__": username = os.environ.get("CHAINLIT_USER") password = os.environ.get("CHAINLIT_PASSWORD") dbhost = os.environ.get("DB_HOST") dbuser = os.environ.get("DB_USER") dbpassword = os.environ.get("DB_PASSWORD") dbname = os.environ.get("DB_NAME") result = register_user(username, password, dbhost, dbuser, dbpassword, dbname) if result == 0: print("Success Add Initial User.") else: print("Initial User Already Exists.")
Chainlitアプリの定義(app.py)
Chainlitアプリの動作を定義するPythonファイルです。今回はLLMとは接続せずに、単純に入力にプリフィックスを付けて返すだけの機能にしています。
データ永続化を実現するうえで一番肝になるのがこちらの行です。
cl_data._data_layer = SQLAlchemyDataLayer(f"postgresql+asyncpg://{conn_str}")
これでChainlitのCustom Data Layerをセットすることで、Chainlit側で会話データなどをDBに記録してくれるようになります。
また、会話再開時に実行する処理は@cl.on_chat_resumeでデコレートされた関数に書きます。この中ではDBから取得した会話の記録が格納されたthreadから会話の履歴を取得し、cl.user_sessionの"chat_history"を復元する処理を実行しています。
ファイル全体は以下の通りです。
import os import asyncpg import bcrypt import chainlit as cl import chainlit.data as cl_data from chainlit.data.sql_alchemy import SQLAlchemyDataLayer from chainlit.types import ThreadDict import psycopg2 username = os.environ.get("CHAINLIT_USER") password = os.environ.get("CHAINLIT_PASSWORD") dbhost = os.environ.get("DB_HOST") dbuser = os.environ.get("DB_USER") dbpassword = os.environ.get("DB_PASSWORD") dbname = os.environ.get("DB_NAME") # ここでChainlitのCustom Data Layerを設定する。 conn_str = f"{dbuser}:{dbpassword}@{dbhost}:5432/{dbname}" cl_data._data_layer = SQLAlchemyDataLayer(f"postgresql+asyncpg://{conn_str}") def login_user(username, password): conn = psycopg2.connect( f"host={dbhost} dbname={dbname} user={dbuser} password={dbpassword}" ) with conn: with conn.cursor() as cur: cur.execute( """ SELECT logins."passwordHash" FROM logins INNER JOIN users ON logins.id=users.id WHERE users.identifier=%s """,(username,) ) pw = cur.fetchone() if not pw: return False if pw and bcrypt.checkpw(password.encode("utf-8"),pw[0].encode("utf-8")): return True else: return False @cl.password_auth_callback async def auth_callback(username: str, password: str): if login_user(username, password): return await cl_data.get_data_layer().get_user(username) else: return None @cl.on_chat_start async def on_chat_start(): cl.user_session.set("chat_history",[]) @cl.on_message async def on_message(message: cl.Message): chat_history = cl.user_session.get("chat_history") chat_history.append({"role": "user", "content": message.content}) return_message_content = f"Received: {message.content}" chat_history.append({"role": "assistant", "content": return_message_content}) await cl.Message( content=return_message_content, ).send() @cl.on_chat_resume async def on_chat_resume(thread: ThreadDict): cl.user_session.set("chat_history", []) chat_history = thread["metadata"]["chat_history"] for message in chat_history: cl.user_session.get("chat_history").append(message)
起動スクリプト(start.sh)
2つのPythonのファイルを実行するためのシェルスクリプトファイルを作成しました。"docker-compose.yml"でdepends_on:を指定したものの、DB側の初期化処理中にアプリ側でユーザ追加処理を実行し、エラーが出てしまうことがあったので、初期化処理を待ってからPythonを実行するようにしています。
until PGPASSWORD=$DB_PASSWORD psql -h "$DB_HOST" -U "$DB_USER" -c '\q'; do >&2 echo "DB起動完了まで待機中・・・" sleep 1 done cd $(dirname $0) python add-initial-user.py chainlit run app.py --host 0.0.0.0
使ってみて
docker compose up --buildで起動し、アプリにアクセス、アプリ側の環境変数にセットしたユーザ情報でログインします。
データ永続化をして特にいいな、と思ったのが、フィードバック機能が利用できるようになったことです。
今まで"いいね"ボタンあるんだーという感じで見ているだけだったのですが、データ永続化が有効になっていると、フィードバックのデータも自動的にDBのfeedbacksというテーブルに記録されていきます。これでLLMが生成したどの出力が良かったのか、もしくは悪かったのかをすぐに抽出することが出来るようになり、とても便利だと感じました。
たとえばデータベースに接続し、次のようなSQLを実行することで応答結果とそれに対するフィードバックを確認することが出来ます。
SELECT steps.output, feedbacks.value, feedbacks.comment FROM steps INNER JOIN feedbacks ON steps."parentId"=feedbacks."forId";
結果
これはすごく便利だと思いました。
まとめ
ということで、今回はChainlitのデータ永続化を、Custom Data Layerを使って有効化する手順について調べ、試してみるところまでやってみたことをご紹介しました。Chainlitアプリのユーザを増やす場合はどうしよう・・・といった課題もありますが、データ永続化を有効化することで、Chainlitの便利な機能を利用出来るようになることが分かって良かったです。
