【UiPath】「アレやっといて！」でロボットを動かしてみる (Assistants API編)

# はじめに
- 本記事ではAzure OpenAI Serviceにてプレビューリリースされた [Assistants API](https://learn.microsoft.com/ja-jp/azure/ai-services/openai/concepts/assistants) を使用して、自然言語によるUnattended Robot実行の実装方法について説明します。
- この実装により標題通り「○○やっといて！」というユーザープロンプトをトリガーとしてロボットによる自動化処理を実行することできます。しかも **プロンプトの文脈に応じてパラメーターを自動抽出して引数を渡すこともできます！**
- Unattended Robotの実行には [APIトリガー](https://docs.uipath.com/ja/orchestrator/automation-cloud/latest/user-guide/managing-api-triggers) を使用して実行用エンドポイントを作成します。[Assistants APIのFunction Calling](https://learn.microsoft.com/ja-jp/azure/ai-services/openai/how-to/assistant-functions?tabs=python) によってUnattended Robot実行のための関数(今回はPythonで実装)と引数を抽出して実行します。

# 処理の流れ
- 今回使用するサービスの一覧です。
    - Azure OpenAI Service
    - UiPath Automation Cloud (Community版でもOK)
    - UiPath Unattended Robot (以降URと略す)
    - Google Colaboratory (以降Colabと略す)

- オブジェクトごとの処理の流れです。青色矢印がメッセージの流れ、赤色矢印がファンクション実行の流れです。

![00a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/98d435f5-5376-8d4d-3062-00909cfd25ab.png)

# デモ動画
- 完成形はこのように動作します。

<iframe width="600" height="440" src="https://www.youtube.com/embed/9wl5s9PN4sQ?si=7A9yW9bbsK8sXd4M" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" allowfullscreen></iframe>

# 実装手順
- 実装手順の大まかな流れについて説明します。

1. **プロセス登録**: UiPath Studioにて自動化処理のワークフローを作成し、Automation Cloud上のOrchestratorにパブリッシュしてプロセスとして登録します。
2. **UR設定**: Automation Cloudにてプロセスの実行環境であるURの設定を行います。
3. **APIトリガー設定**: Automation CloudにてURジョブ実行を容易に行うためにAPIトリガーを設定します。
4. **Assistant作成**: Google ColabにてPythonコードでAssistants APIを実行し、Assistantを作成します。
5. **Python関数作成**: AssistantからのFunction Callingにて実行されるPython関数を作成します。この関数がコールされるとAPIトリガーによってURジョブが実行されるように設定しておきます。
6. **Assistantスレッド作成してプロンプト実行**: Assistants APIにてAssistantスレッドを作成します。この時、ユーザーからのプロンプトを受け付けてその内容に応じてPython関数をコールしてURジョブを実行し、結果を出力します。

- 実装手順の詳細について説明します。

## 1. プロセス登録
- URジョブとして実行するプロセスを登録します。例として下記のようなワークフローをUiPath Studioで作成してAutomation Cloud上のOrchestratorにパブリッシュします。
    - **LogTest**: `message` 入力引数をログ出力するだけのワークフロー
    ![01.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/4d336b5a-7d36-4985-5100-ac77c3a6eef3.jpeg)

    - **RPA_Challenge01**: `browser` 入力引数で指定されたブラウザー(ChromeまたはEdge)で [RPAチャレンジ](https://www.rpachallenge.com/?lang=ja) を実行して結果を出力引数として返すワークフロー
    ![16.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/43ff474e-6fd0-86e9-3e33-c453d4f1c721.jpeg)

- [Automation Cloud](https://cloud.uipath.com) にログインし、Orchestratorのフォルダーに移動します。
- オートメーション > プロセスにて、先ほどパブリッシュしたパッケージをプロセスとして登録します。
![02.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/3faa7236-1c30-b4bc-8f72-24e6e1bc3fe4.jpeg)

## 2. UR設定
- マシンテンプレートを作成し、Production(Unattended)ランタイムを1にセットします。生成されたクライアントIDとクライアントシークレットをメモします。
    ![03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/f40d3411-daad-e523-e692-1d27526e2b38.png)

- Windowsマシンにて [UiPath Studioインストーラー](https://download.uipath.com/UiPathStudio.msi) をダウンロードしてUnattended Robotとしてインストールします。先ほどメモしたクライアントIDとクライアントシークレットを使用してOrchestratorに接続します。
    ![04.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/f96e1213-7021-1ac6-9bbd-b5c7dd5f77c3.jpeg)

- ロボットアカウントを作成し、プロセスを登録したフォルダーに割り当てます。マシンテンプレートも同じフォルダーに割り当てます。
- 手動でURジョブ実行できることを確認します。

## 3. APIトリガー設定
- それぞれのプロセスを容易に実行するためにAPIトリガーを設定します。詳細な手順は下記の記事をご参照ください。

https://qiita.com/hidecha/items/5b0733639dcd1420531a

- 出力引数を受け取るために呼び出しモードは **同期 (ロングポーリング)** を使用します。

    ![05.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/e658314e-dbaa-f552-e08b-4896d4bc2954.jpeg)

    :::note warn
    長時間実行されるURジョブはPythonコードからの呼び出し時にタイムアウトになる可能性があります。**非同期ポーリング** によってタイムアウトは回避できますが出力引数を受け取るためには別途APIコールする必要があります。
:::

## 4. Assistant作成
- Azure OpenAI Serviceのインスタンスをデプロイします。Assistants APIを使用するには2024年2月現在では 米国東部2、オーストラリア東部またはスウェーデン中部リージョンでのみ利用できます。今回はスウェーデン中部(`Sweden Central`)でデプロイします。
    ![06.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/b989d94b-a65d-1945-67a6-bda58b8eb9df.jpeg)

- デプロイが完了しましたら キーとエンドポイントにてキー1をコピーしてメモします。
    ![07.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/a43b62c5-8532-46ef-5ebf-c1bc2c2abb4b.jpeg)

- [Azure AI Studio](https://oai.azure.com/portal)にてGPT-4モデルをデプロイします。デプロイ名は `gpt-4` とします。
    ![08.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/6f23a046-2575-dcef-a3df-456099088473.jpeg)

- **アシスタント(プレビュー)** にてアシスタントプレイグラウンドが表示されることを確認します。
    ![09.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/715d7892-f6c4-b683-b6be-05b9366f9a6b.jpeg)

- Google ColabにてPythonコードを順次実行します。(これらのコードは [Assistants function calling with Bing Search](https://github.com/Azure-Samples/azureai-samples/tree/3b8b49cce137c0f255d8a622d8ccf53e7b8c19d5/scenarios/Assistants/function_calling) を参考にしております)
    - まず必要なライブラリーをインストールします。`!pip install openai requests typing`
    - openaiのライブラリ―バージョンは1.x系が必要となります。古いバージョンを使用している場合には `!pip install openai --upgrade` を実行します。

    ![10.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/130d01b5-ed62-9526-93f3-b029b2a2341b.jpeg)

    - 次のコマンドを実行します。Azure OpenAIリソース名とAPIキーは適宜変更します。
    ```Python
    import os
    
    # 環境に応じて変更
    os.environ["AZURE_OPENAI_ENDPOINT"] = "https://<Azure OpenAI リソース名>.openai.azure.com/"
    os.environ["AZURE_OPENAI_KEY"] = "<Azure OpenAI APIキー>"
    os.environ["AZURE_OPENAI_DEPLOYMENT"] = "gpt-4"
    ```

- Azure OpenAI Serviceをclientとして定義します。
    ```Python
    from openai import AzureOpenAI
    
    client = AzureOpenAI(
        api_key=os.getenv("AZURE_OPENAI_KEY"),
        api_version="2024-02-15-preview",
        azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
    )
    ```

- アシスタントを作成します。アシスタント名、指示を指定します。ツールとしてFuncation Callingで実行する関数と引数を定義します。
    - 関数名: `executeUR`
    - パラメーター:
        - `process`: Automation Cloudでこのあと登録するプロセス名を `enum` で列挙します。(`LogTest` や `RPA-Challenge01` は適宜変更します)
        - `input`: プロセスに渡す引数を定義します。

    ```Python
    # アシスタントの作成
    assistant = client.beta.assistants.create(
        name="Automation-Assistant", # アシスタント名
        instructions="あなたはAutomation Cloudアシスタントです。ユーザーからの指示に従って関数を実行し、結果を答えてください。", # 指示 (アシスタントの定義や目的など)
        model=os.getenv("AZURE_OPENAI_DEPLOYMENT"), # モデルのデプロイ名
    
        # アシスタントが使用できるツールのリスト (ここではFunction callingで1つの外部APIを指定)
        tools=[{
            "type": "function",
            "function": {
                "name": "executeUR",
                "description": "Execute Unattended Robot based on Process name",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "process": {"type": "string", "enum": [
                            "LogTest",
                            "RPA-Challenge01"
                        ]}, # プロセス名を列挙
                        "input": {"type": "string", "description": "Input parameter"}   # 引数
                    },
                    "required": ["process", "input"] # 必須フィールド
                }
            }
        }]
    )
    ```

- アシスタントの作成を確認します。
    ```Python
    # 確認
    for k,v in dict(assistant).items():
        print(k + ": ", v)
    ```
    ![11.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/65bbc2a0-4488-4877-a32a-3c42bac83350.jpeg)

- 先ほど出力された `id` を環境変数にセットします。
- Automation Cloudの個人用アクセストークン(PAT)を環境変数にセットします。PATの取得方法は [こちらの記事](https://qiita.com/hidecha/items/609dc6723dffb3bd81c6) をご覧ください。スコープは `OR.Jobs` を指定します。

    ```Python
    import time
    import json
    import requests
    from typing import Optional
    
    # 環境に応じて変更
    os.environ["AZURE_OPENAI_ASSISTANT"] = "<アシスタントID>"
    os.environ["UIPATH_PAT"] = "<個人用アクセストークン>"
    ```

- なお作成されたアシスタントは [Azure AI Studio](https://oai.azure.com/portal) でも確認できます。
    ![17.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/dafd3466-2e54-162a-ca7b-ffaa257af698.jpeg)

## 5. Python関数作成
- 次にFunction Callingによって実行される関数 `executeUR` を定義します。
    - この関数はプロセス名と引数を受け取り、プロセスごとに定義したAPIトリガーにてURジョブを実行します。(APIトリガーのエンドポイントは適宜変更します)
    - 入力引数名もプロセスごとに適宜変更します。この例では LogTestプロセスは `message`、RPA-Challenge01プロセスは `browser` という入力引数を取ります。
    - APIトリガーのステータスコードは呼び出しモードが同期(ロングポーリング)の時は200(OK)、非同期の時は202(Accepted)となります。
    - 同期呼び出しによるURジョブ正常実行時には出力引数が取得できますので、この関数の戻り値として返します。

    ```Python
    # 外部API
    def executeUR(process, input=''):
        print('>Process: ', process)
        print('>Input: ', input)
    
        # Personal Access Token
        auth = 'Bearer ' + os.getenv("UIPATH_PAT")
    
        # Header
        headers = {
            'Authorization': auth
        }
    
        base_url = ''
        params = {}
        # ProcessごとのAPI Triggerエンドポイント
        if process == 'LogTest':
            base_url = '<LogTestプロセスを実行するAPIトリガーのエンドポイント>'
            params = {
                'message': input
            }
        elif process == 'RPA-Challenge01':
            base_url = '<RPA-Challenge01プロセスを実行するAPIトリガーのエンドポイント>'
            params = {
                'browser': input
            }
        else:
            print(f"Error: Process {process} is not defined!")
            return None
    
        # APIリクエストを送信
        response = requests.get(base_url, headers=headers, params=params)
    
        if response.status_code == 200 or response.status_code == 202:
            # レスポンスが成功した場合、データを解析
            data = response.json()
            result = data['output']
            return result
        else:
            # エラーが発生した場合、ステータスコードを出力
            print(f"Error: Unable to execute Unattended Robot. Status code {response.status_code}")
            return None
    ```

- 関数呼び出しでアシスタントに必要な3つの関数を定義します。これらすべての関数は最後のプロンプト実行時に使用され、URジョブ実行の状態監視やメッセージ・パラメーターの受け渡しなどを行います。

    ```Python
    def poll_run_till_completion(
        client: AzureOpenAI,
        thread_id: str,
        run_id: str,
        available_functions: dict,
        verbose: bool,
        max_steps: int = 10,
        wait: int = 3,
    ) -> None:
        """
        Poll a run until it is completed or failed or exceeds a certain number of iterations (MAX_STEPS)
        with a preset wait in between polls
    
        @param client: OpenAI client
        @param thread_id: Thread ID
        @param run_id: Run ID
        @param assistant_id: Assistant ID
        @param verbose: Print verbose output
        @param max_steps: Maximum number of steps to poll
        @param wait: Wait time in seconds between polls
    
        """
    
        if (client is None and thread_id is None) or run_id is None:
            print("Client, Thread ID and Run ID are required.")
            return
        try:
            cnt = 0
            while cnt < max_steps:
                run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
                if verbose:
                    print("Poll {}: {}".format(cnt, run.status))
                cnt += 1
                if run.status == "requires_action":
                    tool_responses = []
                    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
    
                        for call in tool_calls:
                            if call.type == "function":
                                if call.function.name not in available_functions:
                                    raise Exception("Function requested by the model does not exist")
                                function_to_call = available_functions[call.function.name]
                                tool_response = function_to_call(**json.loads(call.function.arguments))
                                tool_responses.append({"tool_call_id": call.id, "output": tool_response})
    
                    run = client.beta.threads.runs.submit_tool_outputs(
                        thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
                    )
                if run.status == "failed":
                    print("Run failed.")
                    break
                if run.status == "completed":
                    break
                time.sleep(wait)
    
        except Exception as e:
            print(e)
    
    def create_message(
        client: AzureOpenAI,
        thread_id: str,
        role: str = "",
        content: str = "",
        file_ids: Optional[list] = None,
        metadata: Optional[dict] = None,
        message_id: Optional[str] = None,
    ) -> any:
        """
        Create a message in a thread using the client.
    
        @param client: OpenAI client
        @param thread_id: Thread ID
        @param role: Message role (user or assistant)
        @param content: Message content
        @param file_ids: Message file IDs
        @param metadata: Message metadata
        @param message_id: Message ID
        @return: Message object
    
        """
        if metadata is None:
            metadata = {}
        if file_ids is None:
            file_ids = []
    
        if client is None:
            print("Client parameter is required.")
            return None
    
        if thread_id is None:
            print("Thread ID is required.")
            return None
    
        try:
            if message_id is not None:
                return client.beta.threads.messages.retrieve(thread_id=thread_id, message_id=message_id)
    
            if file_ids is not None and len(file_ids) > 0 and metadata is not None and len(metadata) > 0:
                return client.beta.threads.messages.create(
                    thread_id=thread_id, role=role, content=content, file_ids=file_ids, metadata=metadata
                )
    
            if file_ids is not None and len(file_ids) > 0:
                return client.beta.threads.messages.create(
                    thread_id=thread_id, role=role, content=content, file_ids=file_ids
                )
    
            if metadata is not None and len(metadata) > 0:
                return client.beta.threads.messages.create(
                    thread_id=thread_id, role=role, content=content, metadata=metadata
                )
    
            return client.beta.threads.messages.create(thread_id=thread_id, role=role, content=content)
    
        except Exception as e:
            print(e)
            return None
    
    def retrieve_and_print_messages(
        client: AzureOpenAI, thread_id: str, verbose: bool, out_dir: Optional[str] = None
    ) -> any:
        """
        Retrieve a list of messages in a thread and print it out with the query and response
    
        @param client: OpenAI client
        @param thread_id: Thread ID
        @param verbose: Print verbose output
        @param out_dir: Output directory to save images
        @return: Messages object
    
        """
    
        if client is None and thread_id is None:
            print("Client and Thread ID are required.")
            return None
        try:
            messages = client.beta.threads.messages.list(thread_id=thread_id)
            display_role = {"user": "User query", "assistant": "Assistant response"}
    
            prev_role = None
    
            if verbose:
                print("\n\nCONVERSATION:")
            for md in reversed(messages.data):
                if prev_role == "assistant" and md.role == "user" and verbose:
                    print("------ \n")
    
                for mc in md.content:
                    # Check if valid text field is present in the mc object
                    if mc.type == "text":
                        txt_val = mc.text.value
                    # Check if valid image field is present in the mc object
                    elif mc.type == "image_file":
                        image_data = client.files.content(mc.image_file.file_id)
                        if out_dir is not None:
                            out_dir_path = Path(out_dir)
                            if out_dir_path.exists():
                                image_path = out_dir_path / (mc.image_file.file_id + ".png")
                                with image_path.open("wb") as f:
                                    f.write(image_data.read())
    
                    if verbose:
                        if prev_role == md.role:
                            print(txt_val)
                        else:
                            print("{}:\n{}".format(display_role[md.role], txt_val))
                prev_role = md.role
            return messages
        except Exception as e:
            print(e)
            return None
    ```

## 6. Assistantスレッド作成してURジョブ実行
- Assistantスレッドを作成します。この時、ユーザーからのプロンプトを受け付けてその内容に応じてPython関数をコールしてURジョブを実行します。

    ```Python:
    # Main
    prompt = input('Enter your request: ')
    message = {"role": "user", "content": prompt}
    
    available_functions = {"executeUR": executeUR}
    verbose_output = True
    
    thread = client.beta.threads.create()
    create_message(client, thread.id, message["role"], message["content"])
    
    run = client.beta.threads.runs.create(thread_id=thread.id, assistant_id=os.getenv("AZURE_OPENAI_ASSISTANT"))
    
    poll_run_till_completion(
        client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=verbose_output
    )
    messages = retrieve_and_print_messages(client=client, thread_id=thread.id, verbose=verbose_output)
    ```

- プロンプトにて `「テストですよ！」とログ出力しといて` と入力してみます。おおざっぱな指示ですが、アシスタントがパラメーターを自動抽出して関数を `executeUR({"process":"LogTest","input":"テストですよ！"})` と実行してくれます。LogTestプロセスに渡すmessage引数を使ってURジョブ実行されます。
    ![12.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/83cb6fbc-9497-ab68-d64f-306a5d66d989.jpeg)
    - またOrchestratorのURジョブ実行ログにて正しく引数が渡されていることを確認できます。
    ![15.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/4df1f2af-b2f2-29f1-476f-ee63584f9020.jpeg)

- もう一度Assistantスレッド作成のコードを実行します。今度はプロンプトにて `RPAチャレンジをEdgeで実行して結果を表示して！` と入力してみます。今回もアシスタントがパラメーターを自動抽出して関数を `executeUR({"process":"RPA-Challenge01","input":"Edge"})` と実行してくれます。さらに実行結果としての出力引数(正答率や実行時間)も表示してくれます。全体の挙動は [デモ動画](#デモ動画) の通りです。
    ![14.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/c1f9a9ca-dd44-b865-b13d-f80b1b3112dd.jpeg)

- `process` パラメーターはAssistant作成時に `enum` で取り得る値を列挙したため、その中から最も相応しいものが選択されます。**いずれにも合わないような雑すぎる指示にはアシスタントがちゃんとエラーを返してくれます。**
    ![13.PNG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/178312/664be4ce-0064-42d2-431a-197230982778.png)

# 参考記事 
- [Azure OpenAI Service の Assistants API を理解する](https://zenn.dev/microsoft/articles/533488db14e442)
- [How Assistants work](https://platform.openai.com/docs/assistants/how-it-works)

# おわりに
今回はAzure OpenAI ServiceにてAssistants APIを利用してURジョブを実行してみました。これまでもFunction Callingは実装可能でしたが、Assistants APIによって関数の定義とThreadおよびRunによる状態管理が分かりやすくなったと思います。少々あいまいなプロンプトでも適切にパラメーターを自動抽出して関数を実行できるのでとても便利です。アイディア次第でさまざまな自動化に利用できますので是非試してみてください！