使用spring webflux客户端正确处理服务器发送事件的DONE

Question

我尝试使用 spring webflux 客户端调用返回 Server-Sent-Event (chatgpt) 的 REST API。

这是我的客户：

public interface OpenAIAsyncClient {

  @PostExchange(url = "/chat/completions", accept = MediaType.TEXT_EVENT_STREAM_VALUE)
  Flux<ChatCompletionResult> request(@RequestBody AbstractChatCompletionRequest request);
}

所以我会收到这样的 json 对象列表：

{"id":"chatcmpl-id","object":"chat.completion.chunk","created":1732310107,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp","choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}]}
...
{"id":"chatcmpl-id","object":"chat.completion.chunk","created":1732310107,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp","choices":[{"index":0,"delta":{},"logprobs":null,"finish_reason":"stop"}]}
[DONE]

但最后一行是一个 [DONE] 字符串，它破坏了杰克逊的反序列化。 有没有办法用 spring webflux 客户端正确处理这个问题？

Answer 1

好吧，我的方法是重写 Jackson2JsonDecoder :

public class IgnoreDoneJackson2JsonDecoder extends Jackson2JsonDecoder {

  public IgnoreDoneJackson2JsonDecoder() {
    super();
  }

  public IgnoreDoneJackson2JsonDecoder(ObjectMapper mapper, MimeType... mimeTypes) {
    super(mapper, mimeTypes);
  }

  @Override
  public Object decode(DataBuffer buffer, @NotNull ResolvableType targetType, MimeType mimeType,
      Map<String, Object> hints) throws DecodingException {
    if (buffer.capacity() > 5) {
      var begin = buffer.toString(0, 6, Charset.defaultCharset());
      if ("[DONE]".equals(begin)) {
        return null;
      }
    }
    return super.decode(buffer, targetType, mimeType, hints);
  }
}

并在创建weclient时设置它：

var webClient = WebClient.builder()
    .baseUrl(url)
    .codecs(configurer -> configurer.defaultCodecs().serverSentEventDecoder(customDecoder))
    .build();