Python调用ChatGPT API实战指南：从基础实现到生产环境避坑

2次阅读

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

ChatGPT API 基于 GPT 模型提供文本生成能力，主要特点包括：

多轮对话管理 ：支持上下文关联的连续对话
长文本生成 ：单次请求可处理长达 4000 tokens 的内容
格式控制 ：支持 JSON 结构化输出
可调创造性 ：通过 temperature 参数控制生成随机性

典型应用场景：

智能客服对话系统
内容自动生成（邮件 / 报告 / 代码）
语言翻译与润色
知识问答引擎

import requests

headers = {"Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

data = {
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello!"}]
}

response = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers=headers,
    json=data
)

优势：
– 无额外依赖
– 完全控制请求流程

劣势：
– 需要手动处理所有错误
– 缺乏类型提示

import openai

openai.api_key = API_KEY
response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello!"}]
)

选型建议 ：
– 快速验证场景：官方库
– 生产环境定制需求：原生 HTTP 请求

API 密钥管理推荐方案：

# config.py
import os
from dotenv import load_dotenv

load_dotenv()
API_KEY = os.getenv("OPENAI_API_KEY")

# .env 文件（加入.gitignore）OPENAI_API_KEY=your_key_here

import aiohttp

async def async_chat_completion(messages):
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"model": "gpt-3.5-turbo", "messages": messages}
        ) as resp:
            if resp.status != 200:
                error = await resp.text()
                raise Exception(f"API error: {error}")
            return await resp.json()

def stream_response(messages):
    with requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers=headers,
        json={
            "model": "gpt-3.5-turbo",
            "messages": messages,
            "stream": True
        },
        stream=True
    ) as resp:
        for line in resp.iter_lines():
            if line:
                chunk = json.loads(line.decode("utf-8"))
                yield chunk

from dataclasses import dataclass
from typing import List, Literal

@dataclass
class ChatMessage:
    role: Literal["system", "user", "assistant"]
    content: str

@dataclass
class CompletionParams:
    model: str = "gpt-3.5-turbo"
    temperature: float = 0.7
    max_tokens: int = 1000

# 使用示例
params = CompletionParams()
messages = [ChatMessage(role="user", content="Hello")]

async def batch_completion(message_list):
    async with aiohttp.ClientSession() as session:
        tasks = [async_chat_completion(session, messages)
            for messages in message_list
        ]
        return await asyncio.gather(*tasks)

from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=4, max=60))
def call_with_retry(messages):
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages
    )

class Conversation:
    def __init__(self, max_history=5):
        self.history = []
        self.max_history = max_history

    def add_message(self, role, content):
        self.history.append({"role": role, "content": content})
        if len(self.history) > self.max_history:
            self.history.pop(0)

# 使用 AWS KMS 示例
import boto3

kms = boto3.client("kms")

def decrypt_key(encrypted_key):
    return kms.decrypt(CiphertextBlob=base64.b64decode(encrypted_key)
    )["Plaintext"].decode()

proxies = {
    "http": "http://proxy.example.com:8080",
    "https": "http://proxy.example.com:8080"
}

response = requests.post(
    API_ENDPOINT,
    proxies=proxies,
    headers=headers,
    json=data
)

# Prometheus 埋点示例
from prometheus_client import Counter, Histogram

API_CALLS = Counter("openai_calls_total", "API call count")
RESPONSE_TIME = Histogram("openai_response_seconds", "API latency")

@RESPONSE_TIME.time()
def track_call(messages):
    API_CALLS.inc()
    return call_api(messages)

import unittest
from unittest.mock import patch

class TestChatAPI(unittest.TestCase):
    @patch("requests.post")
    def test_api_call(self, mock_post):
        mock_response = unittest.mock.Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {"choices": [{"message": {"content": "Hi"}}]}
        mock_post.return_value = mock_response

        response = call_chatgpt([{"role": "user", "content": "Hello"}])
        self.assertIn("Hi", response["choices"][0]["message"]["content"])