SageMakerで顧客リストの自動スコアリング

SageMakerとは

JupyterLabについて

AWS SageMaker でも標準的にサポートされており、開発にあたってとても助かりました。

開発の第一歩としてのシンプルなアプローチ

import datetime

import pandas as pd
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import OneHotEncoder, StandardScaler

# CSVデータの読み込み
df = pd.read_csv('dummy_leads_v2.csv')

# Convert `前回対応日時` to datetime
df['前回対応日時'] = pd.to_datetime(df['前回対応日時'])

# 基準となる日時を設定
base_datetime = datetime.datetime(2023, 1, 1)

# `前回対応日時` から基準となる日時を引いた秒数を計算
df['前回対応日時_seconds'] = (df['前回対応日時'] - base_datetime).dt.total_seconds()

# Initialize OneHotEncoder
encoder = OneHotEncoder(handle_unknown='ignore', sparse_output=False)

# Fit and transform the `前回対応内容` column
encoded_features = encoder.fit_transform(df[['前回対応内容']])

# Create a new DataFrame with the encoded features
encoded_df = pd.DataFrame(encoded_features)
df = pd.concat([df, encoded_df], axis=1)

# --- ここで列名をstr型に変換 ---
df.columns = df.columns.astype(str)

# 例：線形回帰モデルを使用する場合
model = LinearRegression()

# データの前処理
scaler = StandardScaler()

