ChatGPT聊天记录归档与检索难题的工程化解决方案

1次阅读

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

存储瓶颈问题
ChatGPT 对话日志通常以 JSON 格式存储，单条消息平均占用 2 -4KB
日活 10 万用户的系统每月会产生约 1.2TB 原始数据（按每用户 20 条 / 天计算）
传统 MySQL 的 B + 树索引在数据量超 500 万条后，模糊查询延迟显著增加
查询性能缺陷
LIKE '%keyword%' 查询在千万级数据下响应时间超过 3 秒
上下文关联查询需要多次 JOIN 操作（如查找同一对话中的所有消息）
无法支持语义相似度搜索（如『找和订单取消相关的对话』）
典型业务场景
客服系统需在 30 秒内定位用户历史投诉记录
金融场景要求 6 个月内任意交易咨询对话可追溯
教育应用需要跨会话分析学习者的知识点掌握情况

搜索引擎对比
Elasticsearch：成熟分布式架构，支持混合查询，社区资源丰富
Milvus：专为向量优化，但缺乏文本检索原生支持
Pinecone：全托管服务，成本较高且定制化能力有限
混合索引设计
倒排索引：处理精确匹配（如订单号、日期范围）
HNSW 向量索引：支持 cosine 相似度计算（维度设为 768）

字段组合策略：

{
  "mappings": {
    "properties": {"text": { "type": "text", "analyzer": "ik_max_word"},
      "vector": {"type": "dense_vector", "dims": 768}
    }
  }
}

BERT 向量化流程
使用 sentence-transformers/all-MiniLM-L6-v2 模型
预处理步骤：
1. 合并对话轮次（最多 6 条消息）
2. 清除特殊字符和停用词
3. 中英文统一转小写
批处理加速：每 100 条消息作为 1 个 batch 推理

消息分片存储

def generate_shard_id(msg_id: str, total_shards: int) -> int:
    """SHA-256 哈希分片（避免热点）"""
    return int(sha256(msg_id.encode()).hexdigest(), 16) % total_shards

async def bulk_index(docs: List[Dict], index_name: str):
    """异步批量写入（每批 500 条）"""
    from elasticsearch.helpers import async_bulk
    await async_bulk(es_client, [{
        '_op_type': 'index',
        '_index': f"{index_name}-shard{generate_shard_id(doc['id'], 12)}",
        '_source': doc
    } for doc in docs])

混合查询 DSL

def build_hybrid_query(keywords: str, vector: List[float], boost_ratio=0.7):
    return {
        "query": {
            "bool": {
                "should": [{ "match": { "text": keywords}},
                    { "script_score": {"query": { "match_all": {}},
                        "script": {"source": "cosineSimilarity(params.query_vector,'vector') + 1.0",
                            "params": {"query_vector": vector}
                        }
                    }}
                ],
                "boost": boost_ratio
            }
        },
        "size": 10
    }

存储策略
热数据（30 天内）：SSD 存储，3 副本
冷数据（30 天前）：HDD 存储，1 副本

生命周期管理：

PUT _ilm/policy/hot_cold_policy {
  "phases": {"hot": { "min_age": "0d", "actions": {"rollover": {"max_size": "50gb"}} },
    "cold": {"min_age": "30d", "actions": {"allocate": {"require": {"data": "cold"}}} }
  }
}

性能公式
分片数计算：shards = max(2, ceil(peak_qps / 5000))
存储预估：disk_gb = total_vectors × (4 × dim + 128) / (1024³)
示例：1 亿条 768 维向量约需 286GB

向量维度控制
超过 1024 维会导致查询延迟指数增长
建议方案：
- 先用 PCA 降维（保留 95% 方差）
- 对短文本使用蒸馏模型（如 all-MiniLM-L6-v2）

哈希冲突检测

def detect_hash_collision(shard_id: int, sample_size=10000):
    query = {"size": 0, "aggs": {"unique_ids": {"cardinality": {"field": "id"}}}}
    res = es_client.count(index=f"*-shard{shard_id}", body=query)
    return res['count'] > res['aggregations']['unique_ids']['value']