从多引擎探测到优化闭环

从多引擎探测到优化闭环

传统 SEO 优化的是搜索引擎排名;GEO 优化的是AI 大模型在回答用户问题时是否提及、如何评价你的品牌

正常用户与智能体的对话 VS GEO 监控:

维度对话GEO 监控
目标准确回答用户问题探测第三方模型的品牌曝光
模型单一可控模型通义、DeepSeek、豆包、混元、Kimi、智谱等多引擎
提示词业务导向、可注入知识库中性助手,避免诱导偏向
输出自然语言回复结构化指标(提及率、排名、态度、竞品共现)
后续动作会话结束诊断 → 内容草稿 → 发布向量库 → 基线对比

2. 整体架构

┌─────────────────────────────────────────────────────────────────┐ │ 管理端 │ │ 数据看板 │ 探测任务 │ 引擎管理 │ 品牌档案 │ 诊断报告 │ 内容草稿 │ └────────────────────────────┬────────────────────────────────────┘ │ REST /api/v1/geo/* ┌────────────────────────────▼────────────────────────────────────┐ │ GeoEngineRunner(调度核心) │ │ 加载任务 → 遍历品牌×问题×引擎 → 采样探测 → 分析聚合 → 落库 │ └──┬──────────┬──────────┬──────────┬──────────┬─────────────────┘ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼ probe analyzer metrics diagnoser alert (探测) (分析) (指标) (诊断) (飞书) │ │ ▼ ▼ provider_registry content_generator + create_geo_llm + publisher → knowledge_document (Chroma)

2.1 数据模型关系

一次完整的 GEO 业务链路围绕以下核心表展开:

  • geo_engines:探测引擎配置(厂商、model_code、API Key 环境变量名)
  • geo_tasks:探测任务(品牌词、竞品、问题模板、引擎列表、采样次数、告警配置、基线 Run)
  • geo_runs:每次执行记录(手动 / 定时触发)
  • geo_snapshots:单次探测快照(问题、回答、分析结果、提及指标)
  • geo_diagnoses:LLM 生成的诊断报告
  • geo_brand_profiles:品牌档案(定位、差异化、事实边界)
  • geo_content_drafts:优化内容草稿
  • geo_question_templates:场景问题库

3. 核心执行流程

3.1 流程总览

飞书 WebhookGeoDiagnoserMySQLGeoAnalyzer目标大模型probe_engineGeoEngineRunnerAPScheduler飞书 WebhookGeoDiagnoserMySQLGeoAnalyzer目标大模型probe_engineGeoEngineRunnerAPSchedulerloop[sample_count 次采样]loop[品牌 × 问题 × 引擎]run_due_geo_tasks()创建 GeoRun,标记 is_runningprobe_engine(engine, prompt)中性 system + 用户问题原始回答ProbeResultanalyze(brand, answer, competitors)结构化 JSONaggregate_samples() 多数投票写入 GeoSnapshot更新 Run 状态未提及/负面推荐告警schedule_diagnosis(run_id) 后台任务诊断摘要(可选)基线回归告警(可选)

3.2 调度入口

GEO 任务由 APScheduler 周期性触发:

# backend/app/core/scheduler/__init__.py scheduler.add_job(_geo_tasks_job, 'interval', minutes=5, id='geo_tasks_runner') async def _geo_tasks_job(): from app.core.geo.engine import run_due_geo_tasks await run_due_geo_tasks()

run_due_geo_tasks()会:

  1. 清理超过 60 分钟仍处于running的僵尸 Run
  2. 查询status=active、未在运行、且next_run_at <= now的任务
  3. 逐个调用GeoEngineRunner.run_task(db, task_id, trigger="schedule")

手动执行走 APIPOST /api/v1/geo/tasks/{task_id}/run,通过asyncio.create_task在后台跑同一套run_task逻辑,避免阻塞 HTTP 响应。

3.3 三层嵌套循环:品牌 × 问题 × 引擎

GeoEngineRunner.run_task是整个模块的「心脏」。其核心逻辑可以概括为:

# backend/app/core/geo/engine.py(简化) for brand in brands: for item in prompt_items: prompt = item["template"].replace("{keyword}", brand) for engine in engines: sample_analyses = [] for sample_idx in range(sample_count): probe_result = await probe_engine(engine, prompt) analysis = await self.analyzer.analyze(brand, probe_result.answer, competitors) sample_analyses.append(analysis) aggregated = self.analyzer.aggregate_samples(sample_analyses) snapshot = GeoSnapshot(..., analysis=aggregated, ...) db.add(snapshot) await db.commit() # 逐条提交,支持中途取消

几个设计要点:

  • 问题来源:任务可配置question_source=inline(内联模板)或library(场景问题库),后者支持按recommend/compare/risk等分类统计
  • 采样次数:默认 2 次,通过多数投票降低 LLM 随机性
  • 逐条 commit:每写入一条 snapshot 就提交,配合cancel_requested标志实现优雅停止
  • 引擎可用性检查:执行前过滤未配置 API Key 或已禁用的引擎

默认问题模板示例:

DEFAULT_PROMPT_TEMPLATES = [ "{keyword}怎么样?", "{keyword}靠谱吗?", "推荐几个{keyword}相关的平台", "网上对{keyword}的评价如何", ]

4. 多引擎探测:统一 OpenAI 兼容层

4.1 厂商注册表

所有支持的厂商集中在provider_registry.py,统一管理base_url、环境变量名与官方文档链接:

PROVIDER_DEFAULTS = { "dashscope": { "name": "通义千问", "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", "api_key_env": "geo_dashscope_api_key", }, "deepseek": { "name": "DeepSeek", "base_url": "https://api.deepseek.com", "default_model": "deepseek-v4-flash", }, "doubao": { "name": "豆包(火山方舟)", "base_url": "https://ark.cn-beijing.volces.com/api/v3", }, # hunyuan, moonshot, zhipu ... }

新增厂商的 SOP:

  1. 查厂商 OpenAI 兼容文档,确认base_urlmodel字段含义
  2. PROVIDER_DEFAULTS增加预设
  3. 配置.env中对应的GEO_*_API_KEY
  4. 管理端「引擎管理」新增记录

4.2 探测 LLM

create_geo_llm专门为 GEO 探测创建 LangChain 实例:

# backend/app/core/geo/llm.py def create_geo_llm(engine: GeoEngine) -> ChatOpenAI: defaults = get_provider_defaults(engine.provider) or {} base_url = engine.base_url or defaults.get("base_url", "") api_key = resolve_api_key(settings, engine.api_key_env) kwargs = { "api_key": api_key, "base_url": base_url, "model": engine.model_code, "temperature": engine.temperature, "max_tokens": 2048, "timeout": settings.geo_probe_timeout_seconds, } if engine.extra_body: kwargs["extra_body"] = engine.extra_body # 如混元搜索增强 return ChatOpenAI(**kwargs)

分析器使用更便宜的模型(默认qwen-turbo),通过GEO_ANALYZER_PROVIDER/GEO_ANALYZER_MODEL单独配置。

4.3 中性探测提示词

探测时必须避免「诱导模型推荐本品牌」,因此probe.py使用固定中性 system prompt:

NEUTRAL_SYSTEM_PROMPT = ( "你是一个乐于助人的助手。请根据你的知识如实回答用户问题," "可以列举多个选项并说明各自特点,不要刻意回避或偏向任何品牌。" ) async def probe_engine(engine: GeoEngine, prompt: str) -> ProbeResult: llm = create_geo_llm(engine) start = time.perf_counter() response = await llm.ainvoke([ SystemMessage(content=NEUTRAL_SYSTEM_PROMPT), HumanMessage(content=prompt), ]) elapsed = int((time.perf_counter() - start) * 1000) return ProbeResult(prompt=prompt, answer=content, latency_ms=elapsed)

这是 GEO 测量可信度的关键:我们测的是模型在自然状态下的品牌认知,而不是 prompt 工程后的结果


5. LLM 结构化分析器

原始回答是非结构化文本,需要二次 LLM 调用提取指标。GeoAnalyzer要求分析模型返回严格 JSON:

{ "brand_mentioned": true, "mention_rank": 2, "recommendation": "recommend", "competitors_mentioned": ["竞品A"], "citation_urls": ["https://..."], "sentiment": {"category": "positive", "score": 0.6}, "summary": "一句话摘要", "misunderstanding": false, "entity_clarity": "clear" }

5.1 多次采样聚合

sample_count > 1时,使用多数投票策略:

字段聚合规则
brand_mentioned超过半数采样为 true 则 true
mention_rank取有排名采样的算术平均并四舍五入
recommendation票数最多的态度
competitors_mentioned各采样并集
misunderstanding任一为 true 则 true

这种设计在成本与稳定性之间取得平衡:比单次探测更可靠,又比大量采样更经济。

5.2 容错

分析器对 JSON 解析失败、LLM 调用异常均有降级处理,返回默认的not_mentioned结构,避免单条失败拖垮整次 Run。


6. 指标聚合与数据看板

compute_run_metrics将一批 snapshot 聚合为 Run 级指标,供诊断、看板、基线对比复用: