如何在iOS的Swift中使用OpenAI Completion API读取流式响应

问题原因 #

你当前使用的URLSession.dataTask(with:completionHandler:)是一次性接收完整响应数据后才执行回调，这就导致无法实时获取OpenAI返回的流式内容。要实现逐块接收并处理数据，需要改用支持流式接收的URLSession用法。

另外你的代码还有两个小问题：

• 未定义fieldName和fieldValue变量，直接判断会报错
• SSE格式解析不严谨，OpenAI的流式响应每行以data: 开头，且可能存在分块数据，需要正确拼接和过滤空行

下面提供两种详细解决方案，根据你的iOS版本选择：

方案一：iOS 15+ 异步序列（Async/Await）方案 #

iOS 15及以上版本支持用AsyncSequence逐块读取响应数据，代码更简洁易读，适合新手：

import Foundation

// 定义OpenAI流式响应的结构体（用于解析数据）
struct OpenAIStreamResponse: Codable {
    let id: String
    let choices: [Choice]
    
    struct Choice: Codable {
        let text: String
    }
}

func callStreamingAPI() async {
    guard let completionAPIURL = URL(string: "https://api.openai.com/v1/completions") else {
        print("无效的API地址")
        return
    }
    
    var request = URLRequest(url: completionAPIURL)
    request.httpMethod = "POST"
    request.addValue("application/json", forHTTPHeaderField: "Accept")
    request.addValue("application/json", forHTTPHeaderField: "Content-Type")
    request.addValue("Bearer <你的Token>", forHTTPHeaderField: "Authorization") // 注意Bearer前缀
    
    let requestBody = [
        "model": "<你的模型名>",
        "prompt": "<你的提示词>",
        "max_tokens": 200,
        "stream": true
    ] as [String: Any]
    
    do {
        request.httpBody = try JSONSerialization.data(withJSONObject: requestBody)
    } catch {
        print("请求体序列化失败: \(error)")
        return
    }
    
    do {
        // 创建异步流式任务
        let (bytes, response) = try await URLSession.shared.bytes(for: request)
        
        // 检查响应状态码
        guard let httpResponse = response as? HTTPURLResponse, (200...299).contains(httpResponse.statusCode) else {
            print("HTTP响应错误")
            return
        }
        
        // 逐行读取数据
        var buffer = Data()
        for try await byte in bytes {
            buffer.append(byte)
            
            // 每次读取到换行符时处理一行数据
            if byte == UInt8(ascii: "\n") {
                if let line = String(data: buffer, encoding: .utf8) {
                    processSSELine(line)
                }
                buffer.removeAll()
            }
        }
    } catch {
        print("请求错误: \(error)")
    }
}

// 处理单条SSE消息
func processSSELine(_ line: String) {
    let trimmedLine = line.trimmingCharacters(in: .whitespacesAndNewlines)
    guard trimmedLine.starts(with: "data: ") else {
        return // 跳过非data开头的行
    }
    
    let dataString = String(trimmedLine.dropFirst(6)) // 去掉"data: "前缀
    if dataString == "[DONE]" {
        print("流式响应结束")
        return
    }
    
    // 解析JSON数据
    guard let data = dataString.data(using: .utf8) else {
        return
    }
    
    do {
        let response = try JSONDecoder().decode(OpenAIStreamResponse.self, from: data)
        if let text = response.choices.first?.text {
            print("实时收到内容: \(text)", terminator: "") // terminator: "" 避免自动换行
        }
    } catch {
        print("解析JSON失败: \(error)")
    }
}

// 调用示例（需要在异步上下文执行）
// Task {
//     await callStreamingAPI()
// }

方案一说明 #

1. 使用URLSession.shared.bytes(for:)创建异步流式任务，会逐字节返回数据
2. 通过缓冲区拼接字节，遇到换行符时处理完整一行的SSE消息
3. 解析data: 开头的行，过滤[DONE]标记，解析出实时返回的文本内容
4. 注意要在异步上下文中调用callStreamingAPI()，比如放在Task里

