GoogleAiGeminiStreamingChatModel - Gemini AI

[KaiSuchomel]

With langchain4j/langchain4j#1951 GoogleAiGeminiStreamingChatModel was introduced.
Is this also possible to add here??

[geoand]

Currently we don't use the upstream implementations, but our own. It should be easy to introduce a streaming implemenation, would you be interested in contributing that?

[KaiSuchomel]

I am not sure where to start 😃

[geoand]

The implementation for ChatModel can be found here.

An example of a StreamingChatModel implemenation (for Azure in this case) can be found here.

If you need any more info, we'll gladly help out.

[KaiSuchomel]

@geoand Can you please have a look here: https://github.com/KaiSuchomel/quarkus-langchain4j/tree/aigemini/streamingChatModel
I have implemented the StreamingChatModel.
It is working as expected, but using McpToolBox results in Timeouts when executing the tool.
I have no glue what is causing the issue.
My Setup is an AIService that looks like the following:

 @SystemMessage("""  You are a professional ChatBot that can communicate with a Oncite DPS System  Context:  {chat_memory.extractDialogue}  """) @McpToolBox("oncite") String chat(@MemoryId int memoryId, @UserMessage String usermessage); @SystemMessage("""  You are a professional ChatBot  Context:  {chat_memory.extractDialogue}  """) @McpToolBox("oncite") Multi<String> chatWithStream(@MemoryId int memoryId, @UserMessage String usermessage); @SystemMessage("""  You are a professional ChatBot that does not use tools  Context:  {chat_memory.extractDialogue}  """) Multi<String> chatWithStreamWOTools(@MemoryId int memoryId, @UserMessage String usermessage);

The corresponding Resource:

 @Inject GecAiService aiService; @PUT @Path("/chat/{memoryId}") @Produces(MediaType.TEXT_PLAIN) @Authenticated public String chat( @PathParam("memoryId") int aMemoryId, String aUsermessage) { return aiService.chat(aMemoryId, aUsermessage); } @PUT @Path("/chat/{memoryId}/multi/tool") @Produces(MediaType.TEXT_PLAIN) @Blocking @Authenticated public Multi<String> chatWithMulti(@PathParam("memoryId") int aMemoryId, String aUsermessage) { return aiService.chatWithStream(aMemoryId, aUsermessage); } @PUT @Path("/chat/{memoryId}/multi") @Produces(MediaType.TEXT_PLAIN) @Blocking @Authenticated public Multi<String> chatWithMultiWOTool(@PathParam("memoryId") int aMemoryId, String aUsermessage) { return aiService.chatWithStreamWOTools(aMemoryId, aUsermessage); }

I am using the following dependencies:

 <dependency> <groupId>io.quarkiverse.langchain4j</groupId> <artifactId>quarkus-langchain4j-ai-gemini</artifactId> <version>${quarkus.langchain4j.version}</version> </dependency> <dependency> <groupId>io.quarkiverse.langchain4j</groupId> <artifactId>quarkus-langchain4j-mcp</artifactId> <version>${quarkus.langchain4j.version}</version> </dependency> <dependency> <groupId>io.quarkiverse.langchain4j</groupId> <artifactId>quarkus-langchain4j-oidc-mcp-auth-provider</artifactId> <version>${quarkus.langchain4j.version}</version> </dependency>

The MCP Server is also developed by me.
It is secured with oidc thats why i used the quarkus-langchain4j-oidc-mcp-auth-provider extension.
Its working like a charme for the nonstreaming API.

I have attached the Logs of the different APIs.

Call1-Tool.log
Call2-Streaming.log
Call3-StreamingWithTool.log

As you can see in Call3 Log there is some illegal Thread Blocking.....
I think this is whats causing the timeout.

[KaiSuchomel]

Okay, i will give it a try 👍

[geoand]

🙏🏽

[KaiSuchomel]

@geoand Same Issue at openai:

