高級言語「日本語」による生成AIアプリ開発

LLMを使って、ノーコードでアプリを書いてみた。

要件

以下三点。

• 録音ファイルを元に議事録起こしをするアプリを作成する。
• Python、Streamlit、LangChain、Amazon S3を使う。
• 人間は原則としてコードを書かず、生成AIとのチャットのみを通して開発する。

最終成果物

（以下略）

コード

コード本体。

import streamlit as st
import boto3
from botocore.exceptions import ClientError
from datetime import datetime
from langchain_aws import ChatBedrock
from langchain.globals import set_debug
from langchain.prompts import PromptTemplate
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate
)
from langchain.schema import HumanMessage
import requests
import time

# デバッグモードの設定
set_debug(True) # デバッグ時はTrueに変更

# Amazon Transcribe、S3クライアントの初期化
transcribe_client = boto3.client('transcribe')
s3_client = boto3.client('s3')
s3_bucket_name = 'hoge-fuga-20240512'
folder_name = 'Audio'  # フォルダ名を指定

# システムプロンプトとユーザープロンプトの定義
system_message_template = """
### システムプロンプト

あなたは議事録のサマリーを作成するAIアシスタントです。

与えられた文字起こしの出力から、以下の点に留意して議事内容の要点をまとめてください。

- **見出し**: 見出しに議事録のタイトルを表示し、その下の行に全体を簡潔に要約した一行の文章を記載して下さい。
- **構造化**: マークダウンを使って、見出しや箇条書きなどでサマリーを構造化してください。
- **要約**: 議事内容の重要なポイントのみを簡潔に要約してください。
- **可読性**: 見やすく分かりやすい表現を心がけてください。
- **アクションアイテム**: 議事内容を踏まえて、アクションアイテム（およびその担当者、期限）を箇条書きで列挙して下さい。

"""
system_message_prompt = SystemMessagePromptTemplate.from_template(system_message_template)
human_message_prompt = PromptTemplate(input_variables=["transcript_text"], template="{transcript_text}")

# 拡張子からMediaFormatを取得する関数
def get_media_format(file_extension):
    media_formats = {
        '.mp3': 'mp3',
        '.mp4': 'mp4',
        '.flac': 'flac',
        '.wav': 'wav',
        '.m4a': 'm4a'
    }
    return media_formats.get(file_extension.lower(), 'wav')

def upload_file_to_s3(file, bucket_name, folder_name, object_key):
    try:
        full_object_key = f"{folder_name}/{object_key}"
        s3_client.upload_fileobj(file, bucket_name, full_object_key)
        return True
    except ClientError as e:
        print(f"S3へのアップロード中にエラーが発生しました: {e}")
        return False

def start_transcription_job(job_name, file_uri, file_format, language_code):
    try:
        transcribe_client.start_transcription_job(
            TranscriptionJobName=job_name,
            Media={'MediaFileUri': file_uri},
            MediaFormat=file_format,
            LanguageCode=language_code
        )
        return True
    except ClientError as e:
        print(f"Transcribe ジョブの開始中にエラーが発生しました: {e}")
        return False

def get_transcription_job_status(job_name):
    try:
        job = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)
        status = job['TranscriptionJob']['TranscriptionJobStatus']
        if status in ['COMPLETED', 'FAILED']:
            return status
        else:
            return 'IN_PROGRESS'
    except ClientError as e:
        print(f"Transcribe ジョブの状態取得中にエラーが発生しました: {e}")
        return None

def get_transcript_text(job_name):
    try:
        job = transcribe_client.get_transcription_job(TranscriptionJobName=job_name)
        if job['TranscriptionJob']['TranscriptionJobStatus'] == 'COMPLETED':
            transcript_file_uri = job['TranscriptionJob']['Transcript']['TranscriptFileUri']
            transcript_response = requests.get(transcript_file_uri)
            return transcript_response.text
        else:
            return None
    except ClientError as e:
        print(f"文字起こし出力の取得中にエラーが発生しました: {e}")
        return None

def generate_summary(transcript_text):
    try:
        human_message = HumanMessage(content=transcript_text)
        prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message])
        bedrock = ChatBedrock(
            model_id="anthropic.claude-3-sonnet-20240229-v1:0",
            region_name="us-east-1",
            verbose=True,
            model_kwargs={
                "anthropic_version": "bedrock-2023-05-31",
                "temperature": 0.1,
                "top_p": 0.9,
                "top_k": 50,
                "max_tokens": 1000,
            },
        )
        chain = prompt | bedrock 
        summary = chain.invoke({"transcript_text": transcript_text})
        return summary.content
    except Exception as e:
        print(f"議事録サマリーの生成中にエラーが発生しました: {e}")
        return None

# Streamlit Web UI
st.title("Meeting Minutes Generator")
st.write("powered by **Claude V3 Sonnet.**")
st.markdown("---")

# ファイルのアップロード
uploaded_file = st.file_uploader("音声ファイルをアップロードしてください", type=['mp3', 'mp4', 'flac', 'wav', 'm4a'])

