如何正确处理FastAPI中异步迭代器__aiter__方法抛出的异常

你的代码里，__aiter__方法的异常无法被路由中的try-except捕获，是因为StreamingResponse返回时并没有执行__aiter__里的逻辑——只有当客户端开始接收流数据时，异步迭代器的代码才会被触发。这时候路由的响应已经发送（状态码200），后续抛出的异常只能导致连接中断，无法修改已发送的响应状态。

根据错误发生的阶段，有两种针对性的处理方式：

1. 处理流迭代过程中的异常（中途出错） #

如果错误是在OpenAI API流式返回的过程中发生的，最合理的方式是将错误信息以流的形式发送给前端，而非中断连接。这样前端可以接收错误提示并做对应处理。

# 自定义异常类
class StreamGeneratorError(Exception):
    def __init__(self, message: str):
        self.message = message

class StreamGenerator:
    def __init__(self, prompt: str):
        self.prompt = prompt

    async def __aiter__(self):
        try:
            # 调用OpenAI流式API示例
            stream = await openai.ChatCompletion.acreate(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": self.prompt}],
                stream=True
            )
            async for chunk in stream:
                content = chunk.choices[0].delta.get("content", "")
                if content:
                    yield f"data: {content}\n\n"
        except Exception as e:
            # 以SSE格式发送错误信息
            yield f"data: error: {str(e)}\n\n"
            # 发送流结束信号
            yield "data: [DONE]\n\n"

@router.post("/v1/chat-with-ai")
async def chat_gpt_api_view(body: UserPrompt) -> StreamingResponse:
    stream_gen_instance = StreamGenerator(body.prompt)
    return StreamingResponse(stream_gen_instance, media_type='text/event-stream')

2. 处理初始化阶段的异常（前置错误） #

如果错误是在创建流之前发生的（比如参数验证失败、API密钥错误），可以将初始化逻辑移到路由中，提前捕获异常并返回标准错误响应。

class StreamGenerator:
    def __init__(self, openai_stream):
        self.openai_stream = openai_stream

    async def __aiter__(self):
        async for chunk in self.openai_stream:
            content = chunk.choices[0].delta.get("content", "")
            if content:
                yield f"data: {content}\n\n"

@router.post("/v1/chat-with-ai")
async def chat_gpt_api_view(body: UserPrompt):
    try:
        # 提前执行OpenAI API初始化，出错直接抛出异常
        stream = await openai.ChatCompletion.acreate(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": body.prompt}],
            stream=True
        )
        stream_gen_instance = StreamGenerator(stream)
        return StreamingResponse(stream_gen_instance, media_type='text/event-stream')
    except Exception as e:
        raise StreamGeneratorError(f"Failed to start stream: {str(e)}") from e

# 全局异常处理器，统一返回错误格式
@router.exception_handler(StreamGeneratorError)
async def stream_error_handler(request: Request, exc: StreamGeneratorError):
    return JSONResponse(
        status_code=500,
        content={"detail": exc.message}
    )

• 流中途的错误无法修改已发送的200状态码，只能通过流内容传递错误信息
• 初始化阶段的错误可以提前拦截，返回自定义状态码（如400、500）和响应体

内容的提问来源于stack exchange，提问作者alfik