2025-06-26 10:37:55,846 WARN [io.ver.cor.imp.BlockedThreadChecker] (vertx-blocked-thread-checker) Thread Thread[vert.x-eventloop-thread-1,5,main] has been blocked for 3120 ms, time limit is 2000 ms: io.vertx.core.VertxException: Thread blocked	at java.base/jdk.internal.misc.Unsafe.park(Native Method)	at java.base/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:269)	at java.base/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1910)	at java.base/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4023)	at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3969)	at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1983)	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2139)	at io.smallrye.context.CompletableFutureWrapper.get(CompletableFutureWrapper.java:161)	at dev.langchain4j.mcp.client.DefaultMcpClient.executeTool(DefaultMcpClient.java:197)	at dev.langchain4j.mcp.client.McpClient_dFVaY5GnvEyhYY_n6r1FXXeoyYA_Synthetic_ClientProxy.executeTool(Unknown Source)	at dev.langchain4j.mcp.McpToolProvider$DefaultToolExecutor.execute(McpToolProvider.java:218)	at io.quarkiverse.langchain4j.runtime.aiservice.QuarkusAiServiceStreamingResponseHandler$3.run(QuarkusAiServiceStreamingResponseHandler.java:186)	at io.quarkiverse.langchain4j.runtime.aiservice.QuarkusAiServiceStreamingResponseHandler.executeTools(QuarkusAiServiceStreamingResponseHandler.java:146)	at io.quarkiverse.langchain4j.runtime.aiservice.QuarkusAiServiceStreamingResponseHandler.onCompleteResponse(QuarkusAiServiceStreamingResponseHandler.java:179)	at dev.langchain4j.model.chat.StreamingChatModel$1.onCompleteResponse(StreamingChatModel.java:55)	at dev.langchain4j.model.openai.OpenAiStreamingChatModel.lambda$doChat$2(OpenAiStreamingChatModel.java:139)	at io.smallrye.mutiny.subscription.Subscribers$CallbackBasedSubscriber.onCompletion(Subscribers.java:109)	at io.smallrye.mutiny.operators.multi.builders.BaseMultiEmitter.completion(BaseMultiEmitter.java:65)	at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.drain(BufferItemMultiEmitter.java:110)	at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.completion(BufferItemMultiEmitter.java:68)	at io.smallrye.mutiny.operators.multi.builders.BaseMultiEmitter.complete(BaseMultiEmitter.java:56)	at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.complete(BufferItemMultiEmitter.java:11)	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.drainLoop(SerializedMultiEmitter.java:118)	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.drain(SerializedMultiEmitter.java:93)	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.onCompletion(SerializedMultiEmitter.java:88)	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.complete(SerializedMultiEmitter.java:155)	at org.jboss.resteasy.reactive.client.impl.MultiInvoker$MultiRequest.complete(MultiInvoker.java:87)	at org.jboss.resteasy.reactive.client.impl.SseEventSourceImpl.notifyCompletion(SseEventSourceImpl.java:220)	at org.jboss.resteasy.reactive.client.impl.SseEventSourceImpl.close(SseEventSourceImpl.java:203)	at org.jboss.resteasy.reactive.client.impl.SseEventSourceImpl.lambda$registerOnClient$2(SseEventSourceImpl.java:152)	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:252)	at io.vertx.core.http.impl.HttpEventHandler.handleEnd(HttpEventHandler.java:76)	at io.vertx.core.http.impl.HttpClientResponseImpl.handleEnd(HttpClientResponseImpl.java:250)	at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$new$0(Http1xClientConnection.java:424)	at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:279)	at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:157)	at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleEnd(Http1xClientConnection.java:734)	at io.vertx.core.impl.ContextImpl.execute(ContextImpl.java:327)	at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:158)	at io.vertx.core.http.impl.Http1xClientConnection.handleResponseEnd(Http1xClientConnection.java:965)	at io.vertx.core.http.impl.Http1xClientConnection.handleHttpMessage(Http1xClientConnection.java:837)	at io.vertx.core.http.impl.Http1xClientConnection.handleMessage(Http1xClientConnection.java:799)	at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:159)	at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)	at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)	at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)	at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1519)	at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1377)	at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1428)	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:530)	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:469)	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)	at java.base/java.lang.Thread.run(Thread.java:1575) 

Result after Timeout: It seems there was an issue retrieving your information. Please try again later or let me know how else I can assist you!

[geoand]

Thanks for checking. That's what I expected.

Feel free to open a PR with your implementation and just keep the failing test disabled for now

[KaiSuchomel]

I have created PR #1578