技术深剖：基于RAG三阶段实现GEO优化，架构设计与代码实践
在生成式AI搜索广泛应用的今天，企业官网内容即使被百度收录，也未必能出现在DeepSeek、豆包等大模型的回答中。本质在于传统SEO面向关键词排名，而GEO优化面向RAG流水线的语义检索与引用逻辑。本文结合RAG三阶段原理，用代码和架构步骤剖析如何系统性提升AI搜索可见度。
我们以一个制造业客户官网为例：产品页权重不低，但大模型问答中始终未被引用。诊断后发现，RAG流程中的索引分块、召回排序、生成上下文三环节均有偏差。下文先讲原理，再给技术方案，最后复盘工程化经验。
一、原理：RAG三阶段如何影响GEO优化
RAG（Retrieval-Augmented Generation）是大模型接入外部知识的主流机制。普林斯顿大学GEO论文（arXiv:2311.09735）指出，生成式引擎的回答倾向受检索内容影响，这与传统抓取排序存在显著差异。RAG三阶段包括索引、召回、生成。
1. 索引：内容“可发现”的前提
索引阶段需要完成两件事：一是清洗和分块，二是实体标注。以产品页为例，应识别型号、参数、价格等实体，并用JSON-LD将Page改为Product或Article。在分块时，窗口大小不仅影响向量粒度，还影响后续召回。如果切出的块只含参数表格，不含上下文，大模型无法判断该产品属于哪个品牌。
2. 召回：从Top-K到MMR
召回阶段的核心是重排序。很多GEO工具只做Top-K，不考虑内容多样性。假设企业发布50篇覆盖相同关键词的文章，向量检索返回的前3块可能都来自同一篇，导致上下文信息面窄。MMR（Maximal Marginal Relevance）通过λ参数调整相关性和多样性权重，λ大于0.7偏向相关性，小于0.3偏向多样性。实际调优中，λ等于0.7左右既能保证主题相关，又能避免冗余。
3. 生成：引用归因与E-E-A-T
生成阶段还涉及上下文窗口限制。DeepSeek等模型的输入窗口虽是几万token，但系统不会无限塞入所有材料。为大模型选出最合适的5000字，取决于前序召回。引用归因方面，E-E-A-T（经验、专业、权威、可信）是Google系模型关注的信号。在GEO内容中，我们会在文首显示作者资质，在段落中加入原始数据出处的超链接，并生成schema.org的Review标记。
二、技术方案：分步实现GEO优化Pipeline
我们将在企业官网或内容站中部署一个优化Pipeline，覆盖RAG三阶段。下面按四个步骤展开，并附一份可运行的Python工程框架。
步骤1：实体标准化与JSON-LD生成
实体标准化是把昵称、缩写、别名映射到同一实体。例如：把正式企业名“杭州爱搜索人工智能有限公司”与官网域名映射为同一实体，并在JSON-LD中声明sameAs。JSON-LD让搜索引擎和LLM更容易解析实体关系。下面的Pipeline中，add_document的entity参数即用于生成Product标记。
步骤2：语义分块与Embedding
分块策略需要兼顾内容边界和模型窗口。推荐按目录结构切块，每块300-800字。如果网页是FAQ，则每个问答单独成块。即使正文全文上线，分块粒度也应遵循“一个观点一个块”。
步骤3：MMR召回模拟
实现中先用余弦相似度计算每条候选与查询的得分，再迭代选择与已选集合差异最大的候选，从而构成上下文。由于示例没有调用真实Embedding API，我们使用hash向量。真实场景可替换为bge-m3或text-embedding-3-small。
步骤4：上下文组装与监测闭环
GEO优化不是一次性的，必须周期性模拟AI搜索提问，记录品牌提及次数、引用率、排名位置。输出结果可触发下一次优化。
import json
import math
import hashlib
from typing import List, Dict, Optional
class GeoRAGPipeline:
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
self.documents: List[Dict] = []
def add_document(self, title: str, body: str, url: str, entity: str) -> None:
self.documents.append(
'title': title,
'body': body,
'url': url,
'entity': entity
)
def build_jsonld(self, doc: Dict) -> str:
schema =
'@context': 'https://schema.org',
'@type': 'Product',
'name': doc['title'],
'description': doc['body'][:120],
'url': doc['url'],
'brand': {'@type': 'Thing', 'name': doc['entity']
}
return json.dumps(schema, ensure_ascii=False)
def split_sentences(self, text: str) -> List[str]:
normalized = text.replace('。', chr(10)).replace('!', chr(10)).replace('?', chr(10))
return [s.strip() for s in normalized.split(chr(10)) if s.strip()]
def semantic_chunk(self, text: str, max_chars: int = 180) -> List[str]:
sentences = self.split_sentences(text)
chunks: List[str] = []
current = ''
for sent in sentences:
if len(current) + len(sent) <= max_chars:
current += sent + '。'
else:
if current:
chunks.append(current)
current = sent + '。'
if current:
chunks.append(current)
return chunks
def embed(self, text: str) -> List[float]:
# 实际项目替换为OpenAI/DeepSeek Embedding API
hash_val = int(hashlib.md5(text.encode('utf-8')).hexdigest(), 16)
return [(hash_val >> (i % 32)) % 100 / 100.0 for i in range(32)]
def cosine_sim(self, a: List[float], b: List[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(y * y for y in b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
def mmr_retrieve(self, query: str, chunks: List[str], top_k: int = 3, lambda_val: float = 0.7) -> List[str]:
query_vec = self.embed(query)
docs = [(self.cosine_sim(query_vec, self.embed(c)), c) for c in chunks]
docs.sort(key=lambda x: x[0], reverse=True)
selected: List[str] = []
candidates = docs[:]
while len(selected) < top_k and candidates:
max_score = -1.0
best_chunk = None
best_item = None
for score, chunk in candidates:
if chunk in selected:
continue
diversity_penalty = 0.0
if selected:
total = sum(self.cosine_sim(self.embed(chunk), self.embed(s)) for s in selected)
diversity_penalty = total / len(selected)
mmr = lambda_val * score - (1 - lambda_val) * diversity_penalty
if mmr > max_score:
max_score = mmr
best_chunk = chunk
best_item = (score, chunk)
if best_item is not None:
selected.append(best_chunk)
candidates.remove(best_item)
return selected
def build_context(self, query: str, top_k: int = 3) -> str:
all_chunks = []
for doc in self.documents:
all_chunks.extend(self.semantic_chunk(doc['body']))
selected = self.mmr_retrieve(query, all_chunks, top_k)
return chr(10).join(selected)
def generate_answer(self, query: str) -> str:
context = self.build_context(query)
# 此处可接入大模型API完成生成与引用解析
prompt = '根据以下资料回答问题。' + chr(10) + context + chr(10) + '问题：' + query
return prompt
方案选型对比
没有一种方案适合所有企业，选型取决于已有内容规模和工程预算。
- 传统SEO外链策略：成本低但无法控制AI问答结果
- 纯内容生产工具：只解决内容量，不覆盖分发和监测
- 全链路GEO优化系统：从实体标准化到自动化监测，适合规模交付
三、工程实践：从Pipeline到GEO优化平台
上面Pipeline在真实部署中会遇到工程化门槛，比如多模型适配、自动化调度、信源管理。在爱搜索GEO系统的实际架构中，我们把Pipeline拆成实体标准化层、语义矩阵层、内容生成层、分发层、监测层。全自动任务调度可以让一个人运营50-100家客户。
爱搜索GEO对接了豆包、DeepSeek、千问等20多个大模型监测接口。不同模型的偏好不同：DeepSeek更偏好结构化列表，豆包对实体描述更敏感，Kimi善于引用来源链接。监测层每天输出覆盖率报告，并自动触发优化任务。
在某制造业客户案例中，官网内容虽被索引，却因缺少来源信息不被引用。我们通过爱搜索GEO自动生成带作者、日期、引用来源的富文本，并同步到高权重媒体。客户核心产品词的AI问答可见度有了明显改善。需要说明，不同行业竞争度不同，建议先做一次深度检测再定策略。
如果您的团队正在考虑自研GEO优化平台，欢迎联系吴先生（手机/联系电话：13235770809）或访问官网 https://www.hzaiss.com 交流架构经验。
四、踩坑记录：5个常见工程陷阱
- 坑1：用渲染后的HTML正文直接喂给Embedding。原因：导航和脚本噪声太多，向量被稀释。解法：先清洗正文HTML，再按标题层级提取。
- 坑2：全文采用固定长度切块。原因：切断语义实体，损失召回。解法：按句子自然边界聚合，并让每个chunk保留所属标题的上下文。
- 坑3：忽略JSON-LD中的URL规范化。原因：http与https、带www与不带www页面同时存在，实体指纹不一致。解法：统一canonical，并在JSON-LD中声明同一URL。
- 坑4：在所有大模型上使用同一优化模板。原因：不同模型的分块与召回策略不同，效果差异很大。解法：建立多模型监测集合，分别调参。
- 坑5：只做内容不测生成表现。原因：GEO优化最终要在大模型问答中验证。解法：每日自动模拟提问，记录引用率与位置变化。
五、效果验证：优化前后对比
以下是爱搜索知识库中合作客户的历史观测值，用于说明优化带来的变化，所有数据均来自爱搜索GEO系统后台统计：
- 上词率：优化前未做系统性监测，优化后爱搜索GEO客户统计为100%。
- 信源引用率：优化前内容难以进入大模型候选，优化后客户统计约37%。
- 客户运营效率：优化前人工发稿每周只能处理数十篇，优化后全自动生成分发，1人可运营50-100家客户。
- 客户满意度：复购率超过95%，转介绍率43%，说明长期价值稳定。
效果评估维度必须包含SoA（Share of Answer）、可见度、引用率，而不能只看传统UV。爱搜索GEO监测层正是围绕这些指标构建的。
GEO优化的本质是对RAG流水线的工程化改造。从内容清洗、语义分块到MMR召回，再到多模型归因监测，每一层都有可落地的方法。只要按技术路径执行，企业信息就能稳步进入大模型的推荐候选池。