Calling Flux<ChatResponse>stream() returns unordered results

[hetao0403]

trafficstars

Bug description Calling Flux<ChatResponse>stream() returns unordered results Environment Spring AI version 1.0.0-SNAPSHOT , Java version 17,openai

[Grogdunn]

Funny never spotted, but I've used a lot the streams... Which client?

[qq418745]

@Grogdunn

@RestController
@AllArgsConstructor
@RequestMapping("v1/ai")
public class SimpleAiController {

	private final OpenAiChatClient chatClient;

	@GetMapping("/generateStream")
	public Flux<ChatResponse> generateStream(@RequestParam(value = "message", defaultValue = "Tell me a joke") String message) {
		Prompt prompt = new Prompt(new UserMessage(message));
		return chatClient.stream(prompt);
	}
}

Error：

org.springframework.ai.openai.api.OpenAiApi#chatCompletionStream

There's an extra "data:" here.

[qq418745]

I reviewed the source code at https://github.com/openai/openai-node and saw that the strings have been processed, with 'data' being used as a 'fieldname'.

[qq418745]

org.springframework.ai.openai.api.OpenAiApiIT#chatCompletionStream

I noticed that in the automated testing section of the spring-ai test code, despite the tests should have already been passed successfully, I still encountered issues when executing the test code myself, and it failed to pass the tests. This is very confusing, as logically, there should be no problems.

[Grogdunn]

I've tested against OpenAI service and cannot reproduce the wrong behaviour... can you make a test?

[qq418745]

@Grogdunn
"/v1/chat/completions" stream mode.

I am not directly using the OpenAI API, but rather accessing it through a proxy service. Debugging revealed that Spring WebFlux selects the appropriate MessageReader based on the "content-type". Since the "content-type: text/plain" did not engage the ServerSentEventHttpMessageReader, it failed to process the "data:" prefix. I suspect this is most likely the case, and I am still confirming this issue.

[qq418745]

"/v1/chat/completions" stream mode.

The response header from the proxy is "content-type: text/plain; charset=utf-8", but it should correctly be "text/event-stream".

[qq418745]

@Grogdunn My issue has been resolved; it was because of the response header.

[hetao0403]

Funny never spotted, but I've used a lot the streams... Which client?

OpenAiChatClient,3.5 and 4

[hetao0403]

I've tested against OpenAI service and cannot reproduce the wrong behaviour... can you make a test? Yes, sometimes it happens, not necessarily

[sennoy11012]

I wonder if it is in the chatCompletionStream method of the OpenAiApi class that this is used:

.floatMap((mono) -> {
    return mono;
})

It might be worth trying to change this method to:

.concatMap((mono) -> {
    return mono;
})

[YunaiV]

I have a similar problem

this is my log：

prompt is "1+1=" expect is "1+1=2" actual is "12+1="（wrong）

2024-05-21 17:05:50.035 |  INFO 39472 | reactor-http-nio-4 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content() result()]
2024-05-21 17:05:50.036 |  INFO 39472 | task-8 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content(1) result(1)]
2024-05-21 17:05:50.037 |  INFO 39472 | task-9 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content(2) result(12)]
2024-05-21 17:05:50.037 |  INFO 39472 | task-10 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content(+) result(12+)]
2024-05-21 17:05:50.038 |  INFO 39472 | task-11 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content(1) result(12+1)]
2024-05-21 17:05:50.038 |  INFO 39472 | task-12 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content() result(12+1)]
2024-05-21 17:05:50.039 |  INFO 39472 | task-13 [TID: N/A] c.i.y.m.a.s.impl.AiChatServiceImpl       | [xxx][content(=) result(12+1=)]

[YunaiV]

Bug description Calling Fluxstream() returns unordered results Environment Spring AI version 1.0.0-SNAPSHOT , Java version 17,openai

// 注意：Schedulers.immediate() 目的是，避免默认 Schedulers.parallel() 并发消费 chunk 导致 SSE 响应前端会乱序问题 this work for me

[YunaiV]

Bug description Calling Fluxstream() returns unordered results Environment Spring AI version 1.0.0-SNAPSHOT , Java version 17,openai

// 注意：Schedulers.immediate() 目的是，避免默认 Schedulers.parallel() 并发消费 chunk 导致 SSE 响应前端会乱序问题 this work for me

return streamResponse.publishOn(Schedulers.immediate()).map(chunk -> {

[Grogdunn]

The LLM expecially the generative one, are not good in math. All model currently in production respond with something "similar" a good response, but not the correct response.

Some models (gemini 1.5 pro I think) reach more accuracy, but not correctness at 100%. So :warning: BEWARE!

[YunaiV]

The LLM expecially the generative one, are not good in math. All model currently in production respond with something "similar" a good response, but not the correct response.

Some models (gemini 1.5 pro I think) reach more accuracy, but not correctness at 100%. So ⚠️ BEWARE!

Thank you.

My purpose is mainly to give llm a simple question to test whether spring webflux + Spring-AI will have sse out-of-order problems

[hetao0403]

Bug description Calling Fluxstream() returns unordered results Environment Spring AI version 1.0.0-SNAPSHOT , Java version 17,openai

// 注意：Schedulers.immediate() 目的是，避免默认 Schedulers.parallel() 并发消费 chunk 导致 SSE 响应前端会乱序问题 this work for me

thanks

[hetao0403]

Bug description Calling Fluxstream() returns unordered results Environment Spring AI version 1.0.0-SNAPSHOT , Java version 17,openai

// 注意：Schedulers.immediate() 目的是，避免默认 Schedulers.parallel() 并发消费 chunk 导致 SSE 响应前端会乱序问题 this work for me

thanks

[hetao0403]

Flux<ChatResponse> stream = chatClient.stream(prompt); return stream.publishOn(Schedulers.immediate()).map(chunk -> chunk);