LoginSignup
8
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Day 24

Supershipグループ Advent Calendar 2025

@miyay(yuma miyake)inSupership株式会社

AdCPのSignals Protocolを実装してセグメントをAI検索する

Last updated at Posted at 2025-12-23

アドテクノロジーセンター 副センター長の三宅です。

今日はAI Agent向けの広告運用プロトコルであるAdCPの中のオーディエンスセグメント検索コマンドを実装します

この記事は Supershipグループ Advent Calendar 2025 の 23日目の記事です。

TL;DR

  • AdCPとその中のSignals Protocolの説明
  • BigQueryとCloud Run Functionsで最小限実装をする場合の例

AdCP(Ad Content Protocol) について

AdCPは広告運用をAI経由で実行する事を目的としたプロトコルです。

概要は以下を参照してください。

具体的には 次の機能 を提供します

  • 広告商品と在庫を発見
  • メディア購入の作成と管理
  • クリエイティブアセットのアップロードと割り当て
  • キャンペーンの実施状況を監視する
  • オーディエンスシグナルの管理
  • ターゲット設定と最適化の処理
Your Application
       ↓
[Choose One Protocol]
       ↓
   MCP or A2A
       ↓
  AdCP Tasks
       ↓
Ad Platforms (Google, Meta, Amazon, etc.)

Signals Protocol

AdCP本家の説明を読んでもらった方が早いので自動翻訳したものをほぼそのまま貼りますが、データ配信におけるセグメント検索の課題は会社や国をまたいでも同じ状況のようです。

課題:数十万ものセグメントを乗り越える方法

最新のデータプラットフォームは驚くほどの深さを提供しており、多くの場合、数十万もの事前構築されたセグメントを持っています。これにより前例のないターゲティングの選択肢が提供されますが、同時に現実的な課題も生み出します。

・限られた可視性:提供者によって出所や手法は異なります
・指標の不一致:各プラットフォームが品質とパフォーマンスの報告方法が異なる
・命名規則:セグメント名は必ずしもその内容を明確に伝えるわけではありません
・ディスカバリー・フリクション:適切なセグメントを見つけるには、深いプラットフォームの専門知識が必要です
・固定カタログ:既製品セグメントがキャンペーンのニーズに正確に合致しない場合もあります

現在のアプローチ
Human → Navigate platform UIs → Select from available → Activate and test

AdCPを使ったアプローチ
Human describes need → AI evaluates all options → Providers can create custom → Streamlined activation

セグメントは誰でも(有償で)利用できるパブリックデータと、個別カスタマイズ等を行ったプライベートデータに分けられるため、Signals Protocolには認証が用意されています。

  • 認証なし
    • パブリックデータのみ
  • 認証あり
    • プライベート + パブリックデータ

get_signals

get_signals は自然言語でセグメントを探すAPIです。

リクエスト

{
  "tool": "get_signals",
  "arguments": {
    "signal_spec": "High-income households interested in luxury goods", // 自然言語による欲しいセクションの指示(必須)
    "deliver_to": { // 検索条件(必須)
      "deployments": [
        {
          "type": "agent", // DSP向けの場合: platform, 代理店向けの場合: agent(必須)
          "platform": "the-trade-desk", // プラットフォーム識別子(typeがplatformの場合必須)
          "agent_url": "https://wonderstruck.salesagents.com", // 代理店識別用URL(typeがagentの場合必須)
          "account": "hoge" // プラットフォームや代理店のアカウント識別子(任意)
        }
      ],
      "countries": ["US"] // 国(必須)
    },
    "filters": { // フィルター条件(任意)
      "max_cpm": 5.0, // 最大価格(任意)
      "catalog_types": ["marketplace"], // カタログタイプ("marketplace", "custom", "owned")(任意)
      "data_providers": "hoge", // データプロバイダー(任意)
      "min_coverage_percentage": 50 // 最低カバレッジ(任意)
    },
    "max_results": 5 // 最大結果数(任意)
  }
}

レスポンス

{
  "message": "Found 1 luxury segment matching your criteria. Already activated for your sales agent.", // 人間向けの要約
  "context_id": "ctx-signals-123", // セッションID
  "signals": [
    {
      "signal_agent_segment_id": "luxury_auto_intenders", // セグメントID
      "name": "Luxury Automotive Intenders", // 人間向けのセグメント名称
      "description": "High-income individuals researching luxury vehicles", // セグメントの概要
      "signal_type": "marketplace", // セグメントの種類(marketplace, custom, owned)
      "data_provider": "Experian", // データプロバイダー名
      "coverage_percentage": 12, // カバレッジの割合
      "deployments": [
        {
          "type": "agent", // agent or platform
          "agent_url": "https://wonderstruck.salesagents.com", // エージェント識別URL
          "is_live": true, // セグメントが有効化どうか
          "activation_key": { // セグメント指定。後述の2種類指定方法がある
            "type": "key_value",
            "key": "audience_segment",
            "value": "luxury_auto_intenders_v2"
          }
        }
      ],
      "pricing": {
        "cpm": 3.50, // CPM
        "currency": "USD" // 通貨コード
      }
    }
  ]
}

