LLMで広告クリエイティブを生成し、A/Bテストまで自動化する
LLMでの広告コピー量産は簡単ですが、品質ガードと「覗き見問題」に耐える検定設計がなければ、誤った勝者判定を量産するだけになります。生成→バリアント管理→ベイズ的な早期停止→勝ち手の学習ループまでを、動くPythonコードで一気通貫に組みます。
by Akisame
LLMで広告コピーを量産して、そのままA/Bテストまで自動化できるか。答えは「生成は容易、だが検定設計を省くと事故る」です。LLMで案を100本出すのは一瞬ですが、品質ガードと逐次モニタリングに耐える検定を入れなければ、量産がそのまま「誤った勝者判定」の量産になります。本稿では生成パイプライン、バリアント管理、ベイズ的な早期停止、そして勝ち手の学習ループまでを動くPythonコードで通しで組みます。
ここで扱うのは「生成→配信→検定」の自動化に限定します。CVR予測モデルで事前にクリエイティブを選別する話は機械学習でCVR予測モデルを作りクリエイティブ最適化するに譲ります。本稿の主役は、生成した案を実トラフィックに当てて統計的に勝者を決める部分です。
生成パイプライン（プロンプト設計と品質ガード）
LLMでのコピー生成そのものは数行で書けます。事故るのは、生成物をそのまま信用したときです。媒体の文字数制限（検索広告の見出しは全角15文字相当）を超える、薬機法・景表法に触れる誇大表現が混じる、似たような案が重複する——これらを人手で弾いていては自動化の意味がありません。生成と同じ強さで「品質ガード」を実装します。
まず生成側。出力をJSONに固定し、訴求軸（angle）のラベルも一緒に出させておくと、後段の学習ループで効きます。モデルは高頻度・低コストのパイプライン用途なので claude-sonnet-4-6（速度・コスト重視なら claude-haiku-4-5-20251001）を使います。
import os
import json
from anthropic import Anthropic
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
SYSTEM = """あなたはBtoB SaaSに精通した日本語コピーライター。
出力は必ずJSON配列のみ。前置き・説明・コードフェンスを一切付けない。
各要素は headline / description / cta / angle のキーを持つ。
制約:
- headline は全角15文字以内、description は全角45文字以内。
- 誇大表現（必ず / 絶対 / No.1 / 最安 等）を使わない。
- angle は訴求軸の短いラベル（例: 時短 / コスト削減 / 失敗回避 / 信頼）。
"""
def generate_variants(product_brief: str, n: int = 8,
model: str = "claude-sonnet-4-6") -> list[dict]:
msg = client.messages.create(
model=model,
max_tokens=1500,
system=SYSTEM,
messages=[{
"role": "user",
"content": f"商品概要:\n{product_brief}\n\n"
f"上記の広告クリエイティブ案を{n}件、JSON配列で出力。",
}],
)
raw = msg.content[0].text.strip()
# JSON以外が混ざった場合に備えて配列部分だけ抽出する
start, end = raw.find("["), raw.rfind("]")
return json.loads(raw[start:end + 1])次に品質ガード。ここが自動化の生命線になります。文字数は「全角＝2、半角＝1」で数えないと媒体審査で落ちるので、east_asian_width で幅を測ります。NG語は媒体ポリシーと法令に合わせて辞書化し、CTAの欠落もチェックします。
import unicodedata
BANNED = ["必ず", "絶対", "業界No.1", "日本一", "最安", "誰でも簡単"]
def width(s: str) -> int:
"""全角を2、半角を1として表示幅を返す。"""
return sum(2 if unicodedata.east_asian_width(c) in "WF" else 1 for c in s)
def passes_guard(v: dict, max_headline: int = 30,
max_desc: int = 90) -> tuple[bool, list[str]]:
errs = []
if width(v.get("headline", "")) > max_headline:
errs.append("見出しが長い")
if width(v.get("description", "")) > max_desc:
errs.append("説明文が長い")
text = v.get("headline", "") + v.get("description", "")
hits = [w for w in BANNED if w in text]
if hits:
errs.append("NG語: " + ", ".join(hits))
if not v.get("cta"):
errs.append("CTAが空")
return (len(errs) == 0, errs)最後に重複排除。LLMは温度を上げても表現が収束しやすく、見た目だけ違う実質同一案が混ざります。本文のハッシュで弾きます。
import hashlib
def content_hash(v: dict) -> str:
key = (v["headline"] + "|" + v["description"]).strip()
return hashlib.sha256(key.encode("utf-8")).hexdigest()[:12]
def screen(variants: list[dict]) -> tuple[list[dict], list[tuple]]:
seen, kept, rejected = set(), [], []
for v in variants:
ok, errs = passes_guard(v)
if not ok:
rejected.append((v, errs))
continue
h = content_hash(v)
if h in seen:
rejected.append((v, ["重複"]))
continue
seen.add(h)
v["id"] = h
kept.append(v)
return kept, rejected配信連携とバリアント管理
通過した案は「いつ・どの案を・どこに出し・どんな成績だったか」を一元管理しないと、検定も学習ループも組めません。ここで実装すべきはバリアントレジストリと、媒体ごとに差し替え可能な配信アダプタです。
レジストリはSQLiteで十分です（規模が出たらSupabaseのPostgresに置き換えます）。id は前段のコンテンツハッシュをそのまま主キーにして冪等性を担保します。同じ案を二度登録しても増えません。
import sqlite3
import json
import time
def init_db(path: str = "variants.db") -> sqlite3.Connection:
con = sqlite3.connect(path)
con.execute("""
CREATE TABLE IF NOT EXISTS variants(
id TEXT PRIMARY KEY,
payload TEXT,
status TEXT, -- draft / live / paused / winner / archived
platform_ref TEXT, -- 媒体側のクリエイティブID
created_at REAL
)
""")
con.commit()
return con
def register(con: sqlite3.Connection, v: dict) -> None:
con.execute(
"INSERT OR IGNORE INTO variants VALUES(?,?,?,?,?)",
(v["id"], json.dumps(v, ensure_ascii=False), "draft", None, time.time()),
)
con.commit()配信先（Google Ads、Meta など）はAPIの作法がばらばらなので、抽象インターフェースで切り離します。媒体差は各アダプタに閉じ込め、上位のオーケストレーションは媒体非依存に保ちます。これで「同じ案を複数媒体に同条件で出す」も「媒体を後から足す」も破綻しません。
from abc import ABC, abstractmethod
class AdPlatform(ABC):
@abstractmethod
def push(self, variant: dict) -> str:
"""媒体にクリエイティブを登録し、媒体側IDを返す。"""
class DryRunPlatform(AdPlatform):
"""本番投入前の検証用。実際には送らずログだけ出す。"""
def push(self, variant: dict) -> str:
print("would push:", variant["id"], variant["headline"])
return f"dryrun-{variant['id']}"
def deploy(con: sqlite3.Connection, platform: AdPlatform, v: dict) -> None:
ref = platform.push(v)
con.execute(
"UPDATE variants SET status='live', platform_ref=? WHERE id=?",
(ref, v["id"]),
)
con.commit()実際のGoogle Ads / Meta アダプタは各社SDKを push の中に包むだけです。重要なのは、媒体APIの不安定さ（レート制限・審査落ち・一時エラー）を push の内側でリトライ吸収し、上位ロジックに漏らさないこと。配信は「壊れる前提」で組みます。
A/Bテストの統計設計（検定・早期停止）
ここが本稿の核であり、LLM量産が事故に変わる最大の分岐点です。案を大量に出せるようになると、人は「ダッシュボードを毎日見て、差がついた瞬間に勝者を決める」をやりたくなります。これが**覗き見問題（peeking）**で、固定標本の検定を途中で何度も覗くと、有意水準 p < 0.05 のつもりが、実際の偽陽性率は数倍に膨らみます。差が出たから止めるのではなく、止めてよい設計のもとで止める必要があります。
対処は大きく2つ。ひとつは固定標本設計——必要サンプル数を事前に決め、貯まるまで結論を見ません。もうひとつは逐次的に見てよい設計——ベイズ的モニタリングや群逐次法（アルファ消費）で、途中で見ても妥当性が崩れないようにします。LLMで案が次々入れ替わる運用では後者が現実的ですが、まず物差しとして必要サンプル数を出しておきます。
import numpy as np
from statsmodels.stats.power import NormalIndPower
from statsmodels.stats.proportion import proportion_effectsize
def required_sample_size(baseline: float, mde_rel: float,
alpha: float = 0.05, power: float = 0.8) -> int:
"""各群に必要なサンプル数（mde_rel は相対の最小検出効果、例 0.15 = +15%）。"""
p2 = baseline * (1 + mde_rel)
effect = proportion_effectsize(p2, baseline)
n = NormalIndPower().solve_power(
effect_size=effect, alpha=alpha, power=power,
ratio=1, alternative="two-sided",
)
return int(np.ceil(n))
# 例: ベースラインCVR 2%、相対+20%を検出したい
# required_sample_size(0.02, 0.20) -> 各群で数千〜1万規模が必要この数字を見ると、小さな相対改善を厳密に検出するには大量のトラフィックが必要になると分かります。だからこそ早期停止が効きます——明らかに劣る案を早く切れば、その分の予算を有望な案に回せます。
逐次で見てよい設計として、ベータ二項モデルのベイズA/Bテストを実装します。各案のCVRに事前分布 （一様）を置き、観測で更新し、「変動案が対照を上回る確率」と「採用したときの期待損失」で判断します。期待損失とは、その案を勝者に選んだとき、もし本当は反対側が良かった場合に取り逃すCVRの期待値です。
この期待損失が「許容できるほど小さい」しきい値を下回ったら止めます。覗き見しても破綻しないのがベイズ的モニタリングの利点です。
import numpy as np
from dataclasses import dataclass
@dataclass
class Arm:
name: str
impressions: int = 0 # 試行数（CV判定の母数）
conversions: int = 0 # 成功数
prior_alpha: float = 1.0
prior_beta: float = 1.0
@property
def alpha(self) -> float:
return self.prior_alpha + self.conversions
@property
def beta(self) -> float:
return self.prior_beta + (self.impressions - self.conversions)
def sample(self, size: int, rng) -> np.ndarray:
return rng.beta(self.alpha, self.beta, size=size)
class BayesianABTest:
def __init__(self, control: Arm, variant: Arm, seed: int = 0):
self.control = control
self.variant = variant
self.rng = np.random.default_rng(seed)
def _draw(self, size: int = 200_000):
return self.control.sample(size, self.rng), self.variant.sample(size, self.rng)
def prob_variant_wins(self, size: int = 200_000) -> float:
a, b = self._draw(size)
return float(np.mean(b > a))
def expected_loss(self, size: int = 200_000) -> dict:
"""各案を採用したときに取り逃すCVRの期待値（対照CVRに対する相対）。"""
a, b = self._draw(size)
base = float(np.mean(a)) + 1e-12
return {
"variant": float(np.mean(np.maximum(a - b, 0))) / base,
"control": float(np.mean(np.maximum(b - a, 0))) / base,
}
def decision(self, loss_threshold: float = 0.01,
min_impr: int = 2000) -> dict:
if min(self.control.impressions, self.variant.impressions) < min_impr:
return {"stop": False, "reason": "最小サンプル未達"}
losses = self.expected_loss()
leader = "variant" if losses["variant"] < losses["control"] else "control"
if losses[leader] < loss_threshold:
return {"stop": True, "winner": leader,
"expected_loss": losses[leader],
"prob_variant_wins": self.prob_variant_wins()}
return {"stop": False, "leader": leader,
"expected_loss": losses[leader]}使い方は、配信成績が入るたびに Arm を更新し、decision() を呼ぶだけです。loss_threshold は「これ以下の取り逃しなら確定してよい」というビジネス側の許容値で、対照CVRに対する相対で 1% 程度から始めます。min_impr で初期の偶然のブレによる早すぎる確定を防ぎます。
control = Arm("baseline", impressions=4000, conversions=82) # CVR ~2.0%
variant = Arm("llm_gen", impressions=4000, conversions=104) # CVR ~2.6%
test = BayesianABTest(control, variant)
print(test.prob_variant_wins()) # 変動案が勝つ確率
print(test.decision()) # 止めてよいか／勝者勝ち手の学習ループ
A/Bテストは1回勝者を決めて終わりではありません。勝った案の「何が効いたのか」を抽出して次の生成に還元し、同時に複数案への配信配分を継続的に最適化します。ここで2つの仕組みを足すとループが閉じます。
ひとつめは生成への還元。勝ち案の訴求軸（angle）と特徴を要約し、次回の生成プロンプトに「勝ちパターン」として注入します。前段で angle を出させておいたのがここで効きます。
def build_winning_brief(winners: list[dict]) -> str:
lines = [
f"- 見出し例『{w['headline']}』 / 訴求: {w['angle']}（相対CVR {w['cvr_range']}）"
for w in winners
]
return "過去テストで上位だった案の特徴（参考にしつつ、模倣はしない）:\n" + "\n".join(lines)
# 次回生成時に SYSTEM の末尾へ付与する
def generate_with_memory(product_brief: str, winners: list[dict],
n: int = 8) -> list[dict]:
memory = build_winning_brief(winners)
msg = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1500,
system=SYSTEM + "\n\n" + memory,
messages=[{"role": "user",
"content": f"商品概要:\n{product_brief}\n\n案を{n}件、JSON配列で。"}],
)
raw = msg.content[0].text.strip()
s, e = raw.find("["), raw.rfind("]")
return json.loads(raw[s:e + 1])ふたつめは配信配分の最適化。案が3本以上に増えると、A/Bの二項対決より多腕バンディットの方が無駄が少なくて済みます。トンプソンサンプリングなら、各案の事後分布から1点サンプリングして最大の案に配信を寄せます——有望な案へ自動的に予算が流れ、見込みの薄い案への配信が自然に絞られます。
class ThompsonAllocator:
def __init__(self, arm_names: list[str], seed: int = 0):
self.arms = {n: Arm(n) for n in arm_names}
self.rng = np.random.default_rng(seed)
def choose(self) -> str:
draws = {n: arm.sample(1, self.rng)[0] for n, arm in self.arms.items()}
return max(draws, key=draws.get)
def update(self, name: str, converted: bool) -> None:
arm = self.arms[name]
arm.impressions += 1
arm.conversions += int(converted)
# インプレッションが来るたびに choose() で出す案を決め、結果を update()
allocator = ThompsonAllocator(["a", "b", "c", "d"])
chosen = allocator.choose()
# ... 実配信し、CVの有無を観測したら ...
allocator.update(chosen, converted=True)トンプソンサンプリングは「探索（まだ自信のない案を試す）」と「活用（今いちばん良い案に寄せる）」を事後分布が自動で按分します。ベイズA/Bテストの Arm をそのまま流用できるのもきれいなところです。確定的に1勝者を決めたい局面ではA/Bテストの decision()、運用しながら配分し続けたい局面ではバンディット、と使い分けます。
生成・配信・検定・学習がコードで閉じると、人がやるのは「品質ガードの辞書を磨く」「許容損失としきい値を決める」「探索枠の割合を握る」という判断に絞られます。LLMは案を出すだけで、勝者を決めるのは統計であって生成モデルではありません——この役割分担を崩さないことが、量産を成果に変える条件になります。