Python FastAPI框架使用OpenAI SDK怎么实现异步流式返回

Question

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    try:
        stream = await client.chat.completions.create(
        model="xxx",
        messages=[
            {"role": "system", "content": "你是一个有用的人工智能助手"},
            {"role": "user", "content": "常见的十字花科植物有哪些？"},
        ],
        stream=True
    )
    async for completion in stream:
        print(completion.choices[0].delta.content, end="")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

以上代码报错“object Stream can't be used in 'await' expression”，怎么在Python FastAPI框架使用OpenAI SDK怎么实现异步流式返回呢？

Answer 1

♥ 该回答参考ChatGPT，由吾为汝整理提供，如若有用，还请题主采纳！♥

在使用 FastAPI 和 OpenAI SDK 实现异步流式返回时，您需要确保使用的库支持异步操作。根据您提供的代码，client.chat.completions.create 返回的 stream 对象可能不支持异步迭代，因此会导致错误。

为了实现异步流式返回，您可以使用 asyncio 库来处理流式数据。以下是一个示例代码，展示了如何在 FastAPI 中实现这一点：

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

app = FastAPI()

# 定义请求体
class ChatRequest(BaseModel):
    messages: list

# 初始化 OpenAI 客户端
openai.api_key = "YOUR_API_KEY"

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    try:
        # 创建一个异步生成器
        async def stream_response():
            stream = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=request.messages,
                stream=True
            )
            for completion in stream:
                if 'choices' in completion and len(completion['choices']) > 0:
                    delta_content = completion['choices'][0].get('delta', {}).get('content', '')
                    if delta_content:
                        yield delta_content

        # 使用 StreamingResponse 返回流式数据
        return StreamingResponse(stream_response(), media_type="text/event-stream")

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

关键点说明：

1. 异步生成器：stream_response 函数是一个异步生成器，它会逐步返回 OpenAI 的响应内容。

2. StreamingResponse：使用 StreamingResponse 来返回流式数据，media_type 设置为 text/event-stream，以便客户端可以逐步接收数据。

3. 错误处理：在 try-except 块中处理可能的异常，以确保 API 的稳定性。

请确保您已经安装了 fastapi 和 openai 库，并且在代码中替换 "YOUR_API_KEY" 为您的实际 OpenAI API 密钥。这样，您就可以在 FastAPI 中实现异步流式返回了。