Back to Details

proactive-intelligence

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Proactive Intelligence

主动智能

Part of Agent Skills™ by googleadsagent.ai™
属于由 googleadsagent.ai™ 提供的 Agent Skills™

Description

描述

Proactive Intelligence enables agents to autonomously seek out external information — web searches, API re-pulls, data freshness checks — during analysis without waiting for explicit user requests. Traditional reactive agents only work with the data provided to them. Proactive agents recognize when their current context is insufficient, stale, or contradictory, and take independent action to fill knowledge gaps. This transforms the agent from a passive processor into an active investigator that delivers more accurate, more current, and more comprehensive results.
This skill is modeled on the 
search_web
tool integration in the Buddy™ agent at googleadsagent.ai™, where the agent autonomously searches for competitor data, industry benchmarks, recent Google Ads policy changes, and platform updates when it detects that such information would improve its analysis. When Buddy™ encounters a campaign strategy it hasn't seen before, or metrics that deviate significantly from expected ranges, it proactively searches for context rather than speculating. This behavior is triggered by explicit conditions, not random curiosity, ensuring the additional latency and cost are justified.
The proactive intelligence framework operates on a trigger-search-integrate cycle: the agent evaluates trigger conditions during analysis, dispatches targeted searches when conditions are met, scores the relevance and freshness of results, and integrates verified findings into its ongoing reasoning. Confidence scoring ensures the agent distinguishes between well-supported conclusions and speculative ones.
主动智能使 Agent 能够在分析期间自主查找外部信息——Web 搜索、API 重新拉取、数据新鲜度检查——无需等待用户明确提出请求。传统的反应型 Agent 只处理提供给它们的数据。主动式 Agent 会识别当前上下文何时存在信息不足、数据过期或相互矛盾之处，并独立采取行动填补知识缺口。这会将 Agent 从被动处理器转变为主动调查者，从而给出更准确、更及时、更全面的结果。
此 Skill 的设计参考了 googleadsagent.ai™ 上 Buddy™ Agent 中的 
search_web
工具集成。在该 Agent 中，当检测到某些信息有助于改进分析时，它会自主搜索竞争对手数据、行业基准、近期 Google Ads 政策变更和平台更新。当 Buddy™ 遇到此前未见的推广策略，或发现指标明显偏离预期范围时，会主动搜索上下文，而不是凭猜测作答。这种行为由明确条件触发，而非出于随机好奇，从而确保额外的延迟和成本物有所值。
主动智能框架以“触发—搜索—整合”循环运作：Agent 会在分析期间评估触发条件，在条件满足时分发有针对性的搜索，对结果的相关性和新鲜度进行评分，并将经过验证的发现整合进持续进行的推理。置信度评分可确保 Agent 区分证据充分的结论与推测性结论。

Use When

适用场景

  • The agent analyzes data that may be affected by recent external changes (policy updates, market shifts)
  • Competitive intelligence is needed alongside internal data analysis
  • The agent encounters unexpected patterns that need external context to explain
  • Data freshness is critical and the provided data may be outdated
  • Industry benchmarks or best practices are needed to contextualize performance
  • The agent must fact-check its own assumptions against current sources
  • Agent 分析的数据可能受近期外部变化影响（政策更新、市场转变）
  • 除内部数据分析外，还需要竞争情报
  • Agent 遇到需要外部上下文解释的异常模式
  • 数据新鲜度至关重要，而所提供的数据可能已过期
  • 需要行业基准或最佳实践来解读绩效
  • Agent 必须根据当前来源对自己的假设进行事实核查

How It Works

工作原理

mermaid
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
The trigger evaluator runs continuously during analysis, checking predefined conditions: data anomalies (metrics outside expected ranges), knowledge gaps (encountering unfamiliar strategies or terms), staleness indicators (data older than a threshold), and explicit search cues (user mentions competitors or asks "is this normal"). When a trigger fires, the search strategy selector determines the most efficient information-gathering approach. Results are scored for relevance and freshness before integration, and the agent's confidence score is updated to reflect whether the proactive search strengthened or weakened its conclusions.
mermaid
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
触发器评估器在分析期间持续运行，检查预定义条件：数据异常（指标超出预期范围）、知识缺口（遇到不熟悉的策略或术语）、过期指标（数据早于阈值日期），以及明确的搜索线索（用户提及竞争对手或询问“这正常吗”）。当触发器触发时，搜索策略选择器会确定最高效的信息收集方式。结果在整合前会按相关性和新鲜度评分，并更新 Agent 的置信度分数，以反映主动搜索是强化还是削弱了其结论。

Implementation

实现

Trigger Condition Engine:
python
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')}",
        }
Proactive Search Executor:
python
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))
Integration into Analysis Pipeline:
python
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)
触发条件引擎：
python
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')}",
        }
主动搜索执行器：
python
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))
集成到分析流水线：
python
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)

Best Practices

