Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| # === Writable config dirs === | |
| import os | |
| os.environ.setdefault("APP_DATA_DIR", "/data/app_data" if os.access("/data", os.W_OK) else "/tmp/app_data") | |
| os.environ.setdefault("MPLCONFIGDIR", os.path.join(os.environ["APP_DATA_DIR"], "mplconfig")) | |
| os.makedirs(os.environ["MPLCONFIGDIR"], exist_ok=True) | |
| import json, uuid, random | |
| from datetime import datetime | |
| from typing import Dict | |
| import gradio as gr | |
| import pandas as pd | |
| from app.storage import ( | |
| init_db, insert_variant, upsert_campaign, get_variant, get_metrics, | |
| get_campaign, set_campaign_settings, | |
| get_campaign_value_per_conversion, log_event, | |
| export_csv, reset_all, evaluate_stop_rules, | |
| record_compliance_log, audit, get_variants | |
| ) | |
| from app.bandit import ThompsonBandit | |
| from app.linucb import LinUCB | |
| from app.forecast import SeasonalityModel | |
| from app.compliance import rule_based_check, llm_check_and_fix | |
| from app.openai_client import openai_chat_json | |
| from app.adapters import XAdapter, MetaAdapter, GoogleAdsAdapter | |
| # 初期化 | |
| init_db() | |
| _seasonality_cache: Dict[str, SeasonalityModel] = {} | |
| GENERATE_COLUMNS = ["variant_id", "status", "rejection_reason", "text"] | |
| REPORT_COLUMNS = ["variant_id","impressions","clicks","conversions","ctr","cvr","expected_value"] | |
| GEN_SYSTEM = """ | |
| あなたは日本語広告コピーのプロフェッショナルコピーライターです。 | |
| 出力は**次のJSONオブジェクトのみ**で厳密に返してください。余計な文章・説明・前置きは禁止です。 | |
| 形式: | |
| {"variants":[{"headline":"全角15-25字程度","body":"全角40-90字程度"}, ...]} | |
| ルール: | |
| - 医薬効能の断定、100%、永久、即効、根拠のない数値などの誇大表現は禁止 | |
| - CTAは自然に | |
| - 日本語で、句読点や記号は自然に | |
| """ | |
| GEN_USER_TEMPLATE = """ | |
| ブランド: {brand} | |
| 商品/サービス: {product} | |
| 想定ターゲット: {target} | |
| トーン: {tone} | |
| 制約: {constraints} | |
| 生成本数: {k} | |
| 要件: | |
| - "variants" 配列の要素数は **ちょうど {k}** 件にしてください | |
| - 各要素は {{"headline":"...","body":"..."}} のみ | |
| """ | |
| def _seasonal(campaign_id: str) -> SeasonalityModel: | |
| if campaign_id not in _seasonality_cache: | |
| m = SeasonalityModel(campaign_id) | |
| try: | |
| m.fit() | |
| except Exception: | |
| pass | |
| _seasonality_cache[campaign_id] = m | |
| return _seasonality_cache[campaign_id] | |
| def _safe_get_variants(data, k: int): | |
| items = [] | |
| if isinstance(data, dict) and isinstance(data.get("variants"), list): | |
| items = data["variants"] | |
| elif isinstance(data, list): | |
| items = data | |
| if not items or not all(isinstance(x, dict) for x in items): | |
| return None | |
| out = [] | |
| for it in items[:k]: | |
| out.append({ | |
| "headline": str(it.get("headline", "")).strip(), | |
| "body": str(it.get("body", "")).strip(), | |
| }) | |
| return out | |
| def _local_variants(brand: str, product: str, k: int): | |
| base_head = ["使いやすさで選ばれています","日々の習慣をシンプルに","はじめてでも安心","続けやすいサポートを","いま必要な機能だけを"] | |
| base_body = [ | |
| "{brand}の「{product}」。生活になじむ設計で、今日からムリなく始められます。まずは詳細をご覧ください。", | |
| "毎日を少しラクに。{brand}の{product}が、あなたの習慣づくりを後押しします。今すぐチェック。", | |
| "難しい操作は不要。{brand}の{product}なら、使い始めから自然に続けられます。詳しくはサイトへ。", | |
| "必要な情報をひと目で。{brand}の{product}で、日々の管理をシンプルに。詳細を見る。", | |
| "続けやすさを重視。{brand}の{product}で、小さな一歩から。" | |
| ] | |
| out = [] | |
| for i in range(k): | |
| out.append({"headline": base_head[i % len(base_head)], | |
| "body": base_body[i % len(base_body)].format(brand=brand, product=product)}) | |
| return out | |
| async def ui_generate(campaign_id: str, brand: str, product: str, target: str, tone: str, k_variants: int, | |
| ng_words: str, value_per_conversion: float): | |
| k_variants = int(k_variants) | |
| constraints = {"ng_words": [w.strip() for w in ng_words.splitlines() if w.strip()]} if ng_words else {} | |
| upsert_campaign(campaign_id, brand, product, target, tone, "ja", constraints, value_per_conversion) | |
| user = GEN_USER_TEMPLATE.format( | |
| brand=brand, product=product, target=target, tone=tone, | |
| constraints=json.dumps(constraints, ensure_ascii=False), k=k_variants | |
| ) | |
| items = None | |
| try: | |
| data = await openai_chat_json( | |
| [{"role": "system", "content": GEN_SYSTEM},{"role": "user", "content": user}], | |
| temperature=0.2, max_tokens=1200, | |
| ) | |
| items = _safe_get_variants(data, k_variants) | |
| except Exception: | |
| items = None | |
| if not items: | |
| try: | |
| retry_user = user + "\n\n注意: 'variants' は必ず指定件数、各要素は {\"headline\":\"...\",\"body\":\"...\"} のみ。" | |
| data = await openai_chat_json( | |
| [{"role": "system", "content": GEN_SYSTEM},{"role": "user", "content": retry_user}], | |
| temperature=0.1, max_tokens=1000, | |
| ) | |
| items = _safe_get_variants(data, k_variants) | |
| except Exception: | |
| items = None | |
| if not items: | |
| items = _local_variants(brand, product, k_variants) | |
| rows = [] | |
| for it in items[:k_variants]: | |
| headline, body = it["headline"], it["body"] | |
| text = f"{headline}\n{body}".strip() | |
| vid = str(uuid.uuid4())[:8] | |
| ok_rule, bads = rule_based_check(text, (constraints or {}).get("ng_words")) | |
| status, rejection = "approved", None | |
| ok_llm, reasons, fixed = llm_check_and_fix(text) | |
| if not ok_rule and not ok_llm: | |
| status, rejection = "rejected", "; ".join(bads + reasons) | |
| elif not ok_rule and ok_llm: | |
| text = fixed or text | |
| elif ok_rule and not ok_llm: | |
| text = fixed or text | |
| insert_variant(campaign_id, vid, text, status, rejection) | |
| record_compliance_log(campaign_id, vid, status, bads, ok_llm, reasons, fixed) | |
| rows.append({"variant_id": vid, "status": status, "rejection_reason": rejection or "", "text": text}) | |
| df = pd.DataFrame(rows, columns=GENERATE_COLUMNS) | |
| return df | |
| def _policy_of(campaign_id: str) -> str: | |
| cfg = get_campaign(campaign_id) | |
| return str(cfg["policy"] or "thompson") if cfg else "thompson" | |
| def _holdout_ratio_of(campaign_id: str) -> float: | |
| cfg = get_campaign(campaign_id) | |
| return float(cfg["holdout_ratio"] or 0.0) if cfg else 0.0 | |
| def _context(hour: int, segment: str) -> Dict[str, str | int | None]: | |
| return {"hour": int(hour), "segment": (segment or "").strip() or None} | |
| def ui_set_settings(campaign_id: str, policy: str, holdout: float, stop_min_impr: int, stop_rel_ev: float): | |
| set_campaign_settings(campaign_id, policy, holdout, stop_min_impr, stop_rel_ev) | |
| return f"Updated: policy={policy}, holdout={holdout}, stop_min_impressions={stop_min_impr}, stop_rel_ev_threshold={stop_rel_ev}" | |
| def _uniform_variant(campaign_id: str) -> str | None: | |
| vs = get_variants(campaign_id) | |
| if not vs: return None | |
| return random.choice(vs)["variant_id"] | |
| def ui_serve(campaign_id: str, hour: int, segment: str, aa_min_impr: int): | |
| ctx = _context(hour, segment) | |
| m = _seasonal(campaign_id) | |
| policy = _policy_of(campaign_id) | |
| holdout = _holdout_ratio_of(campaign_id) | |
| mets = get_metrics(campaign_id) | |
| if not mets: | |
| raise gr.Error("配信可能なバリアントがありません。まずは Generate してください。") | |
| # A/Aテスト：各バリアントのimpressionsがしきい値未満なら一様ランダム | |
| if aa_min_impr and aa_min_impr > 0: | |
| for r in mets: | |
| if int(r["impressions"]) < int(aa_min_impr): | |
| vid = _uniform_variant(campaign_id) | |
| row = get_variant(campaign_id, vid) | |
| log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) | |
| ThompsonBandit.update_with_event(campaign_id, vid, "impression") | |
| audit(campaign_id, "serve", {"variant_id": vid, "policy": "AA"}) | |
| return vid, row["text"] | |
| # ホールドアウト：一定確率で一様ランダム | |
| if holdout > 0 and random.random() < float(holdout): | |
| vid = _uniform_variant(campaign_id) | |
| row = get_variant(campaign_id, vid) | |
| log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) | |
| ThompsonBandit.update_with_event(campaign_id, vid, "impression") | |
| audit(campaign_id, "serve", {"variant_id": vid, "policy": "holdout"}) | |
| return vid, row["text"] | |
| # 通常ポリシー | |
| if policy == "linucb": | |
| bandit = LinUCB(campaign_id) | |
| vid, _ = bandit.choose(ctx) | |
| else: | |
| bandit = ThompsonBandit(campaign_id) | |
| vid, _ = bandit.sample_arm(ctx, m.expected_ctr) | |
| if not vid: | |
| raise gr.Error("バリアントが見つかりません。") | |
| row = get_variant(campaign_id, vid) | |
| log_event(campaign_id, vid, "impression", datetime.utcnow().isoformat(), None) | |
| ThompsonBandit.update_with_event(campaign_id, vid, "impression") | |
| audit(campaign_id, "serve", {"variant_id": vid, "policy": policy}) | |
| return vid, row["text"] | |
| def ui_feedback(campaign_id: str, variant_id: str, event_type: str, hour: int, segment: str): | |
| if not variant_id: | |
| raise gr.Error("先に Serve してください。") | |
| ctx = _context(hour, segment) | |
| log_event(campaign_id, variant_id, event_type, datetime.utcnow().isoformat(), None) | |
| ThompsonBandit.update_with_event(campaign_id, variant_id, event_type) | |
| # LinUCBはclickのみで学習（CTRモデル） | |
| if event_type == "click" and _policy_of(campaign_id) == "linucb": | |
| LinUCB(campaign_id).update_click(variant_id, ctx, reward=1.0) | |
| audit(campaign_id, "feedback", {"variant_id": variant_id, "event": event_type}) | |
| return f"{event_type} を記録しました。" | |
| def ui_report(campaign_id: str): | |
| mets = get_metrics(campaign_id) | |
| vpc = get_campaign_value_per_conversion(campaign_id) | |
| rows = [] | |
| for r in mets: | |
| imp = int(r["impressions"]); clk = int(r["clicks"]); conv = int(r["conversions"]) | |
| ctr = (clk / imp) if imp > 0 else 0.0 | |
| cvr = (conv / clk) if clk > 0 else 0.0 | |
| ev = ctr * cvr * vpc | |
| rows.append({ | |
| "variant_id": r["variant_id"], | |
| "impressions": imp, "clicks": clk, "conversions": conv, | |
| "ctr": round(ctr, 4), "cvr": round(cvr, 4), "expected_value": round(ev, 6), | |
| }) | |
| return pd.DataFrame(rows, columns=REPORT_COLUMNS) | |
| def ui_apply_stop(campaign_id: str): | |
| paused = evaluate_stop_rules(campaign_id) | |
| if not paused: | |
| return "No changes." | |
| return "Paused: " + ", ".join([f"{vid}({reason})" for vid, reason in paused]) | |
| def ui_export_csv(campaign_id: str, table: str): | |
| path = export_csv(campaign_id, table) | |
| return path | |
| def ui_reset_db(confirm_text: str): | |
| if (confirm_text or "").strip().upper() != "RESET": | |
| raise gr.Error("タイプミス: RESET と入力してください。") | |
| reset_all() | |
| return "DB was reset." | |
| def ui_adapter_send(campaign_id: str, platform: str, variant_id: str, text: str, hour: int, segment: str): | |
| ctx = _context(hour, segment) | |
| if platform == "x": | |
| ad = XAdapter(campaign_id) | |
| elif platform == "meta": | |
| ad = MetaAdapter(campaign_id) | |
| else: | |
| ad = GoogleAdsAdapter(campaign_id) | |
| res = ad.send(variant_id, text, ctx) | |
| return json.dumps(res, ensure_ascii=False) | |
| with gr.Blocks(title="AdCopy MAB Optimizer", fill_height=True) as demo: | |
| gr.Markdown(""" | |
| # AdCopy MAB Optimizer（HF UI・拡張版） | |
| - 生成→審査→配信（Thompson / LinUCB）→レポート | |
| - A/Aテスト・ホールドアウト、撤退基準 | |
| - CSVエクスポート、DBリセット、外部配信スタブ | |
| """) | |
| with gr.Tab("1) Generate"): | |
| with gr.Row(): | |
| campaign_id = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) | |
| k_variants = gr.Slider(1, 10, value=5, step=1, label="生成本数") | |
| value_per_conv = gr.Number(value=5000, label="value_per_conversion") | |
| brand = gr.Textbox(label="ブランド", value="SFM") | |
| product = gr.Textbox(label="商品/サービス", value="HbA1c測定アプリ") | |
| target = gr.Textbox(label="ターゲット", value="30-50代の健康意識が高い層") | |
| tone = gr.Textbox(label="トーン", value="エビデンス重視で安心感") | |
| ng_words = gr.Textbox(label="NGワード（改行区切り）", value="治る\n奇跡") | |
| btn_gen = gr.Button("広告案を生成＆審査＆保存") | |
| table_gen = gr.Dataframe(headers=GENERATE_COLUMNS, interactive=False) | |
| btn_gen.click(ui_generate, [campaign_id, brand, product, target, tone, k_variants, ng_words, value_per_conv], [table_gen]) | |
| with gr.Tab("2) Settings"): | |
| with gr.Row(): | |
| campaign_id_set = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) | |
| policy = gr.Dropdown(choices=["thompson", "linucb"], value="thompson", label="Policy") | |
| holdout = gr.Slider(0.0, 0.5, value=0.0, step=0.05, label="Holdout ratio") | |
| with gr.Row(): | |
| stop_min_impr = gr.Number(value=200, label="撤退判定の最小impressions") | |
| stop_rel_ev = gr.Slider(0.1, 0.9, value=0.5, step=0.05, label="EVの相対しきい値（劣後停止）") | |
| btn_set = gr.Button("設定を保存") | |
| msg_set = gr.Markdown() | |
| btn_set.click(ui_set_settings, [campaign_id_set, policy, holdout, stop_min_impr, stop_rel_ev], [msg_set]) | |
| with gr.Tab("3) Serve & Feedback"): | |
| with gr.Row(): | |
| campaign_id2 = gr.Textbox(label="campaign_id", value="cmp-demo", scale=1) | |
| hour = gr.Slider(0, 23, value=20, step=1, label="hour") | |
| segment = gr.Textbox(label="segment (任意)") | |
| aa_min_impr = gr.Number(value=0, label="A/Aテスト閾値（各variantのimpressionsがこの数に達するまで均等ランダム）") | |
| btn_serve = gr.Button("Serve Ad（impressionを記録）") | |
| served_vid = gr.Textbox(label="served variant_id", interactive=False) | |
| served_text = gr.Textbox(label="served text", lines=6, interactive=False) | |
| btn_serve.click(ui_serve, [campaign_id2, hour, segment, aa_min_impr], [served_vid, served_text]) | |
| with gr.Row(): | |
| btn_click = gr.Button("Clickを記録") | |
| btn_conv = gr.Button("Conversionを記録") | |
| msg = gr.Markdown() | |
| btn_click.click(lambda cid, vid, h, s: ui_feedback(cid, vid, "click", h, s), [campaign_id2, served_vid, hour, segment], [msg]) | |
| btn_conv.click(lambda cid, vid, h, s: ui_feedback(cid, vid, "conversion", h, s), [campaign_id2, served_vid, hour, segment], [msg]) | |
| gr.Markdown("### 外部配信スタブ") | |
| with gr.Row(): | |
| platform = gr.Dropdown(choices=["x", "meta", "google"], value="x", label="プラットフォーム") | |
| btn_send = gr.Button("この広告を外部に送る（スタブ）") | |
| send_res = gr.Textbox(label="送信結果", lines=3) | |
| btn_send.click(ui_adapter_send, [campaign_id2, platform, served_vid, served_text, hour, segment], [send_res]) | |
| with gr.Tab("4) Report & Ops"): | |
| campaign_id3 = gr.Textbox(label="campaign_id", value="cmp-demo") | |
| with gr.Row(): | |
| btn_rep = gr.Button("レポート更新") | |
| table_rep = gr.Dataframe(headers=REPORT_COLUMNS, interactive=False) | |
| btn_rep.click(ui_report, [campaign_id3], [table_rep]) | |
| with gr.Row(): | |
| btn_stop = gr.Button("撤退基準を適用（自動pause）") | |
| msg_stop = gr.Markdown() | |
| btn_stop.click(ui_apply_stop, [campaign_id3], [msg_stop]) | |
| gr.Markdown("#### CSVエクスポート") | |
| table_sel = gr.Dropdown(choices=["events","metrics","variants","compliance_logs","audit_logs"], value="metrics", label="テーブル") | |
| btn_exp = gr.Button("エクスポート") | |
| file_out = gr.File(label="ダウンロード", interactive=False) | |
| btn_exp.click(ui_export_csv, [campaign_id3, table_sel], [file_out]) | |
| gr.Markdown("#### 管理用：DBリセット（全削除）") | |
| confirm = gr.Textbox(label="タイプ: RESET", value="") | |
| btn_reset = gr.Button("DBを初期化") | |
| msg_reset = gr.Markdown() | |
| btn_reset.click(ui_reset_db, [confirm], [msg_reset]) | |
| if __name__ == "__main__": | |
| demo.queue().launch(server_name="0.0.0.0", server_port=7860) | |