ChatGPT工作流入门指南：从零搭建自动化AI助手

1次阅读

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

ChatGPT 工作流本质上是通过维护对话状态和上下文，实现多轮交互的 AI 应用。与单次请求相比，工作流更适合需要记忆和持续交互的场景。

单次请求 ：适合简单问答、一次性任务
工作流 ：适合客服对话、多步骤任务处理等需要上下文记忆的场景

新手开发者在使用 ChatGPT 工作流时常遇到以下问题：

长对话中上下文丢失
多步骤任务编排困难
API 调用频率限制
token 超限导致请求失败

import openai

class ChatGPT:
    def __init__(self, api_key):
        openai.api_key = api_key
        self.conversation_history = []

    def add_message(self, role, content):
        self.conversation_history.append({"role": role, "content": content})

    def get_response(self, max_tokens=150):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=self.conversation_history,
                max_tokens=max_tokens
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"API 调用失败: {str(e)}")
            return None

chat = ChatGPT("your-api-key")
chat.add_message("user", "你好")
response = chat.get_response()
print(response)

# 继续对话
chat.add_message("assistant", response)
chat.add_message("user", "能告诉我今天的日期吗？")
print(chat.get_response())

import asyncio
import aiohttp

async def async_chat_completion(session, messages):
    async with session.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer your-api-key"},
        json={"model": "gpt-3.5-turbo", "messages": messages}
    ) as resp:
        return await resp.json()

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

import time
import random

def exponential_backoff(func, max_retries=5):
    for i in range(max_retries):
        try:
            return func()
        except Exception as e:
            wait_time = min(2 ** i + random.random(), 60)
            time.sleep(wait_time)
    raise Exception("Max retries exceeded")

from ratelimit import limits, sleep_and_retry

# 每分钟 60 次请求
@sleep_and_retry
@limits(calls=60, period=60)
def api_call():
    # API 调用代码
    pass

def calculate_max_tokens(prompt):
    token_count = len(prompt.split()) * 1.5  # 简单估算
    return min(4096 - int(token_count), 150)  # 模型最大 token 限制

def filter_sensitive_info(text):
    sensitive_keywords = ["密码", "身份证", "银行卡"]
    for keyword in sensitive_keywords:
        if keyword in text:
            return "[敏感信息已过滤]"
    return text

对于更复杂的工作流编排，可以考虑使用 LangChain 框架。它提供了: