Swift Vapor 4流式传输OpenAI完整响应对象的技术问题

你当前的代码要么只能返回单负载响应，要么仅能传输消息内容片段，核心问题是未按Server-Sent Events (SSE) 标准格式封装流式响应块，同时需确保完整编码每个ChatStreamResponse对象。以下是修正后的完整代码：

func chatStream(req: Request) async throws -> Response {
    let chats = try req.content.decode([ChatQuery.ChatCompletionMessageParam].self)
    let userName = req.headers.first(name: "userName")
    guard req.headers.first(name: "deviceUniqueId") != nil else {
        throw Abort(.custom(code: 511, reasonPhrase: "user id required"))
    }
    
    let query = ChatQuery(
        messages: chats,
        model: .gpt4_0125_preview,
        n: 1,
        temperature: 1,
        user: userName
    )
    
    let body = Response.Body(stream: { writer in
        let encoder = JSONEncoder()
        encoder.keyEncodingStrategy = .convertToSnakeCase // 匹配OpenAI字段命名规则
        
        Task {
            do {
                for try await res in req.openAI.chatsStream(query: query) {
                    // 编码完整的响应对象
                    let data = try encoder.encode(res)
                    guard let jsonString = String(data: data, encoding: .utf8) else {
                        req.logger.error("Failed to convert response data to string")
                        continue
                    }
                    // 按SSE标准格式封装每个流块
                    let sseChunk = "data: \(jsonString)\n\n"
                    _ = writer.write(.buffer(.init(string: sseChunk)))
                    // 强制刷新缓冲区，确保客户端及时接收数据
                    try await writer.flush()
                }
                // 发送流结束标记，符合OpenAI流式响应规范
                _ = writer.write(.buffer(.init(string: "data: [DONE]\n\n")))
            } catch {
                req.logger.error("Stream Error: \(error)")
                _ = writer.write(.end)
            }
            _ = writer.write(.end)
        }
    })
    
    var res = Response(status: .ok, body: body)
    // 设置SSE响应头，告知客户端正确解析流
    res.headers.add(name: .contentType, value: "text/event-stream")
    res.headers.add(name: .cacheControl, value: "no-cache")
    res.headers.add(name: .connection, value: "keep-alive")
    return res
}

关键修改说明 #

• SSE格式封装：每个完整响应对象的JSON字符串需用data: 前缀和\n\n后缀包裹，这是SSE的标准分隔格式，客户端可识别每个独立的流块。
• 响应头配置：添加text/event-stream类型头，配合no-cache和keep-alive，确保客户端保持长连接并正确处理流式数据。
• 错误处理优化：将try?替换为try，并显式处理编码失败场景，避免静默丢失数据。
• 流结束标记：发送data: [DONE]\n\n告知客户端流式传输完成，对齐OpenAI的响应规范。

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