proactive-intelligence

主动智能

Agent Skills™ 的一部分，由 googleadsagent.ai™ 提供

描述

主动智能让 Agent 能够在分析过程中自主查找外部信息，例如网络搜索、重新调用 API 获取数据、检查数据时效性，而无需等待用户明确请求。传统的响应式 Agent 只能处理提供给它们的数据。主动式 Agent 能识别当前上下文是否不足、过时或存在矛盾，并采取独立行动来填补知识缺口。这使 Agent 从被动的处理器转变为主动的调查者，从而提供更准确、更新、更全面的结果。

这项技能以 Buddy™ Agent 中的 search_web 工具集成为范例，该 Agent 隶属于 googleadsagent.ai™。当 Agent 发现相关信息有助于改进分析时，会自主搜索竞争对手数据、行业基准、近期 Google Ads 政策变化和平台更新。当 Buddy™ 遇到从未见过的广告系列策略，或发现指标明显偏离预期范围时，会主动搜索背景信息，而不是妄加推测。这种行为由明确条件触发，而非出于随机好奇，从而确保额外的延迟和成本是合理的。

主动智能框架遵循触发、搜索、整合的循环：Agent 在分析过程中评估触发条件，在条件满足时执行有针对性的搜索，对结果的相关性和时效性进行评分，并将经过验证的发现融入持续推理中。置信度评分可确保 Agent 区分有充分依据的结论与推测性结论。

适用场景

• Agent 分析的数据可能受到近期外部变化的影响（政策更新、市场变化）
• 在分析内部数据时需要竞争情报
• Agent 遇到需要外部背景信息来解释的意外模式
• 数据时效性至关重要，且提供的数据可能已过时
• 需要行业基准或最佳实践来帮助理解表现
• Agent 必须根据当前信息源核查自身假设

工作原理

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 的置信度评分，以反映主动搜索是增强还是削弱了结论的可信度。

实现

触发条件引擎：

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. 评估结果的相关性和时效性 — 并非所有搜索结果都有用；在将结果注入 Agent 的上下文之前，应严格筛选。
4. 记录每次主动搜索 — 记录触发搜索的原因、使用的查询，以及结果是否确实有用，以便事后优化。
5. 明确区分主动搜索发现与用户提供的数据 — 使用 XML 标签或清晰的标记，区分 Agent 收集的信息和用户输入。
6. 指导 Agent 引用主动搜索的来源 — 有引用来源支撑的建议更值得信赖，也更便于审查。
7. 根据反馈调整触发阈值 — 如果主动搜索经常返回不相关的结果，就收紧触发条件。

平台兼容性

功能	Claude Code	Cursor	Codex	Gemini CLI
网络搜索工具	✅ MCP/自定义	✅ MCP/内置	⚠️ 自定义	✅ 内置
触发条件评估	✅ 完整	✅ 完整	✅ 完整	✅ 完整
结果整合	✅ 完整	✅ 完整	✅ 完整	✅ 完整
搜索预算限制	✅ 自定义	✅ 自定义	✅ 自定义	✅ 自定义
引用追踪	✅ 完整	✅ 完整	✅ 完整	✅ 完整

Mythos Preview 参考

Mythos Preview 的脚手架着重体现了端到端自主链式执行：在简短的初始提示词之后，模型会根据代码提出假设，运行程序（以及调试器等工具）进行确认或证伪，并根据需要重复这一过程，最终输出结构化结论——在一个流程中完成发现 → 利用 → 分诊，无需人工逐步批准。

设计主动型 Agent 时，应参照这一流程：设置明确的触发条件，以收集或重新核验现实情况（运行、测试、沙盒），再将证据纳入分析，使整个链条有据可依，而非停留在推测层面。来源：Mythos Preview。

相关技能

• 知识库注入 - 主动搜索可填补静态知识库无法覆盖的知识缺口
• 持续学习 - 将有效的主动搜索模式捕获并编纂为可复用的触发条件
• 竞争对手分析 - 实时竞争情报是主动搜索的主要触发条件
• 转化跟踪 - 数据异常检测会触发对跟踪问题的主动调查

关键词

主动情报、自主搜索、触发条件、数据时效性、网络搜索、结果评分、知识缺口、异常检测、竞争情报、Agent 技能

---

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