OpenAI の埋め込みモデルを使って "ヒヤリ度" をスコア化する

はじめに

本記事では、前回の 「OpenAI の埋め込みモデルを使ってみる」 につづいて、「ヒヤリハット事例」の「類似検索」と「スコアリング」をやってみます。

特に「スコアリング」では 「重大インシデント」と「軽微なインシデント」の特徴を埋め込みベクトルで分析し、ヒヤリハット事例の重要度を 自動的に評価する 試みとなります。

※埋め込みモデルは text-embedding-3-small を使いました。
※検証した Notebook は Gist に上げました。

ヒヤリハット事例 の 類似検索

まず、テストデータを用意します。

以下の事例を ChatGPT で 10件ずつ(合計:40件) 作成しました。以下は一部抜粋です。

メール誤送信関連

incident	cause	action
別の部署に送るべき連絡が間違って他部署に届いてしまった	アドレスの自動補完機能を使い、宛先間違いに気づかなかった	送信前に宛先を確認する
添付ファイルを取り違えて送信してしまった	類似したファイル名の資料を選択してしまい誤送信が発生した	ファイル名や内容を送信前に確認し、送信後の追跡管理を強化する
..(他 8件)	..	..

電源に関する事例

incident	cause	action
サーバーの電源が突如切れ、データが破損した	過負荷状態が続き、電源装置が耐えられなくなった	サーバーに適切な電源容量を確保し、温度や負荷の監視を行う
会議中にプロジェクターの電源が入らずプレゼンが中断した	プロジェクターの電源ケーブルが接続されていなかった	会議前に設備の動作確認を行い、予備のケーブルを用意する
..(他 8件)	..	..

リリース関連

incident	cause	action
予定外のタイミングで機能がリリースされ、混乱を招いた	リリーススケジュールの管理が不十分で、意図しないタイミングで公開された	リリース管理ツールを導入し、スケジュールの確認プロセスを強化する
緊急リリースで構成ミスが発生し、システム障害につながった	十分なレビューが行われず、緊急対応で作業が急ぎすぎた	緊急時でもレビューや確認を必須化し、チェックリストを活用する
..(他 8件)	..	..

重大インシデント

