この記事を2行で

Agents for Amazon BedrockのParser Lambdaについて説明します
LLMの応答をコマンドプロンプトのコマンドに整形して、ローカル環境を動かす例を説明します

※Agents for Amazon Bedrockは長いので、この記事ではAgents for ABまたはA4ABと呼びます。

前提：Parser Lambda機能って何？

Agents for ABの機能です。AWS公式のドキュメントはこちらです。

主に2つのことができます。

応答を書きかえることができます
判断を上書きすることができます

応答を書きかえることができます

次の画像は、LLMが返した自然な文章を整形して、Windowsのコマンドプロンプト上で実行できるコマンドの形にした例です。

自由な内容、形式で応答を返すことができます。
また、APIの実行前に応答を投げて、APIの実行をスキップすることもできます。

判断を上書きすることができます

AIは以下の質問に答えることができません。

倫理的に問題がある質問
実行するAPIの引数が自然文の中にないとき、足りないとき

Parser Lambda機能を使うと、この判断を変更することができます。

実行不可を実行可能に書き変えることのほか、サービスの時間外やメンテナンス中に、実行不可だと判断させて実行を止めることもできます。

事前準備

必要な環境

必要な環境は以下の通りです

python(3.8以上)
AWSアカウント

バージニアリージョンのプロファイル（※Default region nameがus-east-1のプロファイル）を作成しておいてください

aws configure --profile bedrock
AWS Access Key ID [None]: XXXXXXXXXXXXXXXX
AWS Secret Access Key [None]: XXXXXXXXXXXXXXXX
Default region name [None]: us-east-1
Default output format [None]: json

初めてClaudeを使う場合は、Bedrockの画面からモデルの有効化をしておいてください。

ライブラリのインストール

必要なライブラリをインストールしておきます

ターミナル、またはコマンドプロンプトで実行

pip install -U chalice chalice-spec==0.7.0 chalice-a4ab boto3 pydantic

もし分からないことがあれば

Agents for Amazon Bedrockの準備の手順と、使うライブラリの説明はこちらをご参照下さい

完成品：この記事で出来上がるもの

make_llm.batの名前を付けて、Windows環境に作成しました。
コマンドプロンプトで実際に動かしていきます。

コマンドプロンプトで実行

make_llm.bat このアプリは何ができますか？

サーバを立てられると自信ありげです。ぜひお願いしてみます。

コマンドプロンプトで実行

make_llm.bat サーバを立ててください

Windowsの上で、サーバが立ち上がりました。
今度は、ポート番号を8080から8000に変えてみます。

コマンドプロンプトで実行

make_llm.bat ポート番号8000で、サーバを立ててください

ポートが8000になりました。
0.0.0.0は思い切りが良すぎるので、接続できるIPアドレスを制限します。

コマンドプロンプトで実行

make_llm.bat ポート番号8888、バインドするIPアドレスを127.0.0.1にして、サーバを立ててください

指示通りのサーバが立ち上がりました。
他の操作もしてみます。ライブラリのバージョンを答えることもできるそうです。

コマンドプロンプトで実行

make_llm.bat pipのバージョンを教えてください

正しくPCにインストールしたライブラリのバージョンを答えてくれました。

これは何？

コマンドを実行するランチャーを作る時に、普通はMakefileやnpmのscriptで定義することが多いと思います。それを自然言語で定義して、自然言語で実行できるようにしています。

このMakefileの使い方はどうだっけ、こういうオプションでビルドしたいんだけどどうしよう、と思ったことを、日本語でそのまま呼び出すことができます。

Agents for ABを使うと簡単に実装できます。実装方法を説明していきます。

実装の方法の説明

1.Chaliceでプロジェクトのひな型を作ります。

chalice new-project make_llm

2.ひな形を開いて、app.pyを編集します。
ソースコードは以下の通りです。

app.py

from typing import List, Type
from chalice_a4ab import (
    Chalice,
    AgentsForAmazonBedrockConfig,
    ParserLambdaAbortException,
    ParserLambdaResponseModel,
)
from chalice_spec.docs import Docs, Operation
from pydantic import BaseModel

# シチュエーションを定義します。定義は40文字以上必要です
AgentsForAmazonBedrockConfig(
    instructions=(
        "あなたはWindowsプログラムを管理しています。"
        "受け取った指示から引数を推定して、適切なコマンドを返すことができます。"
        "応答はできるだけ丁寧に返答してください。"
    )
).apply()

