オープンソースの会話型AI開発フレームワークChainlitを触ってみた

git clone https://github.com/yuu551/sample-cdk-chainlit

# プロジェクトフォルダの階層へ移動
cd sample-cdk-chainlit

# 各種ライブラリインストール
npm ci

# デプロイ実施
cdk deploy

✨  Deployment time: 179.7s

Outputs:
DevSshEc2Stack.InstanceId = <your-instance-id>
Stack ARN:
arn:aws:cloudformation:ap-northeast-1:<your-account-id>:stack/DevSshEc2Stack/xxx

✨  Total time: 186.5s

# システムパッケージの更新
sudo yum update -y

# Python関連パッケージのインストール
sudo yum install -y python3-pip python3-devel

# chainlit用のプロジェクトフォルダを作成
mkdir chainlit-dev
cd chainlit-dev

# 仮想環境の作成
python3 -m venv .venv
source .venv/bin/activate

# 必要なPythonパッケージのインストール
pip install chainlit boto3 langchain langchain_aws

# app.py
import chainlit as cl

@cl.on_chat_start
async def start():
    # チャット開始時の処理
    await cl.Message(
        content="こんにちは！私はシンプルなチャットボットです。メッセージを送ってください。"
    ).send()

@cl.on_message
async def main(message: cl.Message):
    # 固定の応答を返す
    response = "ご質問ありがとうございます。これは固定の応答です。"
    await cl.Message(content=response).send()

# アプリケーションの起動
chainlit run app.py -w

[features.spontaneous_file_upload]
		# falseに変更
    enabled = false
    # Define accepted file types using MIME types
    # Examples:
    # 1. For specific file types:
    #    accept = ["image/jpeg", "image/png", "application/pdf"]
    # 2. For all files of certain type:
    #    accept = ["image/*", "audio/*", "video/*"]
    # 3. For specific file extensions:
    #    accept = { "application/octet-stream" = [".xyz", ".pdb"] }
    # Note: Using "*/*" is not recommended as it may cause browser warnings
    accept = ["*/*"]
    max_files = 20
    max_size_mb = 500

import chainlit as cl
from langchain_aws import ChatBedrock
from langchain.schema.runnable import RunnableConfig
from langchain.schema import StrOutputParser

@cl.on_chat_start
async def start():
    try:
        llm = ChatBedrock(
            model_id="anthropic.claude-3-haiku-20240307-v1:0",
            model_kwargs={"temperature": 0.7},
        )

        # StrOutputParserを使用してチェーンを作成
        chain = llm | StrOutputParser()
        cl.user_session.set("chain", chain)

        await cl.Message(
            content="こんにちは！Amazon Bedrockと接続しました。メッセージを送ってください。"
        ).send()
    except Exception as e:
        await cl.Message(
            content=f"Bedrockとの接続に失敗しました: {str(e)}"
        ).send()

@cl.on_message
async def main(message: cl.Message):
    chain = cl.user_session.get("chain")
    response = cl.Message(content="")

    try:
        async for chunk in chain.astream(
            message.content,
            config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
        ):
            await response.stream_token(chunk)

        await response.send()
    except Exception as e:
        await cl.Message(
            content=f"エラーが発生しました: {str(e)}"
        ).send()

AWS_DEFAULT_REGION=ap-northeast-1
AWS_REGION=ap-northeast-1

import re
import boto3
import chainlit as cl
from chainlit.input_widget import Select, Slider
from langchain_aws import ChatBedrock
from langchain.schema.runnable import RunnableConfig
from langchain.schema import StrOutputParser

PATTERN = re.compile(r'v\d+(?!.*\d[kK]$)')

@cl.on_chat_start
async def start():
    try:
        # 利用可能なモデルの取得
        bedrock = boto3.client("bedrock", region_name="ap-northeast-1")
        response = bedrock.list_foundation_models(byOutputModality="TEXT")

        model_ids = [
            item['modelId']
            for item in response["modelSummaries"]
            if PATTERN.search(item['modelId'])
        ]

        # チャット設定の追加
        settings = await cl.ChatSettings([
            Select(
                id="Model",
                label="Amazon Bedrock - Model",
                values=model_ids,
                initial_index=model_ids.index("anthropic.claude-3-haiku-20240307-v1:0"),
            ),
            Slider(
                id="Temperature",
                label="Temperature",
                initial=0.7,
                min=0,
                max=1,
                step=0.1,
            ),
        ]).send()

        await setup_chain(settings)

        await cl.Message(
            content="モデルを選択して、チャットを開始してください。"
        ).send()

    except Exception as e:
        await cl.Message(
            content=f"設定の初期化に失敗しました: {str(e)}"
        ).send()

