ChatGPT自动化入门指南：从零搭建你的第一个AI工作流

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

ChatGPT 自动化能帮你 7×24 小时处理客服咨询，自动生成日报数据分析，还能批量创作新媒体内容。它把重复性工作交给 AI，让你专注更有价值的事。最关键的是，它用代码代替手工操作，效率直接翻倍。

一、OpenAI API 封装：稳如老狗的基础调用

先装官方包，记得用虚拟环境：

pip install openai

auth 配置要像藏私房钱：

import openai
from functools import wraps

# 建议从环境变量读取 API_KEY
openai.api_key = os.getenv('OPENAI_KEY')  

def handle_errors(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except openai.error.RateLimitError:
            print("触发限流，稍后重试...")
        except openai.error.APIConnectionError:
            print("网络波动，自动重连中...")
    return wrapper

@handle_errors
def chat_completion(prompt):
    """
    封装基础对话功能
    :param prompt: 用户输入的提示词
    :return: API 返回的完整响应
    """
    return openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7  # 控制创意程度
    )

二、异步并发：让 AI 火力全开

用 asyncio 实现每秒处理 10+ 请求：

import asyncio
from aiohttp import ClientSession

async def async_chat(session, prompt):
    """异步单次请求封装"""
    headers = {"Authorization": f"Bearer {os.getenv('OPENAI_KEY')}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": prompt}]
    }

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

async def batch_query(prompts):
    """并发处理任务池"""
    async with ClientSession() as session:
        tasks = [async_chat(session, p) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 使用示例
if __name__ == '__main__':
    questions = ["如何学习 Python", "推荐 AI 书籍", "解释区块链"]
    results = asyncio.run(batch_query(questions))

三、任务持久化：选对工具很重要

Redis 方案（适合轻量级场景）：

import redis

r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

# 存储对话历史
r.rpush("chat_history", json.dumps({"user": "你好", "ai": "Hi~"}))

RabbitMQ 方案（需要可靠消费时）：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='chat_task')

# 发布任务
channel.basic_publish(
    exchange='',
    routing_key='chat_task',
    body=json.dumps({"prompt": "今日新闻摘要"})
)

四、生产环境避坑指南

1. 速率限制应对策略

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5),  # 最多重试 5 次
    wait=wait_exponential(multiplier=1, max=60)  # 指数退避
)
def call_api_with_retry(prompt):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response

2. 对话上下文三套方案

• 简易模式：固定窗口

  # 只保留最近 3 轮对话
  history = history[-6:] 

• 经济模式：Token 计数

  from transformers import GPT2Tokenizer
  tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
  
  def trim_history(history, max_tokens=2048):
      total = 0
      trimmed = []
      for msg in reversed(history):
          total += len(tokenizer.encode(msg["content"]))
          if total > max_tokens:
              break
          trimmed.insert(0, msg)
      return trimmed

• 专业模式：向量检索（需要 pinecone 等库）

3. 敏感词过滤

import re

blacklist = r"(暴力 | 色情 | 政治敏感词)"

def sanitize_input(text):
    return re.sub(blacklist, "[已过滤]", text, flags=re.IGNORECASE)

五、延伸思考

1. 多模型协作：用 LangChain 的 RouterChain 可以把问题分配给最适合的 AI 模型，比如：
2. GPT- 4 处理复杂推理
3. Claude 处理长文本

4. 本地小模型处理简单问答

5. 隐私保护 三步走：

6. 传输层：强制 HTTPS
7. 存储层：AES 加密
8. 使用层：实时脱敏（如把手机号替换成 138**1234）

最后提醒：先用测试 API_KEY 跑通流程，再上生产环境。遇到问题先查 官方状态页，往往比瞎调试更高效。