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

# はじめに

本記事では、前回の [「OpenAI の埋め込みモデルを使ってみる」](https://qiita.com/kodamap/items/c246901cc04d2ee33995) につづいて、「ヒヤリハット事例」の「類似検索」と「スコアリング」をやってみます。

https://qiita.com/kodamap/items/c246901cc04d2ee33995

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

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

https://gist.github.com/kodamap/7b7cf6d81b35771688073e0aa647f1fc


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

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

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

__メール誤送信関連__

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

__電源に関する事例__

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

__リリース関連__

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

__重大インシデント__

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

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

## 1. データ読み込み

※コードは [チュートリアル: Azure OpenAI Service の埋め込みとドキュメント検索を確認する](https://learn.microsoft.com/ja-jp/azure/ai-services/openai/tutorials/embeddings?tabs=python-new%2Ccommand-line&pivots=programming-language-python) を参考にしました


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

```py
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()
```

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

```py
<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行　: 重大インシデント

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

![nearmiss1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/a0726c98-cdbc-32ba-5aff-8e8d92904386.png)

## 2. テキストの正規化

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

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

```py
# 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` を超えるデータは処理対象外とします。
```py
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` は、各テストデータの トークン数 です。



![nearmiss2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/3671350b-44f0-a916-14a0-3576321283cc.png)

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

Azure OpenAI を使用します

```py
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`) に保存します。

```py
# 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次元ベクトル） です。


```py
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)
```


![nearmiss3.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/78f4279d-9e08-3cbd-c0ed-a8d937c9fc8b.png)


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

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

```py
# 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
```

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


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

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

![nearmiss4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/aff1faa4-e975-304a-7368-8e7412fca65d.png)


# "ヒヤリ度" のスコア化

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

参考：[Zero-shot classification with embeddings](https://cookbook.openai.com/examples/zero-shot_classification_with_embeddings)


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

## 準備

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

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

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

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

```py
# スコアを 正規化する関数
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. データ読み込み

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

```py
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. 「重大インシデント」,「軽微なインシデント」の類似度を取得

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

```py
# 「重大インシデント」のクエリとの類似度
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']` に格納します。

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

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

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

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

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

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

## 3. 結果

__「ヒヤリ度」 が高い順__

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

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

![nearmiss6.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/a5e0288e-de45-59bd-c1fb-6e795fe66a27.png)


__「ヒヤリ度」が低い順__

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

![nearmiss7.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/118309/7ee07584-c4b6-0bb7-a436-e7ccc92f4c74.png)


# まとめ

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

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

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

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

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


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




# 参考サイト

- [Semantic_text_search_using_embeddings](https://github.com/openai/openai-cookbook/blob/main/examples/Semantic_text_search_using_embeddings.ipynb)

- [Tutorial: Explore Azure OpenAI Service embeddings and document search](https://learn.microsoft.com/en-us/azure/ai-services/openai/tutorials/embeddings?tabs=python-new%2Ccommand-line&pivots=programming-language-python)

- [Zero-shot classification with embeddings](https://cookbook.openai.com/examples/zero-shot_classification_with_embeddings)