AI 検索が広げたクエリを分母から外して、実質 CTR を出し直します

# fanout_score.py — クエリ文字列だけで「AI 検索の展開クエリらしさ」を採点する
# クリック数・表示回数は一切参照しない（循環参照を避けるため）
import csv
import re
import sys
from dataclasses import dataclass, field
 
# 属性を並列で繋ぐ機能語。人間の検索語では出現頻度が低い
JOINERS = (" vs ", " versus ", " for ", " and ", " with ", " without ")
 
# 人間が困って打つときに現れる語。出たら強く減点する
INTENT_MARKERS = (
    "error", "fix", "not working", "failed", "how to", "why", "cannot", "can't",
    "エラー", "できない", "直し方", "とは", "使い方",
)
 
# 末尾の年号は「最新情報が欲しい人間」のシグナルとして扱う
YEAR = re.compile(r"\b(19|20)\d{2}\b")
 
# 「何ができるか」を列挙させる語。展開クエリ側に偏る
ATTRIBUTE_NOUNS = re.compile(
    r"\b(features|capabilities|use cases|target audience|overview|benefits)\b"
)
 
 
@dataclass
class Verdict:
    query: str
    score: int
    band: str
    reasons: list = field(default_factory=list)
 
 
def fanout_score(query: str) -> Verdict:
    q = query.lower().strip()
    tokens = q.split()
    score, reasons = 0, []
 
    if len(tokens) >= 6:
        score += 1
        reasons.append(f"tokens>=6 ({len(tokens)})")
    if len(tokens) >= 8:
        score += 1
        reasons.append(f"tokens>=8 ({len(tokens)})")
 
    joiners = [j.strip() for j in JOINERS if j in f" {q} "]
    if joiners:
        score += 1
        reasons.append("joiner:" + "/".join(joiners))
 
    hits = [m for m in INTENT_MARKERS if m in q]
    if hits:
        score -= 2
        reasons.append("intent:" + "/".join(hits))
 
    if YEAR.search(q):
        score -= 1
        reasons.append("year-suffix")
 
    if ATTRIBUTE_NOUNS.search(q):
        score += 1
        reasons.append("attribute-noun")
 
    # 3 点以上は展開クエリ、1 点以下は人間、2 点は保留
    band = "fanout" if score >= 3 else ("human" if score <= 1 else "ambiguous")
    return Verdict(query=query, score=score, band=band, reasons=reasons)
 
 
def main(path: str) -> None:
    with open(path, newline="", encoding="utf-8-sig") as f:
        for row in csv.DictReader(f):
            v = fanout_score(row["Top queries"])
            print(f"{v.band:9s} score={v.score:2d}  {v.query}")
            print(f"            {v.reasons}")
 
 
if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "Queries.csv")

ambiguous score= 2  rork.com core business features target audience
            ['tokens>=6 (6)', 'attribute-noun']
fanout    score= 3  rork max swiftui features and ai capabilities
            ['tokens>=6 (7)', 'joiner:and', 'attribute-noun']
ambiguous score= 2  onspace vs rork for complex app logic
            ['tokens>=6 (7)', 'joiner:vs/for']
human     score= 1  expo audio continue playing screen locked react native 2025
            ['tokens>=6 (9)', 'tokens>=8 (9)', 'year-suffix']
fanout    score= 4  rork max swiftui features and native app generation
            ['tokens>=6 (8)', 'joiner:and', 'attribute-noun', 'tokens>=8 (8)']

# adjudicate.py — ambiguous 判定のクエリだけをモデルに回す
from google import genai
from pydantic import BaseModel
 
client = genai.Client(api_key="YOUR_API_KEY")
 
SCHEMA_PROMPT = """あなたは検索ログの分析者です。
与えられたクエリが、人間が検索窓に入力した語である可能性を判定してください。
 
判定基準:
- 人間の検索語は短く、必要な語だけで構成される傾向がある
- 属性名の列挙（features / capabilities / target audience 等）や、
  比較軸を文中に埋め込んだ表現は、AI が内部で生成した下位クエリに多い
- ただし、エラー文をそのまま貼り付けた長いクエリは人間側である
 
クリック数や表示回数は与えません。語の並びだけで判定してください。"""
 
 
class Judgement(BaseModel):
    likely_human: bool
    confidence: float  # 0.0-1.0
    reason: str
 
 
def adjudicate(query: str) -> Judgement:
    response = client.models.generate_content(
        model="gemini-3.7-flash",
        contents=f"{SCHEMA_PROMPT}\n\nクエリ: {query}",
        config={
            "response_mime_type": "application/json",
            "response_schema": Judgement,
        },
    )
    return Judgement.model_validate_json(response.text)
 
 
AMBIGUOUS = [
    "rork.com core business features target audience",
    "onspace vs rork for complex app logic",
]
 
for q in AMBIGUOUS:
    j = adjudicate(q)
    # confidence が低い判定は採用せず、保留のまま次週へ持ち越す
    verdict = "human" if j.likely_human else "fanout"
    status = verdict if j.confidence >= 0.7 else "hold"
    print(f"{status:7s} conf={j.confidence:.2f}  {q}\n        {j.reason}")