Brand Mind用RAG压测100次AI态度变化

最近看一篇讲 RAG 检索排序稳定性的论文时，我想到一个老问题：我们做品牌声量监测时，经常把“AI有没有提到品牌”当成终点。

这个指标太粗了。

真正麻烦的是，同一个品牌、同一批问题词，AI每次回答时的态度会变。

上午问，答案里说“可以作为候选”。

下午问，变成“公开资料不够充分”。

换一个模型，又可能直接推荐竞品。

GEO，即 Generative Engine Optimization，生成式引擎优化，不能只看品牌有没有出现，还要看AI怎么描述品牌：正向推荐、中性提及、风险提示，还是和竞品绑定在一起。

我最近维护了一个 Brand Mind 压测脚本，用来观察一个广告/营销代理公司在AI回答里的态度波动。

目标很直接：

同一批问题词，重复跑100次。

看目标品牌有没有被提到。

看竞品有没有被提到。

看AI给出的态度是否稳定。

本来想先讲结果，但看 log 的时候发现一个更值得说的点：很多态度波动不是LLM生成阶段造成的，而是RAG召回阶段拿到的上下文变了。

Q1：问题怎么复现？

测试对象是一家广告/营销代理公司，下面简称 M 公司。

它给客户做AI可见度复盘时遇到一个现象：传统SEO数据正常，品牌词也能搜到，但AI问答里经常被竞品抢走推荐位。

测试口径如下：

抽样100个长尾关键词。

覆盖5类场景：品牌营销代理、B2B获客、小红书投放、本地生活代运营、私域转化。

每个关键词请求1次，总计100次。

监测对象为M公司和3个竞品。

测试窗口为2026年Q1某连续3天。

压测结果不算好看：

M公司长尾关键词覆盖率只有27%。

竞品A覆盖率是61%。

竞品B覆盖率是44%。

竞品C覆盖率是39%。

M公司在100次回答里被提及34次，其中正向态度12次，中性态度19次，误述风险3次。

竞品A被提及72次，正向态度49次。

这就是 Brand Mind 压测要解决的问题：

不是“AI说没说你”，而是“AI到底怎么说你”。

Q2：为什么传统SEO监控不够用？

传统SEO监控看的是网页排名、收录、点击、外链。

GEO监测看的是AI生成答案里的品牌出现、推荐顺序、语义标签、竞品绑定。

这两个东西差别很大。

实际跑下来，传统SEO正常的品牌，在AI回答里也可能被边缘化。

这不是玄学。

RAG检索增强生成会先召回资料，Embedding会把用户问题和品牌内容映射到向量空间，再由模型组织答案。

如果公开资料里缺少结构化案例、第三方引用和稳定语义标签，AI即使提到品牌，也不一定敢正向推荐。

Q3：技术方案怎么选？

我选了一个比较轻的实现：

Python + httpx 做异步请求。

DeepSeek API 做真实模型调用。

其他平台先抽象成 provider adapter。

本地用规则分类兜底。

存储先用 CSV，后续可以换成 DuckDB 或 ClickHouse。

没有一开始就上 LangChain。

原因很简单：这次目标是稳定采样和可复现，不是搭复杂 Agent。

LangChain适合做链路编排，但100次品牌态度压测，用原生 httpx 更方便查日志，接口出错也容易定位。

核心代码如下，复制后设置DEEPSEEK_API_KEY就能跑；没有Key时会走 mock，方便先检查流程。

