Alibaba Cloud Model Studio:関数の呼び出し

はじめに

この例では、翻訳アシスタントと、アシスタントが使用できる関数 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"Translation not found. In a real production environment, a translation service would be called here.")
    except Exception as e:
        return f"Translation failed: {str(e)}"

説明：

• 翻訳方法：定義済みの辞書を使用して、複数の言語をサポートする翻訳機能をシミュレートします。

• エラー処理：基本的なエラー処理が含まれており、さまざまな条件下で関数が適切な応答を返すことを保証します。

次に、Assistant 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": "指定されたターゲット言語にテキストを翻訳します",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "翻訳するテキスト"
                },
                "target_language": {
                    "type": "string",
                    "description": "ターゲット言語コード (例: 'zh'、'es'、'ja')"
                }
            },
            "required": ["text", "target_language"]
        }
    }
}

説明：

• name：関数の名前は translate_text で、アシスタントは呼び出し時にこれを使用します。

• description：関数の説明です。アシスタントが関数の使用方法を理解するのに役立ちます。

• parameters：関数の パラメーター。元のテキストとターゲット言語を含みます。

ステップ 3：アシスタントを作成する

ここで、定義した翻訳ツールを使用できるアシスタントを作成します。

# アシスタントを作成する
assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Assistant',
    description='異なる言語間でテキストを翻訳できるアシスタント',
    instructions='あなたは翻訳アシスタントです。ユーザーが翻訳をリクエストした場合は、translate_text 関数を使用してください。',
    tools=[translation_tool]
)

説明：

• model：アシスタントが使用するモデル。この例では、言語理解と タスク実行 をサポートする qwen-plus を使用しています。

• name：アシスタントの名前 Translation Assistant です。

• description：アシスタントの機能であるテキスト翻訳について説明します。

• tools：定義した翻訳ツールを統合し、アシスタントがこのツールを使用できるようにします。

ステップ 4：スレッドを作成し、アシスタントと対話する

新しいスレッドを開始し、ユーザーメッセージを追加します。次に、アシスタントを実行してユーザーのクエリを処理します。

# 新しいスレッドを作成する
thread = Threads.create()

# スレッドにユーザーメッセージを追加する
Messages.create(
    thread_id=thread.id, 
    role="user", 
    content="「Hello world」を中国語に翻訳してください。"
)

# アシスタントを実行する
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}")

まとめ

これらの手順に従うことで、翻訳リクエストを処理し、翻訳関数を使用できるアシスタントを構築できました。Assistant API は、複雑でタスク指向のアシスタントの作成を簡素化します。

ツールを追加したり、手順を変更したりすることで、要件に基づいてアシスタントの機能を強化できます。

関数の説明をすばやく生成する

上記の例のステップ 2では、アシスタントのために 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"関数 {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"パラメーター {param.name} の不明な型注釈 {param.annotation}: {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": "指定されたターゲット言語にテキストを翻訳します。\n    これは、定義済みの翻訳を使用した簡単なデモです。\n\n    パラメーター:\n        text (str): 翻訳するテキスト\n        target_language (str): ターゲット言語コード (例: 'zh'、'es'、'ja')\n\n    戻り値:\n        str: 翻訳されたテキスト、またはエラーメッセージ",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string"
                },
                "target_language": {
                    "type": "string"
                }
            },
            "required": [
                "text",
                "target_language"
            ]
        }
    }
}

関数の説明をモデルに次のように渡します。

assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Assistant',
    description='異なる言語間でテキストを翻訳できるアシスタント',
    instructions='あなたは翻訳アシスタントです。ユーザーが翻訳をリクエストした場合は、translate_text 関数を使用してください。',
    tools=[translation_tool]
)

ストリーミング出力を使用する

ストリーミング出力モードでは、Runs がアシスタント イベント ストリームを返すようになったため、ステップ 5：関数の呼び出しを処理し、結果を返すのロジックを変更する必要があります。

アシスタントが呼び出す関数を決定すると、Runs は イベント thread.run.requires_action と、モデルによって提供される入力 パラメーター data.required_action.submit_tool_outputs.tool_calls を返します。この時点で関数の出力を指定する必要があります。

関数出力 run = Runs.submit_tool_outputs を送信するときにストリームを有効にする必要があることに注意してください。

# このコードはデモ用です。ロジックを完全に理解してから、プロジェクトに組み込んでください。
# assistant、thread、および message オブジェクトがすでに作成されていると仮定します

run = Runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        stream=True  # ストリームを有効にする
    )
while True:  # 他のループを追加する
    for event, data in run:  # イベント ストリームと イベント データ
        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 ループは、最新のイベント ストリームを自動的に追跡するのに役立ち、アシスタントは関数呼び出しの結果を取得した後に適切な応答を生成し続けることができます。