我也有同样的错误，没有使用任何库，只是使用我自己的URLSession实现。
我通过添加HTTP头Accept: text/event-stream解决了这个问题。

var request = URLRequest(url: url)
    request.setValue("text/event-stream", forHTTPHeaderField: "Accept")

几年后，现在有了Swift并发，也许有人会发现这很有帮助。
看来原来关于缓冲区填充的问题是一个问题。我发现，通过从连接到SSE端点的服务请求返回AsyncThrowingStream，您可以通过AsyncThrowingStream的bufferingPolicy控制缓冲区。
苹果电脑：AsyncThrowingStream.init(:bufferingPolicy::)
下面的代码是为了可读性。你当然可以使用泛型，使一些东西更可重用，但这得到了重点。
假设你有一个端点，它接受了一个带有主体的POST请求，然后用一个事件流进行响应。正如@Guillaume所提到的，您希望返回text/event-stream，而不是典型的JSON。
请求体中的JSON可能看起来像这样：

{
  "question": "What's the answer?"
}

响应可能会将事件以文本的形式返回，如下所示：

data: {"answer": "This"}
data: {"answer": "This is"}
data: {"answer": "This is the"}
data: {"answer": "This is the answer."}

您需要了解从服务器返回的事件的格式，以便正确解码。它应该有一个JSON字符串在那里的某个地方，因为你将不得不解码的东西，但它可能没有前缀data:。我们将继续假设它是针对这个例子的。
下面是发出此请求的示例代码：

final class Service {
    enum ServiceError: Error {
        case generic
    }

    // 1.
    private var streamTask: Task<Void, Error>?
    private let url = URL(string: "https://example.com/api/events")!
    
    func sendRequest(with question: String) async throws -> AsyncThrowingStream<Response, Error> {
        // 2.
        var request = URLRequest(url: url)
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")
        request.setValue("text/event-stream; charset=utf-8", forHTTPHeaderField: "Accept")
        request.setValue("keep-alive", forHTTPHeaderField: "Connection")
        request.httpMethod = "POST"
        request.httpBody = try JSONEncoder().encode(
            Request(question: question)
        )
        
        // 3.
        let (asyncBytes, response) = try await URLSession.shared.bytes(for: request)
        try Task.checkCancellation()
        
        guard let httpResponse = response as? HTTPURLResponse,
              (200...299).contains(httpResponse.statusCode)
        else {
            throw ServiceError.generic
        }
        
        // 4.
        return AsyncThrowingStream(Response.self, bufferingPolicy: .bufferingNewest(3)) { continuation in
            // 5.
            continuation.onTermination = { [weak self] _ in
                self?.streamTask?.cancel()
                self?.streamTask = nil
            }
            
            // 6.
            streamTask = Task {
                // 7.
                for try await line in asyncBytes.lines {
                    try Task.checkCancellation()
                    
                    // 8.
                    guard line.hasPrefix("data: "),
                          let data = line.dropFirst(6).data(using: .utf8)
                    else {
                        // Seems like we got a partial event or one that doesn't start with
                        // `data: `. No big deal. Just continue to the next event in the stream.
                        continue
                    }
                    
                    // 9.
                    do {
                        let response = try JSONDecoder().decode(Response.self, from: data)
                        continuation.yield(with: .success(response))
                    } catch {
                        continuation.finish(throwing: error)
                    }
                }
                // 10.
                continuation.finish()
            }
        }
    }
}

struct Request: Encodable {
    let question: String
}

struct Response: Decodable {
    let answer: String
}

1.我们需要保留一个非结构化的任务，这样我们可以在需要的时候取消它。这在下面是有意义的。
1.建立我们的URL请求。没什么特别的
1.调用这个bytes(for:delegate:)方法。这个东西很酷，因为它返回(URLSession.AsyncBytes, URLResponse)。AsyncBytes部分是关键。Apple Docs
1.用定义的bufferingPolicy构造AsyncThrowingStream。在这里，我选择将最新的3个事件保存在缓冲区中。做对你有用或有意义的事情。有一个更简单的AsyncThrowingStream(unfolding:)，我们可以使用，但看到GOTCHA在通过这些步骤后，在底部的更多信息。
1.当连续性终止时（即由于取消），我们应该cancel()和零出我们的streamTask。如果你还在想“什么流任务？“这是第六步。
1.这是streamTask。我们创建并保留对非结构化Task的引用来完成我们的工作。通过将它存储在第1步的streamTask属性中，我们可以在第5步中取消它。
1.迭代我们从URLSession调用中得到的asyncBytes.lines。看看我们上面的示例响应中的每个data: {...}事件是如何在一个新行上开始的？这将覆盖每一行。
1.检查line的内容（记住它只是文本，不是JSON）是否以data:开头，然后将这些字符剥离并将其转换为Data。如果不是，我们就有不认识的东西了。继续等待下一个事件。
1.从上一步中取出Data并对其进行解码。如果成功，则yield解码的Response。如果没有，抛出错误，我们就完成了。（您也可以选择不完成，并继续尝试解码流中的下一个事件）。
1.假设我们没有抛出上面的任何东西，一旦我们完成了流中的所有事件，我们将.finish()。
[*] The Gotcha：**
你们中的一些人可能会想：“你在一个令人不安的背景下！为什么要将AsyncThrowingStream与同步build闭包一起使用，后者只需要一个continuation来 Package 一个非结构化的Task？你破坏了自动取消系统！为什么不返回AsyncThrowingStream(unfolding:)？”如果你之前没有这么想，或者现在在想“等等，什么？”，在这里：
我一开始就试着这么做。将// 4.以下的所有内容替换为以下内容，您将获得：

return AsyncThrowingStream {
    for try await line in asyncBytes.lines {
        try Task.checkCancellation()
        
        guard line.hasPrefix("data: "),
              let data = line.dropFirst(6).data(using: .utf8)
        else {
            // Seems like we got a partial event or one that doesn't start with
            // `data: `. No big deal. Just continue to the next event in the stream.
            continue
        }
        
        do {
            return try JSONDecoder().decode(Response.self, from: data)
        } catch {
            throw error
        }
    }
    return nil
}

天哪，这可好多了。不需要坚持一些非结构化的任务。你会自动取消。可以只返回/抛出，而不是弄乱continuation的东西。只是代码更少。

然而，我发现这只是 * 有时 * 中途停止（通常发生在较长的响应中）。我可以在Proxyman中看到服务器仍然在发送事件，但是当我将print(line)放在for try await line in asyncBytes.lines {}的顶部时，它只是停止了中途打印。甚至docs也说“默认情况下，缓冲区限制是Int.max，这意味着它是无界的。”所以要么这不是真的，要么有一些非缓冲区相关的问题发生，而使用continuation的方法不会遇到。

如果您使用AsyncThrowingStream(unfolding:)方法，请确保进行大量测试。
我很想听听是否有人有任何想法，为什么基于continuation的方法是唯一有效的方法。