「メールの件名、毎回考えるのめんどくさいな...」
そう思ったので、AIに件名を考えさせて、A/Bテストして、PDCAを自動で回す仕組みを作ってみました。
やってることはシンプルです。
- OpenAI APIで件名を何パターンか生成
- なるべく違う表現のものを選んでA/Bテスト
- 開封率が良かった件名の特徴を次に活かす
これをぐるぐる回すと、だんだん開封率が上がっていく(はず)。
使ったもの
- blastengine - メール配信API。開封率が取れる。
- OpenAI API - 件名生成とEmbedding。
- Python - つなぎ込み。
全体像
1 OpenAI(件名作って!)
↓
2 Python(似てないやつ選ぶ)
↓
3 blastengine(配信して開封率見る)
↓
4 Python(良かった件名の特徴をメモ)
↓
5 OpenAI(こういうの意識して件名作って!)
↓
2に戻る
実装してみる
まず準備
pip install openai requests numpy
blastengineの認証トークンを作る
blastengineのAPIを叩くにはBearerTokenが必要です。作り方はちょっと独特で、ログインIDとAPIキーをSHA256でハッシュしてBase64エンコードします。
import hashlib
import base64
def generate_bearer_token(login_id: str, api_key: str) -> str:
combined = f"{login_id}{api_key}"
hash_value = hashlib.sha256(combined.encode()).hexdigest().lower()
return base64.b64encode(hash_value.encode()).decode()
BEARER_TOKEN = generate_bearer_token("your_login_id", "your_api_key")
件名をAIに作らせる
件名の生成機能を作ります。ポイントはtemperature=0.9で多様性を出すことと、過去に良かった件名の特徴をプロンプトに混ぜることです。モデルは今回はGPT-4oにしました。
from openai import OpenAI
client = OpenAI()
def generate_subject_candidates(
product_name: str,
target_audience: str,
previous_winners: list[str] = None,
num_candidates: int = 5
) -> list[str]:
"""件名を何パターンか作ってもらう"""
feedback = ""
if previous_winners:
feedback = f"""
ちなみに過去にウケた件名の特徴はこんな感じ:
{chr(10).join(f"- {s}" for s in previous_winners)}
参考にしつつ、新しいのも試してね。
"""
prompt = f"""
メールの件名を{num_candidates}個考えて。
条件:
- 商品: {product_name}
- ターゲット: {target_audience}
- 30文字以内
- バリエーション出して(緊急性、質問形式、数字入りとか)
{feedback}
件名だけリストで出力して。
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.9
)
content = response.choices[0].message.content
subjects = [
line.strip().lstrip("0123456789.-) ")
for line in content.strip().split("\n")
if line.strip()
]
return subjects[:num_candidates]
似てない件名を選ぶ
せっかくA/Bテストするなら、似たような件名を比べても意味がないため、Embeddingで類似度を計算して、なるべく違うやつを選ぶようにします。
import numpy as np
def get_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def cosine_similarity(vec1, vec2):
v1, v2 = np.array(vec1), np.array(vec2)
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
def select_diverse_subjects(subjects: list[str], num_select: int = 2) -> list[str]:
"""できるだけ違う件名を選ぶ"""
if len(subjects) <= num_select:
return subjects
embeddings = [get_embedding(s) for s in subjects]
selected = [0] # 最初の1個は適当に
while len(selected) < num_select:
max_min_distance = -1
best_idx = -1
for i, emb in enumerate(embeddings):
if i in selected:
continue
# 選んだやつとの距離の最小値が一番大きいやつを選ぶ
min_dist = min(
1 - cosine_similarity(emb, embeddings[j])
for j in selected
)
if min_dist > max_min_distance:
max_min_distance = min_dist
best_idx = i
selected.append(best_idx)
return [subjects[i] for i in selected]
blastengineでメール配信
一斉配信の流れは3ステップです。
- 配信を作る(
/deliveries/bulk/begin) - 宛先を追加する(
/deliveries/bulk/update/{id}) - 配信実行(
/deliveries/bulk/commit/{id}/immediate)
import requests
BASE_URL = "https://app.engn.jp/api/v1"
def get_headers():
return {
"Authorization": f"Bearer {BEARER_TOKEN}",
"Content-Type": "application/json"
}
def create_bulk_delivery(subject, text_part, from_email, from_name):
"""配信を作成"""
payload = {
"from": {"email": from_email, "name": from_name},
"subject": subject,
"text_part": text_part
}
r = requests.post(f"{BASE_URL}/deliveries/bulk/begin", headers=get_headers(), json=payload)
r.raise_for_status()
return r.json()["delivery_id"]
def add_recipients(delivery_id, emails):
"""宛先追加"""
payload = {"to": [{"email": e} for e in emails]}
r = requests.put(f"{BASE_URL}/deliveries/bulk/update/{delivery_id}", headers=get_headers(), json=payload)
r.raise_for_status()
def send_now(delivery_id):
"""即時配信"""
r = requests.patch(f"{BASE_URL}/deliveries/bulk/commit/{delivery_id}/immediate", headers=get_headers())
r.raise_for_status()
def get_result(delivery_id):
"""結果取得"""
r = requests.get(f"{BASE_URL}/deliveries/{delivery_id}", headers=get_headers())
r.raise_for_status()
return r.json()
A/Bテストを回す
import random
def run_ab_test(subjects, recipients, text_part, from_email, from_name):
"""受信者を分割してA/Bテスト"""
random.shuffle(recipients)
chunk_size = len(recipients) // len(subjects)
results = []
for i, subject in enumerate(subjects):
start = i * chunk_size
end = start + chunk_size if i < len(subjects) - 1 else len(recipients)
chunk = recipients[start:end]
delivery_id = create_bulk_delivery(subject, text_part, from_email, from_name)
add_recipients(delivery_id, chunk)
send_now(delivery_id)
results.append({
"delivery_id": delivery_id,
"subject": subject,
"count": len(chunk)
})
return results
def check_results(delivery_ids):
"""開封率をチェック"""
results = []
for did in delivery_ids:
r = get_result(did)
total = r.get("total_count", 0)
opens = r.get("open_count", 0)
rate = (opens / total * 100) if total > 0 else 0
results.append({
"subject": r.get("subject"),
"open_rate": round(rate, 2),
"opens": opens,
"total": total
})
return sorted(results, key=lambda x: x["open_rate"], reverse=True)
動かしてみる
def main():
# 設定
product = "クラウド会計ソフト"
target = "中小企業の経営者"
from_email = "news@example.com"
from_name = "会計ソフトNews"
recipients = ["user1@example.com", "user2@example.com", ...] # 実際はDBから
text_part = """
いつもありがとうございます。
新機能のお知らせです...
"""
# 前回良かった件名の特徴(最初は空でOK)
previous_winners = [
"【期間限定】が刺さる",
"数字入れると開封率上がる"
]
# 1. 件名生成
print("件名生成中...")
candidates = generate_subject_candidates(product, target, previous_winners, num_candidates=5)
print(f"候補: {candidates}")
# 2. 違うやつを選ぶ
selected = select_diverse_subjects(candidates, num_select=2)
print(f"選んだやつ: {selected}")
# 3. A/Bテスト配信
print("配信中...")
results = run_ab_test(selected, recipients, text_part, from_email, from_name)
print(f"完了: {results}")
# 4. 結果確認(配信後しばらく経ってから)
# delivery_ids = [r["delivery_id"] for r in results]
# print(check_results(delivery_ids))
if __name__ == "__main__":
main()
シミュレーション検証
件名改善ループが本当に機能するか、検証してみました。
検証のために大量送信するのは良くないのでモックAPIを使います。
方法
- 送信APIをモック化し、特定の特徴(【】強調、数字、緊急性など)を持つ件名に高い開封率を返す
- AIが件名生成するときに高い開封率の特徴に寄せて生成する率が上がるかを調べる
- 20ラウンド × 10件 = 200件のサンプルで統計的に検証
結果
| 指標 | 値 |
|---|---|
| 初回ラウンド平均 | 17.9% |
| 最終ラウンド平均 | 28.0% |
| 改善幅 | +10.1% |
開封率TOP3:
- 54.7% - 【緊急】会計効率化、今すぐ始めませんか?
- 53.7% - クラウド会計無料体験【限定30日】
- 52.5% - 【緊急】会計業務の時間を50%削減!
ラウンド別平均開封率
ラウンド1: 17.9% ████████
ラウンド2: 18.7% █████████
ラウンド3: 17.9% ████████
ラウンド4: 17.0% ████████
ラウンド5: 16.8% ████████
ラウンド6: 22.6% ███████████
ラウンド7: 22.4% ███████████
ラウンド8: 22.6% ███████████
ラウンド9: 24.2% ████████████
ラウンド10: 31.5% ███████████████
ラウンド11: 30.7% ███████████████
ラウンド12: 29.7% ██████████████
ラウンド13: 27.6% █████████████
ラウンド14: 30.1% ███████████████
ラウンド15: 30.0% ███████████████
ラウンド16: 25.9% ████████████
ラウンド17: 31.4% ███████████████
ラウンド18: 30.3% ███████████████
ラウンド19: 25.4% ████████████
ラウンド20: 28.0% █████████████
フィードバックループが効いているっぽい。
モデルごとの差異
GPT-5でもやってみましたが、GPT-4oほどは劇的な改善効果が出ませんでした。GPT-4.1だとまあまあ改善しましたが、それでもGPT-4oほどではなかったです。
モデルごとに「フィードバックを採用するかどうか」で性質の違いがあるかもしれません。
次やりたいこと
- 本番環境で運用しながらの検証
- RAG、LoRA、ローカルLLMなどを使った本格的な自律改善システム