@cl.on_settings_update
async def setup_chain(settings):
    try:
        llm = ChatBedrock(
            model_id=settings["Model"],
            model_kwargs={"temperature": settings["Temperature"]}
        )
        chain = llm | StrOutputParser()
        cl.user_session.set("chain", chain)
    except Exception as e:
        await cl.Message(
            content=f"モデルの設定に失敗しました: {str(e)}"
        ).send()

@cl.on_message
async def main(message: cl.Message):
    chain = cl.user_session.get("chain")
    response = cl.Message(content="")

    try:
        async for chunk in chain.astream(
            message.content,
            config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
        ):
            await response.stream_token(chunk)

        await response.send()
    except Exception as e:
        await cl.Message(
            content=f"エラーが発生しました: {str(e)}"
        ).send()

import re
import boto3
import chainlit as cl
from chainlit.input_widget import Select, Slider
from langchain_aws import ChatBedrock
from langchain.schema.runnable import RunnableConfig
from langchain.schema import StrOutputParser
from langchain.prompts import ChatPromptTemplate

PATTERN = re.compile(r'v\d+(?!.*\d[kK]$)')

＃ システムプロンプトを定義
DEFAULT_SYSTEM_PROMPT = """あなたは親切で丁寧なアシスタントです。
ユーザーの質問に対して、簡潔で分かりやすい回答を心がけてください。
専門用語を使用する場合は、必要に応じて説明を加えてください。"""

@cl.on_chat_start
async def start():
    try:
        # 利用可能なモデルの取得
        bedrock = boto3.client("bedrock", region_name="ap-northeast-1")
        response = bedrock.list_foundation_models(byOutputModality="TEXT")

        model_ids = [
            item['modelId']
            for item in response["modelSummaries"]
            if PATTERN.search(item['modelId'])
        ]

        # チャット設定の追加
        settings = await cl.ChatSettings([
            Select(
                id="Model",
                label="Amazon Bedrock - Model",
                values=model_ids,
                initial_index=model_ids.index("anthropic.claude-3-haiku-20240307-v1:0"),
            ),
            Slider(
                id="Temperature",
                label="Temperature",
                initial=0.7,
                min=0,
                max=1,
                step=0.1,
            )
        ]).send()

        await setup_chain(settings)

        await cl.Message(
            content="モデルを選択して、チャットを開始してください。"
        ).send()

    except Exception as e:
        await cl.Message(
            content=f"設定の初期化に失敗しました: {str(e)}"
        ).send()

@cl.on_settings_update
async def setup_chain(settings):
    try:
        llm = ChatBedrock(
            model_id=settings["Model"],
            model_kwargs={"temperature": settings["Temperature"]}
        )

        # プロンプトテンプレートの作成
        prompt = ChatPromptTemplate.from_messages([
            ("system", DEFAULT_SYSTEM_PROMPT),
            ("human", "{input}")
        ])

        # チェーンの作成
        chain = prompt | llm | StrOutputParser()
        cl.user_session.set("chain", chain)
    except Exception as e:
        await cl.Message(
            content=f"モデルの設定に失敗しました: {str(e)}"
        ).send()

@cl.on_message
async def main(message: cl.Message):
    chain = cl.user_session.get("chain")
    response = cl.Message(content="")

    try:
        async for chunk in chain.astream(
            {"input": message.content},  # プロンプトテンプレートに合わせて入力形式を変更
            config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
        ):
            await response.stream_token(chunk)

        await response.send()
    except Exception as e:
        await cl.Message(
            content=f"エラーが発生しました: {str(e)}"
        ).send()

import re
import boto3
import chainlit as cl
from chainlit.input_widget import Select, Slider
from langchain_aws import ChatBedrock
from langchain.schema.runnable import RunnableConfig
from langchain.schema import StrOutputParser
from langchain.prompts import ChatPromptTemplate
from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage, AIMessage

PATTERN = re.compile(r'v\d+(?!.*\d[kK]$)')

DEFAULT_SYSTEM_PROMPT = """"""

