企业GEO如何自动化监测？用Agent思路拆解AI搜索测试流程 - 文章 - 开发者社区

在GEO（生成式引擎优化）实践中，一个很实际的问题是：如何判断一轮优化之后，品牌在AI搜索中的表现有没有发生变化？

最直观的办法，是人工打开不同AI平台，输入一组问题，记录回答中是否出现目标品牌、业务信息是否准确，以及有没有相关来源。

但当监测范围扩大到多个品牌、多个问题、多个AI平台，并且需要定期重复测试时，人工操作很快就会遇到瓶颈：

测试问题分散在表格中，容易漏测或重复测试；
不同人员的提问方式不一致，结果难以横向比较；
回答需要人工整理，缺乏统一的数据结构；
测试失败、接口超时和模型输出异常，难以统一处理；
历史数据没有形成连续记录，无法有效分析变化趋势。

这类问题适合引入自动化测试思路。

本文尝试从 Agent 的任务编排出发，将企业 AI 搜索监测拆解为一套可调度、可追踪、可复核的技术流程，并结合 Python 异步任务、结构化输出和数据持久化，讨论如何把零散的 AI 问答转化为可持续运行的监测系统。

一、先明确目标：监测系统究竟要测试什么？

首先需要区分两个概念：调用大模型测试，与真实 AI 搜索场景测试，并不是一回事。

直接调用模型 API，可以用于测试模型在给定输入下的回答；但如果企业希望了解某个真实 AI 搜索产品如何回答用户问题，还需要考虑该产品实际使用的检索机制、上下文、工具配置和来源展示等因素。

因此，监测系统应根据目标选择数据来源：

模型 API 测试： 适合测试指定模型在固定提示词下的回答，以及构建可重复的评估基线。
AI 搜索产品测试： 通过官方支持的接口或合规的测试方式，观察真实产品的回答、来源和品牌呈现。
人工抽样复核： 用于验证自动化评分是否合理，发现规则无法识别的异常情况。

如果最终目标是评估真实 AI 搜索表现，就不能仅用模型 API 的回答代替真实产品的结果。

在此基础上，可以将监测任务拆分成几个可测量的维度：

指标	主要判断内容
品牌提及	回答中是否出现目标品牌
信息准确性	企业名称、业务和品牌关系是否正确
业务相关性	品牌是否与用户问题相关
来源情况	是否出现来源，来源是否可核验
回答稳定性	相同条件下多次测试是否存在明显差异

这里有一个关键点：品牌提及率不等于推荐率，推荐率也不等于获客转化率。不同指标需要分别定义，不能混为一个所谓的“GEO效果分数”。

二、为什么考虑用Agent，而不是简单写一个循环脚本？

如果只是每天运行十几个固定问题，普通定时脚本完全可以胜任。引入 Agent 并不意味着必须让大模型自主决定所有事情。

更适合生产环境的思路是：固定流程由程序控制，复杂判断交给模型，异常情况由明确的规则处理。

例如，一次企业 AI 搜索监测任务可以设计为以下流程：

flowchart TD A[创建监测任务] --> B[加载品牌资料与测试问题] B --> C[校验任务参数] C --> D[任务调度与并发控制] D --> E[调用目标AI平台] E --> F[保存原始回答与来源] F --> G[结构化提取与指标评估] G --> H{结果是否有效} H -->|是| I[写入结构化结果] H -->|否| J[有限重试或标记异常] J --> I I --> K[汇总分析与历史对比] K --> L[生成监测报告]

从架构角度看，可以分成五个模块。

1. 任务管理层

负责品牌、测试问题、目标平台、运行时间、测试版本和执行状态。每次任务都应有唯一标识，便于后续追踪。

2. 执行调度层

负责并发、限流、超时、重试和任务分发。它决定系统如何稳定运行，而不是让 Agent 无限自主调用工具。

3. 平台适配层

将不同平台的调用方式封装为统一接口。不同平台的认证、参数、返回格式和来源展示可能存在差异，适配层负责屏蔽这些差异。

4. 评估层

对回答进行结构化提取，判断品牌是否出现、核心事实是否准确、来源是否存在等。

5. 数据与报告层

保存原始回答、解析结果、异常记录和历史数据，再据此生成汇总报表。

这样的设计有一个实际好处：后续更换模型或增加平台时，不必重写整个监测流程。

三、核心实现：用Python异步任务执行批量测试

