例: コンテナモデルアダプタの実装

以下は、コンテナを利用したモデルのために定義された例のモデルアダプタです。使用されるイメージは、/mirror エンドポイントでリッスンしているシンプルな flask サーバーであり、"text" を唯一のフィールドとしてリクエストオブジェクトを受け取ることを想定しています。このモデルアダプタは、レスポンスオブジェクトの "returnedText" フィールドでそのテキストをそのまま返します。

container_context オブジェクトの完全な定義は、API: ModelAdapter リファレンスドキュメントで確認できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import requests
import json
import pandas as pd

import palantir_models as pm


class ExampleIdentityFunctionModelAdapter(pm.ContainerModelAdapter):
    """
        :display-name: Example Identity Function Model Adapter
        :description: コンテナバックのモデルのモデルアダプターの参照例
    """
    def __init__(self, shared_volume_path, model_host_and_port):
        self.shared_volume_path = shared_volume_path  # 共有ボリュームパスを設定
        self.model_host_and_port = model_host_and_port  # モデルのホストとポートを設定

    @classmethod
    def init_container(cls, container_context):
        shared_volume_path = container_context.shared_empty_dir_mount_path
        # 注意：このアダプタは1つのコンテナ名と提供されたサービスURIを期待しています。
        model_host_and_port = list(container_context.services.values())[0][0]
        return cls(shared_volume_path, model_host_and_port)

    @classmethod
    def api(cls):
        inputs = {"input_df": pm.Pandas(columns=[("text", str)])}  # 入力はテキストのデータフレーム
        outputs = {"output_df": pm.Pandas(columns=[("text", str), ("returnedText", str)])}  # 出力もテキストのデータフレーム
        return inputs, outputs

    def predict(self, input_df):
        def run_inference_on_row(row):
            request = {"text": row.text}  # テキストをリクエストとして送信
            response = requests.post("http://" + self.model_host_and_port + "/mirror", json=request)  # リクエストをPOST
            json_res = json.loads(response.content.decode("utf-8"))  # レスポンスをJSON形式でデコード
            return (row.text, json_res["returnedText"])  # 元のテキストと返されたテキストを返す

        results = [run_inference_on_row(row) for row in input_df.itertuples()]  # 入力データフレームの各行で推論を実行
        columns = ["text", "returnedText"]
        return pd.DataFrame(results, columns=columns)  # 結果をデータフレームとして返す