从零实现电脑端ChatGPT接入：Python API集成与避坑指南

3次阅读

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

ChatGPT API 为开发者提供了直接调用大模型能力的通道，能快速为应用添加智能对话功能；其按 token 计费的模式相比传统 NLP 服务更具成本优势；同时支持流式响应和异步调用，适合构建实时交互场景。

原生 API 优势：
更细粒度控制请求参数和响应处理
避免 SDK 版本兼容性问题
适合需要深度定制的场景
SDK 优势：
简化身份验证和错误处理
内置类型提示和文档支持
快速集成（如 openai 库仅需 3 行代码）

import os
from dotenv import load_dotenv  # pip install python-dotenv

# 最佳实践：将 API_KEY 存储在.env 文件，并加入.gitignore
load_dotenv()
API_KEY = os.getenv('OPENAI_API_KEY')  # 从环境变量读取

import openai
from tenacity import retry, stop_after_attempt, wait_exponential  # pip install tenacity

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def chat_completion_with_retry(messages):
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages,
        temperature=0.7,
        stream=True  # 启用流式响应
    )

def handle_stream_response(response):
    full_content = ""
    for chunk in response:
        if chunk.choices[0].delta.get('content'):
            content = chunk.choices[0].delta.content
            full_content += content
            yield content  # 实时生成内容
    logger.info(f"Total tokens used: {full_content}")

import httpx  # pip install httpx

client = httpx.Client(
    limits=httpx.Limits(
        max_connections=100,  # 最大连接数
        max_keepalive_connections=20  # 保持活跃连接数
    ),
    timeout=30.0  # 超时设置(秒)
)

import asyncio

async def async_chat_request(messages):
    async with openai.AsyncOpenAI() as client:
        stream = await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            stream=True
        )
        async for chunk in stream:
            print(chunk.choices[0].delta.get('content', ''), end='')

import logging

class SensitiveDataFilter(logging.Filter):
    def filter(self, record):
        if hasattr(record, 'msg') and 'sk-' in record.msg:
            record.msg = '[API_KEY REDACTED]'
        return True

logger.addFilter(SensitiveDataFilter())