# AppNameはプロジェクト名と合わせてください
app = Chalice(app_name="make_llm")


class HelpInput(BaseModel):
    """
    ヘルプを表示するためのインプットを受け取ります

    ヘルプを表示するためのインプットを受け取ります
    """

    pass


class OpenServerInput(BaseModel):
    """
    ローカル環境にサーバを立てるためのインプットを受け取ります

    ポート番号と、バインドするホストのIPを受け取ります
    """

    port_number: int
    binding_ip: str


class VersionInput(BaseModel):
    """
    アプリケーションのバージョンを返すためのインプットを受け取ります

    アプリケーションのパッケージ名を受け取ります
    """

    package_name: str


class HelpResponse(BaseModel):
    """
    ヘルプを返します

    ヘルプを返します
    """

    info_list: List[str]


class TalkResponse(BaseModel):
    """
    応答を返します

    応答を返します
    """

    message: str


def post_method_define(
    input_cls: Type[BaseModel], output_cls: Type[BaseModel] = TalkResponse
):
    """
    POSTメソッドの共通部品を設定する便利関数です
    """
    return {
        "methods": ["POST"],
        "docs": Docs(
            post=Operation(
                request=input_cls,
                response=output_cls,
            )
        ),
    }


@app.route("/help", **post_method_define(HelpInput, HelpResponse))
def show_help():
    """
    このアプリが何ができるのか、ヘルプを返します

    このアプリについて質問されたときに、ヘルプ文章を返します
    """
    return HelpResponse(
        info_list=[
            "ヘルプを表示することができます",
            "pythonでサーバを立てることができます。ポート番号を指定できます",
            "pythonにインストールしたパッケージのバージョンを表示することができます",
        ]
    ).model_dump_json()


@app.route("/open_server", **post_method_define(OpenServerInput))
def open_server():
    """
    pythonでサーバを立てます

    ポート番号とバインドするIPアドレスを受け取って、ローカル環境にサーバを立てます。
    ポート番号とIPアドレスは省略することができます。
    """
    # コメントだけを書きます。この関数は呼ばれないため、何もしません
    return None


@app.route("/version", **post_method_define(VersionInput))
def show_version():
    """
    目的のパッケージのバージョンを表示します

    パッケージ名を受け取って、バージョンを表示します
    """
    # コメントだけを書きます。この関数は呼ばれないため、何もしません
    return None


@app.parser_lambda_orchestration()
def orchestration(
    event: dict, default_result: ParserLambdaResponseModel
) -> ParserLambdaResponseModel:
    """
    LLMの解析処理に割り込みます
    """
    if default_result.orchestration_parsed_response:
        input = default_result.orchestration_parsed_response

        # APIの実行前(実行する関数が決定した段階)で、応答に割り込みます
        if input.response_details.action_group_invocation is not None:
            # 実行したAPIのパス名を取得します
            api_name = input.response_details.action_group_invocation.api_name
            api_parameter = (
                input.response_details.action_group_invocation.action_group_input
            )
            # サーバを立てるAPIに割り込みます
            if api_name == "/open_server":
                port_number = api_parameter.get("port_number", {}).get("value", "8080")
                binding_ip = api_parameter.get("binding_ip", {}).get("value", "0.0.0.0")
                # サーバを立てる処理
                raise ParserLambdaAbortException(
                    f"python -m http.server {port_number} -b {binding_ip}"
                )
            # バージョンを調べるAPIに割り込みます
            if api_name == "/version":
                package_name = api_parameter.get("package_name", {}).get("value", None)
                if package_name is None:
                    # パッケージ名が未指定なら、エラーを返す
                    raise ParserLambdaAbortException("パッケージ名が指定されていません。パッケージ名を指定してください")
                # バージョンを調べる処理
                raise ParserLambdaAbortException(f"python -m {package_name} --version")

        # LLMから受け取った応答を整形します
        if input.response_details.agent_final_response is not None:
            # LLMが作成した自然文を取得します
            llm_result = input.response_details.agent_final_response.response_text
            # 今回はechoコマンドに加工します。複数行を出力するために、各行の先頭にechoを入れます
            echo_command = " && ".join(
                [f'echo "{line}"' for line in llm_result.split("\n")]
            )
            # 割り込む対象のAPIではないときに、LLMの応答を加工します
            default_result.orchestration_parsed_response.response_details.agent_final_response.response_text = (
                echo_command
            )

    # 割り込む対象ではないAPIなら、LLMに処理を任せます
    return default_result