@cl.on_chat_start
async def start():
    try:
        # 利用可能なモデルの取得
        bedrock = boto3.client("bedrock", region_name="ap-northeast-1")
        response = bedrock.list_foundation_models(byOutputModality="TEXT")

        model_ids = [
            item['modelId']
            for item in response["modelSummaries"]
            if PATTERN.search(item['modelId'])
        ]

        # チャット設定の追加
        settings = await cl.ChatSettings([
            Select(
                id="Model",
                label="Amazon Bedrock - Model",
                values=model_ids,
                initial_index=model_ids.index("anthropic.claude-3-haiku-20240307-v1:0"),
            ),
            Slider(
                id="Temperature",
                label="Temperature",
                initial=0.7,
                min=0,
                max=1,
                step=0.1,
            )
        ]).send()

        # 会話履歴の初期化
        cl.user_session.set("message_history", [])

        await setup_chain(settings)

        await cl.Message(
            content="モデルを選択して、チャットを開始してください。"
        ).send()

    except Exception as e:
        await cl.Message(
            content=f"設定の初期化に失敗しました: {str(e)}"
        ).send()

@cl.on_settings_update
async def setup_chain(settings):
    try:
        llm = ChatBedrock(
            model_id=settings["Model"],
            model_kwargs={"temperature": settings["Temperature"]}
        )

        # プロンプトテンプレートの作成（会話履歴を含める）
        prompt = ChatPromptTemplate.from_messages([
            ("system", DEFAULT_SYSTEM_PROMPT),
            ("human", "これまでの会話内容:\n{chat_history}\n\n現在の質問:\n{input}")
        ])

        # チェーンの作成
        chain = prompt | llm | StrOutputParser()
        cl.user_session.set("chain", chain)
    except Exception as e:
        await cl.Message(
            content=f"モデルの設定に失敗しました: {str(e)}"
        ).send()

@cl.on_message
async def main(message: cl.Message):
    chain = cl.user_session.get("chain")
    message_history = cl.user_session.get("message_history")
    response = cl.Message(content="")

    try:
        # 会話履歴の文字列を作成
        chat_history = "\n".join([
            f"Human: {msg['human']}\nAssistant: {msg['ai']}"
            for msg in message_history
        ])

        async for chunk in chain.astream(
            {
                "input": message.content,
                "chat_history": chat_history
            },
            config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
        ):
            await response.stream_token(chunk)

        # 会話履歴に追加
        message_history.append({
            "human": message.content,
            "ai": response.content
        })
        cl.user_session.set("message_history", message_history)

        await response.send()
    except Exception as e:
        await cl.Message(
            content=f"エラーが発生しました: {str(e)}"
        ).send()

# database.py
import json
from decimal import Decimal
from chainlit.data.dynamodb import DynamoDBDataLayer

class DecimalDynamoDBWrapper:
    """
    ChainlitのDynamoDBデータレイヤーをラップし、Decimal型の変換を処理するクラス

    DynamoDBはPythonのfloat型を直接サポートしていないため、
    Decimal型への変換が必要です。このラッパーは自動的にその変換を行います。
    """
    def __init__(self, data_layer: DynamoDBDataLayer):
        self.data_layer = data_layer
        self._wrap_serialize_item()
        self._wrap_deserialize_item()

    def _wrap_serialize_item(self):
        """
        DynamoDBに保存する前にfloat型をDecimal型に変換
        """
        original_serialize = self.data_layer._serialize_item

        def convert_floats_to_decimal(obj):
            if isinstance(obj, dict):
                return {key: convert_floats_to_decimal(value) for key, value in obj.items()}
            elif isinstance(obj, list):
                return [convert_floats_to_decimal(item) for item in obj]
            elif isinstance(obj, float):
                return Decimal(str(obj))
            return obj

        def wrapped_serialize_item(item):
            converted_item = convert_floats_to_decimal(item)
            return original_serialize(converted_item)

        self.data_layer._serialize_item = wrapped_serialize_item

import chainlit.data as cl_data
from chainlit.data.dynamodb import DynamoDBDataLayer
from chainlit.data.storage_clients.s3 import S3StorageClient
from database import DecimalDynamoDBWrapper

# app.pyでの初期化
storage_client = S3StorageClient(bucket="chainlit-storage-[ACCOUNT_ID]-[REGION]")
data_layer = DynamoDBDataLayer(table_name="ChainlitData", storage_provider=storage_client)

# ラッパーでデータレイヤーを拡張
wrapped_data_layer = DecimalDynamoDBWrapper(data_layer)

# Chainlitのデータレイヤーを置き換え
cl_data._data_layer = wrapped_data_layer.data_layer