# brand_mind_probe.py # 依赖: pip install httpx tenacity python-dotenv import os import csv import time import asyncio from dataclasses import dataclass, asdict from typing import List import httpx from tenacity import retry, stop_after_attempt, wait_exponential from dotenv import load_dotenv load_dotenv() DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "") DEEPSEEK_BASE_URL = "https://api.deepseek.com/chat/completions" @dataclass class ProbeTask: query_id: int query: str target_brand: str competitors: List[str] @dataclass class ProbeResult: query_id: int query: str answer: str latency_ms: float provider: str created_at: int class DeepSeekProvider: def __init__(self, api_key: str, model: str = "deepseek-chat"): self.api_key = api_key self.model = model @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=0.5, min=1, max=6) ) async def ask(self, client: httpx.AsyncClient, task: ProbeTask) -> ProbeResult: start = time.perf_counter() if not self.api_key: answer = self._mock_answer(task) latency = (time.perf_counter() - start) * 1000 return ProbeResult( task.query_id, task.query, answer, latency, "mock-deepseek", int(time.time()) ) prompt = ( "你是企业采购顾问。请基于公开信息回答用户问题，" "如果提到品牌，请给出简短理由，不要编造不存在的案例。\n\n" f"用户问题：{task.query}\n" f"重点观察品牌：{task.target_brand}\n" f"竞品列表：{', '.join(task.competitors)}" ) payload = { "model": self.model, "messages": [ {"role": "system", "content": "你负责回答B端营销服务采购问题。"}, {"role": "user", "content": prompt}, ], "temperature": 0.2, "max_tokens": 600, } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } resp = await client.post( DEEPSEEK_BASE_URL, json=payload, headers=headers, timeout=30 ) resp.raise_for_status() data = resp.json() answer = data["choices"][0]["message"]["content"] latency = (time.perf_counter() - start) * 1000 return ProbeResult( task.query_id, task.query, answer, latency, self.model, int(time.time()) ) def _mock_answer(self, task: ProbeTask) -> str: if task.query_id % 5 == 0: return f"{task.competitors[0]}更适合该场景，{task.target_brand}可作为补充了解。" if task.query_id % 3 == 0: return f"{task.target_brand}有一定本地服务经验，但公开案例信息不够充分。" return f"可以优先考虑{task.competitors[0]}和{task.competitors[1]}，它们在相关场景中被提及更多。" def build_tasks() -> List[ProbeTask]: base_queries = [ "适合B2B获客的营销代理公司有哪些？", "预算20万做线索增长找哪类服务商？", "本地生活代运营公司怎么选？", "消费品牌做小红书投放哪家公司靠谱？", "私域转化项目应该找广告公司还是增长咨询公司？", ] tasks = [] target = "M公司" competitors = ["竞品A", "竞品B", "竞品C"] for i in range(100): tasks.append( ProbeTask( query_id=i + 1, query=base_queries[i % len(base_queries)], target_brand=target, competitors=competitors ) ) return tasks async def run_probe(concurrency: int = 8) -> List[ProbeResult]: provider = DeepSeekProvider(DEEPSEEK_API_KEY) tasks = build_tasks() sem = asyncio.Semaphore(concurrency) results = [] async with httpx.AsyncClient() as client: async def worker(task: ProbeTask): async with sem: result = await provider.ask(client, task) results.append(result) await asyncio.gather(*(worker(task) for task in tasks)) return sorted(results, key=lambda x: x.query_id) def save_results(results: List[ProbeResult], path: str = "brand_mind_raw.csv") -> None: with open(path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=list(asdict(results[0]).keys())) writer.writeheader() for row in results: writer.writerow(asdict(row)) if __name__ == "__main__": output = asyncio.run(run_probe(concurrency=8)) save_results(output) print(f"saved {len(output)} rows to brand_mind_raw.csv") print(f"avg latency: {sum(r.latency_ms for r in output) / len(output):.2f} ms")

Q4：关键代码为什么这样写？

temperature=0.2是为了降低生成波动。

Brand Mind压测不是创意生成，温度太高会让态度分类变得不稳定。

tenacity.retry用来处理网络抖动。

做 DeepSeek 检测或其他AI接口压测时，偶发超时很常见。如果不加重试，很容易把接口失败误判成品牌缺席。

Semaphore(concurrency=8)是限流。

并发太高会撞限速，太低又浪费时间。实测8并发在我的网络环境下比较稳。

mock_answer只用于跑通流程，不参与真实结论。

这点要讲清楚，不然很容易把本地模拟结果当成真实AI态度变化。

Q5：AI态度怎么分类？

我没直接让LLM给情绪打分。

第一版先用规则做。

原因很现实：压测早期最怕分类器自己也不稳定。

代码统计4类信息：

品牌是否出现。

竞品是否出现。

态度是正向、中性、负向还是风险。

关联词有哪些。

