FastAPI集成OpenAI流式ChatCompletion的SSE接口阻塞主线程问题

问题出在你用asyncify包装同步的OpenAI流式调用，虽然发起请求是异步的，但遍历同步流式迭代器的过程会阻塞FastAPI的事件循环，导致其他端点无法处理请求。同时，直接使用同步客户端的流式接口，无法在异步函数中正确遍历，会引发coroutine object is not an iterator错误。

正确的做法是使用OpenAI官方提供的异步客户端，确保整个流式处理流程都是非阻塞的异步操作：

步骤1：确保使用OpenAI异步客户端 #

OpenAI Python库（版本>=1.0.0）支持异步操作，只需初始化AsyncOpenAI客户端：

from openai import AsyncOpenAI

# 初始化异步客户端，可通过环境变量或直接传入api_key
client = AsyncOpenAI(
    api_key="your-api-key-here"  # 也可通过环境变量读取：os.getenv("OPENAI_API_KEY")
)

步骤2：修改FastAPI端点代码 #

将原有的同步调用替换为异步客户端调用，并且在流式遍历的时候使用async for（异步客户端返回的是异步迭代器，必须用async for遍历）：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import json
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(api_key="your-api-key-here")

@app.post("/ai_re/")
async def read_item(request: Request):
    base_prompt = "This is a prompt"
    sources = []
    
    # 使用异步客户端发起流式请求
    response = await client.chat.completions.create(
        model='gpt-3.5-turbo',
        messages=[{"role": "system", "content": base_prompt.strip()}],
        max_tokens=550,
        temperature=0.28,
        stream=True,
        n=1
    )

    async def event_generator():
        # 用async for遍历异步流式响应
        async for event in response:
            event_text = event.choices[0].delta.content or ""
            event_data = {
                "texte": event_text,
                "source": sources
            }
            yield f"data: {json.dumps(event_data)}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

关键说明 #

• 异步客户端的必要性：同步客户端的流式接口返回同步迭代器，在异步函数中直接用for循环会阻塞事件循环；异步客户端返回异步迭代器，async for遍历不会阻塞，保证FastAPI可以同时处理其他请求。
• 移除asyncify：异步客户端本身支持await调用，无需再用asyncify包装同步方法，避免额外线程切换开销和阻塞问题。
• 简化空内容判断：用or ""替代原有的条件判断，逻辑一致且更简洁。

这样修改后，SSE流式响应可正常工作，同时不会阻塞其他端点的请求。

内容的提问来源于stack exchange，提问作者Baudouin Arbarétier