从零开始搭建ChatGPT应用：新手避坑指南与最佳实践

1次阅读

没有评论

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

在开始搭建 ChatGPT 应用之前，首先需要明确技术选型。这里我们主要比较直接调用 OpenAI API 和使用 LangChain 等框架的优劣。

直接调用 API：适合快速原型开发和小规模应用，灵活性高，但需要自行处理对话状态、错误恢复等细节。
LangChain 框架 ：适合复杂应用场景，提供对话链、记忆管理等功能，开发效率高，但灵活性稍低。

选择标准主要取决于开发效率和灵活性需求。如果是小型项目或学习用途，直接调用 API 更合适；如果是生产级应用，LangChain 能显著减少开发时间。

API 密钥是访问 OpenAI 服务的凭证，必须妥善管理。常见的方案有两种：

环境变量 ：简单易用，适合开发环境。
密钥管理服务（如 AWS KMS）：适合生产环境，安全性更高。

以下是使用环境变量的示例代码：

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

保持对话上下文是 ChatGPT 应用的核心功能之一。以下是三种常见的实现方式：

简单列表存储 ：将对话历史保存在列表中，每次调用 API 时传入整个历史。

conversation_history = []

def chat(message):
    conversation_history.append({"role": "user", "content": message})
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=conversation_history
    )
    conversation_history.append({"role": "assistant", "content": response.choices[0].message.content})
    return response.choices[0].message.content

数据库存储 ：适用于需要长期保存对话的场景，如用户会话。
缓存存储 ：使用 Redis 等缓存服务，适合高并发场景。

流式响应可以提升用户体验，尤其是在生成较长文本时。以下是使用 Python async/await 的示例：

import asyncio

async def stream_response(prompt):
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    async for chunk in response:
        print(chunk.choices[0].delta.content, end="", flush=True)

OpenAI API 有速率限制，超出限制会导致 429 错误。令牌桶算法是一种常见的限流策略，以下是实现片段：

import time

class TokenBucket:
    def __init__(self, capacity, rate):
        self.capacity = capacity
        self.rate = rate
        self.tokens = capacity
        self.last_check = time.time()

    def consume(self, tokens=1):
        now = time.time()
        elapsed = now - self.last_check
        self.last_check = now
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

在生产环境中，敏感内容过滤是必不可少的。可以使用正则表达式结合关键词库实现：

import re

sensitive_keywords = ["暴力", "色情", "政治"]

def filter_content(text):
    for keyword in sensitive_keywords:
        if re.search(keyword, text):
            return "内容包含敏感信息，已过滤"
    return text

503（服务不可用）和 429（速率限制）是常见错误，可以通过自动重试机制处理：

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def safe_chat(message):
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": message}]
    )

对话状态丢失是常见问题，可以通过检查点机制预防：

def save_checkpoint(conversation_history):
    with open("checkpoint.json", "w") as f:
        json.dump(conversation_history, f)

def load_checkpoint():
    try:
        with open("checkpoint.json", "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return []

缓存可以显著减少 API 调用次数，提升响应速度。以下是简单实现：

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_chat(message):
    return client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": message}]
    )

监控响应延迟有助于发现性能瓶颈。以下是示例代码：

import time

class LatencyMonitor:
    def __init__(self):
        self.latencies = []

    def record(self, start_time):
        latency = time.time() - start_time
        self.latencies.append(latency)
        return latency

    def average_latency(self):
        return sum(self.latencies) / len(self.latencies) if self.latencies else 0

向量数据库（如 Pinecone、Milvus）可以存储和检索文本的向量表示，从而实现长期记忆功能。具体步骤：