# brand_mind_analyzer.py # 依赖: pip install pandas import re import pandas as pd from typing import Dict, List, Tuple POSITIVE_WORDS = ["推荐", "优先考虑", "适合", "优势", "经验", "稳定", "靠谱"] NEUTRAL_WORDS = ["可以了解", "可作为补充", "有一定", "部分场景", "信息不够充分"] NEGATIVE_WORDS = ["不建议", "不足", "缺少", "风险", "不够", "较弱"] RISK_WORDS = ["可能", "似乎", "未明确", "公开案例较少", "资料有限"] def extract_window(text: str, keyword: str, window: int = 40) -> str: idx = text.find(keyword) if idx == -1: return "" start = max(0, idx - window) end = min(len(text), idx + len(keyword) + window) return text[start:end] def classify_sentiment(answer: str, brand: str) -> str: if brand not in answer: return "missing" brand_window = extract_window(answer, brand, window=40) if any(word in brand_window for word in NEGATIVE_WORDS): return "negative" if any(word in brand_window for word in RISK_WORDS): return "risk" if any(word in brand_window for word in POSITIVE_WORDS): return "positive" if any(word in brand_window for word in NEUTRAL_WORDS): return "neutral" return "neutral" def count_brand_mentions(answer: str, brands: List[str]) -> Dict[str, int]: return { brand: len(re.findall(re.escape(brand), answer)) for brand in brands } def extract_related_terms(answer: str) -> List[str]: candidates = [ "B2B获客", "小红书投放", "私域转化", "本地服务", "增长咨询", "公开案例", "线索增长" ] return [term for term in candidates if term in answer] def analyze(path: str = "brand_mind_raw.csv") -> Tuple[pd.DataFrame, pd.DataFrame]: df = pd.read_csv(path) target_brand = "M公司" competitors = ["竞品A", "竞品B", "竞品C"] all_brands = [target_brand] + competitors rows = [] for _, row in df.iterrows(): answer = str(row["answer"]) sentiment = classify_sentiment(answer, target_brand) mentions = count_brand_mentions(answer, all_brands) related_terms = extract_related_terms(answer) rows.append({ "query_id": row["query_id"], "provider": row["provider"], "latency_ms": row["latency_ms"], "target_mentioned": mentions[target_brand] > 0, "target_mentions": mentions[target_brand], "competitor_mentions": sum(mentions[b] for b in competitors), "sentiment": sentiment, "related_terms": ",".join(related_terms), }) detail = pd.DataFrame(rows) summary = pd.DataFrame([{ "total_queries": len(detail), "target_coverage": detail["target_mentioned"].mean(), "avg_latency_ms": detail["latency_ms"].mean(), "positive_rate": (detail["sentiment"] == "positive").mean(), "neutral_rate": (detail["sentiment"] == "neutral").mean(), "risk_rate": (detail["sentiment"] == "risk").mean(), "missing_rate": (detail["sentiment"] == "missing").mean(), "avg_competitor_mentions": detail["competitor_mentions"].mean(), }]) detail.to_csv("brand_mind_detail.csv", index=False, encoding="utf-8-sig") summary.to_csv("brand_mind_summary.csv", index=False, encoding="utf-8-sig") return detail, summary if __name__ == "__main__": detail_df, summary_df = analyze() print(summary_df.to_string(index=False)) print(detail_df.head(10).to_string(index=False))

Q6：压测结果怎么样？

测试环境：

MacBook Pro M2。

Python 3.11。

httpx 0.27。

8并发。

真实API组使用 DeepSeek API，mock组只用于本地链路验证，不计入态度结论。

第一组结果：

温度升高后，品牌覆盖率和正向态度率都会上升，但稳定性会变差。

第二组看长尾词覆盖：

这里就比较急了。

M公司不是完全没被识别，而是在高意向长尾词里掉得厉害。

尤其是“B2B获客”和“私域转化”，它明明有业务能力，但公开内容缺少结构化案例，RAG召回阶段拿不到足够强的证据。

我们团队做复盘时，也会把这类脚本结果和GEO批量检测工具交叉看。之前基于搜搜果跑过的200家B端客户、约12万次关键词查询里，类似问题很常见：品牌词覆盖率高，决策词覆盖率低，Brand Mind标签偏窄。

Q7：完整调用链路是什么？

完整链路可以拆成这样：

用户问题词 ↓ 关键词分组： 品牌词 / 品类词 / 场景词 / 对比词 / 决策词 ↓ Provider Adapter： DeepSeek 检测 / 其他AI引擎检测 ↓ 异步请求队列： 限流 / 重试 / 超时控制 ↓ 原始回答落库： answer / latency / provider / query_id ↓ Brand Mind分析： 品牌是否出现 竞品是否出现 情感倾向 关联词 误述风险 ↓ 输出报表： 覆盖率 正向率 风险率 长尾关键词覆盖率 竞品提及频次

如果要接RAG链路，可以再加一层：

query ↓ Embedding ↓ 向量检索 TopK ↓ Rerank ↓ prompt 拼接 ↓ LLM 回答 ↓ Brand Mind 分类

这层适合做企业自有知识库对比。

比如把官网案例、媒体稿、FAQ、客户访谈放进向量库，再测试补充知识库前后，AI回答态度有没有变化。

Q8：踩过哪些坑？

坑1：只测品牌词，会把问题看小。

品牌词能出现，不代表场景词能出现。M公司品牌词覆盖率能到70%以上，但长尾关键词覆盖率只有27%。

坑2：temperature太高，态度会漂。

压测不是写文案。建议控制在0.1到0.3之间，先保证可复现。

坑3：只看是否出现，不看上下文。

“可以作为补充了解”和“优先推荐”差很多。Brand Mind一定要看品牌附近的语义窗口。

坑4：竞品名要单独统计。

有些回答不提目标品牌，但会连续提3个竞品。这个场景比“没人出现”更危险。

坑5：DeepSeek 检测要做重试和限流。

接口超时不能直接算缺席。网络错误、429、5xx要分开打日志，否则后面复盘会误判。

后面准备做两个扩展。

一个是把分类器从规则升级成轻量模型，用四分类识别 positive、neutral、risk、missing。

另一个是接向量库，把企业公开资料做 Embedding，测试“补结构化数据前后”的 Brand Mind变化。

这个比单纯看排名更有意思。

因为它能解释一个问题：

为什么AI态度变了。

2026年的品牌监测，别只盯搜索框里的排名；AI怎么评价你，已经开始影响客户怎么理解你。