Merge pull request #266 from MacPaw/bug/261-streaming-does-not-honor-http-status-codes

Bug: Fix error parsing in StreamInterpreter

Author: nezhyborets

Demo/DemoChat/Sources/ChatStore.swift

@@ -10,6 +10,7 @@ import Combine
 import OpenAI
 import SwiftUI
 
+@MainActor
 public final class ChatStore: ObservableObject {
     public var openAIClient: OpenAIProtocol
     let idProvider: () -> String
@@ -61,8 +62,7 @@ public final class ChatStore: ObservableObject {
     func deleteConversation(_ conversationId: Conversation.ID) {
         conversations.removeAll(where: { $0.id == conversationId })
     }
-
-    @MainActor
+    
     func sendMessage(
         _ message: Message,
         conversationId: Conversation.ID,
@@ -78,7 +78,8 @@ public final class ChatStore: ObservableObject {
 
             await completeChat(
                 conversationId: conversationId,
-                model: model
+                model: model,
+                stream: true
             )
             // For assistant case we send chats to thread and then poll, polling will receive sent chat + new assistant messages.
         case .assistant:
@@ -139,11 +140,11 @@ public final class ChatStore: ObservableObject {
             }
         }
     }
-
-    @MainActor
+    
     func completeChat(
         conversationId: Conversation.ID,
-        model: Model
+        model: Model,
+        stream: Bool
     ) async {
         guard let conversation = conversations.first(where: { $0.id == conversationId }) else {
             return
@@ -169,59 +170,88 @@ public final class ChatStore: ObservableObject {
             ))
 
             let functions = [weatherFunction]
-
-            let chatsStream: AsyncThrowingStream<ChatStreamResult, Error> = openAIClient.chatsStream(
-                query: ChatQuery(
-                    messages: conversation.messages.map { message in
-                        ChatQuery.ChatCompletionMessageParam(role: message.role, content: message.content)!
-                    }, model: model,
-                    tools: functions
-                )
+            
+            let chatQuery = ChatQuery(
+                messages: conversation.messages.map { message in
+                    ChatQuery.ChatCompletionMessageParam(role: message.role, content: message.content)!
+                }, model: model,
+                tools: functions
             )
-
-            var functionCalls = [(name: String, argument: String?)]()
-            for try await partialChatResult in chatsStream {
-                for choice in partialChatResult.choices {
-                    let existingMessages = conversations[conversationIndex].messages
-                    // Function calls are also streamed, so we need to accumulate.
-                    choice.delta.toolCalls?.forEach { toolCallDelta in
-                        if let functionCallDelta = toolCallDelta.function {
-                            if let nameDelta = functionCallDelta.name {
-                                functionCalls.append((nameDelta, functionCallDelta.arguments))
-                            }
+            
+            if stream {
+                try await completeConversationStreaming(
+                    conversationIndex: conversationIndex,
+                    model: model,
+                    query: chatQuery
+                )
+            } else {
+                try await completeConversation(conversationIndex: conversationIndex, model: model, query: chatQuery)
+            }
+        } catch {
+            conversationErrors[conversationId] = error
+        }
+    }
+    
+    private func completeConversation(conversationIndex: Int, model: Model, query: ChatQuery) async throws {
+        let chatResult: ChatResult = try await openAIClient.chats(query: query)
+        chatResult.choices
+            .map {
+                Message(
+                    id: chatResult.id,
+                    role: $0.message.role,
+                    content: $0.message.content?.string ?? "",
+                    createdAt: Date(timeIntervalSince1970: TimeInterval(chatResult.created))
+                )
+            }.forEach { message in
+                conversations[conversationIndex].messages.append(message)
+            }
+    }
+    
+    private func completeConversationStreaming(conversationIndex: Int, model: Model, query: ChatQuery) async throws {
+        let chatsStream: AsyncThrowingStream<ChatStreamResult, Error> = openAIClient.chatsStream(
+            query: query
+        )
+
+        var functionCalls = [(name: String, argument: String?)]()
+        for try await partialChatResult in chatsStream {
+            for choice in partialChatResult.choices {
+                let existingMessages = conversations[conversationIndex].messages
+                // Function calls are also streamed, so we need to accumulate.
+                choice.delta.toolCalls?.forEach { toolCallDelta in
+                    if let functionCallDelta = toolCallDelta.function {
+                        if let nameDelta = functionCallDelta.name {
+                            functionCalls.append((nameDelta, functionCallDelta.arguments))
                         }
                     }
-                    var messageText = choice.delta.content ?? ""
-                    if let finishReason = choice.finishReason,
-                       finishReason == .toolCalls
-                    {
-                        functionCalls.forEach { (name: String, argument: String?) in
-                            messageText += "Function call: name=\(name) arguments=\(argument ?? "")\n"
-                        }
+                }
+                var messageText = choice.delta.content ?? ""
+                if let finishReason = choice.finishReason,
+                   finishReason == .toolCalls
+                {
+                    functionCalls.forEach { (name: String, argument: String?) in
+                        messageText += "Function call: name=\(name) arguments=\(argument ?? "")\n"
                     }
-                    let message = Message(
-                        id: partialChatResult.id,
-                        role: choice.delta.role ?? .assistant,
-                        content: messageText,
-                        createdAt: Date(timeIntervalSince1970: TimeInterval(partialChatResult.created))
+                }
+                let message = Message(
+                    id: partialChatResult.id,
+                    role: choice.delta.role ?? .assistant,
+                    content: messageText,
+                    createdAt: Date(timeIntervalSince1970: TimeInterval(partialChatResult.created))
+                )
+                if let existingMessageIndex = existingMessages.firstIndex(where: { $0.id == partialChatResult.id }) {
+                    // Meld into previous message
+                    let previousMessage = existingMessages[existingMessageIndex]
+                    let combinedMessage = Message(
+                        id: message.id, // id stays the same for different deltas
+                        role: message.role,
+                        content: previousMessage.content + message.content,
+                        createdAt: message.createdAt
                     )
-                    if let existingMessageIndex = existingMessages.firstIndex(where: { $0.id == partialChatResult.id }) {
-                        // Meld into previous message
-                        let previousMessage = existingMessages[existingMessageIndex]
-                        let combinedMessage = Message(
-                            id: message.id, // id stays the same for different deltas
-                            role: message.role,
-                            content: previousMessage.content + message.content,
-                            createdAt: message.createdAt
-                        )
-                        conversations[conversationIndex].messages[existingMessageIndex] = combinedMessage
-                    } else {
-                        conversations[conversationIndex].messages.append(message)
-                    }
+                    conversations[conversationIndex].messages[existingMessageIndex] = combinedMessage
+                } else {
+                    conversations[conversationIndex].messages.append(message)
                 }
             }
-        } catch {
-            conversationErrors[conversationId] = error
         }
     }
 

Sources/OpenAI/Private/StreamInterpreter.swift

@@ -16,6 +16,11 @@ class StreamInterpreter<ResultType: Codable> {
     var onEventDispatched: ((ResultType) -> Void)?
 
     func processData(_ data: Data) throws {
+        let decoder = JSONDecoder()
+        if let decoded = try? decoder.decode(APIErrorResponse.self, from: data) {
+            throw decoded
+        }
+        
         guard let stringContent = String(data: data, encoding: .utf8) else {
             throw StreamingError.unknownContent
         }

Tests/OpenAITests/StreamInterpreterTests.swift

@@ -32,6 +32,14 @@ struct StreamInterpreterTests {
         #expect(chatStreamResults.count == 1)
     }
 
+    @Test func parseApiError() throws {
+        do {
+            try interpreter.processData(chatCompletionError())
+        } catch {
+            #expect(error is APIErrorResponse)
+        }
+    }
+    
     // Chunk with 3 objects. I captured it from a real response. It's a very short response that contains just "Hi"
     private func chatCompletionChunk() -> Data {
         "data: {\"id\":\"chatcmpl-AwnboO5ZnaUyii9xxC5ZVmM5vGark\",\"object\":\"chat.completion.chunk\",\"created\":1738577084,\"model\":\"gpt-4-0613\",\"service_tier\":\"default\",\"system_fingerprint\":null,\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"\",\"refusal\":null},\"logprobs\":null,\"finish_reason\":null}]}\n\ndata: {\"id\":\"chatcmpl-AwnboO5ZnaUyii9xxC5ZVmM5vGark\",\"object\":\"chat.completion.chunk\",\"created\":1738577084,\"model\":\"gpt-4-0613\",\"service_tier\":\"default\",\"system_fingerprint\":null,\"choices\":[{\"index\":0,\"delta\":{\"content\":\"Hi\"},\"logprobs\":null,\"finish_reason\":null}]}\n\ndata: {\"id\":\"chatcmpl-AwnboO5ZnaUyii9xxC5ZVmM5vGark\",\"object\":\"chat.completion.chunk\",\"created\":1738577084,\"model\":\"gpt-4-0613\",\"service_tier\":\"default\",\"system_fingerprint\":null,\"choices\":[{\"index\":0,\"delta\":{},\"logprobs\":null,\"finish_reason\":\"stop\"}]}\n\n".data(using: .utf8)!
@@ -44,4 +52,9 @@ struct StreamInterpreterTests {
     private func chatCompletionChunkTermination() -> Data {
         "data: [DONE]\n\n".data(using: .utf8)!
     }
+    
+    // Copied from an actual reponse that was an input to inreptreter
+    private func chatCompletionError() -> Data {
+        "{\n    \"error\": {\n        \"message\": \"The model `o3-mini` does not exist or you do not have access to it.\",\n        \"type\": \"invalid_request_error\",\n        \"param\": null,\n        \"code\": \"model_not_found\"\n    }\n}\n".data(using: .utf8)!
+    }
 }