基于ChatGPT API构建智能对话系统：吴恩达课程实践指南

1次阅读

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

在构建智能对话系统的过程中，开发者常常面临几个核心挑战。首先，上下文管理是一个棘手的问题。传统的对话系统往往难以长期保持对话的连贯性，导致用户体验不佳。其次，响应延迟也是一个关键痛点，特别是当系统需要处理大量并发请求时。最后，成本控制是许多团队关注的重点，尤其是在使用商业 API 时。

目前市面上有多种对话模型可供选择，包括 Rasa、Dialogflow 和 ChatGPT API 等。以下是它们的优缺点对比：

Rasa：开源且高度可定制，但需要大量训练数据和开发资源。
Dialogflow：谷歌提供的商业解决方案，易于使用但灵活性较低。
ChatGPT API：基于强大的 GPT 模型，支持复杂的上下文管理，但需要合理控制成本。

综合考虑灵活性、开发成本和性能，ChatGPT API 是一个不错的选择，尤其适合需要快速上手的项目。

首先，我们需要搭建一个基于 FastAPI 的后端服务。FastAPI 是一个现代、快速的 Web 框架，非常适合构建 API 服务。

from fastapi import FastAPI
from pydantic import BaseModel
import openai

app = FastAPI()

class Message(BaseModel):
    content: str
    session_id: str

@app.post("/chat")
async def chat(message: Message):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": message.content}]
    )
    return {"response": response.choices[0].message.content}

为了保持对话的连贯性，我们需要管理对话的上下文。可以通过在数据库中存储会话历史来实现这一点。

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import openai
import redis

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, db=0)

class Message(BaseModel):
    content: str
    session_id: str

@app.post("/chat")
async def chat(message: Message):
    session_key = f"session:{message.session_id}"
    previous_messages = r.lrange(session_key, 0, -1)
    messages = [{"role": "user", "content": msg.decode("utf-8")} for msg in previous_messages]
    messages.append({"role": "user", "content": message.content})

    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=messages
    )

    r.rpush(session_key, message.content)
    r.rpush(session_key, response.choices[0].message.content)

    return {"response": response.choices[0].message.content}

在 API 调用过程中，可能会遇到各种错误，如网络问题或 API 限制。我们需要合理处理这些错误，并记录日志以便后续分析。

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.post("/chat")
async def chat(message: Message):
    try:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": message.content}]
        )
        return {"response": response.choices[0].message.content}
    except Exception as e:
        logger.error(f"Error in chat: {e}")
        return {"error": str(e)}

对于频繁出现的相似请求，可以使用缓存来减少 API 调用次数。

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import openai
import redis

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, db=0)

class Message(BaseModel):
    content: str
    session_id: str

@app.post("/chat")
async def chat(message: Message):
    cache_key = f"cache:{message.content}"
    cached_response = r.get(cache_key)
    if cached_response:
        return {"response": cached_response.decode("utf-8")}

    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": message.content}]
    )

    r.setex(cache_key, 3600, response.choices[0].message.content)
    return {"response": response.choices[0].message.content}

如果需要处理大量请求，可以考虑使用批处理来优化性能。

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import openai
import asyncio

app = FastAPI()

class Message(BaseModel):
    content: str
    session_id: str

@app.post("/batch_chat")
async def batch_chat(messages: list[Message]):
    tasks = []
    for message in messages:
        tasks.append(
            openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": message.content}]
            )
        )
    responses = await asyncio.gather(*tasks)
    return [{"response": response.choices[0].message.content} for response in responses]

ChatGPT API 有速率限制，可以通过以下方式应对：