ChatGPT归档技术实战：从数据管理到高效检索的完整解决方案

1次阅读

没有评论

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

随着 ChatGPT 应用的普及，用户与 AI 的对话数据呈爆炸式增长。这些数据具有以下特点：

非结构化文本为主，包含大量上下文关联
单次对话可能跨越多个回合，形成会话链
数据量每天可达 TB 级别
用户需要快速检索历史对话

传统的关系型数据库在这种场景下面临严峻挑战：

全文检索效率低下，LIKE 查询无法满足性能需求
海量数据导致索引膨胀，写入性能急剧下降
缺乏语义理解能力，难以实现意图匹配搜索

针对 ChatGPT 归档场景，我们对比了主流存储方案：

方案	优点	缺点	适用场景
PostgreSQL	ACID 支持完善，事务性强	全文检索性能有限	结构化数据占比高的场景
Elasticsearch	分布式搜索性能优异	资源消耗大，运维复杂	纯搜索场景
MongoDB	灵活的模式，水平扩展容易	内存占用高，查询语法复杂	文档型数据存储
混合方案	各取所长，平衡性能与功能	系统复杂度增加	中大型生产环境

推荐采用 Elasticsearch+PostgreSQL 混合方案：

PostgreSQL 存储原始对话数据和元信息
Elasticsearch 专门处理搜索请求
通过消息队列实现数据同步

系统架构分为四层：

数据采集层：从 ChatGPT API 获取原始对话流
处理层：进行数据清洗、标准化和向量化
存储层：双写 PostgreSQL 和 Elasticsearch
查询层：处理用户搜索请求，合并结果

graph TD
    A[ChatGPT API] --> B[数据采集]
    B --> C[消息队列]
    C --> D[数据处理]
    D --> E[PostgreSQL]
    D --> F[Elasticsearch]
    E --> G[查询服务]
    F --> G
    G --> H[前端展示]

from datetime import datetime
from pydantic import BaseModel

def normalize_text(text: str) -> str:
    """标准化对话文本，移除特殊字符和多余空格"""
    return ' '.join(text.strip().split())

class Dialogue(BaseModel):
    """对话数据模型"""
    dialog_id: str
    user_id: str
    created_at: datetime
    turns: list[dict]  # [{role: 'user'|'assistant', content: str}]
    metadata: dict = {}

    def to_es_doc(self):
        """转换为 Elasticsearch 文档格式"""
        return {
            'dialog_id': self.dialog_id,
            'content': '\n'.join([f"{turn['role']}: {normalize_text(turn['content'])}" 
                for turn in self.turns
            ]),
            'embedding': generate_embedding(self.turns[-1]['content']),
            **self.metadata
        }

from elasticsearch.helpers import bulk

def batch_import(data: list[Dialogue], es_client, batch_size=500):
    """批量导入优化，减少网络 IO"""
    actions = [
        {
            '_op_type': 'index',
            '_index': 'chatgpt-dialogues',
            '_source': d.to_es_doc()}
        for d in data
    ]

    # 分批提交避免内存溢出
    for i in range(0, len(actions), batch_size):
        bulk(es_client, actions[i:i+batch_size])
        logger.info(f'已导入 {i+batch_size}/{len(actions)} 条数据')

# Elasticsearch 索引配置
INDEX_SETTINGS = {
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 1,
        "analysis": {
            "analyzer": {
                "dialog_analyzer": {
                    "type": "custom",
                    "tokenizer": "whitespace",
                    "filter": ["lowercase", "stemmer"]
                }
            }
        }
    },
    "mappings": {
        "properties": {"content": {"type": "text", "analyzer": "dialog_analyzer"},
            "embedding": {"type": "dense_vector", "dims": 768},
            "created_at": {"type": "date"}
        }
    }
}

from sentence_transformers import SentenceTransformer

# 初始化模型(提前加载)
encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def hybrid_search(es_client, query: str, user_id: str, size=10):
    """混合搜索: 结合关键词和语义"""
    embedding = encoder.encode(query)

    return es_client.search(
        index='chatgpt-dialogues',
        body={
            "query": {
                "bool": {
                    "must": [{"match": {"content": query}},
                        {"term": {"user_id": user_id}}
                    ],
                    "should": [
                        {"script_score": {"query": {"match_all": {}},
                            "script": {"source": "cosineSimilarity(params.query_vector,'embedding') + 1.0",
                                "params": {"query_vector": embedding.tolist()}
                            }
                        }}
                    ]
                }
            },
            "size": size
        }
    )