Perplexity / Gemini / OpenAI の API で測る AI 検索の実態 — 「質と競合」の調査実装

from dataclasses import dataclass
from datetime import datetime, timezone
 
@dataclass
class Observation:
    engine: str            # "perplexity" | "openai" | "gemini"
    query: str
    answer_text: str       # 回答本文（後で「何と言われたか」の採点に使う）
    cited_urls: list[str]  # 引用元 URL（後で share of voice に使う）
    fetched_at: str        # ISO 8601 (UTC)
 
def _now() -> str:
    return datetime.now(timezone.utc).isoformat()

import os
import httpx
 
PPLX_ENDPOINT = "https://api.perplexity.ai/v1/sonar"
 
def fetch_perplexity(query: str, model: str = "sonar") -> Observation:
    resp = httpx.post(
        PPLX_ENDPOINT,
        headers={"Authorization": f"Bearer {os.environ['PERPLEXITY_API_KEY']}"},
        json={"model": model, "messages": [{"role": "user", "content": query}]},
        timeout=60,
    )
    resp.raise_for_status()
    return parse_perplexity(query, resp.json())
 
def parse_perplexity(query: str, data: dict) -> Observation:
    urls = [r["url"] for r in data.get("search_results") or []]
    if not urls:
        urls = list(data.get("citations") or [])
    return Observation(
        engine="perplexity",
        query=query,
        answer_text=data["choices"][0]["message"]["content"],
        cited_urls=urls,
        fetched_at=_now(),
    )

from openai import OpenAI
 
def fetch_openai(query: str, model: str = "gpt-5.5") -> Observation:
    client = OpenAI()  # OPENAI_API_KEY
    resp = client.responses.create(
        model=model,
        tools=[{"type": "web_search"}],
        input=query,
    )
    return parse_openai(query, resp)
 
def parse_openai(query: str, resp) -> Observation:
    urls: list[str] = []
    texts: list[str] = []
    for item in resp.output:
        if item.type != "message":
            continue
        for block in item.content:
            if block.type != "output_text":
                continue
            texts.append(block.text)
            for ann in block.annotations or []:
                if ann.type == "url_citation":
                    urls.append(ann.url)
    return Observation("openai", query, "\n".join(texts), urls, _now())

from google import genai
 
def fetch_gemini(query: str, model: str = "gemini-3.5-flash") -> Observation:
    client = genai.Client()  # GEMINI_API_KEY
    interaction = client.interactions.create(
        model=model,
        input=query,
        tools=[{"type": "google_search"}],
    )
    return parse_gemini(query, interaction)
 
def parse_gemini(query: str, interaction) -> Observation:
    urls: list[str] = []
    texts: list[str] = []
    for step in interaction.steps or []:
        if step.type != "model_output":
            continue
        for block in step.content or []:
            if getattr(block, "type", None) != "text":
                continue
            texts.append(block.text)
            for ann in block.annotations or []:
                if ann.type == "url_citation":
                    urls.append(ann.url)
    return Observation("gemini", query, "\n".join(texts), urls, _now())

import pandas as pd
from urllib.parse import urlparse
 
TRACKED = {
    "example.com": "自社",
    "competitor-a.example": "競合A",
    "competitor-b.example": "競合B",
}
 
def to_domain(url: str) -> str:
    host = urlparse(url).netloc.lower()
    return host.removeprefix("www.")
 
def share_of_voice(observations: list[Observation]) -> pd.DataFrame:
    total: dict[str, set[str]] = {}
    rows = []
    for o in observations:
        total.setdefault(o.engine, set()).add(o.query)
        for d in {to_domain(u) for u in o.cited_urls}:
            rows.append({"engine": o.engine, "query": o.query, "domain": d})
    df = pd.DataFrame(rows)
    sov = (
        df.groupby(["engine", "domain"])["query"]
        .nunique().rename("queries_cited").reset_index()
    )
    sov["queries_total"] = sov["engine"].map(lambda e: len(total[e]))
    sov["sov"] = sov["queries_cited"] / sov["queries_total"]
    return (
        sov[sov["domain"].isin(TRACKED)]
        .sort_values(["engine", "sov"], ascending=[True, False])
        .reset_index(drop=True)
    )

def citation_volatility(observations: list[Observation]) -> pd.DataFrame:
    rows = []
    grouped: dict[tuple[str, str], list[set[str]]] = {}
    for o in observations:
        grouped.setdefault((o.engine, o.query), []).append(
            {to_domain(u) for u in o.cited_urls}
        )
    for (engine, query), sets in grouped.items():
        if len(sets) < 2:
            continue
        dists = []
        for i in range(len(sets)):
            for j in range(i + 1, len(sets)):
                union = sets[i] | sets[j]
                dists.append(1 - len(sets[i] & sets[j]) / len(union) if union else 0.0)
        rows.append({"engine": engine, "query": query,
                     "samples": len(sets), "volatility": sum(dists) / len(dists)})
    return pd.DataFrame(rows)

import json
 
STANCE_PROMPT = """\
次の AI 検索の回答テキストの中で「{brand}」がどう扱われているかを判定してください。
回答テキスト:
---
{answer}
---
次のキーを持つ JSON だけを返す:
mentioned (bool), stance ("recommended" / "neutral" / "negative" / "absent"),
claim (回答が {brand} について述べた主張の要約。1文。absent なら空文字)"""
 
def score_stance(obs: Observation, brand: str, model: str = "gpt-5.5") -> dict:
    client = OpenAI()
    resp = client.responses.create(
        model=model,
        input=STANCE_PROMPT.format(brand=brand, answer=obs.answer_text),
    )
    return json.loads(resp.output_text)

create table ai_search_observations (
  id bigint generated always as identity primary key,
  engine text not null,
  query text not null,
  answer_text text,
  cited_urls jsonb,
  fetched_at timestamptz not null
);

def gsc_calibration(observations: list[Observation], gsc_pages_csv: str) -> pd.DataFrame:
    api = pd.DataFrame([
        {"page": u, "query": o.query}
        for o in observations if o.engine == "gemini"
        for u in o.cited_urls
    ])
    api_pages = (
        api.groupby("page")["query"].nunique()
        .rename("api_cited_queries").reset_index()
    )
    gsc = pd.read_csv(gsc_pages_csv)  # 列: page, impressions（エクスポート時の列名に合わせる）
    merged = api_pages.merge(gsc, on="page", how="outer").fillna(0)
    merged["api_rank"] = merged["api_cited_queries"].rank(ascending=False)
    merged["gsc_rank"] = merged["impressions"].rank(ascending=False)
    merged["rank_gap"] = (merged["api_rank"] - merged["gsc_rank"]).abs()
    return merged.sort_values("rank_gap", ascending=False)