ChatGPT对话归档技术解析：从存储策略到检索优化

1次阅读

没有评论

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

ChatGPT 对话数据具有高频生成、非结构化、上下文关联的特点。与传统业务数据不同，这些对话数据往往包含大量文本内容，且需要长期保存以便后续检索和分析。直接使用传统关系型数据库存储这类数据会面临几个主要问题：

存储膨胀问题：单次对话可能包含数千字，日积月累会导致数据库体积爆炸式增长
检索效率低下：简单的 SQL LIKE 查询无法满足语义搜索需求，且随着数据量增加性能急剧下降
上下文丢失：传统的分页查询会破坏对话的连贯性，难以还原完整对话脉络

MongoDB 的文档模型天然适合存储非结构化对话数据，其分片 (sharding) 功能可以有效分散存储压力：

优势：
自动水平扩展能力
内置压缩算法(WiredTiger 引擎)
支持时间序列集合(Time Series Collections)
局限：
语义检索能力有限
分片键选择不当会导致热点问题

专为搜索优化的分布式引擎：

优势：
强大的文本分析能力(分词、同义词等)
近实时 (NRT) 索引
丰富的聚合查询
局限：
存储成本较高
向量检索性能一般

专业的向量数据库：

优势：
支持亿级向量毫秒查询
多种相似度计算方式
自动索引构建
局限：
需要额外维护向量数据
内存消耗较大

综合各方案优势，推荐三层存储架构：

原始对话存储层：
使用 S3 存储完整对话历史
采用 gzip 压缩(平均可减少 70% 体积)
按用户 ID+ 时间分区存储
元数据存储层：
PostgreSQL 关系型数据库
存储对话 ID、时间戳、用户标识等结构化数据
建立 BTREE 索引加速时间范围查询
向量索引层：
Milvus 存储对话内容的 embeddings
使用 HNSW 索引算法
定期合并小段对话为语义段落

import boto3
import gzip
from datetime import datetime
from typing import List, Dict

class DialogArchiver:
    def __init__(self, s3_bucket: str):
        self.s3 = boto3.client('s3')
        self.bucket = s3_bucket

    def _compress_content(self, text: str) -> bytes:
        try:
            return gzip.compress(text.encode('utf-8'))
        except Exception as e:
            raise RuntimeError(f"Compression failed: {str(e)}")

    def save_dialog(self, user_id: str, dialog_id: str, turns: List[Dict]) -> str:
        """
        存储对话到 S3 并返回对象路径
        :param turns: [{'role':'user|assistant', 'content':'...'}, ...]
        """
        timestamp = datetime.utcnow().isoformat()
        s3_key = f"dialogs/{user_id}/{timestamp[:10]}/{dialog_id}.json.gz"

        try:
            compressed = self._compress_content(json.dumps(turns))
            self.s3.put_object(
                Bucket=self.bucket,
                Key=s3_key,
                Body=compressed,
                Metadata={'user_id': user_id}
            )
            return s3_key
        except Exception as e:
            raise RuntimeError(f"S3 upload failed: {str(e)}")

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class VectorSearcher:
    def __init__(self, model_name: str='paraphrase-multilingual-MiniLM-L12-v2'):
        self.model = SentenceTransformer(model_name)
        self.index = faiss.IndexFlatIP(384)  # 向量维度
        self.id_map = {}

    def add_dialog(self, dialog_id: str, text: str):
        """将对话内容添加到向量索引"""
        try:
            embedding = self.model.encode(text)
            idx = self.index.ntotal
            self.index.add(np.array([embedding]))
            self.id_map[idx] = dialog_id
        except Exception as e:
            raise RuntimeError(f"Index update failed: {str(e)}")

    def search(self, query: str, k: int=5) -> List[str]:
        """语义搜索返回最相关的对话 ID"""
        try:
            query_embedding = self.model.encode(query)
            distances, indices = self.index.search(np.array([query_embedding]), k
            )
            return [self.id_map[i] for i in indices[0] if i in self.id_map]
        except Exception as e:
            raise RuntimeError(f"Search failed: {str(e)}")