Merge pull request #1876 from ejstuart/feature/intermediate-chat-event

Add intermediate chat event

Author: geoand

core/runtime/src/main/java/io/quarkiverse/langchain4j/runtime/aiservice/ChatEvent.java

@@ -26,6 +26,7 @@
  * @see ContentFetchedEvent
  * @see AccumulatedResponseEvent
  * @see PartialThinkingEvent
+ * @see IntermediateResponseEvent
  */
 public class ChatEvent {
 
@@ -60,7 +61,11 @@ public enum ChatEventType {
         /**
          * Indicates that a partial thinking/reasoning chunk has been received (experimental).
          */
-        PartialThinking
+        PartialThinking,
+        /**
+         * Indicates that an intermediate response has been received during multi-turn interactions.
+         */
+        IntermediateResponse
 
     }
 
@@ -301,4 +306,61 @@ public String getText() {
         }
     }
 
+    /**
+     * Event emitted when an intermediate response is received during multi-turn interactions.
+     * This event occurs when the AI responds with tool execution requests, indicating that
+     * additional turns are needed to complete the conversation.
+     *
+     * <p>
+     * This is particularly useful for:
+     * <ul>
+     * <li>Capturing AI reasoning (thinking) that led to tool execution decisions</li>
+     * <li>Retrieving thinking signatures for validation/auditing</li>
+     * <li>Supporting interleaved thinking (reasoning between tool calls)</li>
+     * <li>Building complete audit trails of AI decision-making</li>
+     * </ul>
+     * </p>
+     *
+     * <p>
+     * Example flow with extended thinking:
+     *
+     * <pre>
+     * User: "What's the weather in Auckland?"
+     * → IntermediateResponseEvent: thinking="I should check the time first", tools=[getCurrentTime]
+     * → ToolExecutedEvent: getCurrentTime → "3:45 PM"
+     * → IntermediateResponseEvent: thinking="Now I know it's afternoon, check weather", tools=[getWeather]
+     * → ToolExecutedEvent: getWeather → "Sunny, 72°F"
+     * → ChatCompletedEvent: text="It's 3:45 PM and sunny in Auckland"
+     * </pre>
+     * </p>
+     *
+     * <p>
+     * <b>Note:</b> This event is essential for interleaved thinking to work, as it provides
+     * access to thinking blocks that occur between tool calls.
+     * </p>
+     */
+    public static class IntermediateResponseEvent extends ChatEvent {
+        private final ChatResponse chatResponse;
+
+        /**
+         * Constructs a new IntermediateResponseEvent with the given chat response.
+         *
+         * @param chatResponse the intermediate chat response containing the AI's thinking,
+         *        tool execution requests, and metadata for this turn
+         */
+        public IntermediateResponseEvent(ChatResponse chatResponse) {
+            super(ChatEventType.IntermediateResponse);
+            this.chatResponse = chatResponse;
+        }
+
+        /**
+         * Returns the intermediate chat response.
+         *
+         * @return the chat response containing thinking, tool requests, and attributes for this turn
+         */
+        public ChatResponse getChatResponse() {
+            return chatResponse;
+        }
+    }
+
 }

core/runtime/src/main/java/io/quarkiverse/langchain4j/runtime/aiservice/QuarkusAiServiceStreamingResponseHandler.java

@@ -59,6 +59,7 @@ public class QuarkusAiServiceStreamingResponseHandler implements StreamingChatRe
     private final Consumer<PartialThinking> partialThinkingHandler;
     private final Consumer<Response<AiMessage>> completionHandler;
     private final Consumer<BeforeToolExecution> beforeToolExecutionHandler;
+    private final Consumer<ChatResponse> intermediateResponseHandler;
     private final Consumer<ToolExecution> toolExecuteHandler;
     private final Consumer<ChatResponse> completeResponseHandler;
     private final Consumer<Throwable> errorHandler;
@@ -79,6 +80,7 @@ public class QuarkusAiServiceStreamingResponseHandler implements StreamingChatRe
             Consumer<String> partialResponseHandler,
             Consumer<PartialThinking> partialThinkingHandler,
             Consumer<BeforeToolExecution> beforeToolExecutionHandler,
+            Consumer<ChatResponse> intermediateResponseHandler,
             Consumer<ToolExecution> toolExecuteHandler,
             Consumer<ChatResponse> completeResponseHandler,
             Consumer<Response<AiMessage>> completionHandler,
@@ -97,6 +99,7 @@ public class QuarkusAiServiceStreamingResponseHandler implements StreamingChatRe
         this.partialResponseHandler = ensureNotNull(partialResponseHandler, "partialResponseHandler");
         this.partialThinkingHandler = partialThinkingHandler;
         this.beforeToolExecutionHandler = beforeToolExecutionHandler;
+        this.intermediateResponseHandler = intermediateResponseHandler;
         this.completeResponseHandler = completeResponseHandler;
         this.completionHandler = completionHandler;
         this.toolExecuteHandler = toolExecuteHandler;
