PageIndex 实现方案全解析：超越 ChatGPT 的替代方案与实战指南

2次阅读

共计 2451 个字符，预计需要花费 7 分钟才能阅读完成。

PageIndex（页面索引）在现代应用开发中扮演着至关重要的角色，尤其是在处理大量文本数据的搜索和分页场景中。许多开发者误以为只能依赖 ChatGPT 来实现 PageIndex，这主要源于对 ChatGPT 文本处理能力的过度依赖以及对其 API 的便捷性认知。然而，传统依赖单一 AI 模型的方案存在明显的局限性：

高成本：ChatGPT API 调用按 token 计费，频繁查询会导致成本快速上升
高延迟：AI 模型需要时间处理请求，不适合实时性要求高的场景
灵活性差：预训练模型的输出难以精确控制，无法满足特定业务需求

优点：

开箱即用的语义理解能力
无需构建复杂的文本处理管道

缺点：

成本随查询量线性增长
响应时间通常在 500ms-2s 之间
无法针对特定领域优化

优点：

专为搜索场景优化，查询延迟 <100ms
支持复杂的分词和评分策略
水平扩展能力强

缺点：

需要额外的基础设施
学习曲线较陡

优点：

完全可控的实现
无外部依赖
内存占用低

缺点：

需要手动实现高级功能
性能优化难度大

// 索引定义
PUT /page_index
{
  "settings": {
    "number_of_shards": 3,
    "analysis": {
      "analyzer": {
        "my_analyzer": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": ["lowercase", "stemmer"]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "content": {
        "type": "text",
        "analyzer": "my_analyzer"
      },
      "page_id": {"type": "keyword"}
    }
  }
}

// 分页查询 DSL
GET /page_index/_search
{
  "query": {
    "match": {"content": "搜索关键词"}
  },
  "from": 0,
  "size": 10,
  "sort": [
    {
      "_score": {"order": "desc"}
    }
  ]
}

from math import log
import numpy as np
from collections import defaultdict

class BM25:
    def __init__(self, k1=1.5, b=0.75):
        self.k1 = k1
        self.b = b
        self.documents = []
        self.avgdl = 0
        self.f = []  # 文档词频
        self.df = defaultdict(int)  # 文档频率
        self.idf = {}

    def add_document(self, document):
        """添加文档到索引"""
        self.documents.append(document)
        freq = defaultdict(int)
        for word in document:
            freq[word] += 1
        self.f.append(freq)
        for word in freq:
            self.df[word] += 1
        # 更新平均文档长度
        self.avgdl = sum(len(d) for d in self.documents) / len(self.documents)
        # 重新计算 IDF
        N = len(self.documents)
        for word in self.df:
            self.idf[word] = log((N - self.df[word] + 0.5) / (self.df[word] + 0.5) + 1)

    def search(self, query, top_n=10):
        """执行搜索"""
        scores = np.zeros(len(self.documents))
        query_words = set(query)

        for i, doc in enumerate(self.documents):
            dl = len(doc)
            for word in query_words:
                if word not in self.f[i]:
                    continue
                # BM25 计算公式
                idf = self.idf[word]
                tf = self.f[i][word]
                numerator = tf * (self.k1 + 1)
                denominator = tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
                scores[i] += idf * numerator / denominator

        # 返回 top_n 结果
        top_indices = np.argsort(scores)[::-1][:top_n]
        return [(i, scores[i]) for i in top_indices]