从零开始：如何将ChatGPT无缝接入你的应用引擎（附完整代码示例）

2次阅读

没有评论

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

在将 ChatGPT 集成到应用引擎的过程中，开发者常常会遇到几个主要挑战：

API 限流：OpenAI 对 API 调用有严格的速率限制，尤其是在免费层。超出限制会导致请求失败，影响用户体验。
长文本处理：ChatGPT 对输入长度有限制（通常是 4096 个 token），处理长文本时需要分段或截断。
上下文管理：在多轮对话中，如何有效地维护对话上下文是一个复杂的问题。
响应延迟：API 调用可能需要几秒钟才能返回结果，这在实时应用中可能造成用户体验问题。
错误处理：网络问题或 API 错误可能导致请求失败，需要有健壮的重试机制。

开发者可以选择以下几种方式接入 ChatGPT：

直接 API 调用
优点：灵活性高，可以直接控制请求和响应的每个细节
缺点：需要自己处理认证、限流、错误重试等逻辑
使用官方 SDK
优点：简化了 API 调用过程，提供了更好的抽象
缺点：可能缺少某些高级功能，更新可能滞后于 API
通过代理服务
优点：可以绕过某些地区的限制，提供额外的缓存层
缺点：增加了一层依赖，可能有安全和隐私问题

对于大多数中级开发者来说，直接 API 调用提供了最佳的控制性和灵活性，这也是本文重点介绍的方法。

以下是一个完整的 Python 示例，展示了如何调用 ChatGPT API：

import openai
import os
from typing import List, Dict

# 初始化 OpenAI 客户端
openai.api_key = os.getenv('OPENAI_API_KEY')

class ChatGPTClient:
    def __init__(self):
        self.conversation_history: List[Dict] = []

    def add_to_history(self, role: str, content: str):
        """添加消息到对话历史"""
        self.conversation_history.append({"role": role, "content": content})

    def truncate_history(self, max_tokens: int = 4000):
        """截断历史记录以避免超出 token 限制"""
        while self._count_tokens() > max_tokens and len(self.conversation_history) > 1:
            self.conversation_history.pop(0)

    def _count_tokens(self) -> int:
        """估算当前对话历史的 token 数量"""
        return sum(len(msg["content"]) // 4 for msg in self.conversation_history)

    def get_response(self, prompt: str, model: str = "gpt-3.5-turbo") -> str:
        """获取 ChatGPT 的响应"""
        self.add_to_history("user", prompt)
        self.truncate_history()

        try:
            response = openai.ChatCompletion.create(
                model=model,
                messages=self.conversation_history,
                temperature=0.7,
                stream=False
            )

            assistant_reply = response.choices[0].message.content
            self.add_to_history("assistant", assistant_reply)
            return assistant_reply
        except Exception as e:
            print(f"API 调用失败: {str(e)}")
            raise

# 使用示例
if __name__ == "__main__":
    client = ChatGPTClient()
    while True:
        user_input = input("你:")
        if user_input.lower() in ['exit', 'quit']:
            break
        response = client.get_response(user_input)
        print(f"AI: {response}")

对话历史管理
我们维护一个 conversation_history 列表来存储对话上下文
每次交互都会添加用户输入和 AI 响应到这个历史中
Token 限制处理
实现了 _count_tokens 方法来估算当前对话使用的 token 数量
truncate_history方法会在 token 数量接近限制时移除最早的对话记录
基本错误处理
使用 try-except 捕获 API 调用异常
在实际应用中，应该实现更完善的错误处理和重试逻辑

对于需要实时显示响应的应用，可以使用流式 API：

def get_streaming_response(self, prompt: str, model: str = "gpt-3.5-turbo"):
    """获取流式响应"""
    self.add_to_history("user", prompt)
    self.truncate_history()

    try:
        response = openai.ChatCompletion.create(
            model=model,
            messages=self.conversation_history,
            temperature=0.7,
            stream=True
        )

        full_response = ""
        for chunk in response:
            content = chunk.choices[0].delta.get("content", "")
            full_response += content
            yield content

        self.add_to_history("assistant", full_response)
    except Exception as e:
        print(f"API 调用失败: {str(e)}")
        raise