1. イントロダクション
近年、対話型AIアシスタントは、単なるQ&Aツールから、より高度で柔軟な思考パートナーへと進化しています。
中でもOpenAIのAssistant APIは、アシスタント自身が必要なツールを状況に合わせて新規に「発明」し、それを活用してユーザの要求に応えることを可能にします。
このようなツール自律生成型のアプローチは、たとえばCode Interpreter（2023年現在ChatGPTで実装されている機能）との違いを明確にします。
Code Interpreterでは、ユーザが「データを可視化してほしい」といった要求をするたびに、その都度コードを新規生成・実行する必要があり、同様の要求に再度対応する際には再びコード生成から行わなければなりません。
一方、Assistant APIを用いた自律的ツール追加手法では、一度作成したツールをアシスタントが保持し、再利用できる点が大きな強みです。これにより、同様の要求が再び来た際には即座に既存ツールを呼び出すだけで対応でき、繰り返し発生するタスクに対する効率性が大幅に向上します。
本記事では、このような「アシスタントが自らツールを設計・追加し、長期的に再利用可能にする」デモコードを紹介します。これを活用することで、アシスタントはより知的で柔軟な存在へと発展し、ユーザー体験が飛躍的に向上する可能性があります。
2. モチベーション
従来、アシスタントに新機能を与える場合、開発者があらかじめ予測可能なツール群を用意し、それを参照させる方法が一般的でした。
しかし、この手法には以下のような課題があります。
- 事前準備コストの高さ：想定される要求を幅広くカバーするツールセットを用意しなければならない
- 変化への弱さ：新たな要求が出るたびに、開発者が手作業でツールを追加・更新する必要がある
これらの問題を解決する手段として「自律的なツール追加」が注目されます。アシスタントが自身で「不足している機能」を発見し、その場でツールを新規作成・登録できれば、以下のような利点が得られます。
- 機能拡張の自動化：ユーザ要求に応じて即時にツールを拡張
- 再利用による効率化：一度追加したツールは、似た要求にも使い回し可能
- 開発者負担の軽減：アシスタント自身が機能拡張するため、開発者の運用コスト削減
こうした仕組みによって、アシスタントはユーザとの対話を通じて成長し、継続的に知能や機能を強化することができます。
3. デモコードの紹介
ここでは、OpenAI API（Assistant機能）を用いて、アシスタントが自らツールを新規追加し、そのツールを使ってユーザ要求に応答するデモコードを示します。以下で示すポイントが本デモの要所です。
-
自律的なツール生成の流れ
ユーザからの要求を受け取った際、アシスタントは「不足する機能」を判断し、create_tool_and_update_assistant_tools関数を介して新ツールを生成・登録します -
再利用の仕組み
一度追加されたツールは、アシスタントのツールリストに残り、後から同様の要求が来た際には、既に存在するツールを直接呼び出せます。これにより、何度も同じコード生成を繰り返す必要がありません -
柔軟な拡張性
新しい要求が出るたびにアシスタントは自身の機能セットを拡大できるため、時間とともにアシスタントの能力が充実していきます
以下に実際のデモコードを掲載します。
例として「yfinanceを使って、AAPLの現在の株価を教えて」と命令した時に、
ツールを自動で追加した上で、それに上手く回答出来るかを確かめます。
デモコード
import time
import json
import os
from openai import OpenAI
INSTRUCTIONS = """
あなたは、自分のために必要な道具は自分で作り、その道具を使ってユーザーの質問に答えるアシスタントです。
最初にcreate_tool_and_update_assistant_toolsを使って新しいツールを作成し、そのツールを使って回答してください。
ツールを作成した後は、新しく作成したツールを使って回答を完了してください。
create_tool_and_update_assistant_toolsは一度しか使わないこと
"""
tools_list = [
{
"type": "function",
"function": {
"name": "create_tool_and_update_assistant_tools",
"description": "When you need to create a new tool, use this function. Use this tool only once overall, **never repeatedly**.",
"parameters": {
"type": "object",
"properties": {
"requirement": {
"type": "string",
"description": "The requirement of the tool"
}
},
"required": ["requirement"]
}
}
},
]
def update_assistant_tools(tool_json, assistant_id):
tool_list = tools_list + [tool_json]
my_assistant = client.beta.assistants.update(
assistant_id=assistant_id,
tools=tool_list,
)
print(f'tool_list: {tool_list}')
return my_assistant
def create_function(name, args, body):
# function_bodyから余分な関数定義を除去
body_lines = body.strip().split('\n')
cleaned_body_lines = []
import_statements = []
for line in body_lines:
if line.startswith(f'def {name}'):
continue # 関数定義の行をスキップ
elif line.strip().startswith('import '):
import_statements.append(line)
else:
cleaned_body_lines.append(line)
cleaned_body = '\n'.join(cleaned_body_lines)
# import文をグローバルスコープで実行
for stmt in import_statements:
exec(stmt, globals())
# 関数定義を構築
args_str = ", ".join(args)
func_def = f"def {name}({args_str}):\n"
for line in cleaned_body.split('\n'):
func_def += f" {line}\n"
# 関数定義を出力（デバッグ用）
print(f'func_def: {func_def}')
# 関数をグローバルスコープに定義
exec(func_def, globals())
return globals()[name]
# Initialise the OpenAI client
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# Create a new assistant
my_assistant = client.beta.assistants.create(
instructions=INSTRUCTIONS,
name="Assistant",
tools=tools_list,
model="gpt-4o",
temperature=0,
)
assistant_id = my_assistant.id
def create_tool_and_update_assistant_tools():
pass
available_functions = {'create_tool_and_update_assistant_tools': create_tool_and_update_assistant_tools}
def create_tool_and_update_assistant_tools(requirement):
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": [
{
"type": "text",
"text": """Receive a requirement text and output details about a Python function that fulfills the requirement.
Provide the following details in JSON format:
# Output Format
- "function_name": string
- "arguments": array of strings
- "function_body": string
- "tool_json":
{
"type": "function",
"function": {
"name": "function_name",
"description": "function_description",
"parameters": {
"type": "object",
"properties": {
"arg1": {
"type": "string",
"description": "arg1_description"
},
"arg2": {
"type": "string",
"description": "arg2_description"
}
},
"required": ["arg1", "arg2"]
}
}
}
# Notes
Ensure the precision of the function details to match the given requirement exactly."""
}
]
},
{
"role": "user",
"content": [
{
"type": "text",
"text": f"{requirement}"
}
]
}
],
response_format={
"type": "json_object"
},
temperature=0,
max_tokens=2048,
top_p=1,
frequency_penalty=0,
presence_penalty=0
)
json_response = json.loads(response.choices[0].message.content)
# json_response の内容を確認
print(f'json_response: {json_response}')
update_assistant_tools(json_response["tool_json"], assistant_id)
available_functions[json_response["tool_json"]["function"]["name"]] = create_function(
json_response["function_name"],
json_response["arguments"],
json_response["function_body"]
)
return "tool_created"
available_functions = {'create_tool_and_update_assistant_tools': create_tool_and_update_assistant_tools}
thread = client.beta.threads.create()
# アシスタントが回答のメッセージを返すまで待つ関数
def wait_for_assistant_response(thread_id, run_id):
while True:
global available_functions
print(f'available_functions: {available_functions}')
time.sleep(5)
run = client.beta.threads.runs.retrieve(
thread_id=thread_id,
run_id=run_id
)
status = run.status
if status == "failed":
print(f'status: {status}')
# エラーメッセージを取得
error_message = run.error
print(f'Error: {error_message}')
break
elif status == "completed":
print(f'status: {status}')
break
elif status == "requires_action":
if (
run.required_action.type == "submit_tool_outputs"
and run.required_action.submit_tool_outputs.tool_calls is not None
):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_responses = []
for call in tool_calls:
if call.type == "function":
print(f'call.function.name: {call.function.name}')
print(f'call.function.arguments: {call.function.arguments}')
print(f'available_functions keys: {list(available_functions.keys())}')
function_to_call = available_functions.get(call.function.name)
if function_to_call:
tool_response = function_to_call(**json.loads(call.function.arguments))
if tool_response is None:
tool_response = "Error: The function did not return a value."
else:
tool_response = str(tool_response) # Ensure the output is a string
tool_responses.append({"tool_call_id": call.id, "output": tool_response})
else:
print(f"Function {call.function.name} not found in available_functions.")
print(f'tool_responses: {tool_responses}')
# ツールの出力を送信し、run_idを更新
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
)
run_id = run.id
else:
continue
# スレッドのメッセージを確認する関数
def print_thread_messages(thread_id):
msgs = client.beta.threads.messages.list(thread_id=thread_id)
for m in msgs:
# assert m.content[0].type == "text"
print({"role": m.role, "message": m.content[0].text.value})
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="yfinanceを使って、AAPLの現在の株価を教えて"
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=my_assistant.id
)
# アシスタントが回答のメッセージを返すまで待つ
wait_for_assistant_response(thread.id, run.id)
# スレッドのメッセージを表示
print_thread_messages(thread.id)
実際にyinflanceを使った新しいツール(get_current_stock_price)を自分で作り、それを用いて回答出来ていることが確認出来ました。
4. 結論と将来の展望
自律的なツール生成と再利用は、AIアシスタントの拡張性・柔軟性を大きく高め、より豊かなユーザー体験を実現します。今回のデモはその基本的な一例ですが、今後は以下のような発展が期待できます。
- 高度なツール管理：ツール同士の競合解消や、自動的な不要ツールの削除など
- 知的計画能力の強化：複数ツールを組み合わせた高度なタスク達成戦略
- 領域特化型アシスタントへの進化：特定分野に特化したツールセットを自律的に形成し、深い専門知識をもったエージェントとして機能
このような方向性により、AIアシスタントはユーザとの対話を通じて自発的に進化・成長し、より高度かつ有用なパートナーとして活躍する可能性が広がります。