Repository: LangStream/langstream (language: python)

Issue:
Parse exceptions when using Azure OpenAI only
Using the latest main branch, I am seeing parse exceptions when using Azure OpenAI. When using native OpenAI I don't see these errors. 

I sometimes get a partial response, like this:

```
You:
> Name the us presidents of the 20th century
✅
..Server:
The Presidents of the United States during the 20th century were:

1. William McKinley (1897–1901)
2. Theodore Roosevelt (1901–1909)
3. William Howard Taft (1909–1913)
4. Woodrow Wilson (1913–1921)
5. Warren G. Harding (1921–1923)
6. Calvin Coolidge (1923–192
```

But then output stops and the logs are spammed with this:

```
20:40:05.297 [pool-4-thread-1] INFO  a.l.kafka.runner.KafkaReaderWrapper -- Received 1 records from Kafka topics [output-topic-0]
20:40:05.301 [pool-4-thread-1] INFO  a.l.kafka.runner.KafkaReaderWrapper -- Received 1 records from Kafka topics [output-topic-0]
20:40:05.300 [pipeline-document-to-json-1-runner-8ce7d553-5282-424e-a013-2311bd01b426] ERROR a.l.r.agent.CompositeAgentProcessor -- Internal Error processing record: KafkaRecord.KafkaConsumerRecord(record=ConsumerRecord(topic = input-topic, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1695069601951, serialized key size = -1, serialized value size = 42, headers = RecordHeaders(headers = [RecordHeader(key = langstream-client-session-id, value = [66, 56, 65, 54, 50, 67, 55, 68, 45, 49, 55, 55, 52, 45, 52, 53, 55, 66, 45, 56, 48, 53, 65, 45, 68, 57, 48, 52, 57, 50, 69, 48, 54, 48, 50, 51])], isReadOnly = false), key = null, value = Name the us presidents of the 20th century))
reactor.core.Exceptions$ReactiveException: com.fasterxml.jackson.databind.JsonMappingException: Illegal unquoted character ((CTRL-CHAR, code 0)): has to be escaped using backslash to be included in name
 at [Source: (String)"{"id":"chatcmpl-80FKER2i7WcyHzoo1PSMUGIlbHo7G","object":"chat.completion.chunk","created":1695069602,"model":"gpt-4","choices":[{"index":0,"finish_reason":null,"delta":{"content":"9"},"content_filter_results":{"hate":{"filtered":false,"severity":"safe"},"self_harm":{"filtered":false,"severity":"safe"},"se\u0000\u0000\u0000\u0000\u0000\u0000xual":\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000"[truncated 4009 chars]; line: 1, column: 308] (through reference chain: com.azure.ai.openai.models.ChatCompletions["choices"]->java.util.ArrayList[0]->com.azure.ai.openai.models.ChatChoice["content_filter_results"])
	at reactor.core.Exceptions.propagate(Exceptions.java:408)
	at reactor.core.publisher.BlockingIterable$SubscriberIterator.hasNext(BlockingIterable.java:169)
	at java.base/java.util.Iterator.forEachRemaining(Iterator.java:132)
	at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1845)
	at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
	at ai.langstream.ai.agents.services.impl.OpenAICompletionService.getChatCompletions(OpenAICompletionService.java:91)
	at com.datastax.oss.streaming.ai.ChatCompletionsStep.processAsync(ChatCompletionsStep.java:149)
	at ai.langstream.ai.agents.GenAIToolKitAgent.processStep(GenAIToolKitAgent.java:138)
	at ai.langstream.ai.agents.GenAIToolKitAgent.processRecord(GenAIToolKitAgent.java:116)
	at ai.langstream.ai.agents.GenAIToolKitAgent.process(GenAIToolKitAgent.java:92)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.lambda$process$0(AgentCodeAndLoader.java:208)
	at ai.langstream.api.runner.code.AgentCodeAndLoader.executeNoExceptionWithContextClassloader(AgentCodeAndLoader.java:56)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.process(AgentCodeAndLoader.java:207)
	at ai.langstream.runtime.agent.CompositeAgentProcessor.invokeProcessor(CompositeAgentProcessor.java:165)
	at ai.langstream.runtime.agent.CompositeAgentProcessor.lambda$invokeProcessor$0(CompositeAgentProcessor.java:202)
	at ai.langstream.api.runner.code.SingleRecordAgentProcessor.process(SingleRecordAgentProcessor.java:40)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.lambda$process$0(AgentCodeAndLoader.java:208)
	at ai.langstream.api.runner.code.AgentCodeAndLoader.executeNoExceptionWithContextClassloader(AgentCodeAndLoader.java:56)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.process(AgentCodeAndLoader.java:207)
	at ai.langstream.runtime.agent.CompositeAgentProcessor.invokeProcessor(CompositeAgentProcessor.java:165)
	at ai.langstream.runtime.agent.CompositeAgentProcessor.process(CompositeAgentProcessor.java:225)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.lambda$process$0(AgentCodeAndLoader.java:208)
	at ai.langstream.api.runner.code.AgentCodeAndLoader.executeNoExceptionWithContextClassloader(AgentCodeAndLoader.java:56)
	at ai.langstream.api.runner.code.AgentCodeAndLoader$3.process(AgentCodeAndLoader.java:207)
	at ai.langstream.runtime.agent.AgentRunner.runProcessorAgent(AgentRunner.java:832)
	at ai.langstream.runtime.agent.AgentRunner.runMainLoop(AgentRunner.java:644)
	at ai.langstream.runtime.agent.AgentRunner.runJavaAgent(AgentRunner.java:412)
	at ai.langstream.runtime.agent.AgentRunner.run(AgentRunner.java:190)
	at ai.langstream.runtime.agent.AgentRunner.runAgent(AgentRunner.java:129)
	at ai.langstream.runtime.tester.LocalApplicationRunner.lambda$executeAgentRunners$2(LocalApplicationRunner.java:208)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Illegal unquoted character ((CTRL-CHAR, code 0)): has to be escaped using backslash to be included in name
 at [Source: (String)"{"id":"chatcmpl-80FKER2i7WcyHzoo1PSMUGIlbHo7G","object":"chat.completion.chunk","created":1695069602,"model":"gpt-4","choices":[{"index":0,"finish_reason":null,"delta":{"content":"9"},"content_filter_results":{"hate":{"filtered":false,"severity":"safe"},"self_harm":{"filtered":false,"severity":"safe"},"se\u0000\u0000\u0000\u0000\u0000\u0000xual":\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000"[truncated 4009 chars]; line: 1, column: 308] (through reference chain: com.azure.ai.openai.models.ChatCompletions["choices"]->java.util.ArrayList[0]->com.azure.ai.openai.models.ChatChoice["content_filter_results"])
	at com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:402)
	at com.fasterxml.jackson.databind.JsonMappingException.wrapWithPath(JsonMappingException.java:361)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.wrapAndThrow(BeanDeserializerBase.java:1826)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:280)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeUsingPropertyBased(BeanDeserializer.java:463)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault(BeanDeserializerBase.java:1405)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:352)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:185)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer._deserializeFromArray(CollectionDeserializer.java:359)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:244)
	at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:28)
	at com.fasterxml.jackson.databind.deser.SettableBeanProperty.deserialize(SettableBeanProperty.java:542)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeWithErrorWrapping(BeanDeserializer.java:564)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeUsingPropertyBased(BeanDeserializer.java:439)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault(BeanDeserializerBase.java:1405)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:352)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:185)
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:323)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4730)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3677)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3645)
	at com.azure.ai.openai.implementation.OpenAIServerSentEvents.handleCurrentLine(OpenAIServerSentEvents.java:98)
	at com.azure.ai.openai.implementation.OpenAIServerSentEvents.lambda$mapByteBuffersToEvents$0(OpenAIServerSentEvents.java:58)
	at reactor.core.publisher.FluxConcatMapNoPrefetch$FluxConcatMapNoPrefetchSubscriber.onNext(FluxConcatMapNoPrefetch.java:183)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:440)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	... 3 common frames omitted
```