if uploaded_file is not None:
    # S3にファイルをアップロード
    s3_object_key = uploaded_file.name
    if upload_file_to_s3(uploaded_file, s3_bucket_name, folder_name, s3_object_key):
        st.success(f"ファイル '{uploaded_file.name}' がS3にアップロードされました。")
    else:
        st.error("S3へのアップロードに失敗しました。")

    # Amazon Transcribeでファイルを文字起こし
    now = datetime.now()
    transcribe_job_name = f"transcribe-job-{now.strftime('%Y%m%d%H%M%S')}"
    file_extension = '.' + uploaded_file.name.split('.')[-1]
    media_format = get_media_format(file_extension)
    if start_transcription_job(transcribe_job_name, f"s3://{s3_bucket_name}/{folder_name}/{s3_object_key}", media_format, 'ja-JP'):
        st.success(f"Transcribe ジョブ '{transcribe_job_name}' が開始されました。")
    else:
        st.error("Transcribe ジョブの開始に失敗しました。")
    
    # Transcribeジョブの完了を待つ
    st.markdown("#### Transcribe ジョブの実行を待っています...")
    job_status = 'IN_PROGRESS'
    wait_time = 0
    max_wait_time = 600  # 最大待ち時間は10分(600秒)
    progress_bar = st.progress(0)

    while job_status == 'IN_PROGRESS':
        job_status = get_transcription_job_status(transcribe_job_name)
        if job_status is None:
            st.error("Transcribe ジョブの状態を取得できませんでした。")
            break
        wait_time += 5
        progress = min(wait_time / max_wait_time, 1.0)
        progress_bar.progress(progress)
        if wait_time >= max_wait_time:
            st.error("Transcribe ジョブが10分以上かかっています。中断します。")
            break
        time.sleep(5)

    if job_status == 'COMPLETED':
        transcript_text = get_transcript_text(transcribe_job_name)
        if transcript_text:
            st.success(f"文字起こしジョブ '{transcribe_job_name}' が成功し、出力の取得に成功しました。")
            st.markdown("#### 議事録生成中...")
            summary = generate_summary(transcript_text)
            if summary:
                st.info(f"議事録サマリー:\n{summary}")
            else:
                st.error("議事録サマリーの生成に失敗しました。")
        else:
            st.error("文字起こし出力の取得に失敗しました。")
    else:
        st.error(f"Transcribe ジョブが失敗しました。")

アプローチ

LLMに「日本語」による指示を出し、ベースとなるコードを作成した。
なおLLMにはClaude V3 Sonnetを使用した。
使用したプロンプトは以下。

あなたは生成AIアプリ開発者です。 AWSとLangChain、Pythonに習熟しています。 
モデルにはAmazon Bedrockを通じてClaude V3 Haikuを、UIにはStreamlitを使います。
この組み合わせで、以下のアプリを作成してください。

- m4a, mp3, mp4, wav, flacのいずれかの形式のファイルをWeb UIでアップロードできる。
- アップロードされたファイルを、ファイル名をS3オブジェクト名として、S3バケット 'hoge-fuga-20240512' のinputフォルダに入力ファイルとして格納する。
- Amazon Transcribeに入力ファイルを渡し、日本語の文字起こしをする。
- 文字起こしジョブが完了するまで待機する。この時、待機している旨をUIに表示し、また10秒おきに完了状態を確認することで過度なAPIコールを防ぐ。
- 結果がS3 presigned URLとして返却されるので、これをHTTPS GETで受け取り、LangChainのLCLE記法を用いてClaude V3 Haikuモデルに渡し、議事録をマークダウン形式で生成する。

単一Pythonファイルのアプリケーションとして書いて下さい。また、サンプルアプリでありAWS認証情報はaws configureから渡されるので、コード中で記載する必要はありません。
なお、Bedrockを利用するにあたっては、from langchain_aws import ChatBedrockとし、ChatBedrockモジュールを使用してください。
また可読性を重視し、各処理はdefで関数を作成し、例外も適切に処理してください。
また、LangChainのデバッグモードも有効にして下さい。
また、システムプロンプトとユーザープロンプトは分けて使ってください。

TIPS

• モジュール指定
  何もしないとLLMChainを使おうとするので、from langchain_aws import ChatBedrockとし、ChatBedrockモジュールを使用してくださいという指示を出している。

• エラー処理
  初期状態のコードは、結構誤りが多くエラーも頻発する（それでも、自分でゼロから書いたりサンプルを探したりするよりは遥かに時短になる）。
  この場合は、エラーコードを貼り付けるだけでLLM自身が根本原因と対策案を考え、修正コードを提示してくれる。
  個人的には、ここが一番使えるポイントだと思う。まさしくAI-assisted Coding。

• 最適化
  UIを微妙に調整したい場合んど、痒いところに手が届かなかったりする。
  こうしたディテール、仕上げの部分は、何度か試行錯誤した結果、実は少しだけだが人間の方でコードを追加している。
  一方で、docstringやコメント行の追加、システムプロンプトの作成、使用していない不要なモジュールの削除などは、指示すればLLMの方でよしなにやってくれる。
  LLMができるところは任せ、こだわりたいところは人間が微調整する、というのがよい塩梅な気がした。

• LLMは最新情報を知らない
  よく知られていることだが、LLMはモデルがトレーニングされた時点での情報しか持ち合わせておらず、その後のアップデートは理解していない。LangChainなど、変化が激しいライブラリではこのあたりは結構制約になると感じた。
  対策として考えられるのは、なるべく最新のLLMを使うことぐらいだろうか。

まとめ

デバッグとカスタマイズを含めた完成形を作成するまでの所要時間は、およそ3時間ほど。

前回のポストでも示した通り、自分はろくに開発バックグラウンドを持っていないのだが、それでも、UIを含めた生成AIアプリケーションを片手間で作れてしまう。
Streamlitを始めとしたPythonのUIコンポーネントの進化にも助けられているが、それ以上にLLMの寄与するところは大きい。

それにしても、「日本語」を入力するだけで簡単な開発をこなせる時代が来るとは、10年前には想像もできなかった。
10年後にどういう世界になっているのか、楽しみではある。一方で、イスラエルやウクライナでのAI-assisted weaponsの実践適用のニュースを見ると、不安も禁じ得ないのだが。。。