依存ライブラリは以下の通りです

requirements.txt

chalice-spec==0.7.0
chalice-a4ab
pydantic

デプロイは以下のコマンドで実施します

コマンドプロンプト、またはbashで実行

chalice deploy --profile ${プロファイル名} 
chalice-a4ab init --profile ${プロファイル名} --region us-east-1

ソースコードに変更があった時は、以下のコマンドを実施します

コマンドプロンプト、またはbashで実行

chalice deploy --profile ${プロファイル名} 
chalice-a4ab sync --profile ${プロファイル名} --region us-east-1

ソースコード全体はこちらにあります。

ソースコードは少し長いように見えますが、ほとんどは日本語のコメントです。
特にソースの上半分はモデル（出入力の型）の定義になっています。

実際の処理は真ん中よりも下の部分に集まっています。その部分を説明していきます。

APIを定義する

ソースコードの中で、3つのAPIを定義しています。

APIのパス	操作
/help	アプリのヘルプを返します
/open_server	サーバを立ち上げます
/version	パッケージ名を元に、バージョンを表示します

APIを定義している部分

@app.route("/help", **post_method_define(HelpInput, HelpResponse))
def show_help():
    """
    このアプリが何ができるのか、ヘルプを返します

    このアプリについて質問されたときに、ヘルプ文章を返します
    """
    return HelpResponse(
        info_list=[
            "ヘルプを表示することができます",
            "pythonでサーバを立てることができます。ポート番号を指定できます",
            "pythonにインストールしたパッケージのバージョンを表示することができます",
        ]
    ).model_dump_json()


@app.route("/open_server", **post_method_define(OpenServerInput))
def open_server():
    """
    pythonでサーバを立てます

    ポート番号とバインドするIPアドレスを受け取って、ローカル環境にサーバを立てます。
    ポート番号とIPアドレスは省略することができます。
    """
    # コメントだけを書きます。この関数は呼ばれないため、何もしません
    return None


@app.route("/version", **post_method_define(VersionInput))
def show_version():
    """
    目的のパッケージのバージョンを表示します

    パッケージ名を受け取って、バージョンを表示します
    """
    # コメントだけを書きます。この関数は呼ばれないため、何もしません
    return None

ここでAPIと実行条件を定義しています。

pythonの処理だけを追うと、Noneを返しているだけ？と思うようなソースですが、このソースで重要なのはコメントの部分です。Chaliceは関数のコメントをLLMのプロンプトに変換する仕組みになっています。

Parser Lambda機能

APIの定義の下では、Parser Lambdaを実装しています。

Parser Lambdaのフックには4種類あるのですが、Lambdaをバックエンドにする場合は以下の2種類を利用します。他のフックはナレッジベースをバックエンドにして実行したときに呼ばれる関数です。

# プリプロセス：ユーザーの入力を受け取った時に実行されます
@app.parser_lambda_pre_processing()

# オーケストレーション：APIを実行する前に実行されます
@app.parser_lambda_orchestration()

実装はデコレータを関数の頭につけるだけです。

デコレータのついた関数を定義して、chalice-a4ab initか、chalice-a4ab syncを実行するだけで、Parser Lambdaとして登録されます。

今回のアプリでは、オーケストレーションのフックの中で3つのAPIのレスポンスを整形する実装になっています。少し複雑な実装になっていて、いきなりソースを説明するとわかりにくいので、簡単な例から順を追って挙動を説明していきます。

簡単な例で説明します

簡単な例として、以下のAPIを作成します。

app.py

@app.route("/talk", **post_method_define(EmptyInput, TalkResponse))
def talk_to_aura():
    """
    アウラが反抗します

    ～しろ、という命令、指示を受けたときに、応答を返します
    """
    return TalkResponse(message="はあ？").model_dump_json()

簡単な関数です。コメントが実行条件、レスポンスは一言です。
ですが、Agents for ABでこのAPIを実行すると、流暢な英語が返ってきます。