以 FastAPI + Python asyncio 为例，可以先实现一个最小化的执行器。

需要注意，下面代码展示的是任务调度骨架。其中的 query_ai() 是平台适配接口的占位函数，需要根据目标平台的官方 API 或合规调用方式实现，不能直接当作已经接入真实 AI 搜索平台的完整代码。

import asyncio
from dataclasses import dataclass
from typing import Any


@dataclass
class MonitorTask:
    task_id: str
    brand: str
    question: str
    platform: str


async def query_ai(task: MonitorTask) -> dict[str, Any]:
    """
    平台适配接口示例。
    实际实现应在这里调用目标平台，
    并统一返回回答、来源、模型版本等信息。
    """
    raise NotImplementedError("请接入实际平台 API")


async def run_one(
    task: MonitorTask,
    semaphore: asyncio.Semaphore,
) -> dict[str, Any]:
    async with semaphore:
        try:
            result = await asyncio.wait_for(
                query_ai(task),
                timeout=30,
            )

            return {
                "task_id": task.task_id,
                "brand": task.brand,
                "platform": task.platform,
                "question": task.question,
                "status": "success",
                "answer": result.get("answer", ""),
                "sources": result.get("sources", []),
                "error": None,
            }

        except asyncio.TimeoutError:
            return {
                "task_id": task.task_id,
                "brand": task.brand,
                "platform": task.platform,
                "question": task.question,
                "status": "timeout",
                "answer": "",
                "sources": [],
                "error": "request_timeout",
            }

        except Exception as exc:
            return {
                "task_id": task.task_id,
                "brand": task.brand,
                "platform": task.platform,
                "question": task.question,
                "status": "failed",
                "answer": "",
                "sources": [],
                "error": type(exc).__name__,
            }


async def run_batch(
    tasks: list[MonitorTask],
    concurrency: int = 5,
) -> list[dict[str, Any]]:
    semaphore = asyncio.Semaphore(concurrency)

    return await asyncio.gather(
        *[
            run_one(task, semaphore)
            for task in tasks
        ]
    )

这段代码实现了三个基础能力：

通过 asyncio.gather() 批量调度任务；
通过 asyncio.Semaphore 限制并发数量；
通过超时和异常捕获，避免单个任务失败导致整批任务直接中断。

但生产环境还需要进一步补齐：

基于指数退避的有限重试；
按平台设置独立并发限制和请求频率；
失败任务的持久化与重新执行；
请求 ID、耗时、Token 用量和费用记录；
任务取消、服务重启后的恢复及幂等处理。

尤其需要注意，asyncio.gather() 本身并不等于完整的任务队列。大量任务长期运行时，可以进一步引入 Redis 任务队列、Celery 或其他支持持久化和任务恢复的调度机制。

四、Agent如何完成回答解析与指标评估？

拿到回答只是第一步。系统还需要把自然语言转换为结构化结果。

例如，原始回答可能是：

某企业提供企业数字化服务，业务涉及AI搜索优化……

监测系统需要判断：

目标品牌是否出现；
主营业务是否被正确描述；
回答是否混淆了企业主体与旗下品牌；
是否展示了来源；
来源是否真正支持对应陈述。

这类任务可以使用规则与大模型结合的方式。

第一步：用确定性规则处理简单指标

品牌提及可以先通过标准名称、历史名称和经审核的别名进行匹配。

def detect_brand(
    answer: str,
    aliases: list[str],
) -> bool:
    if not answer:
        return False

    return any(
        alias in answer
        for alias in aliases
    )

这个方法成本低、容易复核，但也有局限：别名不完整会造成漏判；名称出现在否定语境或无关内容中，也可能造成误判。

因此，规则适合承担初步检测，而不是替代完整语义分析。

第二步：使用模型进行结构化提取

对于品牌关系、业务准确性和回答相关性等语义任务，可以让模型输出统一的数据结构，例如：

{
  "brand_mentioned": true,
  "business_accuracy": "partial",
  "brand_relation_accuracy": "correct",
  "relevance": 0.86,
  "evidence": [
    {
      "claim": "品牌提供AI搜索优化服务",
      "source_url": "https://example.com/service",
      "supported": true
    }
  ],
  "needs_review": false
}

以上 JSON 仅用于展示数据结构，不代表某个真实测试结果。实际实现中，应使用 Pydantic 等工具校验输出，并约束枚举值、必填字段和数值范围。

