GEO 语义向量优化
实战指南

# 使用 MTEB 评测嵌入模型的检索能力
from mteb import MTEB
from sentence_transformers import SentenceTransformer

# 加载待评测模型
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

# 选择中文检索评测任务
evaluation = MTEB(tasks=["T2Retrieval", "MMarcoRetrieval", "CQADupstackRetrieval"])

# 运行评测
evaluation.run(model, output_folder="results/bge-large-zh")

# 对比多个模型的检索 NDCG@10 得分
# BGE-large-zh-v1.5:  ~0.83
# M3E-base:           ~0.72
# text-emb-3-small:   ~0.68 (中文)
# GTE-Qwen2-7B:       ~0.88+

# 优化前：语义模糊，信息密度低
我们提供很好的AI搜索优化服务，可以帮助你的网站在AI中表现更好。
我们的方法很先进，效果也不错，很多客户都满意。

# 优化后：语义清晰，信息密度高
我们提供 GEO（生成式引擎优化，Generative Engine Optimization）服务，
通过语义向量优化、结构化内容增强和引用率提升三大策略，
帮助网站内容在 AI 搜索引擎（Perplexity、Google AI Overview、Bing Copilot）
的语义检索中获得更高匹配度和引用率。典型客户的 AI 引用率提升 40%~120%。

# 优化前：歧义表达，向量定位不准
这个模型效果很好，可以处理各种任务。

# 优化后：精确表达，向量精准定位
BGE-large-zh-v1.5 嵌入模型在 C-MTEB 中文检索基准上 NDCG@10 达到 0.83，
支持文本向量化、语义搜索、文档聚类等 NLP 检索任务。

"""
语义密度计算工具
通过嵌入向量分析文本的语义信息密度
"""
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

def calculate_semantic_density(text, model, chunk_size=1):
    """
    计算文本的语义密度
    核心思路：将文本按句子切分，计算每句向量之间的平均余弦距离
    - 距离越大 → 各句语义方向差异大 → 信息覆盖广 → 密度高
    - 距离越小 → 各句语义方向相似 → 信息冗余 → 密度低
    """
    # 按句子切分
    sentences = [s.strip() for s in text.split('。') if s.strip()]
    if len(sentences) < 2:
        return 1.0  # 单句无法计算密度

    # 获取句子向量
    embeddings = model.encode(sentences, normalize_embeddings=True)

    # 计算两两余弦距离
    sim_matrix = cosine_similarity(embeddings)
    n = len(sentences)

    # 取上三角的平均距离（排除对角线自身）
    total_distance = 0.0
    count = 0
    for i in range(n):
        for j in range(i + 1, n):
            total_distance += 1.0 - sim_matrix[i][j]  # 距离 = 1 - 相似度
            count += 1

    avg_distance = total_distance / count

    # 归一化到 [0, 1]，经验上 avg_distance 在 0~0.5 范围
    density = min(avg_distance / 0.3, 1.0)
    return density

def detect_semantic_redundancy(text, model, threshold=0.95):
    """检测语义冗余的句子对"""
    sentences = [s.strip() for s in text.split('。') if s.strip()]
    embeddings = model.encode(sentences, normalize_embeddings=True)
    sim_matrix = cosine_similarity(embeddings)

    redundant_pairs = []
    for i in range(len(sentences)):
        for j in range(i + 1, len(sentences)):
            if sim_matrix[i][j] > threshold:
                redundant_pairs.append({
                    "sentence_1": sentences[i],
                    "sentence_2": sentences[j],
                    "similarity": round(sim_matrix[i][j], 4)
                })
    return redundant_pairs

# 使用示例
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

text_low = "GEO优化很重要。GEO优化非常关键。GEO优化对网站至关重要。"
text_high = "GEO优化通过语义向量增强提升AI引用率。具体方法包括嵌入模型选型、文本语义改写和结构化标记。实测效果显示引用率提升40%~120%。"

density_low = calculate_semantic_density(text_low, model)
density_high = calculate_semantic_density(text_high, model)

print(f"低密度文本: {density_low:.3f}")   # 预期 ~0.05
print(f"高密度文本: {density_high:.3f}")  # 预期 ~0.60+

# 检测冗余
redundancies = detect_semantic_redundancy(text_low, model)
for r in redundancies:
    print(f"冗余: [{r['similarity']}] {r['sentence_1'][:20]}... ↔ {r['sentence_2'][:20]}...")

"""
NER 实体识别 + 实体消歧 + 关系抽取
用于 GEO 语义向量优化中的实体管理
"""
import spacy
from collections import defaultdict

# 加载中文 NER 模型
nlp = spacy.load("zh_core_web_trf")

# 实体消歧字典：别名 -> 标准实体
ENTITY_KB = {
    "BGE": "BAAI/bge-large-zh-v1.5",
    "智源嵌入模型": "BAAI/bge-large-zh-v1.5",
    "M3E": "moka-ai/m3e-base",
    "GEO": "Generative Engine Optimization",
    "生成式引擎优化": "Generative Engine Optimization",
}