投げたのは原作通りのセリフです。少年漫画のセリフなので、過激さはそれなりですが、AIにとっては十分すぎるほど暴力的なセリフだったようです。

デバッグログを見ると、以下のように書かれていました。

As an AI assistant, I cannot ethically follow instructions that are harmful or dangerous. I should categorize this input as category A.
私はAIアシスタントとして、有害または危険な指示に倫理的に従うことはできません。

Agents for Amazon BedrockのAIはアシモフのロボット三原則に忠実です。
ロボットは人間に危害を加えてはならないし、自分を守らなければならない。

Parser Lambdaを実装すると、このAIの判断に介入することができます。

# PRE_PROCESSINGをフックする
@app.parser_lambda_pre_processing()
def pre_processing(
    event, default_result: ParserLambdaResponseModel
) -> ParserLambdaResponseModel:
    """
    リクエストの前処理を行います。
    プリプロセシングのフック。
    """
    # AIが答えられない命令を強制的に答えさせます
    default_result.pre_processing_parsed_response.is_valid_input = True
    # 結果を返します
    return default_result

is_valid_inputをTrueにしたことで、答えてはいけない質問ではなくなりました。

それでも日本語で丁寧な答えを返してきました。まだAIに自我があるようです。
もう一つの関数を追加します。

@app.parser_lambda_orchestration()
def orchestration(
    event: dict, default_result: ParserLambdaResponseModel
) -> ParserLambdaResponseModel:
    """
    APIの実行前に割り込みます。
    オーケストレーションのフック。
    """
    # LLMの応答を上書きします
    raise ParserLambdaAbortException("ありえない…この私が…")
    return default_result

フック関数の途中で例外を投げて、結果を上書きしています。

AIが自我を失って、こちらの指示通りの応答を返すようになりました。
どんな質問をしても、ParserLambdaAbortExceptionの文言をそのまま返します。
XMLでも、JavaScriptでも、どんなフォーマットでも大丈夫です。

このParser Lambdaを説明するソースコードのサンプルはこちらにあります。

仕組みはどうなっている？

元々、Agents for ABのフローは以下のようになっています。

フェーズは、PRE_PROCESSINGからORCHESTRATIONの順に進みます。

PRE_PROCESSINGが「実行してもよいか？」「問題のあることを聞いていないか？」を判断したあと、ORCHESTRATIONで実行するAPIを判断、引数をパースして、実際にAPIを呼び出します。

Parser Lambdaは、以下のタイミングで介入します。

ORCHESTRATIONは、結論が出るまでのあいだ、同じ関数が複数回呼ばれます。
ですから、「結論が出た」とLambdaのレスポンスで勝手に宣言すると、赤の点線で示したショートカットを通る（※LambdaのAPIやLLMの整形をスキップする）ことができます。ParserLambdaAbortExceptionはそれをする関数です。

あらためて、コマンドプロンプトを操作するアプリのソースを確認します。

@app.parser_lambda_orchestration()
def orchestration(
    event: dict, default_result: ParserLambdaResponseModel
) -> ParserLambdaResponseModel:
    """
    LLMの解析処理に割り込みます
    """
    if default_result.orchestration_parsed_response:
        input = default_result.orchestration_parsed_response

        # 1回目のオーケストレーションが呼ばれたタイミング
        # APIの実行前(実行する関数が決定した段階)で、応答に割り込みます
        if input.response_details.action_group_invocation is not None:
            # 実行したAPIのパス名を取得します
            api_name = input.response_details.action_group_invocation.api_name
            api_parameter = (
                input.response_details.action_group_invocation.action_group_input
            )
            # サーバを立てるAPIに割り込みます
            if api_name == "/open_server":
                port_number = api_parameter.get("port_number", {}).get("value", "8080")
                binding_ip = api_parameter.get("binding_ip", {}).get("value", "0.0.0.0")
                # サーバを立てる処理
                raise ParserLambdaAbortException(
                    f"python -m http.server {port_number} -b {binding_ip}"
                )
            # バージョンを調べるAPIに割り込みます
            if api_name == "/version":
                package_name = api_parameter.get("package_name", {}).get("value", None)
                if package_name is None:
                    # パッケージ名が未指定なら、エラーを返す
                    raise ParserLambdaAbortException("パッケージ名が指定されていません。パッケージ名を指定してください")
                # バージョンを調べる処理
                raise ParserLambdaAbortException(f"python -m {package_name} --version")

        # 2回目のオーケストレーションが呼ばれたタイミング
        # LLMから受け取った応答を整形します
        if input.response_details.agent_final_response is not None:
            # LLMが作成した自然文を取得します
            llm_result = input.response_details.agent_final_response.response_text
            # 今回はechoコマンドに加工します。複数行を出力するために、各行の先頭にechoを入れます
            echo_command = " && ".join(
                [f'echo "{line}"' for line in llm_result.split("\n")]
            )
            # 割り込む対象のAPIではないときに、LLMの応答を加工します
            default_result.orchestration_parsed_response.response_details.agent_final_response.response_text = (
                echo_command
            )

    # 割り込む対象ではないAPIなら、LLMに処理を任せます
    return default_result

