やぁ🙋♂️中島です🙋♂️こんにちは🙋♂️
今回は、GitHubのissueの重複を検知する処理を書いてみました!
重複を検知するために、OpenAIのEmbeddingsの機能を使用しています!
Embeddingsというのは、文字や単語を数値化(数値ベクトルとして表現)することです。
身近な例だと、おすすめ機能などで使用されています!
レポジトリ
https://github.com/91nakashima/issue_checker 今回やったこと
- GitHubのGitHub AppsでGitHubのAPIを叩けるように設定
- APIからissueの一覧を取得
- OpenAIのEmbeddingsを使用し、issueをベクトル化
- 閾値を設定し、重複を検出
- 検出したらメッセージを送信
実運用に向けて追加しなければならないこと
- Rest APIを公開
- GitHubのWebhookにURLを貼り付ける(issueが作成されたタイミングで動けば良い)
GitHubのGitHub AppsでGitHubのAPIを叩けるように設定
これはこのサイトを参考にして設定しました。
詳しく記載してありますので参考にしてみてください!
https://qiita.com/itkr/items/db5ed965bf8cac1cdd4eAPIからissueの一覧を取得
取得は、下記のように記載しました。
今回は、全てのissueを取得したかったので、再帰的に全てのissueを取得してきました。
https://github.com/91nakashima/issue_checker/blob/main/src/main.py#L51-L73
def get_issues(self, page=1, per_page=100) -> list[GithubIssue]:
url = f'https://api.github.com/repos/{self.owner}/{self.repo}/issues?page={page}&per_page={per_page}'
response = requests.get(url, headers=self.headers)
link_str = response.headers.get('Link', '')
next_pages = link_str.split(', ')
is_has_next = False
for next_page in next_pages:
if 'rel="next"' in next_page:
is_has_next = True
link_str = next_page
break
if link_str and is_has_next:
next_page_int: int = int(link_str.split('page=')[1].split('&')[0])
val = [
GithubIssue(**issue) for issue in response.json()
] + self.get_issues(next_page_int, per_page)
return sorted(val, key=lambda x: x.id, reverse=True)
return [GithubIssue(**issue) for issue in response.json()]
OpenAIのEmbeddingsを使用し、issueをベクトル化
ベクトル化する関数は簡単に記載してあります。
https://github.com/91nakashima/issue_checker/blob/main/src/embedding.py#L8-L12
def get_embedding(text) -> list[float]:
model = 'text-embedding-3-small'
text = text.replace("\n", " ")
return client.embeddings.create(input=[text], model=model).data[0].embedding
閾値を設定し、重複を検出
ここでは、閾値を設定し、似ている情報を算出する関数を作成しました。
https://github.com/91nakashima/issue_checker/blob/main/src/embedding.py#L14-L55
def find_embeddings(
base_embeddings: list[float],
compare_embeddings: list[list[float]],
ids: list[int],
threshold: float
) -> list[dict[int, float]]:
"""
Params
---
base_embeddings: list[float]
The embeddings of the base text
compare_embeddings: list[list[float]]
The embeddings of the texts to compare
threshold: float
The threshold to compare the embeddings
Returns
---
list[dict[int, float]]
The list of ids of the texts that are similar to the base text
"""
similar_ids: dict[int, float] = []
for i, compare_embedding in enumerate(compare_embeddings):
# not working
# similarity = client.embeddings.similarity(base_embeddings, compare_embedding).data[0].score
# if similarity > threshold:
# similar_ids.append(ids[i])
similarity = sum([a * b for a, b in zip(base_embeddings, compare_embedding)])
if similarity > threshold:
similar_ids.append({
"val": similarity,
"id": ids[i]
})
# おおきい順にソート
similar_ids = sorted(similar_ids, key=lambda x: x["val"], reverse=True)
return similar_ids
検出したらメッセージを送信
上記の関数を使用して、重複をコメントとして送信しています。
https://github.com/91nakashima/issue_checker/blob/main/main.py#L78-L93
data = find_embeddings(target_issue_embedding, embeddings, ids, 0.7)
data = data[:3]
if not len(data):
return
send_text = '重複を検出しました。\n\n'
for issue_id in data:
find_data = next(filter(lambda x: x.id == issue_id['id'], res), None)
if find_data:
print(find_data.number)
print(find_data.title)
send_text += f'#{find_data.number} {find_data.title}\n'
app.send_comment(issue_number, send_text)
まとめ
GitHubでAPIを叩けるようにするトークンを発行し、
OpenAIのAPIを.envファイルに追加すれば簡単に実装することができるので、ぜひ皆さんも開発が楽になるように参考にしながら使ってみてください!
https://github.com/91nakashima/issue_checkerのREADMEにも記載がありますが、GitHub Appのpemファイルなどをディレクトリに設置し、.envファイルに必要な情報を登録したら、すでに実装してあるので、コマンドを叩けば、重複を検知することができます。
$ python main.py --issue 1111 --repo XXX