@@ -125,6 +128,7 @@ public QuarkusAiServiceStreamingResponseHandler(AiServiceContext context, Invoca
             Consumer<String> partialResponseHandler,
             Consumer<PartialThinking> partialThinkingHandler,
             Consumer<BeforeToolExecution> beforeToolExecutionHandler,
+            Consumer<ChatResponse> intermediateResponseHandler,
             Consumer<ToolExecution> toolExecuteHandler, Consumer<ChatResponse> completeResponseHandler,
             Consumer<Response<AiMessage>> completionHandler,
             Consumer<Throwable> errorHandler, List<ChatMessage> temporaryMemory, TokenUsage sum,
@@ -137,6 +141,7 @@ public QuarkusAiServiceStreamingResponseHandler(AiServiceContext context, Invoca
         this.partialResponseHandler = ensureNotNull(partialResponseHandler, "partialResponseHandler");
         this.partialThinkingHandler = partialThinkingHandler;
         this.beforeToolExecutionHandler = beforeToolExecutionHandler;
+        this.intermediateResponseHandler = intermediateResponseHandler;
         this.toolExecuteHandler = toolExecuteHandler;
         this.completeResponseHandler = completeResponseHandler;
         this.completionHandler = completionHandler;
@@ -245,6 +250,9 @@ public void onCompleteResponse(ChatResponse completeResponse) {
                 @Override
                 public void run() {
                     addToMemory(aiMessage);
+                    if (intermediateResponseHandler != null) {
+                        intermediateResponseHandler.accept(completeResponse);
+                    }
                     for (ToolExecutionRequest toolExecutionRequest : aiMessage.toolExecutionRequests()) {
                         // Call before tool execution handler
                         if (beforeToolExecutionHandler != null) {
@@ -297,6 +305,7 @@ public void run() {
                             partialResponseHandler,
                             partialThinkingHandler,
                             beforeToolExecutionHandler,
+                            intermediateResponseHandler,
                             toolExecuteHandler,
                             completeResponseHandler,
                             completionHandler,

core/runtime/src/main/java/io/quarkiverse/langchain4j/runtime/aiservice/QuarkusAiServiceTokenStream.java

@@ -53,6 +53,7 @@ public class QuarkusAiServiceTokenStream implements TokenStream {
     private Consumer<Throwable> errorHandler;
     private Consumer<Response<AiMessage>> completionHandler;
     private Consumer<BeforeToolExecution> beforeToolExecutionHandler;
+    private Consumer<ChatResponse> intermediateResponseHandler;
     private Consumer<ToolExecution> toolExecuteHandler;
     private Consumer<ChatResponse> completeResponseHandler;
 
@@ -63,6 +64,7 @@ public class QuarkusAiServiceTokenStream implements TokenStream {
     private int onErrorInvoked;
     private int ignoreErrorsInvoked;
     private int beforeToolExecutionInvoked;
+    private int onIntermediateResponseInvoked;
     private int toolExecuteInvoked;
 
     public QuarkusAiServiceTokenStream(List<ChatMessage> messages,
@@ -120,6 +122,13 @@ public TokenStream onToolExecuted(Consumer<ToolExecution> toolExecuteHandler) {
         return this;
     }
 
+    @Override
+    public TokenStream onIntermediateResponse(Consumer<ChatResponse> intermediateResponseHandler) {
+        this.intermediateResponseHandler = intermediateResponseHandler;
+        this.onIntermediateResponseInvoked++;
+        return this;
+    }
+
     @Override
     public TokenStream onCompleteResponse(Consumer<ChatResponse> completionHandler) {
         this.completeResponseHandler = completionHandler;
@@ -156,6 +165,7 @@ public void start() {
                 partialResponseHandler,
                 partialThinkingHandler,
                 beforeToolExecutionHandler,
+                intermediateResponseHandler,
                 toolExecuteHandler,
                 completeResponseHandler,
                 completionHandler,
@@ -209,6 +219,10 @@ private void validateConfiguration() {
             throw new IllegalConfigurationException("onPartialThinking can be invoked on TokenStream at most 1 time");
         }
 
+        if (onIntermediateResponseInvoked > 1) {
+            throw new IllegalConfigurationException("onIntermediateResponse can be invoked on TokenStream at most 1 time");
+        }
+
         if (onErrorInvoked + ignoreErrorsInvoked != 1) {
             throw new IllegalConfigurationException("One of [onError, ignoreErrors] " +
                     "must be invoked on TokenStream exactly 1 time");

core/runtime/src/main/java/io/quarkiverse/langchain4j/runtime/aiservice/TokenStreamMulti.java

@@ -70,6 +70,7 @@ private void createTokenStream(UnicastProcessor<ChatEvent> processor) {
                     processor.onComplete();
                 })
                 .onRetrieved(content -> processor.onNext(new ChatEvent.ContentFetchedEvent(content)))
+                .onIntermediateResponse(response -> processor.onNext(new ChatEvent.IntermediateResponseEvent(response)))
                 .beforeToolExecution(
                         beforeExecution -> processor.onNext(new ChatEvent.BeforeToolExecutionEvent(beforeExecution.request())))
                 .onToolExecuted(execution -> processor.onNext(new ChatEvent.ToolExecutedEvent(execution)))

docs/modules/ROOT/pages/guide-streamed-responses.adoc

@@ -66,9 +66,10 @@ When using a `Multi<ChatEvent>` each emitted item is a subclass of `ChatEvent` w
 * `BeforeToolExecutionEvent` - Emitted before a tool is executed
 * `ToolExecutedEvent` - Tool execution metadata
 * `ContentFetchedEvent` - Content retrieved from RAG sources
+* `IntermediateResponseEvent` - An intermediate response during multi-turn interactions with tool execution
 * `ChatCompletedEvent` - The final LLM response
 
-NOTE: When tools are used, the event flow is: `BeforeToolExecutionEvent` → (tool execution) → `ToolExecutedEvent`. This allows monitoring both before and after tool execution phases.
+NOTE: When tools are used, the event flow is: `IntermediateResponseEvent` → `BeforeToolExecutionEvent` → (tool execution) → `ToolExecutedEvent` → (recursive LLM call) → `ChatCompletedEvent`. The `IntermediateResponseEvent` captures the AI's thinking and tool execution requests before tools are executed.
 
 [source,java]
 ----