Swift流式请求数据丢失问题排查及Alamofire流取消方案咨询

问题背景 #

我正在开发一款Swift版简易ChatGPT应用，通过流式方式请求API数据。API返回数据正确，但接收时出现文字和字符丢失的情况，调试后仍未明确URL请求无法返回完整数据的原因。

现有两种实现方案：

• 第一种采用基础HTTP请求（存在数据丢失问题），支持任务取消，但不确定第二种Alamofire方案如何中途取消流。
• 第二种方案可完整返回所有数据，实际开发中更倾向于使用健壮的Alamofire，但需保留中途取消流的能力。

请问为何第一种方案无法返回完整数据？

第一种方案（存在数据丢失问题） #

func sendMessageStream(Question_To_Be_Asked: String) async throws -> AsyncThrowingStream<String, Error> {
    var urlRequest = self.urlRequest
    urlRequest.httpBody = try jsonBody(text: Question_To_Be_Asked)
    
    let (result, response) = try await urlSession.bytes(for: urlRequest)
    try Task.checkCancellation()
    
    guard let httpResponse = response as? HTTPURLResponse else {
        throw "Invalid response"
    }
    
    guard 200...299 ~= httpResponse.statusCode else {
        var errorText = ""
        for try await line in result.lines {
            try Task.checkCancellation()
            errorText += line
        }
        
        if let data = errorText.data(using: .utf8), let errorResponse = try? jsonDecoder.decode(ErrorRootResponse.self, from: data).error {
            errorText = "\n\(errorResponse.message)"
        }
        
        throw "Bad Response: \(httpResponse.statusCode), \(errorText)"
    }
    
    var responseText = ""
    return AsyncThrowingStream { [weak self] in
        guard let self else { return nil }
        for try await line in result.lines {
            //print(line) <- incomplete data
            try Task.checkCancellation()
            if line.hasPrefix("data: "), let data = line.dropFirst(6).data(using: .utf8), let response = try? self.jsonDecoder.decode(StreamCompletionResponse.self, from: data), let text = response.choices.first?.delta.content {
                
                responseText += text
                return text
            }
        }
        return nil
    }
}

第二种方案（可正常运行） #

func sendStreamMessage(messages: [Message]) -> DataStreamRequest{
    let openAIMessages = messages.map({OpenAIChatMessage(role: $0.role, content: $0.content)})
    let body = OpenAIChatBody(model: "gpt-4", messages: openAIMessages, stream: true)
    let headers: HTTPHeaders = [
        "Authorization": "Bearer \(Constants.openAIApiKey)"
    ]
    
    return AF.streamRequest(endpointUrl, method: .post, parameters: body, encoder: .json, headers: headers)
}

func sendMessage(question: String)  {
    let messages = [Message(id: UUID().uuidString, role: .user, content: question, createAt: Date())]
    currentInput = ""
    
    sendStreamMessage(messages: messages).responseStreamString { [weak self] stream in
        guard let self = self else { return }
        switch stream.event {
        case .stream(let response):
            switch response {
            case .success(let string):
                let streamResponse = self.parseStreamData(string)
                
                streamResponse.forEach { newMessageResponse in
                    guard let messageContent = newMessageResponse.choices.first?.delta.content else {
                        return
                    }
                    //here messageContent is final complete string from stream
                }
            case .failure(_):
                print("Something failes")
            }
            print(response)
        case .complete(_):
            print("COMPLETE")
        }
    }
}

func parseStreamData(_ data: String) ->[ChatStreamCompletionResponse] {
    let responseStrings = data.split(separator: "data:").map({$0.trimmingCharacters(in: .whitespacesAndNewlines)}).filter({!$0.isEmpty})
    let jsonDecoder = JSONDecoder()
    
    return responseStrings.compactMap { jsonString in
        guard let jsonData = jsonString.data(using: .utf8), let streamResponse = try? jsonDecoder.decode(ChatStreamCompletionResponse.self, from: jsonData) else {
            return nil
        }
        return streamResponse
    }
}

struct ChatStreamCompletionResponse: Decodable {
    let id: String
    let choices: [ChatStreamChoice]
}

struct ChatStreamChoice: Decodable {
    let delta: ChatStreamContent
}

struct ChatStreamContent: Decodable {
    let content: String
}

struct Message: Decodable, Hashable {
    let id: String
    let role: SenderRole
    let content: String
    let createAt: Date
    
    func hash(into hasher: inout Hasher) {
        hasher.combine(id)
    }
}

struct OpenAIChatBody: Encodable {
    let model: String
    let messages: [OpenAIChatMessage]
    let stream: Bool
}

struct OpenAIChatMessage: Codable {
    let role: SenderRole
    let content: String
}

enum SenderRole: String, Codable {
    case system
    case user
    case assistant
}

问题解答 #

1. 第一种方案数据丢失的原因 #

基础HTTP请求方案存在两个核心问题，导致流式数据接收不完整：

• AsyncThrowingStream闭包逻辑错误：在生成闭包中，遍历result.lines时匹配到有效数据就return text，这会直接中断当前循环迭代，后续的line处理被跳过。下一次迭代器调用时，循环从下一次开始，导致部分流式分片被遗漏。
• 忽略非"data:"前缀的必要行：OpenAI流式响应中，除了data: 开头的内容，还可能包含空行或data: [DONE]标记。代码直接跳过这些行，若出现网络传输导致的分片拆分异常，或遗漏了[DONE]前的最后几块数据，就会造成内容丢失。

另外，result.lines基于换行符分割异步序列，网络传输中可能出现分割不规范的情况，而Alamofire的responseStreamString会更健壮地处理流式数据的拼接与分割，这也是第二种方案能完整接收的原因之一。

2. Alamofire方案实现中途取消流的方法 #

Alamofire的DataStreamRequest原生支持取消操作，只需保留请求引用，在需要取消时调用cancel()方法：

// 保存当前流式请求的引用
var currentStreamRequest: DataStreamRequest?

func sendMessage(question: String)  {
    let messages = [Message(id: UUID().uuidString, role: .user, content: question, createAt: Date())]
    currentInput = ""
    
    // 保存请求引用
    currentStreamRequest = sendStreamMessage(messages: messages).responseStreamString { [weak self] stream in
        guard let self = self else { return }
        switch stream.event {
        case .stream(let response):
            switch response {
            case .success(let string):
                let streamResponse = self.parseStreamData(string)
                
                streamResponse.forEach { newMessageResponse in
                    guard let messageContent = newMessageResponse.choices.first?.delta.content else {
                        return
                    }
                    // 处理消息内容
                }
            case .failure(_):
                print("请求失败")
            }
        case .complete(_):
            print("流已完成")
            currentStreamRequest = nil // 完成后清空引用
        }
    }
}

// 取消当前流式请求的方法
func cancelCurrentStream() {
    currentStreamRequest?.cancel()
    currentStreamRequest = nil
}

调用cancelCurrentStream()即可中途终止流式请求，Alamofire会自动断开连接并停止接收后续数据。

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