从零开始：如何自己开发一个ChatGPT对接接口（新手避坑指南）

4次阅读

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

ChatGPT API 提供了强大的自然语言处理能力，可以实现智能对话、文本生成和语言理解等功能。典型应用场景包括客服机器人、内容创作助手和编程辅助工具。通过 API 调用，开发者可以快速集成这些能力到自己的应用中。

直接使用 OpenAI 官方 SDK 是最快捷的方式，但自行封装接口能带来更多灵活性。

官方 SDK 优点 ：简单易用，内置认证和错误处理
自行封装优点 ：
更精细的 token 使用控制，节省成本
可自定义超时和重试策略
便于添加业务特定的预处理和后处理

永远不要将 API 密钥硬编码在代码中。推荐两种安全存储方式：

# 方式 1：环境变量
import os
api_key = os.environ.get('OPENAI_API_KEY')

# 方式 2：HashiCorp Vault（更安全）import hvac
client = hvac.Client(url='https://vault.example.com')
api_key = client.read('secret/openai')['data']['api_key']

使用 aiohttp 实现高效异步调用：

import aiohttp

async def chat_completion(prompt):
    headers = {'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }

    payload = {
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}],
        'temperature': 0.7  # 控制生成随机性，0- 2 之间
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers=headers,
            json=payload
        ) as resp:
            return await resp.json()

处理大文本时使用流式响应避免内存问题：

async def stream_response(prompt):
    payload = {
        'model': 'gpt-3.5-turbo',
        'messages': [{'role': 'user', 'content': prompt}],
        'stream': True  # 开启流式
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(
            'https://api.openai.com/v1/chat/completions',
            headers=headers,
            json=payload
        ) as resp:
            # 注意：这里要使用异步迭代器
            async for chunk in resp.content:
                yield chunk.decode('utf-8')  # 逐块处理

实现指数退避重试机制：

import asyncio
import random

async def request_with_retry(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await chat_completion(prompt)
        except aiohttp.ClientResponseError as e:
            if e.status == 429:
                wait = min((2 ** attempt) + random.random(), 60)
                await asyncio.sleep(wait)
                continue
            raise

确保日志中不泄露敏感信息：

import logging

def safe_log(response):
    sanitized = response.copy()
    sanitized.pop('api_key', None)
    sanitized['messages'] = '[REDACTED]'
    logging.info(f'API response: {sanitized}')

使用 tiktoken 库精确计算 token 使用量：

import tiktoken

def count_tokens(messages):
    encoding = tiktoken.encoding_for_model('gpt-3.5-turbo')
    return sum(len(encoding.encode(msg['content'])) for msg in messages)