ChatGPT数据收集实战指南：从零搭建高效采集系统

1次阅读

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

在使用 ChatGPT API 进行大规模数据收集时，开发者常遇到三类典型问题：

API 速率限制 ：OpenAI 对免费和付费 API 都有严格的每分钟请求次数（RPM）和每分钟 Tokens 限制，直接同步请求极易触发限流
数据格式不一致 ：API 返回的 JSON 结构可能随版本更新变化，且不同对话模式（completion/chat）的响应格式差异显著
存储瓶颈 ：单机环境下高频写入可能导致：
内存溢出（OOM）
磁盘 IO 竞争
检索性能下降

同步请求 ：
优点：实现简单，调试方便
缺点：吞吐量低（约 10-20 QPS）
适用场景：小批量测试数据采集
异步请求 ：
优点：理论吞吐量提升 5 -10 倍
缺点：需要处理协程调度和异常传递
适用场景：生产环境持续采集

类型	写入速度	查询效率	持久化	适用场景
Redis	10w+ ops/s	O(1)	可选	实时去重缓存
SQLite	1k-5k ops/s	O(log n)	是	中小规模结构化存储
Elasticsearch	5k-10k ops/s	O(1)~O(log n)	是	全文检索场景

import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt

class AsyncChatGPTCollector:
    def __init__(self, api_key, max_concurrent=50):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = aiohttp.ClientSession(headers={"Authorization": f"Bearer {api_key}"}
        )

    @retry(stop=stop_after_attempt(3))
    async def fetch_data(self, prompt):
        async with self.semaphore:
            try:
                async with self.session.post(
                    "https://api.openai.com/v1/chat/completions",
                    json={"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}]}
                ) as resp:
                    return await resp.json()
            except Exception as e:
                print(f"Request failed: {str(e)}")
                raise

时间复杂度分析 ：
– 单次请求：O(1) 网络 IO + O(1)JSON 解析
– 整体吞吐：O(n)/ 线程，实际受限于网络带宽和 API 限额

from pybloom_live import ScalableBloomFilter
import hashlib

class DeduplicationEngine:
    def __init__(self, initial_capacity=1000000, error_rate=0.001):
        self.filter = ScalableBloomFilter(
            initial_capacity=initial_capacity, 
            error_rate=error_rate
        )

    def is_duplicate(self, text):
        text_hash = hashlib.sha256(text.encode()).hexdigest()
        if text_hash in self.filter:
            return True
        self.filter.add(text_hash)
        return False

空间复杂度 ：O(n)，实际占用约 1.44nlog2(1/ε) bits（ε 为错误率）

import sqlite3
from datetime import datetime
import os

class ShardedDB:
    def __init__(self, base_path="data", shard_size=100000):
        self.shard_size = shard_size
        self.base_path = base_path
        os.makedirs(base_path, exist_ok=True)

    def get_shard_name(self, idx):
        shard_num = idx // self.shard_size
        return f"{self.base_path}/shard_{shard_num}.db"

    def insert_data(self, data):
        shard_path = self.get_shard_name(data['id'])
        with sqlite3.connect(shard_path) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS responses (
                    id INTEGER PRIMARY KEY,
                    prompt TEXT,
                    response TEXT,
                    timestamp DATETIME
                )
            ''')
            cursor.execute('''INSERT INTO responses VALUES (?, ?, ?, ?)''', (data['id'], data['prompt'], data['response'], datetime.now()))

import time

class AdaptiveRateLimiter:
    def __init__(self, initial_delay=0.1):
        self.last_request_time = 0
        self.current_delay = initial_delay

    async def throttle(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.current_delay:
            adjust_factor = 1 + (self.current_delay - elapsed) / self.current_delay
            self.current_delay *= min(adjust_factor, 2.0)
            await asyncio.sleep(self.current_delay - elapsed)
        else:
            self.current_delay *= 0.9  # 逐渐加速
        self.last_request_time = time.time()

import psutil
import logging

class MemoryMonitor:
    def __init__(self, threshold=0.8):
        self.threshold = threshold

    def check_memory(self):
        mem = psutil.virtual_memory()
        if mem.percent > self.threshold * 100:
            logging.warning(f"Memory usage {mem.percent}% exceeds threshold"
                f"{self.threshold*100}%, triggering GC"
            )
            return True
        return False

严格遵守 OpenAI 使用政策
避免采集：
个人隐私信息
违法内容
暴力 / 仇恨言论
建议添加内容过滤层：

def content_filter(text):
    forbidden_phrases = ["credit card", "SSN", "password"]
    return any(phrase in text.lower() for phrase in forbidden_phrases)

当部署多采集节点时：