import hashlib
import boto3
import chainlit as cl
import os
from dotenv import load_dotenv
import datetime
import uuid

# .envファイルから環境変数を読み込み
load_dotenv()

class UserAuth:
    def __init__(self, chainlit_table_name, auth_table_name):
        self.dynamodb = boto3.resource('dynamodb')
        self.chainlit_table = self.dynamodb.Table(chainlit_table_name)
        self.auth_table = self.dynamodb.Table(auth_table_name)
        self._ensure_admin_exists()

    def hash_password(self, password: str) -> str:
        return hashlib.sha256(password.encode()).hexdigest()

    def _ensure_admin_exists(self):
        """管理者ユーザーが存在しない場合は作成する"""
        try:
            admin_username = os.getenv('ADMIN_USERNAME')
            admin_password = os.getenv('ADMIN_PASSWORD')

            if not admin_username or not admin_password:
                print("Warning: Admin credentials not found in environment variables")
                return

            # 管理者ユーザーの存在確認
            admin_auth = self.get_user_auth(admin_username)

            if not admin_auth:
                # 管理者ユーザーが存在しない場合は作成
                try:
                    self.create_user(admin_username, admin_password, role="admin")
                    print(f"Admin user '{admin_username}' created successfully")
                except Exception as e:
                    print(f"Error creating admin user: {str(e)}")
            else:
                print(f"Admin user '{admin_username}' already exists")

        except Exception as e:
            print(f"Error in _ensure_admin_exists: {str(e)}")

    def get_user_auth(self, username: str):
        """認証情報の取得"""
        response = self.auth_table.get_item(
            Key={
                'username': username
            }
        )
        return response.get('Item')

    def create_user(self, username: str, password: str, role: str = "user"):
        """ユーザーを作成する"""
        if role != "admin":
            raise PermissionError("Only admin users can be created")

        hashed_password = self.hash_password(password)

        # 認証情報をAuthテーブルに保存
        auth_item = {
            'username': username,
            'password': hashed_password,
            'role': role,
            'createdAt': datetime.datetime.now(datetime.timezone.utc).isoformat()
        }

        self.auth_table.put_item(Item=auth_item)

        # ChainlitDataテーブルへの書き込み
        self.chainlit_table.put_item(
            Item={
                'PK': f"USER#{username}",
                'SK': "USER",
                'id': str(uuid.uuid4()),
                'identifier': username,
                'createdAt': datetime.datetime.now(datetime.timezone.utc).isoformat(),
                'metadata': {}
            }
        )

    def verify_user(self, username: str, password: str):
        """ユーザー認証"""
        user_auth = self.get_user_auth(username)
        if not user_auth:
            return None

        hashed_password = self.hash_password(password)

        if user_auth['password'] == hashed_password:
            print("test")
            return cl.User(
                identifier=username,
                metadata={"role": user_auth['role'], "provider": "credentials"}
            )
        return None

ADMIN_USERNAME=XXX
ADMIN_PASSWORD=YYY

# app.py
import chainlit as cl
from auth import UserAuth

# 認証インスタンスの初期化
user_auth = UserAuth(
    chainlit_table_name="ChainlitData",
    auth_table_name="UserAuth"
)

@cl.password_auth_callback
async def auth_callback(username: str, password: str):
    """
    Chainlitの認証コールバック

    Args:
        username (str): ユーザー名
        password (str): パスワード

    Returns:
        User | None: 認証成功時はユーザーオブジェクト、失敗時はNone
    """
    user = user_auth.verify_user(username, password)

    if user:
        return await cl_data.get_data_layer().get_user(username)
    return None

chainlit create-secret

Copy the following secret into your .env file. Once it is set, changing it will logout all users with active sessions.
CHAINLIT_AUTH_SECRET="XXX"
WARNING: All log messages before absl::InitializeLog() is called are written to STDERR
E0000 00:00:1739249012.261197   15438 init.cc:232] grpc_wait_for_shutdown_with_timeout() timed out.

CHAINLIT_AUTH_SECRET="XXX"

@cl.on_message
async def main(message: cl.Message):
    chain = cl.user_session.get("chain")
    # 会話履歴のセッション情報を取得
    message_history = cl.user_session.get("message_history")
    response = cl.Message(content="")

    try:
        chat_history = "\n".join([
            f"Human: {msg['human']}\nAssistant: {msg['ai']}"
            for msg in message_history
        ])

        async for chunk in chain.astream(
            {
                "input": message.content,
                "chat_history": chat_history
            },
            config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
        ):
            await response.stream_token(chunk)

        # 会話履歴に追加
        message_history.append({
            "human": message.content,
            "ai": response.content
        })
        # 会話履歴をセッションに追加。
        cl.user_session.set("message_history", message_history)

        await response.send()

