アシスタント API は関数呼び出しをサポートしています。この機能により、エージェントは外部関数を自動的に呼び出して、テキスト翻訳などのタスクを実行できます。このトピックでは、簡単な「翻訳エージェント」の例を通して、関数呼び出しの基本を素早く理解できるように説明します。
重要アシスタント API は非推奨になりつつあります。代替として Responses API への移行を検討してください。Responses API には、複数の組み込みツールが含まれており、マルチターンコンテキスト管理をサポートしています。
クイックスタート
この例では、翻訳エージェントと、そのエージェントが呼び出すことができる translate_text という名前の関数を作成します。次に、エージェントに "Hello world" を中国語に翻訳するように依頼します。
事前準備
次のコマンドを実行して、requests や dashscope などの必要な依存関係ライブラリをインストールできます。
pip install requests dashscope
ステップ 1：「translate_text」関数の作成
まず、簡単な翻訳関数を作成します。この関数は、デモンストレーション目的で事前定義された翻訳テーブルを使用します。
def translate_text(text, target_language):
"""
指定されたターゲット言語にテキストを翻訳します。
これは、事前定義された翻訳を使用する簡単なデモンストレーションです。
パラメーター：
text (str)：翻訳するテキスト。
target_language (str)：ターゲット言語コード (例：'zh'、'es'、'ja')。
戻り値：
str：翻訳されたテキストまたはエラーメッセージ。
"""
# デモンストレーション用の翻訳辞書。
mock_translations = {
('Hello world', 'zh'): '你好世界',
('Hello world', 'es'): '¡Hola Mundo!',
('Hello world', 'ja'): 'こんにちは世界',
('How are you?', 'zh'): '你好吗？',
('How are you?', 'es'): '¿Cómo estás?',
('How are you?', 'ja'): 'お元気ですか？'
}
try:
return mock_translations.get((text, target_language),
f"翻訳が見つかりません。本番環境では、ここで翻訳サービスが呼び出されます。")
except Exception as e:
return f"翻訳に失敗しました：{str(e)}"
- 翻訳機能：事前定義された翻訳テーブルを使用して翻訳機能をシミュレートし、多言語間の変換をサポートします。
- エラー処理：この関数には、あらゆる状況で適切な応答を返すことを保証するための基本的なエラー処理メカニズムが含まれています。
これで、アシスタント API を使用してエージェントを作成できます。このエージェントは、ユーザーのクエリを自動的に処理し、定義された translate_text 関数を呼び出して翻訳サービスを提供します。
ステップ 2：「translate_text」関数の記述
エージェントに translate_text 関数を記述する必要があります。エージェントはこの記述を使用して関数を正しく呼び出します。
from dashscope import Assistants, Messages, Runs, Threads
import json
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
# 翻訳ツールを定義します
translation_tool = {
"type": "function",
"function": {
"name": "translate_text",
"description": "Translates text into the specified target language",
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "The text to translate"
},
"target_language": {
"type": "string",
"description": "The target language code (for example, 'zh', 'es', or 'ja')"
}
},
"required": ["text", "target_language"]
}
}
}
- name：関数の名前は
translate_textです。エージェントはこの名前を使用して関数を呼び出します。 - description：エージェントがその目的を理解するのに役立つツールの説明です。
- parameters：翻訳するテキストとターゲット言語を含む、関数のパラメーターを定義します。
ステップ 3：エージェントの作成
これで、アシスタントインスタンスを作成できます。このインスタンスは、定義した翻訳ツールを使用するエージェントです。
# アシスタントを作成します
assistant = Assistants.create(
model='qwen-plus',
name='Translation Agent',
description='An agent that can translate text between different languages',
instructions='You are a translation agent. When a user requests a translation, use the translate_text function to help them.',
tools=[translation_tool]
)
- model：使用するモデルを指定します。この例では、言語理解とタスク処理をサポートする
qwen-plusを使用します。 - name：エージェントの名前です。値を「Translation Assistant」に設定します。
- description：ユーザーのテキスト翻訳を支援するというエージェントの目的の説明です。
- tools：以前に定義した
translation_toolを登録します。これにより、エージェントがツールを呼び出すことができます。
ステップ 4：会話スレッドを作成し、エージェントと対話する
新しい会話スレッドを作成し、それにユーザーメッセージを追加してから、エージェントを実行してユーザーのクエリを処理します。
# 新しいスレッドを作成します
thread = Threads.create()
# スレッドにユーザーメッセージを追加します
Messages.create(
thread_id=thread.id,
role="user",
content="Please translate 'Hello world' into Chinese."
)
# アシスタントを実行します
run = Runs.create(thread_id=thread.id, assistant_id=assistant.id)
# 実行が完了するのを待ちます
run = Runs.wait(thread_id=thread.id, run_id=run.id)
- Threads.create()：後続のメッセージのために新しい会話スレッドを作成します。
- Messages.create()：スレッドにユーザーメッセージを追加します。この場合、ユーザーは「Hello world」を中国語に翻訳するように依頼します。
- Runs.create()：エージェントをトリガーして、ユーザーメッセージの処理を開始します。
- Runs.wait()：エージェントの処理が完了するのを待ちます。
ステップ 5：関数呼び出しを処理し、結果を返す
エージェントが処理中にツールを呼び出す必要がある場合、translate_text 関数が呼び出され、結果が返されます。
# 関数呼び出しが必要かどうかを確認します
if run.required_action:
for tool_call in run.required_action.submit_tool_outputs.tool_calls:
if tool_call.function.name == "translate_text":
args = json.loads(tool_call.function.arguments)
translation = translate_text(args["text"], args["target_language"])
# ツール出力を送信します
Runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=[{"tool_call_id": tool_call.id, "output": translation}]
)
# 新しい実行が完了するのを待ちます
run = Runs.wait(thread_id=thread.id, run_id=run.id)
- 関数呼び出しの確認：エージェントが関数を呼び出す必要がある場合、コードは要求された関数が
translate_textであるかどうかを確認し、以前に定義された関数を使用して翻訳を実行します。 - 結果の送信：
Runs.submit_tool_outputsを使用して翻訳結果をエージェントに送信し、エージェントの次の応答を待ちます。
ステップ 6：エージェントの応答を取得する
エージェントの処理が完了したら、会話スレッドからエージェントの応答を取得し、ユーザーに表示できます。
# アシスタントの応答を取得します
messages = Messages.list(thread_id=thread.id)
for message in messages.data:
if message.role == "assistant":
print(f"Assistant: {message.content[0].text.value}")
まとめ
これらのステップに従うことで、ユーザーの翻訳リクエストを処理し、翻訳関数を使用してテキスト変換を実行できるエージェントを正常に作成できました。アシスタント API を使用すると、複雑なタスク駆動型のエージェントを簡単かつ効率的に構築できます。
必要に応じて、ツールの追加やエージェントの動作命令の変更など、エージェントの機能を拡張できます。
ビジネス関数の説明の迅速な生成
クイックスタートの例では、エージェントに「translate_text」関数を記述する必要があります。このプロセスは面倒な場合があります。そのため、ビジネス関数を迅速に記述するのに役立つ簡単な変換関数を提供します。
import inspect
def function_to_schema(func) -> dict:
# Python の型を JSON スキーマの型にマッピングします
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
type(None): "null",
}
# 関数のシグネチャを取得しようとします
try:
signature = inspect.signature(func)
except ValueError as e:
# シグネチャの取得に失敗した場合、エラーメッセージとともにエラーを発生させます
raise ValueError(
f"Failed to get signature for function {func.__name__}: {str(e)}"
)
# パラメーターの型を格納する辞書を初期化します
parameters = {}
# 関数のパラメーターを反復処理し、その型をマッピングします
for param in signature.parameters.values():
try:
param_type = type_map.get(param.annotation, "string")
except KeyError as e:
# パラメーターの型アノテーションが不明な場合、エラーを発生させます
raise KeyError(
f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
)
parameters[param.name] = {"type": param_type}
# 必須パラメーターのリストを作成します (デフォルト値がないもの)
required = [
param.name
for param in signature.parameters.values()
if param.default == inspect._empty
]
# 関数のスキーマを辞書として返します
return {
"type": "function",
"function": {
"name": func.__name__,
"description": (func.__doc__ or "").strip(), # 関数の説明 (docstring) を取得します
"parameters": {
"type": "object",
"properties": parameters, # パラメーターの型
"required": required, # 必須パラメーターのリスト
},
},
}
たとえば、クイックスタートの translate_text 関数を考えてみましょう。
translation_tool = function_to_schema(translate_text)
print(json.dumps(translation_tool, indent=4, ensure_ascii=False))
translate_text 関数は自動的に次のように変換されます。
{
"type": "function",
"function": {
"name": "translate_text",
"description": "Translates text into the specified target language.\n This is a simple demonstration that uses a predefined translation.\n\n Parameters:\n text (str): The text to translate.\n target_language (str): The target language code (for example, 'zh', 'es', or 'ja').\n\n Returns:\n str: The translated text or an error message.",
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string"
},
"target_language": {
"type": "string"
}
},
"required": [
"text",
"target_language"
]
}
}
}
これで、関数の説明をモデルに渡すことができます。
assistant = Assistants.create(
model='qwen-plus',
name='Translation Agent',
description='An agent that can translate text between different languages',
instructions='You are a translation agent. When a user requests a translation, use the translate_text function to help them.',
tools=[translation_tool]
)
ストリーミング出力の使用
ストリーミング出力を使用する場合、ステップ 5：関数呼び出しを処理し、結果を返すのコードロジックを変更する必要があります。これは、Runs オブジェクトがアシスタントイベントストリームを返すようになったためです。
アシスタントが関数を呼び出すことを決定すると、Runs オブジェクトは thread.run.requires_action イベントと、大規模言語モデル (LLM) によって提供される入力パラメーター data.required_action.submit_tool_outputs.tool_calls を返します。この時点で関数出力を送信する必要があります。
run = Runs.submit_tool_outputs で関数出力を送信する際にも、ストリーミング出力を有効にする必要があることに注意してください。
# このコードはデモンストレーション専用です。ロジックを完全に理解した上で、ご自身のプロジェクトに統合してください。
# assistant、thread、message オブジェクトは作成済みであると仮定します。
# ツール関数のマッピングを定義します
tools_map = {
"translate_text": translate_text, # 翻訳関数
}
run = Runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
stream=True # ストリーミング出力を有効にします
)
while True: # 外側のループを追加します
for event, data in run: # イベントストリームとイベントデータの詳細については、アシスタント API のストリーミング出力に関するドキュメントをご参照ください。
if event == 'thread.run.requires_action': # アシスタントがツールを呼び出し、関数出力を待機しています。
tool_outputs = [] # 出力を送信するメソッドは、ステップ 5 のものと似ています。
for tool in data.required_action.submit_tool_outputs.tool_calls:
name = tool.function.name
args = json.loads(tool.function.arguments)
output = tools_map[name](**args)
tool_outputs.append({
"tool_call_id": tool.id,
"output": output,
})
run = Runs.submit_tool_outputs( # 関数出力を送信します
thread_id=thread.id,
run_id=data.id,
tool_outputs=tool_outputs,
stream=True # ここでもストリーミング出力を有効にする必要があります。
)
break # 現在の for ループを中断します。次のループで新しい Runs オブジェクトをポーリングします。
else:
break # 最初の for ループが関数呼び出しをトリガーせずに正常に終了した場合、while ループを中断します。
イベントストリームを処理する for ループの外側に追加の while ループがあることにお気づきかもしれません。これは、関数出力を送信すると、システムが新しい Runs オブジェクトを生成するためです。while ループは、最新のイベントストリームを自動的に追跡するのに役立ちます。これにより、アシスタントは関数呼び出しの結果を受け取った後も応答の生成を続けることができます。