方案二：委托模式（兼容iOS 13+） #

如果需要兼容iOS 13及以下版本，需要使用URLSession的委托方法来逐块接收数据：

import Foundation

class StreamAPIManager: NSObject, URLSessionDataDelegate {
    private var buffer = Data()
    private var session: URLSession!
    
    override init() {
        super.init()
        session = URLSession(configuration: .default, delegate: self, delegateQueue: .main)
    }
    
    func startStreamingRequest() {
        guard let completionAPIURL = URL(string: "https://api.openai.com/v1/completions") else {
            print("无效的API地址")
            return
        }
        
        var request = URLRequest(url: completionAPIURL)
        request.httpMethod = "POST"
        request.addValue("application/json", forHTTPHeaderField: "Accept")
        request.addValue("application/json", forHTTPHeaderField: "Content-Type")
        request.addValue("Bearer <你的Token>", forHTTPHeaderField: "Authorization")
        
        let requestBody = [
            "model": "<你的模型名>",
            "prompt": "<你的提示词>",
            "max_tokens": 200,
            "stream": true
        ] as [String: Any]
        
        do {
            request.httpBody = try JSONSerialization.data(withJSONObject: requestBody)
        } catch {
            print("请求体序列化失败: \(error)")
            return
        }
        
        let task = session.dataTask(with: request)
        task.resume()
    }
    
    // 逐块接收数据的委托方法
    func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
        buffer.append(data)
        
        // 拆分缓冲区中的完整行
        while let newlineIndex = buffer.firstIndex(of: UInt8(ascii: "\n")) {
            let lineData = buffer[..<newlineIndex]
            buffer.removeSubrange(...newlineIndex)
            
            if let line = String(data: lineData, encoding: .utf8) {
                processSSELine(line)
            }
        }
    }
    
    // 请求完成的委托方法
    func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
        if let error = error {
            print("请求结束，错误: \(error)")
        } else {
            print("请求正常结束")
        }
        buffer.removeAll()
    }
    
    // 处理SSE行的方法（和方案一相同）
    func processSSELine(_ line: String) {
        let trimmedLine = line.trimmingCharacters(in: .whitespacesAndNewlines)
        guard trimmedLine.starts(with: "data: ") else {
            return
        }
        
        let dataString = String(trimmedLine.dropFirst(6))
        if dataString == "[DONE]" {
            print("流式响应结束")
            return
        }
        
        guard let data = dataString.data(using: .utf8) else {
            return
        }
        
        do {
            let response = try JSONDecoder().decode(OpenAIStreamResponse.self, from: data)
            if let text = response.choices.first?.text {
                print("实时收到内容: \(text)", terminator: "")
            }
        } catch {
            print("解析JSON失败: \(error)")
        }
    }
}

// 定义OpenAI响应结构体
struct OpenAIStreamResponse: Codable {
    let id: String
    let choices: [Choice]
    
    struct Choice: Codable {
        let text: String
    }
}

// 调用示例
// let manager = StreamAPIManager()
// manager.startStreamingRequest()

方案二说明 #

1. 创建StreamAPIManager类并实现URLSessionDataDelegate协议
2. 通过urlSession(_:dataTask:didReceive:)委托方法实时接收每一块数据
3. 用缓冲区拼接数据，拆分出完整的SSE行进行处理
4. 委托方法默认在delegateQueue指定的队列执行，示例中用了主队列，方便更新UI

关键注意事项 #

• 替换代码中的<你的Token>、<你的模型名>、<你的提示词>为实际内容
• OpenAI的Authorization头部需要加Bearer 前缀，不要漏掉
• SSE格式的响应每行以data: 开头，空行用于分隔消息，解析时要过滤无效行
• 如果需要更新UI，要确保回到主队列执行（异步方案可以用DispatchQueue.main.async，委托方案已经指定了主队列）

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