(「ウィルス蔓延, スマホ紛失, 本番環境で障害発生, データ消失」など）

incident	cause	action
従業員がスマホやクレジットカードを紛失し、個人情報が流出した可能性が発生	紛失時に迅速な対応が取られず、不正利用のリスクが高まった	紛失対応マニュアルの整備と紛失防止用の追跡アプリを導入する
本番環境でシステム障害が発生し、サービスが数時間停止した	テスト不足により、リリースしたコードに重大なバグが含まれていた	リリース前にステージング環境での徹底的なテストを実施する
..(他 8件)	..	..

1. データ読み込み

※コードは チュートリアル: Azure OpenAI Service の埋め込みとドキュメント検索を確認する を参考にしました

テストデータを読み込みます。data/nearmiss_incident_poc.csv は、40件の事例を記載したファイルです。

import pandas as pd
import tiktoken

embedding_model = "text-embedding-3-small"
embedding_encoding = "cl100k_base"
max_tokens = 8192  # the maximum for text-embedding-3-small is 8191

# load & inspect dataset
input_datapath = "data/nearmiss_incident_poc.csv"  # to save space, we provide a pre-filtered dataset
df = pd.read_csv(input_datapath)
df.info()

各データごとに、「事例」、「原因」、「対応」のカラムがあります。

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 40 entries, 0 to 39
Data columns (total 3 columns):
 #   Column    Non-Null Count  Dtype 
---  ------    --------------  ----- 
 0   incident  40 non-null     object
 1   cause     40 non-null     object
 2   action    40 non-null     object
dtypes: object(3)
memory usage: 1.1+ KB

また、テストデータは、結果を確認しやすくするためカテゴリごとに並べました

• 0 ~ 9 行 : メール誤送信関連
• 10 ~ 19行 : 電源に関する事例
• 20 ~ 29行　: リリース関連
• 30 ~ 39行　: 重大インシデント

df = df[["incident", "cause", "action"]]
df = df.dropna()
df.head(3)

2. テキストの正規化

冗長な空白、句読点を、不必要な情報を削除し埋め込みを取得するためのデータクリーニングを行います。

• 空白や改行を取り除く
• 全角英数字を半角文字に変換する
• モデルの最大トークン数を超えるデータは対象外とする
• 埋め込みを取得したい各カラムを結合

# normalize text
import re
import unicodedata

#https://pandas.pydata.org/pandas-docs/stable/user_guide/indexing.html#evaluation-order-matters
pd.options.mode.chained_assignment = None 

# s is input text
def normalize_text(s, sep_token = " \n "):
    s = re.sub(r'\s+',  ' ', s).strip()
    s = re.sub(r". ,","",s)
    # Convert full-width alphanumeric characters to half-width characters
    s = unicodedata.normalize("NFKC", s)
    # remove all instances of multiple spaces
    s = s.replace("..",".")
    s = s.replace(". .",".")
    s = s.replace("\r\n", "")
    s = s.replace("\n", "")
    # remove url
    url_pattern = r'https?://\S+'
    s = re.sub(url_pattern, "", s)
    s = s.strip()
    return s

def combine_text(df):
    return df["combined"] = (
        df.incident.str.strip() + ";" + 
        df.cause.str.strip() + ";" + 
        df.action.str.strip()
    )
    
def drop_long_token_records(df, n_tokens=max_tokens):
    tokenizer = tiktoken.get_encoding("cl100k_base")
    df['n_tokens'] = df.combined.apply(lambda x: len(tokenizer.encode(x)))
    return df[df.n_tokens<n_tokens] 

テストデータには 事象, 原因, 対応案 カラムがあります。それらを正規化して結合します。

df = drop_long_token_records(df) の部分で、text-embedding-3-small で処理可能な最大トークン数　8191 を超えるデータは処理対象外とします。

df['incident']= df['incident'].apply(lambda x : normalize_text(x))
df['cause']= df['cause'].apply(lambda x : normalize_text(x))
df['action']= df['action'].apply(lambda x : normalize_text(x))
df = combine_text(df)
# MAX Token を超えるデータは削除
df = drop_long_token_records(df)
# 正規化したテキストは、`nearmiss_incident_poc_for_embeddings.csv`　に保存
df.to_csv("data/nearmiss_incident_poc_for_embeddings.csv", encoding="utf_8_sig", index=False)
df[['combined','n_tokens']].head(3)

combinedは結合したデータ、n_tokens は、各テストデータの トークン数 です。

3. 埋め込み（ベクトル）を取得する

Azure OpenAI を使用します

import os
import openai
from openai import AzureOpenAI
import pandas as pd

client = AzureOpenAI(
  api_key = os.getenv("AZURE_OPENAI_API_KEY"),  
  api_version = "2023-05-15",
  azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")
)

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get_embedding(text: str, model="text-embedding-3-small"):
    return client.embeddings.create(input = [text], model=model).data[0].embedding

埋め込み（ベクトル）を取得するコードです。取得したベクトルは再利用するため、ファイル(nearmiss_incident_poc_with_embeddings.csv) に保存します。

# reuse embeddings which is saved earlier
datafile_path = "data/nearmiss_incident_poc_for_embeddings.csv"
df = pd.read_csv(datafile_path)
# model = "deployment_name"
embedding_model = "text-embedding-3-small"

# get embeddings (API access)
df['embedding'] = df.combined.apply(lambda x : get_embedding(x, model=embedding_model)) 
# save embeddings for reuse
df.to_csv("data/nearmiss_incident_poc_with_embeddings.csv", encoding="utf_8_sig", index=False)

ベクトル取得後の csv の中身です。embedding 列が が APIで取得した 「埋め込み」(1536次元ベクトル） です。

import numpy as np
from ast import literal_eval

datafile_path = "data/nearmiss_incident_poc_with_embeddings.csv"
df = pd.read_csv(datafile_path)
df["embedding"] = df.embedding.apply(literal_eval).apply(np.array)
df.head(3)

4. ヒヤリハット事例を検索する

類似度を求める関数を用意します。

# search through the docs for a user query
import numpy as np
from ast import literal_eval

def search_docs(query, n=3, prob=0.4, embedding_model="text-embedding-3-small", sort=True):
    try: 
        print(f"embedding_model: {embedding_model}")
        embedding = get_embedding(query, model=embedding_model)
    except NotFoundError as err:
        print("Error")
        return
    
    datafile_path = "data/nearmiss_incident_poc_with_embeddings.csv"
    df = pd.read_csv(datafile_path)
    df["embedding"] = df.embedding.apply(literal_eval).apply(np.array)
    df["similarity"] = df.embedding.apply(lambda x: cosine_similarity(x, embedding))
    df = df.query('similarity > @prob')
    if sort:
        df = df.sort_values("similarity", ascending=False).head(n)
    return df

「メール誤送信」の事例を検索してみます。（類似度が高い順にデータを出力）

embedding_model="text-embedding-3-small"
search_docs("メールを間違って送信した事例",
    n=10, 
    prob=0.4, 
    embedding_model=embedding_model)

No.0 ~ 9 が メール誤送信関連。メール誤送信関連の事例が高い類似度でマッチしました。（画面は Jupyter notebookの結果で、途切れているので 4件）

"ヒヤリ度" のスコア化

OpenAIのドキュメントに Zero-shot classification があり、Positive と Negative ラベル と の埋め込みとレビューの 類似度から、Zero-Shotで 二値分類する例がありました。

参考：Zero-shot classification with embeddings

このアイデアを参考に、「"重大インシデント"」の 特徴 を利用して、その類似度を算出すれば、ヒヤリハット事例のスコア化ができるのではないかと考えました。

準備

「重大インシデント」, 「軽微なインシデント」 のラベルをフレーズを連ねて用意します

INCIDENT_QUERY = """
重大なシステム停止、
機密情報の漏洩、
データ消失、
ランサムウェア攻撃、
ウィルス感染。
"""

NON_INCIDENT_QUERY =  """
影響なしの未遂事例、
軽微な問題が発生。
遅刻した。
"""

類似度は埋め込みモデルやデータによってスケールが異なるため、 0 ~ 5の範囲で スコアを正規化する関数を用意します。

# スコアを 正規化する関数
def normalize_to_range(data, min=0, max=5):
    data_min = np.min(data)
    data_max = np.max(data)
    # スケーリング（0から1へ）
    normalized = (data - data_min) / (data_max - data_min)
    # 目標範囲へのスケーリング
    scaled = normalized * (max - min) + min
    return scaled

1. データ読み込み

埋め込み取得済みのテストデータを読み込みます。

import numpy as np
from ast import literal_eval

# 使用モデル
embedding_model = "text-embedding-3-small"

# 保存した埋め込みデータを読み込み
datafile_path = "data/nearmiss_incident_poc_with_embeddings.csv"
df = pd.read_csv(datafile_path)
df["embedding"] = df.embedding.apply(literal_eval).apply(np.array)

2. 「重大インシデント」,「軽微なインシデント」の類似度を取得

先に用意した 「重大インシデント」, 「軽微なインシデント」 のラベルをフレーズと、各テストデータの類似度を求めます。

# 「重大インシデント」のクエリとの類似度
df["similarity_major"] = search_docs(
    INCIDENT_QUERY, n=40, prob=0.0, embedding_model=embedding_model,sort=False)["similarity"]
# 「非重大インシデント」のクエリとの類似度
df["similarity_minor"] = search_docs(
    NON_INCIDENT_QUERY, n=40, prob=0.0, embedding_model=embedding_model,sort=False)["similarity"]

「重大インシデント と 「軽微インシデント」の類似度差分を df['diff_similarity'] に格納します。

df['diff_similarity'] = df["similarity_major"] - df["similarity_minor"]

スコア(rating) の計算は、以下のように行いました。

• 類似度差分が正の場合: （「重大インシデント」のスコアが高い場合）には、similarity_major をそのまま使用。
• 類似度差分が負の場合: その負の値（diff_similarity）をペナルティとして加算（実質的に引き下げ）

# 重大インシデントと軽微インシデントの類似度差分を 重大インシデントの類似度に反映
df["rating"] = np.where(
    df["diff_similarity"] > 0,
    df["similarity_major"],
    df["similarity_major"] + df["diff_similarity"]
    )

スコア(rating) を 0 ~ 5 の範囲で正規化します

# 0 ~ 5 の範囲で正規化
scaled = normalize_to_range(df["rating"], min=0, max=5) 
df["rating_nomalized"] = scaled.apply(lambda x: f"{x:.1f}")

3. 結果

「ヒヤリ度」 が高い順

columns = ["incident",
           "cause",
           "action",
           "rating",
           "rating_nomalized"]
df.sort_values("rating_nomalized", ascending=False)[columns]

• No. 30 ~ 39 は 「重大インシデント」のテストデータであり、ほぼ網羅されている
• No. 10 の "電源関連" 事例も「重大インシデント」として上位にある

「ヒヤリ度」が低い順

• メール誤送信関連など 「軽微なインシデント」の スコア化が "概ね" できている
• 「動作しなくなった」などの事例が、「軽微なインシデント」と判定されたのは改善の余地あり

まとめ

人が手動でデータをランク付けする方法は正確ですが、データが増えると非常に手間がかかります。さらに、新しいデータが追加されたり、評価基準が変更された場合、全体の再評価が必要になります。

今回紹介したスコア化の手法は、埋め込みモデルを利用した類似度検索が基本なので、学習プロセスを必要としません。これにより、以下の利点があります：

• 新しいデータが追加されても、再度埋め込みを生成するだけで対応可能
• 評価基準の変更時には、クエリを変更するだけで調整が可能

一方で、「ラベルの調整」や「非ラベルデータへのペナルティ」など、色々と改善の余地はあります。

本手法は、「埋め込み（ベクトル）」を用いた シンプルな計算でスコア化を実現する実用的な１つのアイデアです。「Zero-Shot分類」の他、「スコア化する方法」の１つとして活用できればと思います。

読んでくださった方の何かの役に立てば幸いです。

参考サイト

• Semantic_text_search_using_embeddings

• Tutorial: Explore Azure OpenAI Service embeddings and document search

• Zero-shot classification with embeddings