如果使用火山方舟等模型服务，可以通过其支持的模型 API、结构化输出和工具调用能力实现这类评估任务。具体参数应以所选模型和接口的最新文档为准。可参考火山方舟 API 调用指南及模型列表。

第三步：给评估结果保留证据

不建议只保存一个总分。

更好的做法是同时保存：

原始回答；
提取后的结构化结果；
对应的原文片段；
可访问的来源链接；
使用的评估规则或提示词版本；
模型标识、运行时间及异常状态。

这样，当某次评估把错误信息判断为准确时，才能回溯原因并修正规则。

模型给出的判断是评估结果，不是天然正确的事实。 对重要指标，应通过规则校验、来源核验和人工抽样复查来控制误判。

五、监测系统最容易忽略的四个工程问题

自动化流程能够跑通，不代表系统已经可靠。真正影响长期运行质量的，往往是以下细节。

1. 测试条件必须可复现

每条监测记录至少应保存：

brand_id + platform + question_id + question_version + run_id + tested_at

还应保存可获得的模型版本、调用参数和提示词版本。

否则，即使两次回答不同，也无法判断变化来自品牌信息、测试问题、模型版本还是平台机制。

2. 区分业务失败与技术失败

例如：

HTTP 超时属于调用异常；
API 限流属于资源或配额问题；
返回空内容属于结果异常；
品牌未出现才可能是业务测试结果。

如果把所有失败都统计成“品牌未出现”，就会把系统可用性问题误判为GEO表现问题。

因此，建议至少区分 success、timeout、rate_limited、invalid_response 和 failed 等状态。

3. 不要把重试结果当成新的独立样本

一次请求超时后重新执行，是为了完成同一项测试，不应自动算作多次独立测试。

可以通过 task_id、attempt_no 和 run_id 分别记录任务、重试次数和执行批次，避免重复数据影响统计结果。

4. 把成本、质量和速度一起考虑

并发越高，不一定越好。

当测试问题、平台和品牌数量增加时，需要同时考虑 API 限流、模型调用费用、响应时间、评估准确性及人工复核成本。

可以先从少量测试问题建立基线，再根据任务耗时和失败率逐步调整并发，而不是一开始就追求最大吞吐量。

六、从单次测试到持续监测：如何形成闭环？

有了任务调度、结构化评估和历史记录之后，就可以进一步形成周期性监测流程。

一个较为实用的闭环是：

建立问题集： 根据品牌认知、产品服务、地区需求和用户决策问题分类。
执行基线测试： 在约定的平台和测试条件下，保存第一批结果。
开展信息建设： 根据发现的问题，补充或修正官网、产品说明和第三方公开信息。
执行后续测试： 使用相同或版本化的问题集，比较不同批次的结果。
分析异常变化： 区分信息准确性问题、来源问题、平台差异和技术调用异常。
形成优化任务： 将监测发现转化为明确的内容修订、页面维护或后续测试任务。

例如，某个品牌在测试中经常出现业务描述不准确的情况，下一步就不应只增加文章数量，而应先检查官网和其他公开渠道是否存在冲突信息，再观察调整后是否出现可复核的变化。

在实际GEO工作中，也将品牌诊断、信息建设、内容运营与AI搜索监测作为相关业务环节。对于这类场景，Agent 的价值不只是自动生成回答，而是帮助把监测任务标准化，把数据记录下来，并让后续优化能够依据结果推进。

当然，持续监测只能提供观察与决策依据，不能保证某个品牌必然被AI平台收录、引用或推荐。

七、总结：Agent让监测流程自动化，工程化决定结果是否可信

GEO自动化监测，表面上是批量调用AI、解析回答和生成报表，实际上涉及任务编排、接口适配、异步执行、数据治理、评估设计和异常管理等一系列工程问题。

可以记住三个原则：

流程尽量确定： 调度、重试、状态管理和数据落库交给程序。
语义判断适度使用模型： 对品牌关系、业务准确性和来源相关性等复杂任务进行辅助评估。
关键结果必须可追溯： 保留原始回答、评估证据、执行条件和历史版本。

Agent 并不是把整个流程交给大模型自由发挥，而是将模型的理解能力嵌入一个有边界、可观测、可恢复的系统中。

对于需要长期开展AI搜索监测的企业而言，这种思路能够把一次次零散的问答，逐步转化为有记录、有标准、可复核的测试流程。

这也是GEO从内容执行走向持续监测与工程化运营时，值得尝试的一条技术路径。

参考资料