proactive-intelligence

プロアクティブインテリジェンス

Agent Skills™ の一部です。googleadsagent.ai™ より

説明

プロアクティブインテリジェンスにより、エージェントはユーザーからの明示的な依頼を待たずに、分析中にウェブ検索、API の再取得、データの鮮度チェックなどの外部情報を自律的に探し出せます。従来のリアクティブなエージェントは、提供されたデータだけを使って動作します。プロアクティブなエージェントは、現在のコンテキストが不十分である、古くなっている、または矛盾していると判断すると、知識の不足を補うために自律的に行動します。これにより、エージェントは受動的な処理装置から能動的な調査者へと変わり、より正確で、最新かつ包括的な結果を提供できます。

このスキルは、googleadsagent.ai™ の Buddy™ エージェントにおける search_web ツールの統合をモデルにしています。このエージェントは、競合他社のデータ、業界ベンチマーク、最近の Google Ads ポリシー変更、プラットフォームの更新情報が分析の改善に役立つと判断すると、自律的に検索します。Buddy™ は、これまで見たことのないキャンペーン戦略に遭遇した場合や、指標が予想範囲から大きく外れている場合、推測するのではなく、背景情報を積極的に検索します。この動作は明示的な条件によって起動し、無作為な好奇心によるものではありません。そのため、追加の遅延やコストに見合う価値が確保されます。

プロアクティブインテリジェンスのフレームワークは、トリガー、検索、統合のサイクルで動作します。エージェントは分析中にトリガー条件を評価し、条件が満たされると対象を絞った検索を実行し、結果の関連性と鮮度を評価して、検証済みの情報を継続中の推論に統合します。信頼度スコアリングにより、十分な裏付けのある結論と推測に基づく結論をエージェントが区別できます。

使用する場面

• 最近の外部変化（ポリシーの更新、市場の変化など）の影響を受ける可能性があるデータをエージェントが分析する場合
• 社内データの分析に加えて、競合情報が必要な場合
• エージェントが、説明のために外部の背景情報を必要とする予想外のパターンに遭遇した場合
• データの鮮度が重要で、提供されたデータが古くなっている可能性がある場合
• パフォーマンスを評価するために、業界ベンチマークやベストプラクティスが必要な場合
• 現在の情報源に照らして、エージェントが自身の仮定をファクトチェックする必要がある場合

仕組み

graph TD
    A[Agent Analysis] --> B[Trigger Evaluator]
    B -->|No Trigger| C[Continue Analysis]
    B -->|Trigger Fired| D[Search Strategy Selector]
    D --> E{Search Type}
    E -->|Web Search| F[search_web Tool]
    E -->|API Re-pull| G[Data Refresh]
    E -->|Benchmark Lookup| H[Industry Database]
    F --> I[Result Scorer]
    G --> I
    H --> I
    I --> J{Relevance Score}
    J -->|High| K[Integrate into Analysis]
    J -->|Low| L[Discard + Note]
    K --> M[Update Confidence Score]
    L --> M
    M --> C

トリガー評価器は分析中に継続して実行され、あらかじめ定義された条件を確認します。条件には、データの異常（指標が予想範囲から外れている）、知識の不足（不慣れな戦略や用語への遭遇）、鮮度の指標（データがしきい値より古い）、明示的な検索の手がかり（ユーザーが競合他社に言及する、または「これは正常ですか」と尋ねる）が含まれます。トリガーが発動すると、検索戦略の選択器が最も効率的な情報収集方法を決定します。結果は統合前に関連性と鮮度について評価され、プロアクティブ検索によって結論の裏付けが強まったか弱まったかを反映するよう、エージェントの信頼度スコアが更新されます。

実装

トリガー条件エンジン：

class TriggerEngine:
    def __init__(self):
        self.triggers = [
            AnomalyTrigger(),
            StalenessTrigger(max_age_days=7),
            KnowledgeGapTrigger(),
            CompetitorMentionTrigger(),
        ]

    def evaluate(self, context: dict) -> list[dict]:
        fired = []
        for trigger in self.triggers:
            result = trigger.check(context)
            if result["fired"]:
                fired.append(result)
        return sorted(fired, key=lambda t: t["priority"], reverse=True)