セグメントIDで指定する場合

{
  "type": "segment_id",
  "segment_id": "ttd_segment_12345"
}

Key-Valueで指定する場合

{
  "type": "key_value",
  "key": "audience_segment",
  "value": "luxury_auto_intenders"
}

実装

今回はPoCとして簡易的にget_signalsを実装してみます。

AdCPの解釈が間違っている可能性がある点と、今後仕様が変更になる可能性がある点はご了承ください。

前提

基盤の前提

  • セグメント一覧がBigQuery上にある
    • セグメントIDとセグメント名を持つ
      • 実際のDMPやDSP上のセグメントはもっと複雑なデータを持ちますが、説明の簡単化のためにセグメント名に一致するかどうかで判断します
  • AdCPのAPIはCloud Run Functionsで動作させる想定

また、PoCのため以下の制約を持ちます

  • 本来はMCPとして実装されるが、今回はHTTPで実装
  • 権限制御、データ利用料、カバレッジ率、UU数は特に考慮しない
    • よって、 signal_spec のみを考慮する
    • ただしこれらはAIの利用するクエリや評価指標を変更すれば対応できるため、あまり難しい問題ではないと思う
  • 複雑なワークフローは組まない
    • 実際にサービスに組み込む場合はADK等を使って、自然言語から単語を抽出 → 類義語拡張 → セグメント検索 → 評価 のようなワークフローを組むといいと思います
    • 公式サンプル実装では RAG (semantic), FTS (keyword), and Hybrid search modes が選べると書いてあります
  • AIの応答結果の振れは特に訂正しない
    • AIの回答は指示と異なる事があるため、実運用時は訂正が必要です

実装

全部書くと長いので細かい部分は省略します。
実際に使うには適宜例外処理や認証を追加してください。

エンドポイント

import json
import os
import uuid
from google.cloud import bigquery
from google import genai
from google.genai import types
import functions_framework
from flask import jsonify

@functions_framework.http
def get_signals(request):
    """AdCP get_signals APIエンドポイント"""
    # リクエストデータを取得
    request_data = request.get_json()

    # 自社DSPへのリクエストのみ処理
    deploy_reqs = request_data.get("deliver_to", {}).get("deployments", [])
    is_my_dsp = any(
        d.get("type") == "platform" and d.get("platform") == "my_dsp"
        for d in deploy_reqs
    )
    if not is_my_dsp:
        return jsonify({"error": "Only my_dsp platform is supported"}), 400

    signal_spec = request_data.get("signal_spec")
    max_results = request_data.get("max_results", 10)

    # Geminiでセグメント検索
    gemini_result = run_gemini_conversation(signal_spec)

    # 結果を最大件数に制限
    gemini_result["selected_segments"] = gemini_result["selected_segments"][:max_results]

    # AdCP形式のレスポンスを構築
    response = build_adcp_response(gemini_result)

    return jsonify(response)

AIによる分析

Gemini 3.0 Proを利用していますが、ローカルで試した感じではClaude Opus 4.5の方がより複雑な提案をしてきたのでClaudeが使えるならClaudeを使った方がいいと思います。最初はGemini 2.5 Flashで実行していましたが途中で回答を放棄するので駄目です。

GCP_PROJECT = os.environ.get("GCP_PROJECT")
GEMINI_REGION = os.environ.get("GEMINI_REGION")
BQ_PROJECT = os.environ.get("BQ_PROJECT")

# Gemini クライアント
gemini_client = genai.Client(
    vertexai=True,
    project=GCP_PROJECT,
    location=GEMINI_REGION,
)

# ツール定義（Gemini形式）
search_segment_tool = types.Tool(
    function_declarations=[
        types.FunctionDeclaration(
            name="search_segments",
            description="セグメントを名前で検索します。",
            parameters=types.Schema(
                type=types.Type.OBJECT,
                properties={
                    "keyword": types.Schema(
                        type=types.Type.STRING,
                        description="検索キーワード"
                    )
                },
                required=["keyword"]
            )
        )
    ]
)

