[2023年4月版]Auto-GPTのコードを解き明かす。驚異のAIハック！～その３
引き続き Auto-GPT のコードをざっくりと読み解いてみます
第１回、第２回に引き続き、ローカルに構築したAuto-GPTが、どのようなことをやっているのか、元のPythonのコードに対してGPT-4にコメントをつけてもらいながら読み解いていきたいと思います。
今回も驚異の処理がザクザクです。ご期待ください！
各コマンドの実装を棚卸し
今回はプロンプトで与えられる以下のコマンドの実装についてコメントと実装コードをご紹介いたします。
1. Google Search: "google", args: "input": "<search>"
5. Browse Website: "browse_website", args: "url": "<url>", "question": "<what_you_want_to_find_on_website>"
6. Start GPT Agent: "start_agent", args: "name": "<name>", "task": "<short_task_desc>", "prompt": "<prompt>"
7. Message GPT Agent: "message_agent", args: "key": "<key>", "message": "<message>"
8. List GPT Agents: "list_agents", args: ""
9. Delete GPT Agent: "delete_agent", args: "key": "<key>"
10. Write to file: "write_to_file", args: "file": "<file>", "text": "<text>"
11. Read file: "read_file", args: "file": "<file>"
12. Append to file: "append_to_file", args: "file": "<file>", "text": "<text>"
13. Delete file: "delete_file", args: "file": "<file>"
14. Search Files: "search_files", args: "directory": "<directory>"
15. Evaluate Code: "evaluate_code", args: "code": "<full_code_string>"
16. Get Improved Code: "improve_code", args: "suggestions": "<list_of_suggestions>", "code": "<full_code_string>"
17. Write Tests: "write_tests", args: "code": "<full_code_string>", "focus": "<list_of_focus_areas>"
18. Execute Python File: "execute_python_file", args: "file": "<file>"
19. Execute Shell Command, non-interactive commands only: "execute_shell", args: "command_line": "<command_line>".
20. Task Complete (Shutdown): "task_complete", args: "reason": "<reason>"
21. Generate Image: "generate_image", args: "prompt": "<prompt>"
22. Do Nothing: "do_nothing", args: ""
番号に欠番があり、以下にご紹介する項番はこれに合わせているので、番号が飛んでます。
申し訳ないですが、ご注意ください。
では以下に列挙していきます！
1. Google Search: "google"
まず最初は "ググる" 処理です。
GoogleのAPIキーを使う場合と使わない場合で検索のAPIが分かれております。
1-1. DuckDuckGo
"ググる"コマンドですがシンプル版(GoogleのAPIキーを使わない版)では DuckDuckGo を使って検索してます。
def google_search(query, num_results=8):
"""Google検索の結果を返す"""
search_results = []
# DuckDuckGoを使って検索を行い、結果を取得する
for j in ddg(query, max_results=num_results):
search_results.append(j)
# 結果をJSON形式に変換して返す
return json.dumps(search_results, ensure_ascii=False, indent=4)
一応、プライバシーはバッチリの模様。
DuckDuckGoのAPIを叩いて検索結果をJSON形式に変換して返却しております。
シンプルでございます。
1-2. Google Search API
Google API キーを用いた場合は正規の Google Search API を使って "専用の検索エンジン" による検索を行います。
def google_official_search(query, num_results=8):
"""公式のGoogle APIを使ってGoogle検索の結果を返す"""
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import json
try:
# 設定ファイルからGoogle APIキーとカスタム検索エンジンIDを取得する
api_key = cfg.google_api_key
custom_search_engine_id = cfg.custom_search_engine_id
# カスタム検索APIサービスを初期化する
service = build("customsearch", "v1", developerKey=api_key)
# 検索クエリを送信し、結果を取得する
result = service.cse().list(q=query, cx=custom_search_engine_id, num=num_results).execute()
# レスポンスから検索結果のアイテムを抽出する
search_results = result.get("items", [])
# 検索結果からURLのみのリストを作成する
search_results_links = [item["link"] for item in search_results]
except HttpError as e:
# API呼び出しのエラーを処理する
error_details = json.loads(e.content.decode())
# エラーが無効または欠落しているAPIキーに関連しているかどうかを確認する
if error_details.get("error", {}).get("code") == 403 and "invalid API key" in error_details.get("error", {}).get("message", ""):
return "Error: 提供されたGoogle APIキーが無効または不足しています。"
else:
return f"Error: {e}"
# 検索結果のURLのリストを返す
return search_results_links
Google公式の検索APIからカスタムの検索エンジンを使用して検索を行っています。
検索結果を取得し、URLのみのリストにして検索結果として返却しております。
このように、公開されている"google.com"サイトでは検索を行わずに、クローズドの検索を採用してくれています。
一応、プライバシーにはちゃんときをつかってくれているようですね。素晴らしいです。
5. Browse Website: "browse_website"
Webサイトを閲覧するコマンドです。要約とリンクURLの抽出を行います。
URL はGoogleなどで検索した結果のURLですが、questionには"WEB検索して見つけたいモノ"が入ってます。
def browse_website(url, question):
"""ウェブサイトを閲覧し、要約とリンクを返す"""
summary = get_text_summary(url, question)
links = get_hyperlinks(url)
# リンクは５個までとする
if len(links) > 5:
links = links[:5]
result = f"""Website Content Summary: {summary}\n\nLinks: {links}"""
return result
この要約ってのが非常にクセモノでございまして、手順を追って Auto-GPT内部でどうやって要約を生成しているかを深堀りしてい行きます。
5-1. テキストの抽出
まず、テキストの抽出です。
簡単かと思いきや、HTMLからのテキスト抽出だけでも自力でプログラム組むのははっきり言って無理です。
HTMLがまともなXMLじゃないのでXMLパーサーは使えないし、タグも壊れている場合が多いので補正とかしてると本末転倒です。
というわけで最初の関数は、指定されたURLのウェブページからテキストをスクレイピング（抽出）するプログラムです。
def scrape_text(url):
"""ウェブページからテキストを抽出する"""
response, error_message = get_response(url)
if error_message:
return error_message
# BeautifulSoupを使ってHTMLを解析
soup = BeautifulSoup(response.text, "html.parser")
# スクリプトとスタイルタグを削除
for script in soup(["script", "style"]):
script.extract()
# テキストを抽出
text = soup.get_text()
# テキストを行ごとに分割し、余分な空白を削除
lines = (line.strip() for line in text.splitlines())
# 各行のフレーズを空白で分割し、余分な空白を削除
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
# 空でないチャンクを連結
text = '\n'.join(chunk for chunk in chunks if chunk)
# 抽出したテキストを返す
return text
まず、URLからレスポンスを取得し、エラーメッセージがある場合はエラーメッセージを返して終了です。
次に、BeautifulSoupを使ってHTMLを解析し、スクリプトやスタイルタグを削除します。
テキストを抽出し、行ごとに分割して余分な空白を削除します。各行のフレーズを空白で分割し、さらに余分な空白を削除します。
最後に、空でないチャンクを連結して抽出したテキストを返します。
はい、HTMLから本文テキスト抜き出すだけでも割とめんどいです。BeautifulSoup、ありがとう！！
5-2. テキストの要約を行う処理全体
つづいて本文がとれたのでようやく、要約を行えるのですが…
次のステップのプログラムは、OpenAI を使用して、与えられたテキストを要約しようとしてます。
def summarize_text(text, question):
"""LLMモデルを使用してテキストを要約する"""
# テキストの長さを計算し、表示する
if not text:
return "Error: No text to summarize"
# テキストの長さを計算し、表示する
text_length = len(text)
print(f"Text length: {text_length} characters")
# 要約されたチャンクを格納するための空のリストを作成
summaries = []
# テキストをチャンクに分割し、リストに格納
chunks = list(split_text(text))
# 各チャンクに対してループ処理を行う
for i, chunk in enumerate(chunks):
# 現在のチャンク番号を表示する
print(f"Summarizing chunk {i + 1} / {len(chunks)}")
# チャンクと質問を含むプロンプトメッセージを作成
messages = [create_message(chunk, question)]
# LLMモデルを使用してチャンクの要約を作成
summary = create_chat_completion(
model=cfg.fast_llm_model,
messages=messages,
max_tokens=300,
)
# 要約をsummariesリストに追加
summaries.append(summary)
# すべてのチャンクが要約されたことを表示する
print(f"Summarized {len(chunks)} chunks.")
# すべての要約を結合する
combined_summary = "\n".join(summaries)
# 結合された要約を含むプロンプトメッセージを作成
messages = [create_message(combined_summary, question)]
# LLMモデルを使用して最終的な要約を作成
final_summary = create_chat_completion(
model=cfg.fast_llm_model,
messages=messages,
max_tokens=300,
)
# 最終的な要約を返す
return final_summary
最初に要約したい本文が空でないことを確認し、テキストの長さを計算して表示します。
次に、テキストをいくつかのチャンクに分割し、それぞれのチャンクを要約します。それぞれの要約は、summaries リストに追加されます。
すべてのチャンクが要約されたら、それらを結合して、さらに要約を行います。
最終的に、この要約が返されます。このプログラムでは、最大トークン数を300に制限しています。
いっぺんに全文を要約するのはトークン足りなくなりそうなのでチャンクに分割してそれぞれを要約し、要約をまとめて要約させてます。
全部を要約ではなく、部分部分の要約を結合した要約、を作成して"全体の要約"とみなしています。
このプログラムはまだ要約の全体フローなので、さらに細かく見ていきます。
5-3. チャンクに分ける
本文テキストをチャンクに分ける処理です。
以下の関数は、与えられたテキストを最大長ごとのチャンク(細切れ)に分割します。
def split_text(text, max_length=8192):
"""最大長さを持つチャンクにテキストを分割する"""
# テキストを改行コードで段落ごとに分割する
paragraphs = text.split("\n")
# 現在のチャンクの長さを初期化
current_length = 0
# 現在のチャンクを格納するための空のリストを作成
current_chunk = []
# 各段落に対してループ処理を行う
for paragraph in paragraphs:
# 現在のチャンクの長さに段落の長さを加算したものが最大長を超えない場合
if current_length + len(paragraph) + 1 <= max_length:
# 現在のチャンクに段落を追加
current_chunk.append(paragraph)
# 現在のチャンクの長さを更新
current_length += len(paragraph) + 1
# 最大長を超える場合
else:
# 現在のチャンクを改行コードで結合し、yieldで返す
yield "\n".join(current_chunk)
# 新しいチャンクを開始し、現在の段落を追加
current_chunk = [paragraph]
# 現在のチャンクの長さを更新
current_length = len(paragraph) + 1
# 最後のチャンクが空でない場合、改行コードで結合してyieldで返す
if current_chunk:
yield "\n".join(current_chunk)
まず、テキストを改行コードで段落ごとに分割し、それぞれの段落をチャンクに追加します。
チャンクの長さが最大長さを超えない場合、段落は現在のチャンクに追加されます。
最大長さを超える場合、現在のチャンクを結合し、yieldを使って返し、新しいチャンクを開始します。
最後のチャンクが空でない場合、結合してyieldで返します。
ここでは、デフォルトのチャンクの最大長は8192文字に設定されています。
このように段落ごとに文字数をチェックしていきながら、最大長を超えてたら超える前のチャンクをyieldで返してしまって、次のチャンクの文字列長チェックに移っています。
こういうのもちゃんと作るの大変なんすよね…
5-4. テキストをAIに要約させる
はい、チャンクにしたテキストを実際に要約させるところです。
以下の関数は、"与えられたテキストのチャンクと質問"をもとに、"ユーザーがそのチャンクを要約するよう求めるメッセージ"を作成するためのものでございます。…はい。
def create_message(chunk, question):
"""テキストのチャンクを要約するためのユーザーメッセージを作成する"""
# ユーザーの役割とコンテンツを含むメッセージを作成
return {
"role": "user",
# 上記のテキストを使用して、次の質問に答えてください: \"{question}\" -- もし質問がテキストを使って答えられない場合は、テキストを要約してください。
"content": f"\"\"\"{chunk}\"\"\" Using the above text, please answer the following question: \"{question}\" -- if the question cannot be answered using the text, please summarize the text."
}
役割が「ユーザー」であるメッセージを作成し、そのコンテンツにはテキストのチャンクと質問を含めます。
質問がテキストを使って答えられない場合は、テキストの要約が求められます。
このメッセージは、言語モデルに要約タスクを依頼するために使用されます。
"質問"は"WEB検索して見つけたいモノ"が入ってます。これに対して答えがコンテンツ内にあればそれを答え、なければ要約を求めます。
なかなか強烈なプロンプトですね…
5-5. リンクURLの抽出
要約に比べれば、余裕のコマンドです。
与えられたURLのウェブページからリンクURLのみを抽出します。
def scrape_links(url):
"""ウェブページからリンクURLを抽出する"""
response, error_message = get_response(url)
if error_message:
return error_message
# BeautifulSoupでHTMLを解析
soup = BeautifulSoup(response.text, "html.parser")
# スクリプトやスタイルタグを除去
for script in soup(["script", "style"]):
script.extract()
# ハイパーリンクを抽出
hyperlinks = extract_hyperlinks(soup)
# ハイパーリンクを整形して返す
return format_hyperlinks(hyperlinks)
まず、URLからレスポンスとエラーメッセージを取得します。エラーがある場合は、エラーメッセージを返します。
次に、BeautifulSoupを使ってHTMLを解析し、スクリプトやスタイルタグを除去します。その後、抽出されたハイパーリンクを整形して返します。この関数は、ウェブページのリンク情報を取得する際に使用されます。
ちなみに成型後のリンクは "リンクテキスト (URL)" の形式に整形されます。
いや～BeautifulSoup万歳！
6. Start GPT Agent: "start_agent"
前回の終わりにご紹介しました通り、このAuto-GPTの"メインのインスタンス"はプロンプトが固定されているので、それ以外のゴールやタスクについて考えさせるために、"サブ"となるエージェントを別途、作成してコンテキストを保存します。
さて以下のプログラムは、指定された名前、タスク、プロンプトを持つエージェントを開始するものです。
def start_agent(name, task, prompt, model=cfg.fast_llm_model):
"""与えられた名前、タスク、プロンプトを使ってエージェントを開始する"""
global cfg
# 名前からアンダースコアを削除
voice_name = name.replace("_", " ")
# エージェントに最初のメッセージを送る
first_message = f"""You are {name}. Respond with: "Acknowledged"."""
agent_intro = f"{voice_name} here, Reporting for duty!"
# エージェントの音声を読み上げ
if cfg.speak_mode:
speak.say_text(agent_intro, 1)
# エージェントを作成
key, ack = agents.create_agent(task, first_message, model)
if cfg.speak_mode:
speak.say_text(f"Hello {voice_name}. Your task is as follows. {task}.")
# タスク（プロンプト）を割り当て、レスポンスを取得
agent_response = message_agent(key, prompt)
return f"Agent {name} created with key {key}. First response: {agent_response}"
まず、エージェント名の調整（アンダースコアを削除）し、エージェントに最初のメッセージを送ります。
つづいて次エージェントを作成し、設定が音声モードである場合は、音声でタスクの内容を伝えます。
最後に、タスク（プロンプト）を割り当て、エージェントからのレスポンスを取得して、エージェントの作成情報を返します。
続いてコアとなる処理もご紹介します。
6-1. エージェントの生成処理
実際にAIを叩いてエージェントを誕生させる処理です。
以下の関数は、新しいエージェントを作成し、そのキーを返す機能を提供します。
def create_agent(task, prompt, model):
"""新しいエージェントを作成し、そのキーを返す"""
global next_key
global agents
messages = [{"role": "user", "content": prompt}, ]
# GPTインスタンスを開始
agent_reply = create_chat_completion(
model=model,
messages=messages,
)
# メッセージ履歴全体を更新
messages.append({"role": "assistant", "content": agent_reply})
key = next_key
# エージェントが削除された場合でもキーを一意にするため、len(agents)の代わりにカウンタをカウントアップする
next_key += 1
agents[key] = (task, messages, model)
return key, agent_reply
まず、ユーザーからのプロンプトに基づいてメッセージを作成します。
つづいてGPTインスタンスを開始し、エージェントにメッセージを送信します。
エージェントからの応答を受け取り、メッセージ履歴を更新して、エージェントの役割として応答を追加します。
新しいエージェントのキーを生成し、キーが一意であることを保証するために、次のキーをインクリメントします。
最後に、エージェントの情報（タスク、メッセージ履歴、モデル）を格納し、キーとエージェントからの最初の応答を返します。
この関数で、新しいエージェントを作成し、エージェントとの対話を可能となります。
6-2. エージェントにプロンプトを処理させる
上で誕生させたエージェントにプロンプトを処理させる部分です。
ここではエージェントにメッセージを送信し、エージェントからの応答を返します。
def message_agent(key, message):
"""エージェントにメッセージを送り、その応答を返す"""
global agents
task, messages, model = agents[int(key)]
# メッセージ履歴にユーザーメッセージを追加し、エージェントに送信
messages.append({"role": "user", "content": message})
# GPTインスタンスを開始
agent_reply = create_chat_completion(
model=model,
messages=messages,
)
# メッセージ履歴全体を更新
messages.append({"role": "assistant", "content": agent_reply})
return agent_reply
まず、与えられたキーを使ってエージェントの情報（タスク、メッセージ履歴、モデル）を取得し、ユーザーからのメッセージをメッセージ履歴に追加します。
その後、GPTのCompletion APIを呼び出し、エージェントとしての役割でメッセージを送信します。エージェントからの応答を受け取り、メッセージ履歴にエージェントの役割としての応答を追加します。
最後に、エージェントからの応答を返します。
この関数は、エージェントとしての対話を行いその応答を取得します。
6-3. 音声で発話
音声で読み上げる部分の処理を紹介します。
以下の関数は、テキストを音声で読み上げる機能を実装しています。音声インデックスに応じて異なる音声を使用することができます。
def say_text(text, voice_index=0):
"""
テキストを音声で読み上げる関数。音声インデックスに応じて異なる音声を使用することができる。
パラメータ:
text (str): 読み上げるテキスト。
voice_index (int): 使用する音声のインデックス。デフォルトは 0。
"""
# 音声読み上げの別スレッド処理用に関数を定義
def speak():
# ElevenLabs の API キーが存在しない場合
if not cfg.elevenlabs_api_key:
# macOS の TTS を使用する場合
if cfg.use_mac_os_tts == 'True':
macos_tts_speech(text, voice_index)
# そうでない場合、gTTS を使用する
else:
gtts_speech(text)
# ElevenLabs の API キーが存在する場合
else:
success = eleven_labs_speech(text, voice_index)
# ElevenLabs が失敗した場合は gTTS を使用する
if not success:
gtts_speech(text)
# セマフォのリリース
queue_semaphore.release()
# セマフォを取得してスレッドを開始する
queue_semaphore.acquire(True)
thread = threading.Thread(target=speak)
thread.start()
構成に応じて、ElevenLabs の音声、macOS の TTS、または Google Text-to-Speech (gTTS) を使用してテキストを読み上げます。
スレッドを使用して音声読み上げをバックグラウンドで実行し、セマフォを使用してタスクの並行実行を制御します。
バックグラウンドでも複数のスレッド＝音声が流れないように、前の音声が終わるまで待って次の音声が流れるようになっております。
まさかセマフォが出てくるとは…こんなところも勉強になりますね。
7. Message GPT Agent: "message_agent"
エージェントにメッセージを送って実行させるコマンドです。
コアな処理は上で紹介してしまいましたが、コマンドから呼び出される部分を紹介いたします。
以下の関数は指定されたキーとメッセージを使用してエージェントにメッセージを送ります。
def message_agent(key, message):
"""与えられたキーとメッセージでエージェントにメッセージを送る"""
global cfg
# キーが有効な整数であるかどうかを確認
if is_valid_int(key):
agent_response = agents.message_agent(int(key), message)
# キーが有効な文字列であるかどうかを確認
elif isinstance(key, str):
agent_response = agents.message_agent(key, message)
else:
return "Invalid key, must be an integer or a string."
# レスポンスを話す
if cfg.speak_mode:
speak.say_text(agent_response, 1)
return agent_response
まず、キーが有効な整数もしくは文字列であるかを確認し、エージェントにメッセージを送ります。
そのエージェントからのレスポンスを受け取り、レスポンスを返します。
また音声モードが有効な場合はレスポンスなどを読み上げてもらいます。
音声読み上げの処理も上記で紹介したものです。
8. List GPT Agents: "list_agents"
エージェントのリストアップはサクッとまいりましょう。
まずはコマンドで呼び出す部分。
def list_agents():
"""すべてのエージェントをリストアップする"""
return agents.list_agents()
この agents.list_agents() は以下のようになってます。
def list_agents():
"""Return a list of all agents"""
global agents
# Return a list of agent keys and their tasks
return [(key, task) for key, (task, _, _) in agents.items()]
このようにすべてのエージェントのキーとタスクを一覧として表示します。
9. Delete GPT Agent: "delete_agent"
エージェントの削除処理です。
def delete_agent(key):
"""指定されたキーを持つエージェントを削除する"""
result = agents.delete_agent(key)
if not result:
return f"Agent {key} does not exist."
return f"Agent {key} deleted."
指定されたキーでエージェントを削除します。キーが無ければその旨のメッセージを表示して終わりです。
agents.delete_agent(key)について以下に紹介します。
def delete_agent(key):
"""Delete an agent and return True if successful, False otherwise"""
global agents
try:
del agents[int(key)]
return True
except KeyError:
return False
ハッシュテーブルから要素の削除をおこなっています。以上です。
10. Write to file: "write_to_file"
ファイルへの書き込み処理です。
指定したファイル名と作業ディレクトリのパスを結合してちゃんと書き込めるならテキストを書き込む、処理です。
def write_to_file(filename, text):
"""Write text to a file"""
try:
filepath = safe_join(working_directory, filename)
directory = os.path.dirname(filepath)
if not os.path.exists(directory):
os.makedirs(directory)
with open(filepath, "w") as f:
f.write(text)
return "File written to successfully."
except Exception as e:
return "Error: " + str(e)
ファイル操作は特にツッコむところはないですが、たまには基本的な処理で人の書いたコードを見てみるのも勉強になりますね。
11. Read file: "read_file"
続いてファイルの読み取りです。
def read_file(filename):
"""Read a file and return the contents"""
try:
filepath = safe_join(working_directory, filename)
with open(filepath, "r", encoding='utf-8') as f:
content = f.read()
return content
except Exception as e:
return "Error: " + str(e)
ここも特に注意点はないでしょう。
12. Append to file: "append_to_file"
ファイルの追加書き込み記処理です。
def append_to_file(filename, text):
"""Append text to a file"""
try:
filepath = safe_join(working_directory, filename)
with open(filepath, "a") as f:
f.write(text)
return "Text appended successfully."
except Exception as e:
return "Error: " + str(e)
open(filepath, "a") で追記モードですね。はい。
13. Delete file: "delete_file"
ファイルの削除です。
def delete_file(filename):
"""Delete a file"""
try:
filepath = safe_join(working_directory, filename)
os.remove(filepath)
return "File deleted successfully."
except Exception as e:
return "Error: " + str(e)
ここも特に触れることはないですね…
14. Search Files: "search_files"
フォルダの検索処理です。
作業フォルダ以下の特定のパス以下のファイルリストを作成してます。
def search_files(directory):
found_files = []
if directory == "" or directory == "/":
search_directory = working_directory
else:
search_directory = safe_join(working_directory, directory)
for root, _, files in os.walk(search_directory):
for file in files:
if file.startswith('.'):
continue
relative_path = os.path.relpath(os.path.join(root, file), working_directory)
found_files.append(relative_path)
return found_files
os.walk(search_directory) でこのディレクトリ以下のすべてのファイルとディレクトリを捜査してます。
ついでに root, _, files と受け取っているのでディレクトリ名は無視してファイル名だけを取り出して、さらに相対パスにしてます。
絶対パスではないので、ユーザー名など流出しないようにしてくれてます。便利。
15. Evaluate Code: "evaluate_code"
はい、出ました。AIにコードの改善をやらせちゃうパターンです。
この関数は、与えられたコードをAIに評価させ、コードの改善に関する提案のリストを返します。
def evaluate_code(code: str) -> List[str]:
"""
文字列を受け取り、create_chat_completion API呼び出しからの応答を返す関数です。
パラメータ:
code (str): 評価されるコード。
戻り値:
create_chat_completionからの結果文字列。コードを改善するための提案のリスト。
"""
function_string = "def analyze_code(code: str) -> List[str]:"
args = [code]
# 与えられたコードを分析し、改善のための提案のリストを返します。
description_string = """Analyzes the given code and returns a list of suggestions for improvements."""
# Chat Completion API で評価させる
result_string = call_ai_function(function_string, args, description_string)
return result_string
"AIに関数としての役割をふるまわせる"関数=AI関数 function_stringを定義し、与えられたコードをリストにしてargsに格納します。
続いてdescription_stringに関数の説明を格納します。
ここでは def analyze_code(code: str) -> List[str]: というシグネチャと "与えられたコードを分析し、改善のための提案のリストを返します。" という説明を定義しております。
最後にcall_ai_function 関数を使ってAI関数を呼び出し、その結果を返却します。
call_ai_function 関数については前回の記事の"GPTに "Python関数" 役をさせる"の節で紹介しています。ぜひ、再度、参照してください。
function_stringで定義したシグネチャとdescription_stringの説明文から類推される振る舞いから結果だけを返してもらう、強烈な処理です。
16. Get Improved Code: "improve_code"
続いて上記で指摘した内容を元に、改善されたコードの生成もAIにやらせちゃいます。
以下はコードと改善提案を受け取り、改善されたコードを返します。
def improve_code(suggestions: List[str], code: str) -> str:
"""
コードと提案を受け取り、create_chat_completion API呼び出しからの応答を返す関数です。
パラメータ:
suggestions (List): 改善が必要な点に関する提案のリスト。
code (str): 改善されるべきコード。
戻り値:
create_chat_completionからの結果文字列。応答で改善されたコード。
"""
function_string = (
"def generate_improved_code(suggestions: List[str], code: str) -> str:"
)
args = [json.dumps(suggestions), code]
# 提供された提案に基づいて提供されたコードを改善し、それ以外の変更は行わないでください。
description_string = """Improves the provided code based on the suggestions provided, making no other changes."""
result_string = call_ai_function(function_string, args, description_string)
return result_string
上の処理と同様、"AIに関数としての役割をふるまわせる"関数=AI関数 function_stringを定義し、与えられたコードをリストにしてargsに格納します。
続いてdescription_stringに関数の説明を格納します。
今回は def generate_improved_code(suggestions: List[str], code: str) -> str: というシグネチャと "提供された提案に基づいて提供されたコードを改善し、それ以外の変更は行わないでください。" という説明を定義しております。
最後にcall_ai_function 関数を使ってAI関数を呼び出し、その結果を返却します。
17. Write Tests: "write_tests"
続いて上記で改善されたコードと指摘された内容を元に、テストコードの生成もAIにやらせちゃいます。
以下の関数は、コードと指摘されたピックを入力として受け取り、作成されたテストケースを返す関数です。
def write_tests(code: str, focus: List[str]) -> str:
"""
コードと観点のトピックを入力として受け取り、create_chat_completion API コールの応答を返す関数。
パラメータ:
focus (List): 改善が必要な箇所に関する提案のリスト。
code (str): テストケースが生成される対象のコード。
戻り値:
create_chat_completionからの結果文字列。応答で提出されたコードのテストケース。
"""
function_string = (
"def create_test_cases(code: str, focus: Optional[str] = None) -> str:"
)
args = [code, json.dumps(focus)]
# 既存のコードに対してテストケースを生成してください。必要に応じて特定の観点に着目したテストケースを作成してください。
description_string = """Generates test cases for the existing code, focusing on specific areas if required."""
result_string = call_ai_function(function_string, args, description_string)
return result_string
上の処理と同様、"AIに関数としての役割をふるまわせる"関数=AI関数 function_stringを定義し、与えられたコードをリストにしてargsに格納します。
続いてdescription_stringに関数の説明を格納します。
今回は def create_test_cases(code: str, focus: Optional[str] = None) -> str: というシグネチャと "既存のコードに対してテストケースを生成してください。必要に応じて特定の観点に着目したテストケースを作成してください。" という説明を定義しております。
最後にcall_ai_function 関数を使ってAI関数を呼び出し、その結果の生成されたテストケースを返却します。
call_ai_functionが以下に強力か… おわかりいただけただろうか？
18. Execute Python File: "execute_python_file"
続いてのコマンドは、Pythonの実行です。
以下の関数はDockerコンテナ内でPythonファイルを実行し、出力を返す処理です。
def execute_python_file(file):
"""
Dockerコンテナ内でPythonファイルを実行し、出力を返す関数
パラメータ:
file (str): 実行するPythonファイルの名前。
"""
print (f"Executing file '{file}' in workspace '{WORKSPACE_FOLDER}'")
# ファイルが.py形式であることを確認
if not file.endswith(".py"):
return "Error: Invalid file type. Only .py files are allowed."
file_path = os.path.join(WORKSPACE_FOLDER, file)
# ファイルが存在するか確認
if not os.path.isfile(file_path):
return f"Error: File '{file}' does not exist."
try:
client = docker.from_env()
image_name = 'python:3.10'
# Dockerイメージがローカルに存在するか確認
try:
client.images.get(image_name)
print(f"Image '{image_name}' found locally")
except docker.errors.ImageNotFound:
# イメージが存在しない場合、Docker Hubからプル
print(f"Image '{image_name}' not found locally, pulling from Docker Hub")
low_level_client = docker.APIClient()
for line in low_level_client.pull(image_name, stream=True, decode=True):
status = line.get('status')
progress = line.get('progress')
if status and progress:
print(f"{status}: {progress}")
elif status:
print(status)
# Dockerコンテナを作成して、Pythonファイルを実行
container = client.containers.run(
image_name,
f'python {file}',
volumes={
os.path.abspath(WORKSPACE_FOLDER): {
'bind': '/workspace',
'mode': 'ro'}},
working_dir='/workspace',
stderr=True,
stdout=True,
detach=True,
)
# 実行結果を取得し、コンテナを削除
output = container.wait()
logs = container.logs().decode('utf-8')
container.remove()
return logs
except Exception as e:
return f"Error: {str(e)}"
ちょっと長いですが、やっていることは大したことはないです。
まず与えられたPythonファイルが.py形式であり、指定されたワークスペースフォルダに存在することを確認します。
ファイルがあった場合、Pythonイメージをローカルで探し、Dockerイメージがローカルに存在しない場合、Docker Hubからプルします。
続いてコンテナを作成し、Pythonファイルを実行します。
実行が完了したら、実行結果を取得し、コンテナを削除します。最後に、実行結果を返します。
これは何かの参考になりますね～。
19. Execute Shell Command, non-interactive commands only: "execute_shell"
続いてシェルでのコマンド実行です。
なかなか危険ですが…まず以下のコマンドの分岐の箇所です。
elif command_name == "execute_shell":
if cfg.execute_local_commands:
return execute_shell(arguments["command_line"])
else:
return "You are not allowed to run local shell commands. To execute shell commands, EXECUTE_LOCAL_COMMANDS must be set to 'True' in your config. Do not attempt to bypass the restriction."
このように execute_local_commands の設定で実際にシェルのコマンドを実行させるかどうかを指定できます。デフォルトでは False となっています。
で、実際の処理は以下の通りです。
def execute_shell(command_line):
"""
シェルコマンドを実行し、その出力を返す関数
パラメータ:
command_line (str): 実行するシェルコマンド
"""
current_dir = os.getcwd()
# 必要に応じて、ワークスペースフォルダに移動
if not WORKSPACE_FOLDER in current_dir: # Change dir into workspace if necessary
work_dir = os.path.join(os.getcwd(), WORKSPACE_FOLDER)
os.chdir(work_dir)
print (f"Executing command '{command_line}' in working directory '{os.getcwd()}'")
# シェルコマンドを実行し、出力をキャプチャ
result = subprocess.run(command_line, capture_output=True, shell=True)
output = f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
# 以前の作業ディレクトリに戻る
os.chdir(current_dir)
return output
まず、現在の作業ディレクトリを取得し、必要に応じてワークスペースフォルダに移動します。
続いてシェルコマンドを実行し、標準出力と標準エラー出力をキャプチャします。
最後に、元の作業ディレクトリに戻り、実行結果を返します。
なかなか危険ですが、このように別プロセスを立ち上げてコマンドを実行させることも可能です。
20. Task Complete (Shutdown): "task_complete"
完了のタスクです。
elif command_name == "task_complete":
shutdown()
目的がすべて完遂できた場合にはAuto-GPTのシャットダウンを行います。
21. Generate Image: "generate_image"
最後の大物、画像の生成処理です。もちろんAIを使います。
ここでは、与えられたプロンプトに基づいて画像を生成し、ディスクに保存する関数です。
def generate_image(prompt):
"""
与えられたプロンプトに基づいて画像を生成し、それをディスクに保存する関数。
パラメータ:
prompt (str): 画像生成のためのプロンプト。
"""
filename = str(uuid.uuid4()) + ".jpg"
# DALL-E
if cfg.image_provider == 'dalle':
openai.api_key = cfg.openai_api_key
# DALL-Eを使って画像を生成するAPI呼び出し
response = openai.Image.create(
prompt=prompt,
n=1,
size="256x256",
response_format="b64_json",
)
print("Image Generated for prompt:" + prompt)
# 画像データをbase64からデコードする
image_data = b64decode(response["data"][0]["b64_json"])
# 画像データをディスクに保存
with open(working_directory + "/" + filename, mode="wb") as png:
png.write(image_data)
return "Saved to disk:" + filename
# STABLE DIFFUSION
elif cfg.image_provider == 'sd':
API_URL = "https://api-inference.huggingface.co/models/CompVis/stable-diffusion-v1-4"
headers = {"Authorization": "Bearer " + cfg.huggingface_api_token}
# Stable Diffusion API呼び出し
response = requests.post(API_URL, headers=headers, json={
"inputs": prompt,
})
# 画像データを取得し、Imageオブジェクトを作成
image = Image.open(io.BytesIO(response.content))
print("Image Generated for prompt:" + prompt)
# 画像データをディスクに保存
image.save(os.path.join(working_directory, filename))
return "Saved to disk:" + filename
else:
return "No Image Provider Set"
まずこのプログラムは、設定に応じて、DALL-EまたはStable Diffusionを使って画像を生成しております。
DALL-Eの場合、openai.Image.createメソッドを使ってAPI呼び出しを行い、画像を生成します。画像データは、base64形式でエンコードされているため、デコードしてディスクに保存します。
Stable Diffusionの場合、requests.postを使ってAPI呼び出しを行い、画像を生成します。画像データはImage.openで読み込み、ディスクに保存します。
どちらの場合も、生成された画像はディスクに保存され、保存されたファイル名が返されます。
ここで画像を生成するプロンプトはGPTのCompletion APIによって生成されています。
ここでもメタなAIの使い方してますね～…
22. Do Nothing: "do_nothing"
最後はお馴染みのNOOPに相当するコマンドです。
elif command_name == "do_nothing":
return "No action performed."
はい、アクションは何も実行されません。
お疲れさまでした！！
まとめ
というわけでAuto-GPTのメインの処理はほぼ、網羅いたしました。いかがだったでしょうか。
AIと対話しながらタスクをこなしていく、仕事をさせつつ、できないことはこちらでやりつつ、いかにプロジェクトの状態を与えるか、少ないトークンで与えるか。言葉じゃなくてもいいんだろうけど。
どちらにしても今のところはプロンプトの構築技術がキモかなと、改めて実感したコードリーディングでした。
最後までお付き合いいただきありがとうございました！
Discussion