diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt index 498e275ee..e98631407 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/streaming/StreamingPromptRunner.kt @@ -87,6 +87,13 @@ interface StreamingPromptRunner : PromptRunner { * Create a reactive stream with both objects and thinking content. * Provides access to the LLM's reasoning process alongside the results. * + * Enables application-level thinking on the Interaction + * ([com.embabel.common.ai.model.Thinking.withExtraction] / `extractThinking`) so the + * stream injects prompt format instructions and returns reasoning blocks. This is + * independent of any provider model budget (`Thinking.withTokenBudget(...)`), which is + * only needed when the provider requires a budget (e.g. Anthropic extended thinking) — + * not as a prerequisite for this API. + * * @param itemClass The class of objects to create * @return Flux emitting StreamingEvent instances for objects and thinking */ diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt index d353bb541..6a7cf3984 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/api/common/support/OperationContextDelegate.kt @@ -357,7 +357,9 @@ internal data class OperationContextDelegate( val streamingLlmOperations = streamingFactory().createStreamingOperations(llm) return streamingLlmOperations.createObjectStreamWithThinking( messages = messages, - interaction = streamingInteraction(), + // Enable application-level thinking extraction (format instructions + extractThinking) + // when needed, while preserving any caller-configured model thinking budget. + interaction = streamingInteractionForThinkingIfNecessary(), outputClass = itemClass, agentProcess = context.processContext.agentProcess, action = action, @@ -403,6 +405,29 @@ internal data class OperationContextDelegate( ) } + /** + * Streaming interaction for [createObjectStreamWithThinking]. + * + * Turns on application-level thinking on the Interaction via [Thinking.extractThinking] + * (same idea as non-streaming [thinkingInteraction] / [Thinking.withExtraction]), without + * requiring a provider token budget. Existing budget is preserved with [Thinking.applyExtraction]. + * + * SPI streaming then reads [Thinking.extractThinking] to decide whether to inject prompt + * format instructions — no separate "thinking format" flag. Propagation is entirely through + * [LlmInteraction.llm.thinking]. + * + * This is *not* LLM-native reasoning (provider thinking channels; see #1716). + * Provider budget remains optional: `LlmOptions.withThinking(Thinking.withTokenBudget(...))`. + */ + private fun streamingInteractionForThinkingIfNecessary(): LlmInteraction { + val base = streamingInteraction() + val thinking = when (val existing = llm.thinking) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() + } + return base.copy(llm = llm.withThinking(thinking)) + } + private fun streamingFactory(): StreamingLlmOperationsFactory { val llmOperations = context.agentPlatform().platformServices.llmOperations return llmOperations as? StreamingLlmOperationsFactory diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt index 529ecd971..97172165e 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperations.kt @@ -33,6 +33,7 @@ import com.embabel.agent.spi.support.springai.toSpringAiMessage import com.embabel.agent.spi.support.springai.toSpringToolCallbacks import com.embabel.chat.Message import com.embabel.common.ai.converters.streaming.StreamingJacksonOutputConverter +import com.embabel.common.ai.model.Thinking import com.embabel.common.core.streaming.StreamingEvent import org.slf4j.LoggerFactory import org.springframework.ai.chat.messages.SystemMessage @@ -234,6 +235,8 @@ internal class StreamingChatClientOperations( ): Flux { return doTransformObjectStreamInternal( messages = messages, + // Object-only stream: leave Interaction thinking as-is. Format instructions follow + // Thinking.extractThinking (application-level), not provider tokenBudget (Thinking.enabled). interaction = interaction, outputClass = outputClass, llmRequestEvent = llmRequestEvent, @@ -295,7 +298,9 @@ internal class StreamingChatClientOperations( ): Flux> { return doTransformObjectStreamInternal( messages = messages, - interaction = interaction, + // *WithThinking*: ensure Interaction carries application-level Thinking + // (extractThinking). Format instructions follow that flag — no separate SPI param. + interaction = withApplicationLevelThinkingIfNecessary(interaction), outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, @@ -303,6 +308,26 @@ internal class StreamingChatClientOperations( ) } + /** + * Ensure [Thinking.extractThinking] is set on the interaction for application-level + * (prompt-instructed) thinking streams. Preserves any existing provider budget + * ([Thinking.enabled] / [Thinking.tokenBudget]) via [Thinking.applyExtraction]. + * + * This is *not* LLM-native reasoning (provider thinking channels — see #1716). + */ + private fun withApplicationLevelThinkingIfNecessary(interaction: LlmInteraction): LlmInteraction { + val existing = interaction.llm.thinking + val thinking = when (existing) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() + } + return if (thinking === existing) { + interaction + } else { + interaction.copy(llm = interaction.llm.withThinking(thinking)) + } + } + /** * Internal unified streaming implementation - workhorse -that handles the complete transformation pipeline. * @@ -330,6 +355,9 @@ internal class StreamingChatClientOperations( * **Performance Characteristics:** * - Streaming-friendly: no blocking operations * + * Prompt thinking format follows [Thinking.extractThinking] on [interaction] (application-level). + * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget] and is independent. + * * @return Unified Flux> that public methods can filter as needed */ private fun doTransformObjectStreamInternal( @@ -348,6 +376,9 @@ internal class StreamingChatClientOperations( // Chat Options, additional potential option "streaming" val chatOptions = requireSpringAiLlm(llm).convertOptions(interaction.llm) + // Application-level thinking format: Thinking.extractThinking (not provider tokenBudget / enabled). + val includeApplicationLevelThinking = interaction.llm.thinking?.extractThinking == true + // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any; // erase O via Class for the construction, cast result back at use sites. @Suppress("UNCHECKED_CAST") @@ -357,7 +388,7 @@ internal class StreamingChatClientOperations( clazz = outputClassAny, objectMapper = chatClientLlmOperations.objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = interaction.llm.thinking?.enabled ?: false, + thinkingEnabled = includeApplicationLevelThinking, ) as StreamingJacksonOutputConverter // signature compatibility for downstream Flux/StreamingEvent uses // Build prompt using helper methods, including streaming format instructions diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt index aefdd863d..28cd06290 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/streaming/StreamingLlmOperationsImpl.kt @@ -37,6 +37,7 @@ import com.embabel.agent.spi.support.guardrails.validateUserInput import com.embabel.chat.Message import com.embabel.chat.UserMessage import com.embabel.common.ai.converters.streaming.StreamingJacksonOutputConverter +import com.embabel.common.ai.model.Thinking import com.embabel.common.core.streaming.StreamingEvent import tools.jackson.databind.ObjectMapper import org.slf4j.LoggerFactory @@ -152,6 +153,7 @@ internal class StreamingLlmOperationsImpl( ): Flux { return doTransformObjectStreamInternal( messages = messages, + // Object-only: format instructions follow Thinking.extractThinking on Interaction. interaction = interaction, outputClass = outputClass, llmRequestEvent = llmRequestEvent, @@ -172,7 +174,8 @@ internal class StreamingLlmOperationsImpl( ): Flux> { return doTransformObjectStreamInternal( messages = messages, - interaction = interaction, + // *WithThinking*: ensure Interaction has application-level Thinking.extractThinking. + interaction = withApplicationLevelThinkingIfNecessary(interaction), outputClass = outputClass, llmRequestEvent = llmRequestEvent, agentProcess = agentProcess, @@ -180,6 +183,23 @@ internal class StreamingLlmOperationsImpl( ) } + /** + * Ensure [Thinking.extractThinking] is set for application-level (prompt-instructed) thinking. + * Preserves provider budget via [Thinking.applyExtraction]. Not LLM-native reasoning (#1716). + */ + private fun withApplicationLevelThinkingIfNecessary(interaction: LlmInteraction): LlmInteraction { + val existing = interaction.llm.thinking + val thinking = when (existing) { + null, Thinking.NONE -> Thinking.withExtraction() + else -> if (existing.extractThinking) existing else existing.applyExtraction() + } + return if (thinking === existing) { + interaction + } else { + interaction.copy(llm = interaction.llm.withThinking(thinking)) + } + } + // ======================================== // Internal implementation // ======================================== @@ -191,6 +211,9 @@ internal class StreamingLlmOperationsImpl( * 1. Raw LLM chunks from [LlmMessageStreamer] * 2. Line buffering via [rawChunksToLines] * 3. Event generation via [StreamingJacksonOutputConverter] + * + * Prompt thinking format follows [Thinking.extractThinking] on [interaction]. + * Provider model budget remains [Thinking.enabled] / [Thinking.tokenBudget]. */ private fun doTransformObjectStreamInternal( messages: List, @@ -204,6 +227,7 @@ internal class StreamingLlmOperationsImpl( // Create converter for JSONL parsing. // Spring AI 2.0's StreamingJacksonOutputConverter requires T : Any; // erase O via Class for the construction, cast back for downstream Flux/StreamingEvent. + val includeApplicationLevelThinking = interaction.llm.thinking?.extractThinking == true @Suppress("UNCHECKED_CAST") val outputClassAny = outputClass as Class @Suppress("UNCHECKED_CAST") @@ -211,7 +235,7 @@ internal class StreamingLlmOperationsImpl( clazz = outputClassAny, objectMapper = objectMapper, fieldFilter = interaction.fieldFilter, - thinkingEnabled = interaction.llm.thinking?.enabled ?: false, + thinkingEnabled = includeApplicationLevelThinking, ) as StreamingJacksonOutputConverter // Build prompt contributions with streaming format instructions diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt index 584f4a811..f6139ff46 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/springai/streaming/StreamingChatClientOperationsTest.kt @@ -15,6 +15,7 @@ */ package com.embabel.agent.spi.support.springai.streaming +import com.embabel.agent.api.common.InteractionId import com.embabel.agent.core.Action import com.embabel.agent.core.AgentProcess import com.embabel.agent.core.support.LlmInteraction @@ -22,9 +23,12 @@ import com.embabel.agent.core.internal.streaming.StreamingLlmOperations import com.embabel.agent.spi.support.springai.ChatClientLlmOperations import com.embabel.agent.spi.support.springai.SpringAiLlmService import com.embabel.chat.UserMessage +import com.embabel.common.ai.model.LlmOptions import tools.jackson.module.kotlin.jacksonObjectMapper +import io.mockk.CapturingSlot import io.mockk.every import io.mockk.mockk +import io.mockk.slot import io.mockk.verify import org.junit.jupiter.api.Assertions.* import org.junit.jupiter.api.BeforeEach @@ -37,6 +41,9 @@ import org.springframework.ai.tool.ToolCallback import reactor.core.publisher.Flux import reactor.test.StepVerifier import java.time.Duration +import com.embabel.common.ai.model.OptionsConverter +import com.embabel.common.ai.model.Thinking +import com.embabel.common.ai.prompt.PromptContributor /** * Unit tests for StreamingChatClientOperations. @@ -79,7 +86,7 @@ class StreamingChatClientOperationsTest { every { mockChatClientLlmOperations.createChatClient(mockLlm) } returns mockChatClient every { mockInteraction.promptContributors } returns emptyList() every { mockLlm.promptContributors } returns emptyList() - val mockOptionsConverter = mockk(relaxed = true) + val mockOptionsConverter = mockk(relaxed = true) every { mockLlm.optionsConverter } returns mockOptionsConverter every { mockOptionsConverter.convertOptions(any(), any()) } returns mockk(relaxed = true) every { mockInteraction.llm } returns mockk(relaxed = true) @@ -161,8 +168,8 @@ class StreamingChatClientOperationsTest { mockAction ) - // Then - verify { mockChatClientLlmOperations.getLlm(mockInteraction) } + // Then: Interaction may be a copy with Thinking.extractThinking enabled + verify { mockChatClientLlmOperations.getLlm(any()) } } @Test @@ -468,16 +475,82 @@ class StreamingChatClientOperationsTest { } - private fun mockChatClientForStreaming(chunkFlux: Flux) { + private fun mockChatClientForStreaming(chunkFlux: Flux): CapturingSlot { val mockRequestSpec = mockk(relaxed = true) val mockContentStreamSpec = mockk(relaxed = true) + val promptSlot = slot() - every { mockChatClient.prompt(any()) } returns mockRequestSpec + every { mockChatClient.prompt(capture(promptSlot)) } returns mockRequestSpec every { mockRequestSpec.tools(any>()) } returns mockRequestSpec every { mockRequestSpec.options(any()) } returns mockRequestSpec every { mockRequestSpec.stream() } returns mockContentStreamSpec every { mockContentStreamSpec.content() } returns chunkFlux + return promptSlot + } + + @Nested + inner class ThinkingFormatInstructionTests { + + @Test + fun `createObjectStreamWithThinking includes thinking format without LlmOptions thinking config`() { + // Given: real Interaction with no thinking budget / extraction (the #1799 pre-req). + // SPI enables Thinking.extractThinking on the Interaction for *WithThinking. + val interaction = LlmInteraction( + id = InteractionId("test-thinking-format"), + llm = LlmOptions(), + ) + val promptSlot = mockChatClientForStreaming( + Flux.just("{\"name\":\"Item1\",\"value\":1}\n") + ) + + // When + streamingOperations.createObjectStreamWithThinking( + messages = listOf(UserMessage("test")), + interaction = interaction, + outputClass = TestItem::class.java, + agentProcess = mockAgentProcess, + action = mockAction + ).collectList().block(Duration.ofSeconds(2)) + + // Then: format instructions still ask for blocks (driven by extractThinking) + assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called") + val promptText = promptSlot.captured.contents + assertTrue( + promptText.contains(""), + "createObjectStreamWithThinking should inject thinking format without Thinking.withTokenBudget" + ) + } + + @Test + fun `createObjectStream omits thinking format for provider budget only`() { + // Provider budget (Thinking.enabled + tokenBudget) alone must not inject + // application-level format instructions — that follows extractThinking only. + val interaction = LlmInteraction( + id = InteractionId("test-budget-only"), + llm = LlmOptions().withThinking(Thinking.withTokenBudget(8000)), + ) + val promptSlot = mockChatClientForStreaming( + Flux.just("{\"name\":\"Item1\",\"value\":1}\n") + ) + + // When + streamingOperations.createObjectStream( + messages = listOf(UserMessage("test")), + interaction = interaction, + outputClass = TestItem::class.java, + agentProcess = mockAgentProcess, + action = mockAction + ).collectList().block(Duration.ofSeconds(2)) + + // Then + assertTrue(promptSlot.isCaptured, "expected ChatClient.prompt to be called") + val promptText = promptSlot.captured.contents + assertFalse( + promptText.contains(""), + "tokenBudget alone should not inject application-level thinking format" + ) + } } /** @@ -520,7 +593,7 @@ class StreamingChatClientOperationsTest { @Test fun `should prepend prompt contributions as system message`() { // Given - val mockContributor = mockk() + val mockContributor = mockk() every { mockContributor.contribution() } returns "System contribution" every { mockInteraction.promptContributors } returns listOf(mockContributor) diff --git a/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java b/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java index 675eaa55f..c937a86d6 100644 --- a/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java +++ b/embabel-agent-autoconfigure/models/embabel-agent-ollama-autoconfigure/src/test/java/com/embabel/agent/config/models/ollama/LLMOllamaStreamingBuilderIT.java @@ -22,6 +22,8 @@ import com.embabel.agent.autoconfigure.models.ollama.AgentOllamaAutoConfiguration; import com.embabel.agent.spi.LlmService; import com.embabel.agent.spi.support.springai.SpringAiLlmService; +import com.embabel.common.ai.model.LlmOptions; +import com.embabel.common.ai.model.Thinking; import com.embabel.common.core.streaming.StreamingEvent; import org.junit.jupiter.api.Test; import org.slf4j.Logger; @@ -218,6 +220,155 @@ void realStreamingOllamaIntegrationWithReactiveCallbacks() { logger.info("Integration streaming test completed successfully with {} total events", receivedEvents.size()); } + /** + * Validates application-level thinking retrieval vs provider token budget (#1799 / #1853). + *

+ * {@code createObjectStreamWithThinking} must enable thinking extraction/format via + * {@code Thinking.extractThinking} without requiring {@code Thinking.withTokenBudget}. + * Object-only streaming with a provider budget alone must still complete (budget is orthogonal). + */ + @Test + void createObjectStreamWithThinkingRetrievalIndependentOfTokenBudget() { + reactor.util.Loggers.useVerboseConsoleLoggers(); + + // --- Path A: *WithThinking* without any LlmOptions thinking / token budget --- + // Format instructions and extractThinking are applied by createObjectStreamWithThinking. + PromptRunner extractionOnlyRunner = ai.withLlm("qwen3:latest") + .withToolObject(new Tooling()); + assertTrue(extractionOnlyRunner.supportsStreaming(), "Test LLM should support streaming"); + + List extractionEvents = new CopyOnWriteArrayList<>(); + AtomicReference extractionError = new AtomicReference<>(); + AtomicBoolean extractionCompleted = new AtomicBoolean(false); + + String thinkingPrompt = + "What are exactly two of the hottest months in Florida and their highest temperatures. " + + "Think step by step before returning the JSONL objects."; + + Flux> extractionStream = new StreamingPromptRunnerBuilder(extractionOnlyRunner) + .streaming() + .withPrompt(thinkingPrompt) + .createObjectStreamWithThinking(MonthItem.class); + + extractionStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(event -> { + if (event.isThinking()) { + extractionEvents.add("THINKING: " + event.getThinking()); + logger.info("Extraction-only path thinking: {}", event.getThinking()); + } else if (event.isObject()) { + MonthItem obj = event.getObject(); + extractionEvents.add("OBJECT: " + obj.getName()); + logger.info("Extraction-only path object: {}", obj.getName()); + } + }) + .doOnError(error -> { + extractionError.set(error); + logger.error("Extraction-only stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> extractionCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(extractionError.get(), "createObjectStreamWithThinking must work without tokenBudget"); + assertTrue(extractionCompleted.get(), "Extraction-only stream should complete"); + assertFalse(extractionEvents.isEmpty(), "Should receive at least one streaming event without tokenBudget"); + assertTrue( + extractionEvents.stream().anyMatch(e -> e.startsWith("OBJECT:")), + "Extraction-only path should still produce object events" + ); + logger.info( + "Extraction-only createObjectStreamWithThinking completed with {} events (thinking events may vary by model)", + extractionEvents.size() + ); + + // --- Path B: *WithThinking* with provider token budget (budget preserved + extractThinking applied) --- + PromptRunner budgetAndExtractionRunner = ai.withLlm( + LlmOptions.withModel("qwen3:latest") + .withThinking(Thinking.withTokenBudget(100))) + .withToolObject(new Tooling()); + assertTrue(budgetAndExtractionRunner.supportsStreaming(), "Budget-configured LLM should support streaming"); + + List budgetEvents = new CopyOnWriteArrayList<>(); + AtomicReference budgetError = new AtomicReference<>(); + AtomicBoolean budgetCompleted = new AtomicBoolean(false); + + Flux> budgetStream = new StreamingPromptRunnerBuilder(budgetAndExtractionRunner) + .streaming() + .withPrompt(thinkingPrompt) + .createObjectStreamWithThinking(MonthItem.class); + + budgetStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(event -> { + if (event.isThinking()) { + budgetEvents.add("THINKING: " + event.getThinking()); + logger.info("Budget+extraction path thinking: {}", event.getThinking()); + } else if (event.isObject()) { + MonthItem obj = event.getObject(); + budgetEvents.add("OBJECT: " + obj.getName()); + logger.info("Budget+extraction path object: {}", obj.getName()); + } + }) + .doOnError(error -> { + budgetError.set(error); + logger.error("Budget+extraction stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> budgetCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(budgetError.get(), "createObjectStreamWithThinking must work with tokenBudget (applyExtraction)"); + assertTrue(budgetCompleted.get(), "Budget+extraction stream should complete"); + assertFalse(budgetEvents.isEmpty(), "Should receive events when tokenBudget coexists with extractThinking"); + assertTrue( + budgetEvents.stream().anyMatch(e -> e.startsWith("OBJECT:")), + "Budget+extraction path should still produce object events" + ); + logger.info( + "Budget+extraction createObjectStreamWithThinking completed with {} events", + budgetEvents.size() + ); + + // --- Path C: object-only stream with provider budget only (no *WithThinking*) --- + // Token budget alone must not require thinking events; stream of typed objects completes. + PromptRunner budgetOnlyRunner = ai.withLlm( + LlmOptions.withModel("qwen3:latest") + .withThinking(Thinking.withTokenBudget(100))) + .withToolObject(new Tooling()); + + List objectOnlyEvents = new CopyOnWriteArrayList<>(); + AtomicReference objectOnlyError = new AtomicReference<>(); + AtomicBoolean objectOnlyCompleted = new AtomicBoolean(false); + + String objectOnlyPrompt = + "Return exactly two JSONL objects for the hottest months in Florida with name and temperature."; + + Flux objectOnlyStream = new StreamingPromptRunnerBuilder(budgetOnlyRunner) + .streaming() + .withPrompt(objectOnlyPrompt) + .createObjectStream(MonthItem.class); + + objectOnlyStream + .timeout(Duration.ofSeconds(150)) + .doOnNext(obj -> { + objectOnlyEvents.add("OBJECT: " + obj.getName()); + logger.info("Object-only + budget path object: {}", obj.getName()); + }) + .doOnError(error -> { + objectOnlyError.set(error); + logger.error("Object-only stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> objectOnlyCompleted.set(true)) + .blockLast(Duration.ofSeconds(6000)); + + assertNull(objectOnlyError.get(), "Object-only stream with tokenBudget should not error"); + assertTrue(objectOnlyCompleted.get(), "Object-only stream should complete"); + assertFalse(objectOnlyEvents.isEmpty(), "Object-only path should receive typed objects"); + logger.info( + "Object-only stream with tokenBudget completed with {} objects (thinking retrieval independent of budget)", + objectOnlyEvents.size() + ); + } + @Test void rawTextStreamingOllamaIntegrationWithReactiveCallbacks() { // Enable Reactor debugging diff --git a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc index f69b5c278..9f439be35 100644 --- a/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc +++ b/embabel-agent-docs/src/main/asciidoc/reference/streaming/page.adoc @@ -37,10 +37,42 @@ Provider integrations override `LlmMessageStreamer.streamInference` by emitting the assembled assistant message. The completed message must include provider-assigned tool-call IDs, names, and complete arguments. Provider adapters must not execute those calls themselves. +==== Thinking: application-level retrieval vs model budget + +Embabel streaming currently deals with *application-level thinking*: the model is prompted +(via thinking *format instructions*) to emit reasoning blocks that Embabel parses into +``StreamingEvent.Thinking`` entries. Control is through ``Thinking`` on the Interaction / +``LlmOptions``: + +* *Application-level* — ``Thinking.withExtraction()`` sets ``extractThinking``. That flag + turns on prompt format instructions and block extraction. ``createObjectStreamWithThinking`` + enables this automatically (no separate SPI "format" parameter). +* *Provider model budget* — ``Thinking.withTokenBudget(n)`` sets ``enabled`` + budget for + providers that need it (e.g. Anthropic). Independent of application-level extraction; + combine via ``applyExtraction()`` when you want both. + +This is separate from *LLM-native reasoning* (provider-specific reasoning channels returned as +a dedicated response property), which Embabel does **not** extract today +(see https://github.com/embabel/embabel-agent/issues/1716[#1716]). + +* *Retrieving thinking events* — Call ``createObjectStreamWithThinking(...)``. + You do *not* need a token budget just to use this API. +* *Model thinking budget* — Optionally ``LlmOptions.withThinking(Thinking.withTokenBudget(n))`` + when the provider requires it. + +Object-only streams (``createObjectStream`` / ``generateStream``) only inject thinking format +instructions when ``extractThinking`` is already set on the Interaction (a token budget alone +does not). + ==== Example - Simple Thinking and Object Streaming with Callbacks [source,java] ---- + // Optional: only when the provider needs a thinking budget (e.g. Anthropic). + // Not required just to receive thinking StreamingEvents. + // LlmOptions options = new LlmOptions().withThinking(Thinking.withTokenBudget(8000)); + // PromptRunner runner = ai.withLlm(options).withToolObject(Tooling.class); + PromptRunner runner = ai.withDefaultLlm() // Example uses qwen3:latest .withToolObject(Tooling.class);