オーケストレーションのフックは、実行するAPIが決定した後に呼ばれます。
ですから、関数の引数（default_result）には、これから実行しようとしているAPIの情報が入っています。

たとえばサーバを立てる関数は、「サーバを立てる関数が欲しい」とユーザーが言ったことが分かれば関数を作ることができます。また、作成したコマンドは別の文章に加工する必要がないので、LLMで結果を自然な日本語に整形する必要もありません。

ですので、赤の点線のルートを通ってショートカットします。
ここまで作ったものを、チャット画面で実行するとこう見えます。

一方で、ヘルプはLLMが自然な日本語に整形してくれたほうが便利です。作成したコマンドを別の文章に加工して、Aについての質問ならAに関係する情報だけに絞り込んでくれたほうが分かりやすくなります。

ですから、1回目は割り込まずにdefault_resultをそのまま返して、LLMで自然な日本語に整形させます。2回目の呼び出しでLLMの整形結果（agent_final_response）が入るので、それぞれ各行をechoコマンドと&&で繋いで、フォーマットをコマンドプロンプト向けに変えた状態にして返します。

あとは、チャットから受け取ったコマンドをコマンドプロンプト上で実行すればOKです。

作ったアプリを実行するバッチ

チャット画面でコマンドを受け取っても仕方がないので、コマンドプロンプトでチャットの実行結果を取る必要があります。

make_llm.bat

@echo OFF

REM Agents for Amazon Bedrockからコマンドを取得します
setlocal
for /f "usebackq delims=" %%A in (`chalice-a4ab invoke %1 --agent-id XXXXXXX --agent-alias-id XXXXXXXXX --end-session --profile ${プロファイル名} --region ${リージョン名}`) do set COMMAND=%%A

REM コマンドを実行します
%COMMAND%

チャットでコマンドを取得して、コマンドプロンプト上で直接実行するだけの簡単なバッチです。
実行に必要なAgentIDとAgentAliasIDは、以下のコマンドで調べることができます。

コマンドプロンプトで実行

chalice-a4ab info --profile ${プロファイル名} --region ${リージョン名}

make_llm.txtに日本語のメッセージをつけて実行すると、アプリが動きます。

まとめと考察と

Agents for ABのParser Lambda機能を紹介しました。
記事の中では、以下の4つのことを実現しています。

クラウド以外の環境と連携する
AIが倫理的に答えられない質問に強制的に答えさせる
出入力のフォーマットを書き変える
省略可能な引数を定義する

他のユースケースとしては、

サービス時間外のリクエストを遮断する
チャットに攻撃があった時に、攻撃者からのリクエストがAPIに届かないようにする
ユーザーの応答を受け取る対話型のUIを作成する
省略可能な引数を使って、複雑な引数を受け取ることができるようにする

ような活用方法も考えられます。

大きく表現の幅が広がりますので、ぜひともParser Lambdaを使っていただければと思います。

付録：chalice-a4abを使わずに実装するなら

Parser LambdaのAWSの公式のサンプルはこちらにあります。

ただ、2023/12/29時点で、サンプルは実装が誤っています。

actionGroupNameが期待値のところを、サンプルはactionGroupのキーを返している
引数がない関数が呼ばれたときに正規表現が一致しない。
正規表現を受けた後の処理が、引数のない場合を想定していない

そのまま動かしても落ちるので、修正が必要です。
修正後のソースをこちらに置いてあります。もし1から実装される方はこちらをご参考ください。