最佳实践

  1. Define explicit trigger conditions — proactive search should fire for specific, justified reasons (anomalies, staleness, knowledge gaps), not randomly or on every query.
  2. Cap searches per session — limit to 3-5 proactive searches per session to control latency and cost; prioritize by trigger importance.
  3. Score results for relevance and freshness — not all search results are useful; filter aggressively before injecting into the agent's context.
  4. Log every proactive search — record what triggered the search, the query used, and whether the results were actually useful for post-hoc optimization.
  5. Clearly separate proactive findings from user-provided data — use XML tags or clear labels to distinguish agent-gathered information from user input.
  6. Instruct the agent to cite proactive sources — recommendations grounded in cited sources are more trustworthy and auditable.
  7. Adjust trigger thresholds based on feedback — if proactive searches frequently return irrelevant results, tighten the trigger conditions.
  1. 定义明确的触发条件——主动搜索应因具体且合理的原因（异常、过期、知识缺口）而触发，而不是随机触发或对每个查询都执行。
  2. 限制每个会话的搜索次数——将每个会话的主动搜索限制为 3–5 次，以控制延迟和成本；并按触发器的重要性确定优先级。
  3. 按相关性和新鲜度对结果评分——并非所有搜索结果都有用；在注入 Agent 上下文之前应严格过滤。
  4. 记录每次主动搜索——记录触发搜索的原因、使用的查询，以及结果实际上是否有用，以便后续优化。
  5. 清晰区分主动发现与用户提供的数据——使用 XML 标签或清晰标签，区分 Agent 收集的信息与用户输入。
  6. 指示 Agent 引用主动搜索来源——以已引用来源为依据的建议更可信，也更易于审计。
  7. 根据反馈调整触发阈值——如果主动搜索经常返回无关结果，应收紧触发条件。

Platform Compatibility

平台兼容性

FeatureClaude CodeCursorCodexGemini CLI
Web search tool✅ MCP/custom✅ MCP/built-in⚠️ Custom✅ Built-in
Trigger evaluation✅ Full✅ Full✅ Full✅ Full
Result integration✅ Full✅ Full✅ Full✅ Full
Search budget limits✅ Custom✅ Custom✅ Custom✅ Custom
Citation tracking✅ Full✅ Full✅ Full✅ Full
功能Claude CodeCursorCodexGemini CLI
Web 搜索工具✅ MCP/自定义✅ MCP/内置⚠️ 自定义✅ 内置
触发器评估✅ 完整✅ 完整✅ 完整✅ 完整
结果整合✅ 完整✅ 完整✅ 完整✅ 完整
搜索预算限制✅ 自定义✅ 自定义✅ 自定义✅ 自定义
引用追踪✅ 完整✅ 完整✅ 完整✅ 完整

Mythos Preview Reference

Mythos Preview 参考

Mythos Preview scaffolding highlights end-to-end autonomous chaining: after a brief initial prompt, the model hypothesizes from code, runs the program (and tooling like debuggers) to confirm or refute, repeats as needed, then emits a structured conclusion—discovery → exploitation → triage in one flow—without per-step human approval.
When designing proactive agents, mirror that flow: explicit triggers to gather or re-check reality (runs, tests, sandboxes), then fold evidence back into analysis so the chain stays grounded rather than speculative. Source: Mythos Preview.
Mythos Preview 脚手架突出了 端到端自主链式执行：在收到简短的初始提示后，模型会基于代码提出假设、运行程序（以及 debugger 等工具）来确认或推翻假设，按需重复上述过程，然后输出结构化结论，在一次流程中完成 discovery → exploitation → triage，无需在每一步都获得人工批准。
设计主动式 Agent 时，应参照这一流程：设置明确的触发器以收集信息或重新核实现实情况（运行、测试、沙箱），然后将证据反馈到分析中，使整个链条立足于事实而非臆测。来源：Mythos Preview。

Related Skills

相关 Skill

  • Knowledge Base Injection - Proactive search fills knowledge gaps that the static knowledge base cannot cover
  • Continuous Learning - Successful proactive search patterns are captured and codified into reusable triggers
  • Competitor Analysis - Real-time competitive intelligence is a primary proactive search trigger
  • Conversion Tracking - Data anomaly detection triggers proactive investigation of tracking issues
  • Knowledge Base Injection - 主动搜索可填补静态知识库无法覆盖的知识缺口
  • Continuous Learning - 成功的主动搜索模式会被记录并编码为可复用的触发器
  • Competitor Analysis - 实时竞争情报是主动搜索的主要触发因素
  • Conversion Tracking - 数据异常检测会触发对转化跟踪问题的主动调查

Keywords

关键词

proactive-intelligence, autonomous-search, trigger-conditions, data-freshness, web-search, result-scoring, knowledge-gaps, anomaly-detection, competitive-intelligence, agent-skills
© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License
proactive-intelligence, autonomous-search, trigger-conditions, data-freshness, web-search, result-scoring, knowledge-gaps, anomaly-detection, competitive-intelligence, agent-skills
© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License