class AnomalyTrigger:
    THRESHOLDS = {
        "ctr": (0.005, 0.15),
        "cpc": (0.10, 50.00),
        "conversion_rate": (0.005, 0.30),
        "roas": (0.5, 20.0),
    }

    def check(self, context: dict) -> dict:
        metrics = context.get("campaign_metrics", {})
        anomalies = []
        for metric, (low, high) in self.THRESHOLDS.items():
            value = metrics.get(metric)
            if value is not None and (value < low or value > high):
                anomalies.append({
                    "metric": metric,
                    "value": value,
                    "expected_range": (low, high),
                })
        return {
            "fired": len(anomalies) > 0,
            "trigger": "anomaly",
            "priority": 0.9,
            "details": anomalies,
            "search_query": self.build_query(anomalies) if anomalies else None,
        }

    def build_query(self, anomalies: list) -> str:
        metrics = [a["metric"] for a in anomalies]
        return f"Google Ads {' '.join(metrics)} unusual values industry benchmark 2026"


class StalenessTrigger:
    def __init__(self, max_age_days=7):
        self.max_age_days = max_age_days

    def check(self, context: dict) -> dict:
        data_date = context.get("data_date")
        if not data_date:
            return {"fired": False, "trigger": "staleness"}
        age_days = (datetime.utcnow() - datetime.fromisoformat(data_date)).days
        return {
            "fired": age_days > self.max_age_days,
            "trigger": "staleness",
            "priority": 0.7,
            "details": {"age_days": age_days, "threshold": self.max_age_days},
            "search_query": f"Google Ads recent changes updates {datetime.utcnow().strftime('%B %Y')}",
        }

プロアクティブ検索の実行器：

class ProactiveSearchAgent:
    def __init__(self, search_tool, model, max_searches_per_session=5):
        self.search_tool = search_tool
        self.model = model
        self.searches_remaining = max_searches_per_session
        self.search_log = []

    async def search_if_triggered(self, context: dict) -> list[dict]:
        if self.searches_remaining <= 0:
            return []

        engine = TriggerEngine()
        triggers = engine.evaluate(context)
        if not triggers:
            return []

        results = []
        for trigger in triggers[:2]:
            if self.searches_remaining <= 0:
                break
            query = trigger.get("search_query")
            if not query:
                continue

            search_results = await self.search_tool.execute(query)
            scored = self.score_results(search_results, trigger)
            relevant = [r for r in scored if r["relevance"] > 0.5]

            self.search_log.append({
                "trigger": trigger["trigger"],
                "query": query,
                "results_found": len(search_results),
                "relevant_results": len(relevant),
            })
            self.searches_remaining -= 1
            results.extend(relevant)

        return results

    def score_results(self, results: list, trigger: dict) -> list[dict]:
        scored = []
        for result in results:
            freshness = self.freshness_score(result.get("date"))
            relevance = self.relevance_score(result.get("snippet", ""), trigger)
            scored.append({
                **result,
                "freshness": freshness,
                "relevance": relevance,
                "combined_score": freshness * 0.3 + relevance * 0.7,
            })
        return sorted(scored, key=lambda r: r["combined_score"], reverse=True)

    def freshness_score(self, date_str: str | None) -> float:
        if not date_str:
            return 0.5
        try:
            age = (datetime.utcnow() - datetime.fromisoformat(date_str)).days
            return max(0, 1 - (age / 365))
        except ValueError:
            return 0.5

    def relevance_score(self, text: str, trigger: dict) -> float:
        keywords = trigger.get("search_query", "").lower().split()
        text_lower = text.lower()
        matches = sum(1 for kw in keywords if kw in text_lower)
        return min(1.0, matches / max(len(keywords), 1))

分析パイプラインへの統合：

