ChatGPT使用方法全解析：从API调用到生产环境最佳实践

1次阅读

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

ChatGPT API 是基于 GPT 模型构建的对话式 AI 服务接口，其核心是通过 HTTP 请求与云端模型交互。与直接使用网页版不同，API 允许开发者将 AI 能力集成到自己的应用中，实现定制化对话流程。典型使用场景包括：

客服自动化应答系统
内容生成工具（如邮件 / 报告撰写）
编程辅助工具（代码解释 / 补全）
多轮对话应用（教育 / 娱乐领域）

API 采用请求 - 响应模式，主要参数包括：

{
  "model": "gpt-3.5-turbo",  # 模型版本
  "messages": [  # 对话历史
    {"role": "user", "content": "你好"}
  ],
  "temperature": 0.7  # 控制输出随机性
}

GPT 模型有 token 数量限制（如 gpt-3.5-turbo 的 4096 tokens），处理长文档时需要特殊策略：

文本分块处理 ：将长文本按语义分割成多个段落
摘要递归 ：对前文生成摘要再续写后续内容
关键信息提取 ：先用其他 API 提取核心信息再处理

Python 示例实现分块处理：

from transformers import GPT2Tokenizer
tokenizer = GPT2Tokenizer.from_pretrained("gpt2")

def chunk_text(text, max_tokens=3000):
    chunks = []
    current_chunk = []
    current_length = 0

    for sentence in text.split('.'):
        sentence = sentence.strip()
        if not sentence:
            continue

        tokens = tokenizer.encode(sentence)
        if current_length + len(tokens) > max_tokens:
            chunks.append('.'.join(current_chunk) + '.')
            current_chunk = [sentence]
            current_length = len(tokens)
        else:
            current_chunk.append(sentence)
            current_length += len(tokens)

    if current_chunk:
        chunks.append('.'.join(current_chunk) + '.')
    return chunks

OpenAI 内置内容过滤系统，但可能误判正常内容。应对方案：

提前检测用户输入的敏感词
设置白名单允许特定领域术语
对 API 返回结果做二次过滤

Node.js 敏感词检测示例：

const bannedWords = ['暴力', '仇恨言论'];
unction containsBannedWords(text) {
  return bannedWords.some(word => 
    text.toLowerCase().includes(word.toLowerCase())
  );
}

// 调用前检查
if (containsBannedWords(userInput)) {return { error: '包含受限内容'};
}

降低成本的关键策略：

缓存高频问题的回答
使用更小的模型版本（如 text-davinci-003 → gpt-3.5-turbo）
监控 token 使用情况

Python 成本计算工具：

def calculate_cost(prompt, response, model="gpt-3.5-turbo"):
    # 价格表（美元 / 千 token）prices = {"gpt-3.5-turbo": {"input": 0.0015, "output": 0.002},
        "gpt-4": {"input": 0.03, "output": 0.06}
    }

    input_tokens = len(prompt) / 4  # 近似估算
    output_tokens = len(response) / 4

    cost = (input_tokens * prices[model]["input"] + 
           output_tokens * prices[model]["output"]) / 1000
    return round(cost, 4)

import openai
from typing import Iterator

def stream_response(prompt: str) -> Iterator[str]:
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    for chunk in response:
        content = chunk["choices"][0].get("delta", {}).get("content")
        if content:
            yield content

# 使用示例
for chunk in stream_response("讲个程序员笑话"):
    print(chunk, end='', flush=True)

class ChatSession {constructor() {this.history = [];
  }

  addMessage(role, content) {this.history.push({ role, content});
    // 保持历史记录不超过 5 轮
    if (this.history.length > 10) {this.history = this.history.slice(-10);
    }
  }

  async getResponse(prompt) {this.addMessage('user', prompt);

    const response = await openai.createChatCompletion({
      model: "gpt-3.5-turbo",
      messages: this.history
    });

    const aiReply = response.data.choices[0].message.content;
    this.addMessage('assistant', aiReply);
    return aiReply;
  }
}

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
import openai

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((openai.error.APIError, openai.error.Timeout)
    )
)
def reliable_chat_completion(messages):
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages
    )

测试环境：AWS t3.xlarge 实例，100 次 API 调用平均耗时

调用方式	平均响应时间	吞吐量（req/s）
同步调用	1.2s	8
异步调用	0.8s	15
流式响应	0.5s（首块）	20

异步调用 Python 实现：

import asyncio
import openai

async def async_completion(prompt):
    resp = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return resp["choices"][0]["message"]["content"]

# 批量处理示例
async def process_batch(prompts):
    tasks = [async_completion(p) for p in prompts]
    return await asyncio.gather(*tasks)

OpenAI API 有以下限制（可能调整）：

免费用户：20 次 / 分钟
付费用户：3500 次 / 分钟

推荐策略：

实现漏桶算法控制请求速率
监控 429 错误码并自动降级
考虑多 API 密钥轮询

行业特定术语可能被误判，建议：

whitelist = {"医疗": ["注射", "手术"],
    "金融": ["比特币", "杠杆"]
}

def is_false_positive(text, industry):
    return any(term in text for term in whitelist.get(industry, [])
    )

存储日志前必须处理：