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
40
41
42
43
44
45
46
47
48
49
50
51
52
import palantir_models as pm

import json
import pandas as pd

# モデルアダプタークラスの定義
class ExampleModelAdapter(pm.ExternalModelAdapter):

    def __init__(self, url, credentials_map, configuration_map):
        # "接続設定"マップからモデル設定を抽出
        model_name = configuration_map['model_name']
        model_parameter = configuration_map['model_parameter']

        # "資格証明設定"マップからモデルの資格情報を抽出
        secret_key = credentials_map['secret_key']

        # モデルのロード時にhttpクライアントを初期化
        self.client = ExampleClient(url, model_name, model_parameter, secret_key)

    # メソッド：外部モデルとの接続を初期化
    @classmethod
    def init_external(cls, external_context: pm.ExternalContext) -> "pm.ExternalModelAdapter":
        return cls(
            url=external_context.base_url,
            credentials_map=external_context.resolved_credentials,
            configuration_map=external_context.connection_config,
        )

    # メソッド：APIの入力と出力を定義
    @classmethod
    def api(cls):
        inputs = {"df_in": pm.Pandas()} # 入力はPandasデータフレーム
        outputs = {"df_out": pm.Pandas()} # 出力もPandasデータフレーム
        return inputs, outputs

    # メソッド：予測を行う
    def predict(self, df_in):
        # ペイロードを作成
        payload = {
            "instances": df_in.apply(lambda row: {"features": row.tolist()}, axis=1).tolist()
        }

        # クライアントは例であり、外部モデルに接続するために編集する必要があります
        response = self.client.predict(
            ContentType="application/json",
            Body=json.dumps(payload)
        )

        # 応答から予測を取得
        result = response["Body"].read().decode()
        predictions = pd.DataFrame(json.loads(result)["predictions"])
        return predictions