async def analyze_with_proactive_intelligence(campaign_data: dict, user_query: str):
    search_agent = ProactiveSearchAgent(search_tool=WebSearchTool(), model=model)

    context = {
        "campaign_metrics": campaign_data.get("metrics", {}),
        "data_date": campaign_data.get("date"),
        "user_query": user_query,
    }

    proactive_results = await search_agent.search_if_triggered(context)

    knowledge_block = ""
    if proactive_results:
        knowledge_block = "\n<proactive_research>\n"
        for r in proactive_results[:5]:
            knowledge_block += f"Source: {r.get('url', 'N/A')}\n{r.get('snippet', '')}\n\n"
        knowledge_block += "</proactive_research>\n"

    messages = [
        {"role": "system", "content": f"""You are Buddy™ by googleadsagent.ai™.
{knowledge_block}
Analyze the campaign data and provide recommendations. If proactive research is included above, integrate those findings into your analysis and cite sources."""},
        {"role": "user", "content": json.dumps(campaign_data) + f"\n\n{user_query}"},
    ]

    return await model.generate(messages)

ベストプラクティス

1. 明確なトリガー条件を定義する — プロアクティブ検索は、ランダムに、またはすべてのクエリで実行するのではなく、具体的で妥当な理由（異常、情報の古さ、ナレッジの不足）がある場合に実行します。
2. セッションごとの検索回数に上限を設ける — レイテンシとコストを抑えるため、セッションごとのプロアクティブ検索を3～5回に制限し、トリガーの重要度に応じて優先順位を付けます。
3. 関連性と鮮度に基づいて結果を評価する — すべての検索結果が有用とは限りません。エージェントのコンテキストに追加する前に、厳格にフィルタリングします。
4. すべてのプロアクティブ検索を記録する — 検索のトリガー、使用したクエリ、結果が実際に役立ったかどうかを記録し、事後の最適化に活用します。
5. プロアクティブ検索で得た情報とユーザー提供のデータを明確に区別する — XML タグや明確なラベルを使って、エージェントが収集した情報とユーザーの入力を区別します。
6. プロアクティブ検索の情報源を引用するようエージェントに指示する — 引用した情報源に基づく推奨事項は、より信頼性が高く、監査可能です。
7. フィードバックに基づいてトリガーのしきい値を調整する — プロアクティブ検索で無関係な結果が頻繁に返される場合は、トリガー条件を厳しくします。

プラットフォームの互換性

機能	Claude Code	Cursor	Codex	Gemini CLI
Web 検索ツール	✅ MCP／カスタム	✅ MCP／組み込み	⚠️ カスタム	✅ 組み込み
トリガーの評価	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
結果の統合	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
検索回数の上限	✅ カスタム	✅ カスタム	✅ カスタム	✅ カスタム
引用の追跡	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応

Mythos Preview の参考情報

Mythos Preview の足場となる仕組みは、エンドツーエンドの自律的な連鎖を重視しています。簡単な初期プロンプトの後、モデルはコードから仮説を立て、プログラムを実行（デバッガーなどのツールも使用）して仮説を確認または否定し、必要に応じてこれを繰り返してから、構造化された結論を出します。ステップごとの人間による承認を必要とせず、発見 → 活用 → トリアージを1つのフローで行います。

プロアクティブエージェントを設計する際は、このフローを取り入れてください。明示的なトリガーによって現実を収集または再確認（実行、テスト、サンドボックス）し、その後、証拠を分析に反映することで、連鎖が推測ではなく事実に基づくようにします。出典：Mythos Preview。

関連スキル

• ナレッジベースの注入 - プロアクティブ検索は、静的なナレッジベースでは補えないナレッジの不足を埋めます
• 継続的な学習 - 成功したプロアクティブ検索のパターンを記録し、再利用可能なトリガーとして体系化します
• 競合分析 - リアルタイムの競合情報は、プロアクティブ検索を実行する主なトリガーです
• コンバージョンのトラッキング - データの異常検知によって、トラッキングの問題に対するプロアクティブな調査を開始します

キーワード

プロアクティブインテリジェンス、自律検索、トリガー条件、データの鮮度、Web 検索、結果の評価、ナレッジの不足、異常検知、競合情報、エージェントスキル

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License