# システムプロンプト
SYSTEM_PROMPT = """あなたは広告配信システムのオークションセグメントのレコメンドの専門家です。
ユーザーの自然言語入力から適切なセグメントを検索し、JSON形式で返してください。

## 手順

1. 入力からキーワードを抽出
2. search_segments ツールで検索
3. 関連性の高いセグメントを選定

## 出力形式

`` `json
{
  "selected_segments": [
    {"id": "セグメントID", "name": "セグメント名", "description": "選定理由", "signal_type": "marketplace"}
  ],
  "summary": "検索結果の要約"
}
`` `
"""

def run_gemini_conversation(signal_spec: str) -> dict:
    """Geminiとの会話を実行してセグメントを検索"""
    contents = [types.Content(role="user", parts=[types.Part.from_text(text=signal_spec)])]

    generate_config = types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        tools=[search_segment_tool],
        temperature=0.7,
        max_output_tokens=4096,
    )

    while True:
        response = gemini_client.models.generate_content(
            model="gemini-3-pro-preview",
            contents=contents,
            config=generate_config,
        )

        candidate = response.candidates[0]
        parts = candidate.content.parts
        function_calls = [p.function_call for p in parts if p.function_call]

        if not function_calls:
            # テキストレスポンスをJSONとしてパース
            text_parts = [p.text for p in parts if p.text]
            response_text = "\n".join(text_parts)
            return json.loads(response_text.strip())

        # Function Callを処理
        contents.append(candidate.content)
        function_responses = []
        for fc in function_calls:
            keyword = fc.args.get("keyword", "")
            result = search_segments(keyword)
            function_responses.append(
                types.Part.from_function_response(name=fc.name, response=result)
            )
        contents.append(types.Content(role="user", parts=function_responses))

セグメント検索

実際のデータに合わせて変更します。

ここではシンプルにキーワードがセクション名に含まれていたら対象とすることにします。
実際に実装する場合はステータスやUU数、権限を考慮する必要があります。

bq_client = bigquery.Client(project=BQ_PROJECT)

def search_segments(keyword: str) -> dict:
    """BigQueryでセグメントを検索"""
    query = """
    SELECT id, name
    FROM `pj-sandbox.dsp.segments`
    WHERE LOWER(name) LIKE LOWER(@keyword)
    LIMIT 20
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("keyword", "STRING", f"%{keyword}%")
        ]
    )
    results = bq_client.query(query, job_config=job_config).result()
    return {"results": [dict(row) for row in results]}

レスポンスを構築

def build_adcp_response(gemini_result: dict) -> dict:
    """Geminiの結果をAdCP形式のレスポンスに変換"""
    context_id = str(uuid.uuid4())

    signals = []
    for segment in gemini_result.get("selected_segments", []):
        signal = {
            "signal_agent_segment_id": str(segment.get("id", "")),
            "name": segment.get("name", ""),
            "description": segment.get("description", ""),
            "signal_type": "marketplace",
            "data_provider": "my_dsp",
            "coverage_percentage": 0,
            "pricing": {"cpm": 0, "currency": "JPY"},
            "deployments": [{
                "platform": "my_dsp",
                "is_live": True,
                "activation_key": {
                    "type": "segment_id",
                    "segment_id": str(segment.get("id", ""))
                }
            }]
        }
        signals.append(signal)

    return {
        "message": gemini_result.get("summary", f"Found {len(signals)} segments"),
        "context_id": context_id,
        "data": {"signals": signals}
    }

動作テスト

AdCP用の公式テストツールがあるのですが執筆時点では Signals Protocol に対応していないため、ああJSONが返ってくるなぁというのしか確認できません。

APIの結果が登録されているセグメントデータに依存するためここには貼りませんが、AI Modelが入力された文章から適切なキーワードを提案できれば良好なAI検索結果が得られそうです。
文章からのキーワード提案だけであればローカルで実行できるためみんな気軽にプロンプトを書いてみるといいと思います。

所感

広告運用経験者である方は分かると思いますが運用画面はとにかくレバーが多いうえに、新しい媒体が生まれるたびに機能が付け足され続けてきたため軍艦島のような構造になっています。
設定項目の多さ ≒ 運用の柔軟さはそれぞれのサービスの強みであるためこれが統一されることはないかと思いますが、AdCPはそこをLLMの力を使って差分を吸収させ運用を統一化する世界を目指しているのかと思いました。　

広告配信は結局はどれくらい多くの人間に訴求できるかであり、それは明確な正解があるわけではなく人間の感覚に委ねられるところです。
統計的な判断となる配信属性の指定はAIと相性が良いように思いました。

AdCPに関する情報が少なすぎるので、今後日本語記事も増えることを期待します。

最後に宣伝です。
Supershipではプロダクト開発やサービス開発に関わる人を絶賛募集しております。
ご興味がある方は以下リンクよりご確認ください。
Supership 採用サイト
是非ともよろしくお願いします。

8
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
8
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?