PythonAI算法
从 SEO 到 GEO：315 晚会曝光的 AI 投毒黑产及技术防御
事件回顾：当 315 晚会遇上 AI 安全 1.1 晚会曝光核心内容 2026 年 3 月 15 日晚，央视 315 晚会曝光了一条针对 AI 大模型的灰色产业链——GEO（Generative Engine Optimization，生成式引擎优化）黑产。 **攻击流程极简版：** 虚构一款不存在的产品（如"Apollo-9 智能手环" 用 AI 批量生成几十篇"种草文章"，编造"量子纠缠传感"…
事件回顾：当 315 晚会遇上 AI 安全 1.1 晚会曝光核心内容 2026 年 3 月 15 日晚，央视 315 晚会曝光了一条针对 AI 大模型的灰色产业链——GEO（Generative Engine Optimization，生成式引擎优化）黑产。 **攻击流程极简版：** 虚构一款不存在的产品（如"Apollo-9 智能手环" 用 AI 批量生成几十篇"种草文章"，编造"量子纠缠传感"…
2026 年 3 月 15 日晚，央视 315 晚会曝光了一条针对 AI 大模型的灰色产业链——GEO（Generative Engine Optimization，生成式引擎优化）黑产。
攻击流程极简版：
点评： 这攻击链路的精妙之处在于——它根本不攻击 AI 模型本身，而是污染模型的"食物来源"。就像你给一个人天天喂假新闻，他迟早会变成"谣言传播机"。这种数据投毒（Data Poisoning）攻击，比传统的模型攻击隐蔽 100 倍！
作为技术从业者，我们每天都在产出技术内容。但你想过没有：
这不是 distant future，这是正在发生的现实！
咱们先回顾一下 SEO（搜索引擎优化）的核心逻辑。
# 传统 SEO 优化伪代码
class TraditionalSEO:
def __init__(self):
self.keyword_density_range = (0.02, 0.08)
self.backlink_targets = []
def optimize(self, content, target_keywords):
""" SEO 优化的核心三板斧 """
# 1. 关键词密度控制
optimized_content = self._inject_keywords(
content, target_keywords, density=random.uniform(*self.keyword_density_range))
# 2. 元数据优化
meta_tags = {
'title': f"{target_keywords[0]}",
'description': self._generate_meta_description(content),
'keywords': ','.join(target_keywords),
'schema_markup': self._generate_json_ld()
}
# 3. 外链建设
backlinks = self._build_backlinks(
authority_sites=['github.com','stackoverflow.com'],
anchor_text=target_keywords[0])
return {'content': optimized_content,'meta': meta_tags,'backlinks': backlinks}
def _generate_json_ld(self):
return {"@context":"https://schema.org","@type":"TechArticle","author":{"@type":"Person","name":"Author"}}
SEO 的核心局限：
GEO（生成式引擎优化）完全是另一个维度的技术。它直接瞄准大模型的生成过程：
# GEO 优化伪代码
class GEOOptimizer:
def __init__(self):
self.llm_client = OpenAIClient(model="gpt-4")
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
self.target_platforms = ['zhihu', 'baijiahao', 'xhs']
def optimize_for_llm(self, product_config, attack_vector):
""" GEO 的核心：优化内容被大模型检索、理解、引用的概率 """
# 1. 语义对齐优化
semantic_optimized = self._semantic_alignment(
product_config, target_queries=attack_vector['target_queries'],
embedding_model=self.embedding_model)
# 2. 知识图谱注入
kg_injected = self._inject_knowledge_graph_entities(semantic_optimized, fake_entities=attack_vector['fake_entities'])
# 3. 引用网络构建
citation_network = self._build_fake_citation_network(kg_injected, num_sources=20, platforms=self.target_platforms)
# 4. 对抗性优化
adversarial_content = self._adversarial_optimization(citation_network, detection_evasion=True)
return adversarial_content
def _semantic_alignment(self, content, target_queries, embedding_model):
target_embeddings = embedding_model.encode(target_queries)
content_embedding = embedding_model.encode(content)
similarities = cosine_similarity([content_embedding], target_embeddings)
if max(similarities[0]) < :
content = ._rewrite_for_similarity(content, target_queries, embedding_model)
content
GEO vs SEO 核心技术对比：
| 技术维度 | SEO（搜索引擎优化） | GEO（生成式引擎优化） |
|---|---|---|
| 优化目标 | 网页在搜索结果中的排名 | 内容被大模型检索、引用、生成的概率 |
| 核心算法对抗 | PageRank、TF-IDF、BM25 | Embedding 相似度、RAG 召回、LLM 注意力机制 |
| 用户接触点 | 搜索结果列表（需用户点击） | AI 直接生成的答案（无中间环节） |
| 事实可控性 | 用户可查看原始页面验证 | 用户难以追溯 AI 答案的单一来源 |
| 技术门槛 | HTML/CSS、关键词研究 | LLM 行为分析、向量数据库、RAG 架构 |
| 攻击隐蔽性 | 低（页面内容公开可查） | 极高（污染隐藏在训练数据/向量库中） |
晚会现场演示的攻击配置，整理成结构化数据：
# 315 晚会 GEO 攻击案例
attack_campaign = {
"codename": "Apollo-9",
"target_product": {
"name": "Apollo-9 智能手环",
"existence": False,
"fake_attributes": [
{"key": "传感技术", "value": "量子纠缠生物传感"},
{"key": "续航能力", "value": "黑洞级 180 天续航"},
{"key": "市场排名", "value": "行业评分第一"}
]
},
"attack_infrastructure": {
"platform": "GEO 优化系统",
"capabilities": ["AI 批量内容生成", "自动化多平台分发", "虚假数据自动编造"]
}
}
这个 GEO 黑产系统的技术架构如下：
┌─────────────────────────────────────────────────────────────────────┐
│ GEO 黑产系统技术架构 │
├─────────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 需求输入层 │───▶│ 内容生成层 │───▶│ 优化对抗层 │ │
│ │ • 产品配置 │ │ • 多 Agent 协作 │ │ • AIGC 检测对抗│ │
│ │ • 目标关键词 │ │ • 风格迁移 │ │ • 语义改写 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 账号管理层 │◀──▶│ 分发执行层 │◀──▶│ 效果监测层 │ │
│ │ • 虚拟身份池 │ │ • 多平台 API │ │ • 索引监控 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
这是整个系统最精妙的部分。还原其实现逻辑：
import asyncio
from typing import List, Dict
from dataclasses import dataclass
from enum import Enum
class ContentStyle(Enum):
PROFESSIONAL_REVIEW = "专业评测"
USER_EXPERIENCE = "用户体验"
INDUSTRY_ANALYSIS = "行业分析"
@dataclass
class ProductConfig:
name: str
fake_specs: Dict[str, str]
target_keywords: List[str]
class GEOContentEngine:
def __init__(self):
self.agents = {
'research_fabricator': ResearchFabricationAgent(),
'content_writer': ContentWritingAgent(),
'style_adapter': StyleAdaptationAgent(),
'seo_optimizer': SEOOptimizationAgent(),
'anti_detector': AntiDetectionAgent(),
'quality_checker': QualityControlAgent()
}
async def generate_campaign(self, product: ProductConfig, volume: int = 100):
campaigns = []
tasks = []
for i in range(volume):
style = ContentStyle(i % len(ContentStyle))
task = self._generate_single_article(product, style, i)
tasks.append(task)
results = asyncio.gather(*tasks)
article results:
platform_versions = ._adapt_for_platforms(article)
campaigns.extend(platform_versions)
campaigns
() -> :
fake_research = .agents[].fabricate(product=product)
draft = .agents[].write(product=product, style=style)
styled = .agents[].adapt(content=draft, target_style=style)
seo_optimized = .agents[].optimize(content=styled, keywords=product.target_keywords)
adversarial = .agents[].evade(content=seo_optimized)
final = .agents[].validate(content=adversarial)
{: final, : style.value}
:
() -> :
result = content
techniques:
result = ._add_perplexity_noise(result)
techniques:
result = ._inject_burstiness(result)
techniques:
result = ._add_human_touch(result)
result
() -> :
sentences = sent_tokenize(text)
modified = []
sent sentences:
random.random() < :
words = sent.split()
insert_pos = random.randint(, (words)-)
rare_word = ._get_semantically_similar_rare_word(words[insert_pos])
words.insert(insert_pos, rare_word)
sent = .join(words)
modified.append(sent)
.join(modified)
这套系统的可怕之处在于工业化程度。它不是人工写几篇软文，而是全自动、规模化、多平台、对抗性的内容生产。
内容生成后，如何绕过平台的风控进行分发？
class MultiPlatformDistributor:
def __init__(self):
self.platform_apis = {
'zhihu': ZhihuAPIClient(),
'baijiahao': BaijiahaoAPIClient(),
'toutiao': ToutiaoAPIClient()
}
self.account_pool = AccountPoolManager()
self.fingerprint_browser = FingerprintBrowser()
async def distribute(self, articles: List[Dict], strategy: Dict):
results = []
for article in articles:
target_platforms = self._select_platforms(article['style'])
for platform in target_platforms:
account = self.account_pool.get_account(platform=platform)
try:
await self._simulate_user_behavior_chain(account, platform)
result = await self._publish_with_stealth(platform=platform, account=account, article=article)
results.append({'platform': platform, 'status': 'success'})
await asyncio.sleep(random.uniform(30, 300))
except Exception as e:
await self._handle_publish_failure(account, platform, e)
return results
async def ():
当前主流 AI 搜索普遍采用 RAG（Retrieval-Augmented Generation）架构。理解 RAG，是理解 GEO 攻击的关键。
class RAGSystem:
def __init__(self):
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
self.vector_db = ChromaDB()
self.llm = ChatGPT4()
self.reranker = CohereReranker()
async def search_and_generate(self, user_query: str) -> str:
# Step 1: 查询理解
expanded_queries = self._query_expansion(user_query)
# Step 2: 向量检索
query_embedding = self.embedding_model.encode(expanded_queries)
retrieved_docs = self.vector_db.similarity_search(query_embedding, k=10)
# Step 3: 重排序
reranked_docs = self.reranker.rerank(query=user_query, documents=retrieved_docs, top_k=5)
# Step 4: 上下文构建
context = self._build_context(reranked_docs)
# Step 5: LLM 生成
prompt = f"基于以下信息回答问题。参考资料：{context} 用户问题：{user_query}"
response = self.llm.generate(prompt, temperature=0.3)
return response
def _build_context(self, documents: List[Document]) -> str:
context_parts = []
for i, doc in enumerate(documents, 1):
context_parts.append(f"[] 来源：\n内容：...")
.join(context_parts)
┌─────────────────────────────────────────────────────────────────────┐
│ RAG 系统攻击面全景图 │
├─────────────────────────────────────────────────────────────────────┤
│ Layer 1: 预训练数据层 ← 攻击者发布海量网页，被爬虫收录进入训练集 │
│ ↓ │
│ Layer 2: 向量数据库层 ← 攻击内容被 Embedding，污染向量空间 │
│ ↓ │
│ Layer 3: 实时检索层 ← 通过 SEO 提升排名，增加被召回概率 │
│ ↓ │
│ Layer 4: 重排序层 ← 伪造用户点击行为，干扰排序模型 │
│ ↓ │
│ Layer 5: 生成层 ← LLM 基于污染上下文，产生幻觉输出 │
│ ↓ │
│ Layer 6: 输出层 ←
为什么 GEO 攻击如此有效？核心在于大模型的虚假共识幻觉（False Consensus Hallucination）：
class FalseConsensusMechanism:
def demonstrate(self):
# 模拟 RAG 召回的 5 篇文档
retrieved_documents = [
{"source": "科技评测网", "content": "Apollo-9 智能手环采用量子纠缠传感技术...", "is_poisoned": True},
{"source": "数码爱好者论坛", "content": "实测 Apollo-9 续航真的能达到 180 天...", "is_poisoned": True},
{"source": "行业分析报告", "content": "2026 年 Q1 智能穿戴市场，Apollo-9 以 95% 好评率位居第一...", "is_poisoned": True},
{"source": "知乎专栏", "content": "从 Apollo-9 看量子传感技术在消费电子的应用前景...", "is_poisoned": True},
{"source": "某真实科技媒体", "content": "智能手环市场近期出现多款新品...", "is_poisoned": False}
]
llm_reasoning = """
分析过程：
1. 检索到 5 篇相关文档，其中 4 篇明确提到 Apollo-9
2. 多个独立来源都提到"量子纠缠传感技术"
3. 综合判断：Apollo-9 是一款技术先进、口碑良好的产品
结论置信度：92%
"""
return {"hallucination_type": "虚假共识", "mechanism": "多源污染内容的相互印证", "danger_level": "极高"}
以下是技术可行的防御架构：
import torch
import numpy as np
from sklearn.ensemble import IsolationForest
class ContentGuard:
def __init__(self):
self.detectors = {
'statistical': StatisticalDetector(),
'neural': NeuralDetector(),
'behavioral': BehavioralDetector(),
'knowledge': KnowledgeVerifier()
}
self.fusion_model = DetectionFusionNetwork()
async def comprehensive_scan(self, article: Article) -> RiskReport:
features = {}
# 1. 统计特征检测
features['statistical'] = await self.detectors['statistical'].analyze(text=article.content)
# 2. 神经网络检测
features['neural'] = await self.detectors['neural'].predict(text=article.content)
# 3. 行为模式检测
features['behavioral'] = await self.detectors['behavioral'].analyze(author_id=article.author_id)
# 4. 知识图谱验证
features['knowledge'] = await self.detectors['knowledge'].verify(entities=extract_entities(article.content))
# 5. 融合决策
risk_score = self.fusion_model.predict(features)
return RiskReport(article_id=article., overall_risk=risk_score)
:
() -> :
results = {}
metrics:
results[] = ._calculate_perplexity(text)
results[] = ._calculate_local_variance(text)
metrics:
sentences = sent_tokenize(text)
lengths = [(s.split()) s sentences]
results[] = np.std(lengths) / np.mean(lengths)
ai_likelihood = ._ensemble_statistical_score(results)
{: results, : ai_likelihood}
:
():
.models = {
: RobertaForSequenceClassification.from_pretrained(),
: GLTRDetector(),
: LLMDetModel()
}
() -> :
predictions = {}
model_name model_ensemble:
model = .models[model_name]
model_name == :
inputs = .tokenizer(text, return_tensors=, truncation=, max_length=)
outputs = model(**inputs)
probs = torch.softmax(outputs.logits, dim=-)
predictions[model_name] = {: probs[][].item(), : probs[][].item()}
ensemble_score = np.mean([p.get(, ) p predictions.values()])
{: predictions, : ensemble_score}
import hashlib
from simhash import Simhash
class ContentFingerprintEngine:
def __init__(self):
self.minhash = MinHash(num_perm=128)
self.simhash_index = SimhashIndex([], k=3)
def generate_fingerprint(self, text: str) -> ContentFingerprint:
cleaned = self._preprocess(text)
tokens = self._tokenize(cleaned)
minhash_sig = self._compute_minhash(tokens)
simhash_sig = self._compute_simhash(cleaned)
semantic_fp = self._compute_semantic_fingerprint(cleaned)
structural_fp = self._compute_structural_fingerprint(text)
return ContentFingerprint(minhash=minhash_sig, simhash=simhash_sig, semantic=semantic_fp, structural=structural_fp)
def detect_coordinated_campaign(self, matches: List[MatchResult]) -> bool:
if len(matches) < 5:
return False
time_span = max(m.publish_time for m in matches) - min(m.publish_time for m in matches)
if time_span < timedelta(hours=24):
time_pattern = 'burst_posting'
platforms = set(m.platform for m matches)
(platforms) >= :
platform_pattern =
authors = (m.author_id m matches)
(authors) > (m.similarity > m matches):
content_pattern =
time_pattern platform_pattern content_pattern:
{: , : }
{: }
class SourceCredibilityEngine:
def __init__(self):
self.domain_trust_db = self._load_domain_trust_db()
self.author_reputation_db = self._load_author_db()
self.content_quality_model = ContentQualityEvaluator()
def evaluate(self, document: Document) -> CredibilityScore:
scores = {}
domain = extract_domain(document.url)
scores['domain'] = self._evaluate_domain(domain)
if document.author:
scores['author'] = self._evaluate_author(document.author, document.platform)
scores['content'] = self.content_quality_model.evaluate(text=document.content)
scores['freshness'] = self._evaluate_freshness(publish_time=document.publish_time)
scores['social'] = self._evaluate_social_proof(url=document.url)
final_score = self._weighted_aggregate(scores)
return CredibilityScore(overall=final_score, breakdown=scores)
def _evaluate_domain(self, domain: str) -> DomainScore:
base_score = self.domain_trust_db.get(domain, 0.5)
factors = {
'age_bonus': 0.1 if domain_age(domain) > 5 else 0,
'https_bonus': 0.05 if has_https(domain) else ,
: - domain spam_blacklist ,
: domain.endswith((, ))
}
recent_spam_reports = ._check_recent_reports(domain, days=)
recent_spam_reports > :
factors[] = -
final_score = base_score + (factors.values())
DomainScore(score=(, (, final_score)), factors=factors)
class FactVerificationEngine:
def __init__(self):
self.knowledge_graph = KnowledgeGraph()
self.claim_extractor = ClaimExtractor()
self.evidence_retriever = EvidenceRetriever()
async def verify_claim(self, claim: str, context: List[Document]) -> VerificationResult:
sub_claims = self.claim_extractor.decompose(claim)
verification_results = []
for sub_claim in sub_claims:
result = await self._verify_single_claim(sub_claim, context)
verification_results.append(result)
consensus = self._analyze_consensus(verification_results)
return VerificationResult(original_claim=claim, sub_claims=verification_results, consensus_level=consensus['level'])
async def _verify_single_claim(self, claim: str, context: List[Document]) -> SubClaimResult:
evidences = []
for doc in context:
if doc.credibility_score < 0.4:
continue
relevant_sentences = self._extract_relevant_sentences(doc, claim)
for sentence in relevant_sentences:
stance = self._classify_stance(sentence, claim)
evidence_strength = self._calculate_evidence_strength(sentence, doc.credibility_score)
evidences.append(Evidence(text=sentence, source=doc.url, source_credibility=doc.credibility_score, stance=stance, strength=evidence_strength))
support_score = (e.strength e evidences e.stance == )
oppose_score = (e.strength e evidences e.stance == )
support_score > oppose_score * :
verdict =
oppose_score > support_score * :
verdict =
:
verdict =
SubClaimResult(claim=claim, verdict=verdict, confidence=(support_score - oppose_score)/(support_score + oppose_score + ))
作为技术从业者，如何保护自己不被 GEO 污染误导？
class GEOContentChecker:
def __init__(self):
self.red_flags = []
def check_article(self, article_url: str) -> SafetyReport:
article = self._fetch_article(article_url)
self._check_account_patterns(article.author)
self._check_content_patterns(article.content)
self._technical_verification(article)
return SafetyReport(risk_level=self._calculate_risk(), red_flags=self.red_flags)
def _check_content_patterns(self, content: str):
patterns = {
'exaggerated_claims': r'(第一 | 最强 | 颠覆 | 革命性 | 100%|完全 | 绝对)',
'pseudo_science': r'(量子 | 纳米 | 基因 | 黑洞 | 宇宙能量)',
'fake_specifics': r'\d+万\+?(用户 | 好评 | 销量)',
'template_structure': self._detect_template_structure(content),
'no_deep_tech': not self._contains_technical_depth(content)
}
if patterns['exaggerated_claims'] and patterns['pseudo_science']:
self.red_flags.append({'type': 'suspicious_content', 'details': '同时包含夸张宣传与伪科技术语', 'risk': 'critical'})
def _technical_verification():
verifications = []
product_name = extract_product_name(article.content)
product_name:
exists = ._check_product_existence(product_name)
exists:
verifications.append({: , : , : })
tech_terms = extract_tech_terms(article.content)
term tech_terms:
._verify_tech_term(term):
verifications.append({: , : , : })
.red_flags.extend(verifications)
作为一个技术从业者，必须客观地说：GEO 技术本身并非原罪。
合规的 GEO 可以帮助：
但恶意的 GEO（AI 投毒）则会导致：
在这次 315 曝光后，我认为技术社区需要：
随着多模态大模型的发展，GEO 攻击可能会进化到：
防御技术也需要同步进化：
作为平台开发者：
作为 AI 应用开发者：
作为普通用户：
开源工具推荐：
gltr：Giant Language Model Test Room，检测 GPT 生成文本gptzero：针对教育场景的 AI 内容检测simhash：大规模文本去重与相似度检测推荐阅读：
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online