Auto-GPTがどのようにコマンド実行の判断しているのかを探る
イントロ
最近のAI界隈の盛り上がりの中、個人的に最も盛り上がっている興味は ChatGPT やAIを使って副作用のある活動を自動化するところにある。「副作用のある活動」とは、例えば以下のようなもの。
ローカルのコードを指示通りに改変する
改変したコードをコミットする
特定のブランチを元にPRを作る
PRの内容にレビュー (コメント) する
AIとのディスカッションを元にローカルのドキュメントを改善する
AIとのディスカッションに関して、要点や決まったことを自動でまとめてディスカッションの中で逐次更新していく
ようはAIで作業を自動化したい。「後やっといて」「りょ」で数時間後にできている、しかもその活動が追える状態が理想的である。
業務では難しいかも知れないが、個人のプロダクトならいくらでもできる。やりたい。スプラ3したい。
そこで出会った ChatGPT を利用する Auto-GPT には、「これかもしれない」と期待を禁じ得なかった。とうとう「俺の思いついた最強プロダクト (仮) を勝手に作ってくれる環境が来たかも」と。さっそく試してみた。上手く扱えなかった。
で、悔しかったので自分で作ってみることにした。名前は aias アイアス。
ChatGPT があるとはいえ、やはり書いたことない Python でプロダクトを作ることは容易ではない。まあ走りだしなので言うほど難しくもないけど。
ここで疑問が生じる。正確には前から疑問ではあったけど書くことになったので必要になった。
「ChatGPT からの出力をコマンドやコードの部分だけ正確に出力するにはどうしたらいいんだ？」
コードを出力するにも、ファイルの一部分だけをアウトプットしたとすると、どのように置き換えているのだろう。AIに聞くのも良いけど読むことにする。
対象のコード
$ git log | head
commit 5c93c1a58be93395c35ba4c62fd1cc8b9b17361f
Merge: ee8276c 93c2582
Author: Toran Bruce Richards <toran.richards@gmail.com>
Date: Mon Apr 10 13:31:07 2023 +0100
Merge pull request #457 from kinance/master
Fix the debug mode flag to set it in the config object and check properlyちょうど master ブランチへコミットがゴリゴリ積まれるタイミングだったようで、ゴリゴリ更新されている
scripts/main.py タッチポイント
このコードは、AI と対話しながらコマンドを実行するプログラムです。全体の流れは以下の通りです：
1. 必要な関数やライブラリをインポートし、事前に設定した値を読み込みます。
2. 使用する AI モデルの名前を設定し、プロンプトを構築します。
3. 変数を初期化し、メッセージ履歴を保持するための空のリストを作成します。
4. メモリを取得し、それが空であることを確認します。これは、後でインデックスや参照に使われるためです。
5. 無限ループを開始し、AI との対話を行います。
6. AI からの応答を受け取り、それを解析してコマンド名と引数を取得します。
7. コンフィグレーションによって、ユーザーからの確認を求めるか、連続モードで実行するかを判断します。
8. ユーザーの入力に応じて、コマンドを実行し、結果をメッセージ履歴に追加します。
9. コマンドの結果やユーザーからのフィードバックをメモリに追加します。
10. 7-9 の手順を繰り返し実行し、ユーザーが終了を選択するまでプログラムを実行します。
主な目的は、AI との対話を通じてコマンドを生成し、それらのコマンドを実行して結果を表示することです。また、ユーザーからのフィードバックを受け取り、それをメモリに追加することも可能です。
前準備 1 ~ 4
# TODO: fill in llm values here
check_openai_api_key()
cfg = Config()
parse_arguments()
ai_name = ""
prompt = construct_prompt()
# print(prompt)
# Initialize variables
full_message_history = []
result = None
next_action_count = 0
# Make a constant:
user_input = "Determine which next command to use, and respond using the format specified above:"
# Initialize memory and make sure it is empty.
# this is particularly important for indexing and referencing pinecone memory
memory = get_memory(cfg, init=True)
print('Using memory of type: ' + memory.__class__.__name__)5. AI と対話するための無限ループを開始
# Interaction Loop
while True:無限ループを回してインタラクションなやりとりを開始する。
以後の概要
大きく以下のセクションに分かれる
ChatGPTへのリクエスト
ChatGPTのレスポンスを整形 (THOUGHTS:, REASONING:, CRITICISM:) して出力
ChatGPTのレスポンスからコマンドを抽出
分岐a:
継続モードではなく自動実行許可回数 y -n の残数が0だった場合 (アクションにユーザー許可が必要な場合)
後述
継続モードだった場合 (--continuous フラグ or 自動実行許可回数が残っている)
"NEXT ACTION:" として抽出したコマンドを出力
分岐b:
コマンドが error だった場合
コマンド実行結果にエラーを代入
コマンドが自由記述 human_feedback だった場合
コマンド実行結果に自由記述を代入
実行可能なコマンド (要検証) だった場合
コマンドを実行
自動実行許可回数を -1
これまでのやりとりをメモリに追加する
分岐c:
コマンド実行結果が存在する場合
全文履歴に system プロプトとして保存
"SYSTEM:" としてコマンド実行結果を出力
コマンド実行結果がなかった場合 (コード上起こりえないはず)
全文履歴に system プロプトとして "Unable to execute command" を保存
"SYSTEM:" としてコマンド実行できなかったことを出力
アクションにユーザー許可が必要な場合の挙動
「4. 分岐a」の継続モードではなく自動実行許可回数 y -n の残数が0だった場合
if not cfg.continuous_mode and next_action_count == 0:
### GET USER AUTHORIZATION TO EXECUTE COMMAND ###
# Get key press: Prompt the user to press enter to continue or escape
# to exit
user_input = ""
print_to_console(
"NEXT ACTION: ",
Fore.CYAN,
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
print(
f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {ai_name}...",
flush=True)
while True:
console_input = input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
if console_input.lower() == "y":
user_input = "GENERATE NEXT COMMAND JSON"
break
elif console_input.lower().startswith("y -"):
try:
next_action_count = abs(int(console_input.split(" ")[1]))
user_input = "GENERATE NEXT COMMAND JSON"
except ValueError:
print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
continue
break
elif console_input.lower() == "n":
user_input = "EXIT"
break
else:
user_input = console_input
command_name = "human_feedback"
break
if user_input == "GENERATE NEXT COMMAND JSON":
print_to_console(
"-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
Fore.MAGENTA,
"")
elif user_input == "EXIT":
print("Exiting...", flush=True)
break順番に見ていく
print_to_console(
"NEXT ACTION: ",
Fore.CYAN,
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
print(
f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {ai_name}...",
flush=True)"NEXT ACTION:" とヒントを出力
while True:
console_input = input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
if console_input.lower() == "y":
user_input = "GENERATE NEXT COMMAND JSON"
break
elif console_input.lower().startswith("y -"):
try:
next_action_count = abs(int(console_input.split(" ")[1]))
user_input = "GENERATE NEXT COMMAND JSON"
except ValueError:
print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
continue
break
elif console_input.lower() == "n":
user_input = "EXIT"
break
else:
user_input = console_input
command_name = "human_feedback"
break
無限ループを開始してユーザー入力のため待機。入力の内容によって4つの処理に分岐
"y" のみ
"y -n" と自動実行許可回数を指定した場合
"n" だった場合
それ以外 (自由記述)
if user_input == "GENERATE NEXT COMMAND JSON":
print_to_console(
"-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
Fore.MAGENTA,
"")
elif user_input == "EXIT":
print("Exiting...", flush=True)
breakユーザー入力が "y" か "y -n" だった場合、許可されたことを出力する
ユーザー入力が "n" だった場合、AIとの対話を終了する
全体像が見えてきたところで、本来読みたかったところを確認していく。
「ChatGPT からの出力をコマンドやコードの部分だけ正確に出力するにはどうしたらいいんだ？」
Auto-GPTの場合だと、概要3の「ChatGPTのレスポンスからコマンドを抽出」の部分
# Get command name and arguments
try:
command_name, arguments = cmd.get_command(assistant_reply)
except Exception as e:
print_to_console("Error: \n", Fore.RED, str(e))scripts/commands.py AIレスポンスからコマンドを抽出
最新版はこちら
https://github.com/Torantulino/Auto-GPT/blob/master/scripts/commands.py
コマンド抽出処理は以下
def get_command(response):
"""Parse the response and return the command name and arguments"""
try:
response_json = fix_and_parse_json(response)
if "command" not in response_json:
return "Error:" , "Missing 'command' object in JSON"
command = response_json["command"]
if "name" not in command:
return "Error:", "Missing 'name' field in 'command' object"
command_name = command["name"]
# Use an empty dictionary if 'args' field is not present in 'command' object
arguments = command.get("args", {})
if not arguments:
arguments = {}
return command_name, arguments
except json.decoder.JSONDecodeError:
return "Error:", "Invalid JSON"
# All other errors, return "Error: + error message"
except Exception as e:
return "Error:", str(e)`fix_and_parse_json(response)` で以下の形式の JSON 形式に変換し、
{
"command": {
"name": "command name",
"args":{
"arg name": "value"
}
}
}コマンド名とその引数を返却するシンプルな関数だった。神髄は #fix_and_parse_json(response) にありそうなのでさらに読んでいく
scripts/json_parser.py AIレスポンスから JSON を生成
最新版
https://github.com/Torantulino/Auto-GPT/blob/master/scripts/json_parser.py
def fix_and_parse_json(
json_str: str,
try_to_fix_with_gpt: bool = True
) -> Union[str, Dict[Any, Any]]:
"""Fix and parse JSON string"""
try:
json_str = json_str.replace('\t', '')
return json.loads(json_str)
except json.JSONDecodeError as _: # noqa: F841
json_str = correct_json(json_str)
try:
return json.loads(json_str)
except json.JSONDecodeError as _: # noqa: F841
pass
# Let's do something manually:
# sometimes GPT responds with something BEFORE the braces:
# "I'm sorry, I don't understand. Please try again."
# {"text": "I'm sorry, I don't understand. Please try again.",
# "confidence": 0.0}
# So let's try to find the first brace and then parse the rest
# of the string
try:
brace_index = json_str.index("{")
json_str = json_str[brace_index:]
last_brace_index = json_str.rindex("}")
json_str = json_str[:last_brace_index+1]
return json.loads(json_str)
except json.JSONDecodeError as e: # noqa: F841
if try_to_fix_with_gpt:
print("Warning: Failed to parse AI output, attempting to fix."
"\n If you see this warning frequently, it's likely that"
" your prompt is confusing the AI. Try changing it up"
" slightly.")
# Now try to fix this up using the ai_functions
ai_fixed_json = fix_json(json_str, JSON_SCHEMA)
if ai_fixed_json != "failed":
return json.loads(ai_fixed_json)
else:
# This allows the AI to react to the error message,
# which usually results in it correcting its ways.
print("Failed to fix ai output, telling the AI.")
return json_str
else:
raise e- この関数は、与えられた JSON 文字列を修正し、Python の辞書に変換します。
- まず、タブ文字を削除し、JSON 文字列を Python の辞書に変換しようとします。失敗した場合、correct_json() 関数を使って修正を試みます。
- さらに失敗した場合、最初と最後の中括弧を探して、その範囲の文字列で再試行します。
- それでも失敗した場合、try_to_fix_with_gpt が True の場合、AI を使って修正を試みます。
- AI を使っても失敗した場合、エラーを出力し、JSON 文字列をそのまま返します。
面白いのは4回JSONの取得を試みていること。
普通に parse
#correct_json という関数 (scripts/json_utils.py) を実行して parse
ChatGPT が JSON テキストの前後に自然言語を添えている可能性を考慮して parse
ChatGPT に JSON テキストの修正を試みさせて #fix_json から parse
もちろんそれぞれで修正方法を変えているが、当コマンドの画期的な部分でもあるので力が入っているように見える。
#correct_json はこちら。今回は主軸ではないので掲載だけ。エラーコードに依って修正方法を変えてみたり、波括弧の数を調整してみたりと工夫が見える。
correct_json 関数は、一般的なJSONエラーを修正することを試みます。具体的な処理は以下の通りです。
1. 与えられたJSON文字列（json_str）を json.loads() 関数で解析を試みます。解析が成功した場合、JSON文字列にエラーがないと判断し、そのまま返します。
2. 1. json.JSONDecodeError が発生した場合、エラーメッセージを取得して、どの種類のエラーが発生したかを判断します。
3. エラーメッセージが 'Invalid \escape' で始まる場合、fix_invalid_escape 関数を呼び出して、無効なエスケープシーケンスを修正しようとします。
4. エラーメッセージが 'Expecting property name enclosed in double quotes' で始まる場合、add_quotes_to_property_names 関数を呼び出して、プロパティ名にダブルクォーテーションを追加しようとします。その後、再度 json.loads() 関数で解析を試みます。解析が成功した場合、修正されたJSON文字列を返します。
5. 上記のエラー修正が解決しない場合、balance_braces 関数を呼び出して、括弧のバランスを修正しようとします。括弧が修正された場合、修正されたJSON文字列を返します。
6. すべての修正が失敗した場合、与えられたJSON文字列をそのまま返します。
この関数は、一般的なJSONエラーを修正することを目的としており、完全な解決策ではありませんが、多くの場合で有効です。
#fix_json
def fix_json(json_str: str, schema: str) -> str:
"""Fix the given JSON string to make it parseable and fully complient with the provided schema."""
# Try to fix the JSON using gpt:
function_string = "def fix_json(json_str: str, schema:str=None) -> str:"
args = [f"'''{json_str}'''", f"'''{schema}'''"]
description_string = "Fixes the provided JSON string to make it parseable"\
" and fully complient with the provided schema.\n If an object or"\
" field specified in the schema isn't contained within the correct"\
" JSON, it is ommited.\n This function is brilliant at guessing"\
" when the format is incorrect."
# If it doesn't already start with a "`", add one:
if not json_str.startswith("`"):
json_str = "```json\n" + json_str + "\n```"
result_string = call_ai_function(
function_string, args, description_string, model=cfg.fast_llm_model
)
if cfg.debug:
print("------------ JSON FIX ATTEMPT ---------------")
print(f"Original JSON: {json_str}")
print("-----------")
print(f"Fixed JSON: {result_string}")
print("----------- END OF FIX ATTEMPT ----------------")
try:
json.loads(result_string) # just check the validity
return result_string
except: # noqa: E722
# Get the call stack:
# import traceback
# call_stack = traceback.format_exc()
# print(f"Failed to fix JSON: '{json_str}' "+call_stack)
return "failed"- この関数は、与えられたJSON文字列を修正し、指定されたスキーマに従ってフォーマットを整えることを目的としています。
- まず、AIによるJSON修正を試みます。これは、call_ai_function() を使ってAIに関数を実行させることによって行われます。
- AIによって修正されたJSONが有効であることを確認し、有効であればそのJSON文字列を返します。そうでなければ、"failed" 文字列を返します。
ChatGPT にリクエストしているのは #call_ai_function (system prompt を使って) なので、その前後を見る
result_string = call_ai_function(
function_string, args, description_string, model=cfg.fast_llm_model
)# This is a magic function that can do anything with no-code. See
# https://github.com/Torantulino/AI-Functions for more info.
まず、system prompt (function_string, description_string) と user prompt (args) を生成している箇所。
function_string = "def fix_json(json_str: str, schema:str=None) -> str:"
args = [f"'''{json_str}'''", f"'''{schema}'''"]
description_string = "Fixes the provided JSON string to make it parseable"\
" and fully complient with the provided schema.\n If an object or"\
" field specified in the schema isn't contained within the correct"\
" JSON, it is ommited.\n This function is brilliant at guessing"\
" when the format is incorrect."Fixes the provided JSON string to make it parseable and fully complient with the provided schema.
If an object or field specified in the schema isn't contained within the correct JSON, it is ommited.
This function is brilliant at guessing when the format is incorrect.
提供されたJSON文字列を解析可能で、提供されたスキーマに完全に準拠するように修正する。
スキーマで指定されたオブジェクトやフィールドが正しいJSONに含まれていない場合、それは省略されます。
この関数は、フォーマットが間違っていることを推測するのに優れています。
次に JSON テキストを ``` で囲う処理。
# If it doesn't already start with a "`", add one:
if not json_str.startswith("`"):
json_str = "```json\n" + json_str + "\n```"とはいえ、以後 json_str を使うのは debug 中のみなので、この処理は不要か、ここに記述されるべきではないのだろう。ここで開始地点を明記する ```json とするのであれば、arg 代入の部分も指定したい気持ちがある
args = [f"'''JSON{json_str}'''", f"'''SCHEMA{schema}'''"]とか。
残りの処理は JSON parse を試みて、成功なら JSON テキストを、ダメなら "failed" を返す。
で、この処理結果が #fix_and_parse_json で JSON parse され、#get_command でコマンドを取得するという流れになる。
なるほど、以下の疑問について、大分理解が進んだ。
「ChatGPT からの出力をコマンドやコードの部分だけ正確に出力するにはどうしたらいいんだ？」
取得したコマンドはどうなるのか
もちろん実行される訳だが。関数は #execute_command
実行できるコマンドは 2023/04/11 時点で以下の通り。
"google"
"memory_add"
"start_agent"
"message_agent"
"list_agents"
"delete_agent"
"get_text_summary"
"get_hyperlinks"
"read_file"
"write_to_file"
"append_to_file"
"delete_file"
"search_files"
"browse_website"
"evaluate_code"
"improve_code"
"write_tests"
"execute_python_file"
"generate_image"
"do_nothing"
"task_complete"
なるほど、append_to_file や improve_code がコードを書き換える感じなのだろうか？明日以降見ていこうと思う。
余談
`scripts/commands.py` の `if not arguments:` のくだりがいらなくね？と思ったので PullRequests を建てた
ai-shell という別プロダクトもあり、こちらも気になっているので近いうちに読む
この記事が気に入ったらサポートをしてみませんか？