Discussion / hints:
I have tried with both gpt-4 and gpt-3.5turbo and the issue is the same. 
If I roll back to 0.19, I don't see this issue.
I am taking a look
it seems that the LLM is answering with spam content like ""se\u0000\u0000\u0000\u0000\u0000\u0000xual"
it looks like this problem
https://github.com/Azure/azure-sdk-for-java/issues/35522
there is no spam, this is a typical answer


`{"index":0,"finish_reason":null,"message":null,"delta":{"role":null,"content":" diesel","name":null,"function_call":null,"context":null},"content_filter_results":{"sexual":{"severity":"safe","filtered":false},"violence":{"severity":"safe","filtered":false},"hate":{"severity":"safe","filtered":false},"self_harm":{"severity":"safe","filtered":false}}} chunks for answer chatcmpl-80RJkxHITH9BRqMq8B5fcEZejLA4D`

so "sexual" is part of the response.

The fact is that the output is scrambled, like some ByteBuffer used concurrently
I can reproduce the issue using "docker run"  but not in a unit test
also on minikube the error doesn't reproduce, maybe there is something in the way "docker run" executes the HTTP calls and Netty processes the values.

I have observed other differences in the way the Java OpenAI runs in docker, and I had to upgrade the OpenAI client to see it working on docker (see #419 ) 

I am investigating
The problem is that in "docker run" we are adding this library in the core classpath and that doesn't work well

```
<dependency>
      <groupId>com.azure</groupId>
      <artifactId>azure-core-http-okhttp</artifactId>
      <version>1.11.12</version>
    </dependency>
```

it is imported by the Azure blob storage
    in kubernetes we have this lib, in the .nar file, and the boot classpath is clean
```
     <groupId>com.azure</groupId>
          <artifactId>azure-core-http-netty</artifactId>
```
          
          
          
The short term fix is to remove the library from the docker image (who cares about Azure Code Storage with docker run ??)



Fix patch:
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java
index d309ae380..79ea7d253 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java
@@ -35,7 +35,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Flux;
@@ -84,11 +83,21 @@ public CompletableFuture<ChatCompletions> getChatCompletions(
             Flux<com.azure.ai.openai.models.ChatCompletions> flux =
                     client.getChatCompletionsStream(
                             (String) options.get("model"), chatCompletionsOptions);
-            Stream<com.azure.ai.openai.models.ChatCompletions> model = flux.toStream();
+
             ChatCompletionsConsumer chatCompletionsConsumer =
                     new ChatCompletionsConsumer(
                             streamingChunksConsumer, minChunksPerMessage, finished);
-            model.forEach(chatCompletionsConsumer);
+
+            flux.doOnError(
+                            error -> {
+                                log.error(
+                                        "Internal error while processing the streaming response",
+                                        error);
+                                finished.completeExceptionally(error);
+                            })
+                    .doOnNext(chatCompletionsConsumer)
+                    .subscribe();
+
             return finished.thenApply(
                     ___ -> {
                         result.setChoices(
@@ -101,8 +110,7 @@ public CompletableFuture<ChatCompletions> getChatCompletions(
         } else {
             com.azure.ai.openai.models.ChatCompletions chatCompletions =
                     client.getChatCompletions((String) options.get("model"), chatCompletionsOptions)
-                            .toFuture()
-                            .get();
+                            .block();
             result.setChoices(
                     chatCompletions.getChoices().stream()
                             .map(c -> new ChatChoice(convertMessage(c)))


Test patch:
diff --git a/langstream-runtime/langstream-runtime-tester/pom.xml b/langstream-runtime/langstream-runtime-tester/pom.xml
index da105f731..21c32ac13 100644
--- a/langstream-runtime/langstream-runtime-tester/pom.xml
+++ b/langstream-runtime/langstream-runtime-tester/pom.xml
@@ -47,6 +47,10 @@
           <groupId>${project.groupId}</groupId>
           <artifactId>langstream-pulsar-runtime</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>${project.groupId}</groupId>
+          <artifactId>langstream-codestorage-azure-blob-storage</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
@@ -54,6 +58,12 @@
       <artifactId>langstream-webservice</artifactId>
       <version>${project.version}</version>
       <classifier>original</classifier>
+      <exclusions>
+        <exclusion>
+          <groupId>${project.groupId}</groupId>
+          <artifactId>langstream-codestorage-azure-blob-storage</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>