29 · 流式输出（Streaming）

开发者可以选择用 Embabel 的流式能力，逐步地从 LLM 输送数据。

除了流式输出 LLM 的原始文本，Embabel 的流还可以包含 LLM 推理事件（也就是所谓的 "thinking"）， 以及 LLM 创建的对象流。这一特性与 Embabel 面向对象编程模型的定位高度契合。

概念（Concepts）

• StreamingEvent —— 包装 Thinking 或用户 Object；
• StreamingPromptRunnerBuilder —— 具备流式能力的 runner；
• Spring Reactive Programming Support：底层基础设施基于 Spring AI ChatClient；
• 所有响应式回调（doOnNext、doOnComplete 等）都在开发者手中。

示例：带回调的 thinking 与对象流

PromptRunner runner = ai.withDefaultLlm() // 该示例使用 qwen3:latest
        .withToolObject(Tooling.class);

String prompt = "What are exactly two the most hottest months in Florida and their respective highest temperatures";

// 用 StreamingPromptBuilder 代替 Kotlin 扩展函数
Flux<StreamingEvent<MonthItem>> results = new StreamingPromptRunnerBuilder(runner)
        .streaming()
        .withPrompt(prompt)
        .createObjectStreamWithThinking(MonthItem.class);

// 用 builder 模式订阅真实响应式回调
results
        .timeout(Duration.ofSeconds(150))
        .doOnSubscribe(subscription -> {
            logger.info("Stream subscription started");
        })
        .doOnNext(event -> {
            if (event.isThinking()) {
                String content = event.getThinking();
                receivedEvents.add("THINKING: " + content);
                logger.info("Integration test received thinking: {}", content);
            } else if (event.isObject()) {
                MonthItem obj = event.getObject();
                receivedEvents.add("OBJECT: " + obj.getName());
                logger.info("Integration test received object: {}", obj.getName());
            }
        })
        .doOnError(error -> {
            errorOccurred.set(error);
            logger.error("Integration test stream error: {}", error.getMessage());
        })
        .doOnComplete(() -> {
            completionCalled.set(true);
            logger.info("Integration test stream completed successfully");
        })
        .blockLast(Duration.ofSeconds(6000));

示例：带回调的原始文本流

PromptRunner runner = ai.withDefaultLlm() // 该示例使用 qwen3:latest

String prompt = "What is the highest building in Paris?";

// 用 StreamingPromptBuilder 代替 Kotlin 扩展函数
Flux<String> results = new StreamingPromptRunnerBuilder(runner)
        .streaming()
        .withPrompt(prompt)
        .generateStream();

// 用 builder 模式订阅真实响应式回调
results
        .timeout(Duration.ofSeconds(150))
        .doOnSubscribe(subscription -> {
            logger.info("Stream subscription started");
        })
        .doOnNext(content -> {
            receivedTextChunks.add(content);
            logger.info("Integration test received text chunk: {}", content);
        })
        .doOnError(error -> {
            errorOccurred.set(error);
            logger.error("Integration test stream error: {}", error.getMessage());
        })
        .doOnComplete(() -> {
            completionCalled.set(true);
            logger.info("Integration test stream completed successfully");
        })
        .blockLast(Duration.ofSeconds(6000));