@cl.on_chat_resume
async def on_chat_resume(thread):
    # 会話履歴の初期化
    message_history = []

    # スレッドのメタデータから会話履歴を復元
    if thread.get("metadata") and thread["metadata"].get("message_history"):
        message_history = thread["metadata"]["message_history"]

    # セッションに会話履歴を設定
    cl.user_session.set("message_history", message_history)

    # チャット設定を復元
    settings = cl.user_session.get("chat_settings")
    if settings:
        await setup_chain(settings)

{
  "PK": {
    "S": "THREAD#48dc21cf-e78f-436e-8e48-5a72132a7db9"
  },
  "SK": {
    "S": "STEP#d0f7fe7c-ec94-49d9-90f3-29302c1c858e"
  },
  "createdAt": {
    "S": "2025-02-11T08:01:08.620675Z"
  },
  "end": {
    "S": "2025-02-11T08:01:11.385050Z"
  },
  "feedback": {
    "M": {
      "comment": {
        "S": "最高のアドバイスです！！"
      },
      "forId": {
        "S": "d0f7fe7c-ec94-49d9-90f3-29302c1c858e"
      },
      "id": {
        "S": "THREAD#48dc21cf-e78f-436e-8e48-5a72132a7db9::STEP#d0f7fe7c-ec94-49d9-90f3-29302c1c858e"
      },
      "threadId": {
        "S": "48dc21cf-e78f-436e-8e48-5a72132a7db9"
      },
      "value": {
        "N": "1"
      }
    }
  },
  "generation": {
    "NULL": true
  },
  "id": {
    "S": "d0f7fe7c-ec94-49d9-90f3-29302c1c858e"
  },
  "input": {
    "S": "手作りすると本当に美味しくなるのは何が美味しくなるんでしたっけ？\n質問を忘れまして・・・"
  },
  "isError": {
    "BOOL": false
  },
  "language": {
    "NULL": true
  },
  "metadata": {
    "M": {}
  },
  "name": {
    "S": "on_message"
  },
  "output": {
    "S": ""
  },
  "parentId": {
    "S": "59293aa3-bcde-4975-af34-5ddba60919fa"
  },
  "showInput": {
    "S": "json"
  },
  "start": {
    "S": "2025-02-11T08:01:08.620694Z"
  },
  "streaming": {
    "BOOL": false
  },
  "tags": {
    "NULL": true
  },
  "threadId": {
    "S": "48dc21cf-e78f-436e-8e48-5a72132a7db9"
  },
  "type": {
    "S": "run"
  }
}

#!/bin/bash

# 引数チェック
if [ "$#" -ne 5 ]; then
    echo "Usage: $0 <username> <password> <role> <chainlit_table> <auth_table>"
    exit 1
fi

USERNAME=$1
PASSWORD=$2
ROLE=$3
CHAINLIT_TABLE=$4
AUTH_TABLE=$5

# 現在時刻（ISO 8601形式）
TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
# UUID生成
UUID=$(uuidgen)
# パスワードのSHA256ハッシュ化
HASHED_PASSWORD=$(echo -n "$PASSWORD" | sha256sum | awk '{print $1}')

# Authテーブルにユーザー認証情報を登録
aws dynamodb put-item \\
    --table-name $AUTH_TABLE \\
    --item '{
        "username": {"S": "'"$USERNAME"'"},
        "password": {"S": "'"$HASHED_PASSWORD"'"},
        "role": {"S": "'"$ROLE"'"},
        "createdAt": {"S": "'"$TIMESTAMP"'"}
    }'

# ChainlitDataテーブルにユーザー情報を登録
aws dynamodb put-item \\
    --table-name $CHAINLIT_TABLE \\
    --item '{
        "PK": {"S": "USER#'"$USERNAME"'"},
        "SK": {"S": "USER"},
        "id": {"S": "'"$UUID"'"},
        "identifier": {"S": "'"$USERNAME"'"},
        "createdAt": {"S": "'"$TIMESTAMP"'"},
        "metadata": {"M": {}}
    }'

echo "User $USERNAME created successfully"

sh ./create_user.sh <username> <password> <role> <chainlit_table> <auth_table>

cdk destroy