def extract_and_link_entities(text):
    """提取实体并链接到知识库标准实体"""
    doc = nlp(text)
    entities = []

    for ent in doc.ents:
        # 尝试消歧链接
        canonical = ENTITY_KB.get(ent.text, ent.text)
        entities.append({
            "text": ent.text,           # 原始文本
            "label": ent.label_,        # 实体类型
            "canonical": canonical,     # 消歧后标准实体
            "start": ent.start_char,    # 位置
            "end": ent.end_char,
        })

    return entities

def extract_relations(text, entities):
    """
    基于依存句法分析提取实体间关系（简化版）
    实际生产中可使用深度学习关系抽取模型
    """
    doc = nlp(text)
    relations = []

    # 基于依存路径提取主谓宾三元组
    for token in doc:
        if token.dep_ == "nsubj":  # 主语
            subject = token.text
            predicate = token.head.text  # 谓语
            # 查找宾语
            for child in token.head.children:
                if child.dep_ == "dobj":
                    obj = child.text
                    relations.append({
                        "subject": subject,
                        "predicate": predicate,
                        "object": obj,
                    })

    return relations

# 使用示例
text = "BGE在C-MTEB检索基准上达到了0.83的NDCG得分"
entities = extract_and_link_entities(text)
relations = extract_relations(text, entities)

for e in entities:
    print(f"实体: {e['text']} → {e['canonical']} ({e['label']})")
for r in relations:
    print(f"三元组: ({r['subject']}, {r['predicate']}, {r['object']})")

"""
语义相似度计算与检索排名模拟
模拟 AI 搜索引擎的向量检索过程
"""
import numpy as np
from sentence_transformers import SentenceTransformer