# 特徴量を指定
scaled_features = scaler.fit_transform(df[['対応回数', '前回対応日時_seconds', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11', '12']])

# --- ここでモデルを学習させる ---
model.fit(scaled_features, df['スコア'])  # 'スコア' を目的変数として学習

# スコアの算出
scores = model.predict(scaled_features)

# スコアをデータフレームに追加
df['score'] = scores

# CSV ファイルに書き込む
df.to_csv('output.csv', index=False)

# 最初の5行を表示
print(df.head().to_markdown(index=False, numalign="left", stralign="left"))

出力結果は画像のようになりました。

「スコア」を学習させただけあって、もともと持っていた情報に近い値が出力されました。

SageMaker JumpStartを使ってAIモデルの自動スコアリングに挑戦

Amazon自身が提供しているモデルや、Meta,Deepseek,nvidiaが提供しているモデルなど600件ほどモデルが選択できました。

モデルの選定

Meta Llama Prompt Guard 86Mを選定

import datetime
import pandas as pd
import boto3
import json
import ast
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.linear_model import LinearRegression

# CSVデータの読み込み
df = pd.read_csv('dummy_leads_v2.csv')

# Convert `前回対応日時` to datetime
df['前回対応日時'] = pd.to_datetime(df['前回対応日時'])

# --- ここで列名をstr型に変換 ---
df.columns = df.columns.astype(str)

# SageMakerエンドポイント名
endpoint_name = 'my-endopoint-dummy1'  # エンドポイント名に置き換えてください

# SageMakerランタイムクライアントの作成
runtime = boto3.client('sagemaker-runtime')

# スコアリング関数
def llama_score(text):
    """
    Llamaモデルでテキストをスコアリングする関数
    """
    payload = json.dumps({"inputs": text})  # 入力テキストを JSON 形式に変換

    response = runtime.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Body=payload
    )

    result = json.loads(response["Body"].read().decode())  # レスポンスを JSON から読み込む

    # スコアの抽出 (モデルの出力形式に合わせて調整が必要)
    # Llamaモデルの出力形式に合わせて、スコアを抽出する処理を実装
    # 例: 出力が確率分布の場合、特定のカテゴリの確率をスコアとする
    # 例: 出力が数値の場合、そのままスコアとする
    # ここでは、例として最初の要素をスコアと仮定
    score = result[0]  # モデルの出力構造に合わせて変更

    return score

# スコア算出用の特徴量を作成 (例: 前回対応内容)
df['llama_input'] = df['前回対応内容']

# Llamaモデルでスコアリング
df['score'] = df['llama_input'].apply(llama_score)


# Llamaモデルの出力形式に合わせて、スコアを抽出する処理を実装
# ラベルとスコア抽出関数

def extract_label_and_score(result):
    try:
        # 文字列を辞書に変換
        if isinstance(result, str):
            result = ast.literal_eval(result)  # 安全に辞書化

        # 辞書の場合、ラベルとスコア抽出
        if isinstance(result, dict):
            label = result.get('label', 'UNKNOWN')
            score = float(result.get('score', 0)) * 100  # パーセント表示
            return label, score
    except Exception as e:
        print(f"Error parsing result: {e}")  # デバッグ用
    return "UNKNOWN", 0.0  # 失敗時のデフォルト



# スコアを抽出してデータフレームに追加
df[['label', 'risk_score']] = df['score'].apply(
    lambda x: pd.Series(extract_label_and_score(x))
)


# CSV ファイルに書き込む
df.to_csv('output2.csv', index=False)


print("ok")

左の画像のような結果となった

簡単にラベル部分の解説をすると、下記のようになる
BENIGN：正常
INJECTION：不正指示
JAILBREAK：回避

もともと、AIへの指示文言の評価に使用されるモデルのため、このような形になってます。

Meta Llama 2 7B Chat で自然言語からスコアを生成する

import datetime
import pandas as pd
import boto3
import json
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.linear_model import LinearRegression
import re

# CSVデータの読み込み
df = pd.read_csv('dummy_leads_v2.csv')

# Convert `前回対応日時` to datetime
df['前回対応日時'] = pd.to_datetime(df['前回対応日時'])


# --- ここで列名をstr型に変換 ---
df.columns = df.columns.astype(str)

# SageMakerエンドポイント名
endpoint_name = 'my-endpoint-dummy2'  # エンドポイント名に置き換えてください

# SageMakerランタイムクライアントの作成
runtime = boto3.client('sagemaker-runtime')

# スコアリング関数
def llama_score(payload):
    """
    Llamaモデルのエンドポイントを呼び出す関数
    """
    response = runtime.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Body=payload
    )
    result = json.loads(response["Body"].read().decode())
    return result

# プロンプトの定義
prompt = """以下の営業リード情報を0〜100点で評価してください。禁止事項が含まれる場合は-100点。

- 対応内容: {前回対応内容}
- 対応回数: {対応回数}
- 会社名: {会社名}
- 部署: {部署}
- 役職: {役職}

スコア:"""

# Llamaモデルでスコアリング
def process_row(row):
    text = prompt.format(
        前回対応内容=row['前回対応内容'],
        対応回数=row['対応回数'],
        部署=row['部署'],
        役職=row['役職']
    )
    payload = json.dumps({"inputs": text})
    result = llama_score(payload)
    return result

df['llama_output'] = df.apply(process_row, axis=1)


# Llamaモデルの出力形式に合わせて、スコアを抽出する処理を実装
# ラベルとスコア抽出関数

def extract_numeric_score(result):
    """
    'generated_text' から 0～100 の数字だけを抽出する関数
    """
    try:
        text = result.get('generated_text', '')
        # 最初に出てくる数字（1～3桁の整数）
        match = re.search(r'[0-9]{1,3}', text)
        if match:
            score = int(match.group())
            if 0 <= score <= 100:
                return score
        return 0  # 見つからない or 範囲外なら0
    except Exception as e:
        print(f"Error extracting score: {e}")
        return 0
        
# DataFrameに適用
df['score'] = df['llama_output'].apply(lambda x: extract_numeric_score(x))



# CSV ファイルに書き込む
df.to_csv('output3.csv', index=False)


print("ok")

概ね問題なくスコア生成することができました。

うまくプロンプトを理解できていないのか、点数はついていても、点数のコメント部分がプロンプトを英訳しただけの内容になってしまっているので、プロンプトの調整やモデルのトレーニング、もしくはモデルの再選定が必要そうでした。

まとめ