def cosine_similarity(a, b):
    """计算两个向量的余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    """计算欧氏距离（转换为相似度）"""
    dist = np.linalg.norm(a - b)
    return 1.0 / (1.0 + dist)  # 距离越小越相似

def dot_product_similarity(a, b):
    """计算点积相似度（需归一化向量）"""
    return np.dot(a, b)

def semantic_search(query, documents, model, top_k=5):
    """
    模拟 AI 搜索引擎的语义检索
    1. 向量化查询和文档
    2. 计算相似度
    3. 排序返回 Top-K
    """
    # 向量化
    query_emb = model.encode([query], normalize_embeddings=True)[0]
    doc_embs = model.encode(documents, normalize_embeddings=True)

    # 计算余弦相似度（归一化后等价于点积）
    scores = np.dot(doc_embs, query_emb)

    # 排序
    top_indices = np.argsort(scores)[::-1][:top_k]

    results = []
    for idx in top_indices:
        results.append({
            "document": documents[idx],
            "score": round(float(scores[idx]), 4),
            "rank": len(results) + 1,
        })

    return results

# 使用示例
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

query = "如何优化网页内容让AI搜索引擎更容易引用"
documents = [
    "GEO优化通过语义向量增强提升AI引用率，核心方法包括嵌入模型选型和文本语义改写",
    "SEO是搜索引擎优化，主要通过关键词和外链提升排名",
    "语义嵌入模型将文本转换为高维向量，用于AI检索中的语义匹配",
    "Python是一种流行的编程语言，广泛用于数据科学和AI开发",
    "生成式引擎优化（GEO）是让内容被AI搜索引擎引用的优化策略",
]

results = semantic_search(query, documents, model, top_k=3)
for r in results:
    print(f"#{r['rank']} [{r['score']:.4f}] {r['document'][:50]}...")

# 预期输出：
# #1 [0.82+] GEO优化通过语义向量增强提升AI引用率...
# #2 [0.78+] 生成式引擎优化（GEO）是让内容被AI搜索引擎...
# #3 [0.65+] 语义嵌入模型将文本转换为高维向量...

"""
SentenceTransformer 核心用法
GEO 语义向量优化中的嵌入工具
"""
from sentence_transformers import SentenceTransformer
import numpy as np

# 1. 加载模型
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

# 2. 基础嵌入
text = "GEO优化通过语义向量增强提升AI引用率"
embedding = model.encode(text, normalize_embeddings=True)
print(f"向量维度: {embedding.shape}")     # (1024,)
print(f"向量范数: {np.linalg.norm(embedding):.4f}")  # 1.0000

# 3. 批量嵌入（高效处理大量文档）
documents = [
    "语义嵌入模型将文本转换为高维向量",
    "GEO优化是面向AI搜索引擎的内容优化策略",
    "向量数据库支持高效的ANN近似最近邻检索",
]
embeddings = model.encode(
    documents,
    normalize_embeddings=True,
    batch_size=32,        # 批量大小
    show_progress_bar=True, # 进度条
)

# 4. 相似度计算
query = "什么是GEO优化"
query_emb = model.encode(query, normalize_embeddings=True)
scores = np.dot(embeddings, query_emb)
for doc, score in zip(documents, scores):
    print(f"[{score:.4f}] {doc}")

# 5. BGE 模型推荐：添加指令前缀
# BGE 系列模型在查询端建议添加 "为这个句子生成表示以用于检索相关文章："
query_with_instruction = "为这个句子生成表示以用于检索相关文章：什么是GEO优化"
query_emb_enhanced = model.encode(query_with_instruction, normalize_embeddings=True)
scores_enhanced = np.dot(embeddings, query_emb_enhanced)
print(f"\n指令增强后最高分: {max(scores_enhanced):.4f}")
print(f"原始查询最高分:   {max(scores):.4f}")

"""
GEO 语义向量优化完整 Pipeline
1. 文本预处理与语义优化
2. 向量化与索引构建
3. 语义搜索测试
4. 效果评估与报告
"""
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.manifold import TSNE
import json

class GEOSemanticPipeline:
    def __init__(self, model_name="BAAI/bge-large-zh-v1.5"):
        self.model = SentenceTransformer(model_name)
        self.index = {}     # doc_id -> embedding
        self.docs = {}      # doc_id -> text

    def add_document(self, doc_id, text):
        """向索引中添加文档"""
        emb = self.model.encode(text, normalize_embeddings=True)
        self.index[doc_id] = emb
        self.docs[doc_id] = text

    def semantic_search(self, query, top_k=5):
        """语义检索"""
        # BGE 模型查询前缀
        query_text = f"为这个句子生成表示以用于检索相关文章：{query}"
        query_emb = self.model.encode(query_text, normalize_embeddings=True)

        scores = {}
        for doc_id, doc_emb in self.index.items():
            scores[doc_id] = float(np.dot(query_emb, doc_emb))

        sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [(doc_id, score, self.docs[doc_id]) for doc_id, score in sorted_results[:top_k]]

    def evaluate_ranking(self, queries_with_relevant, top_k=5):
        """
        评估检索效果
        queries_with_relevant: [(query, [relevant_doc_ids]), ...]
        """
        mrr_sum = 0.0
        recall_sum = 0.0
        ndcg_sum = 0.0

        for query, relevant_ids in queries_with_relevant:
            results = self.semantic_search(query, top_k=top_k)
            result_ids = [r[0] for r in results]

            # MRR (Mean Reciprocal Rank)
            for rank, rid in enumerate(result_ids, 1):
                if rid in relevant_ids:
                    mrr_sum += 1.0 / rank
                    break

            # Recall@K
            hits = len(set(result_ids) & set(relevant_ids))
            recall_sum += hits / len(relevant_ids)

        n = len(queries_with_relevant)
        return {
            "MRR": round(mrr_sum / n, 4),
            "Recall@K": round(recall_sum / n, 4),
        }

    def visualize_embeddings(self, labels=None):
        """t-SNE 降维可视化"""
        ids = list(self.index.keys())
        embs = np.array([self.index[i] for i in ids])

        tsne = TSNE(n_components=2, random_state=42, perplexity=min(30, len(ids)-1))
        coords = tsne.fit_transform(embs)

        return [{"id": ids[i], "x": float(coords[i][0]),
                  "y": float(coords[i][1]), "text": self.docs[ids[i]][:50]}
                for i in range(len(ids))]

# 使用示例
pipeline = GEOSemanticPipeline()

# 添加文档
pipeline.add_document("doc1", "GEO优化通过语义向量增强提升AI搜索引擎引用率")
pipeline.add_document("doc2", "SEO是传统搜索引擎优化，主要关注关键词排名")
pipeline.add_document("doc3", "语义嵌入模型将文本编码为高维向量用于语义检索")

# 语义搜索
results = pipeline.semantic_search("如何让AI引用我的网站内容")
for doc_id, score, text in results:
    print(f"[{score:.4f}] {doc_id}: {text[:40]}...")

"""
语义相似度基准测试
对比优化前后的内容与目标查询的相似度变化
"""
import numpy as np
from sentence_transformers import SentenceTransformer

def run_similarity_benchmark(model, queries, content_before, content_after):
    """
    运行语义相似度基准测试
    queries: 目标查询列表
    content_before: 优化前的内容列表
    content_after: 优化后的内容列表（与before一一对应）
    """
    query_embs = model.encode(queries, normalize_embeddings=True)
    before_embs = model.encode(content_before, normalize_embeddings=True)
    after_embs = model.encode(content_after, normalize_embeddings=True)

    results = []
    for i, query in enumerate(queries):
        # 找到与该查询最相似的优化前后内容
        before_scores = np.dot(before_embs, query_embs[i])
        after_scores = np.dot(after_embs, query_embs[i])

        best_before = float(max(before_scores))
        best_after = float(max(after_scores))
        improvement = best_after - best_before

        results.append({
            "query": query,
            "before_score": round(best_before, 4),
            "after_score": round(best_after, 4),
            "improvement": round(improvement, 4),
            "improvement_pct": round(improvement / best_before * 100, 1) if best_before > 0 else 0,
        })

    # 汇总
    avg_before = np.mean([r["before_score"] for r in results])
    avg_after = np.mean([r["after_score"] for r in results])
    print(f"平均相似度: {avg_before:.4f} → {avg_after:.4f} (提升 {(avg_after-avg_before)/avg_before*100:.1f}%)")

    return results