Embabel智能体AI框架中的LLM流式处理
[LOADING...]
1. 概述
LLM 流式传输会在生成输出的过程中逐 token 地传递结果。这降低了延迟，因为用户能在几毫秒内看到界面中的首批结果，而无需等待数秒才能获得完整答案。
对于具有结构化输出和推理能力的代理，流式传输应该支持增量处理：
- 在响应到达时将其解析为 Java 对象，以及
- 在思考块到达时也对其进行处理
然而，当前主流框架并不支持这些功能。
在本教程中，我们将：
- 简要回顾这些框架的局限性
- 并展示如何使用 Embabel Agentic AI 框架 在对象和思考块随流到达时对其进行增量处理
我们将涵盖四种场景：
- 无推理的单个对象
- 无推理的多个对象
- 有推理但无工具调用的多个对象
- 同时包含推理和工具调用的多个对象
2. 结构化流式传输：框架的局限性
用于 LLM 集成的主流 Java 框架 Spring AI 和 LangChain4j 对流式传输的支持程度有限。当模型生成文本时，我们可以获得实时的文本流，但这两个框架都无法开箱即用地将该文本流转换为类型化对象的集合。
Spring AI 的 entity() 方法可以将模型输出转换为 Java 对象，但它适用于阻塞式 call() 方法，却不适用于 stream() 方法。ChatClient API 参考直接说明了这一点：“未来，我们将提供一个便捷方法，让你可以通过响应式 stream() 方法返回 Java 实体。在此期间，你应该使用结构化输出转换器（Structured Output Converter）。”
LangChain4j 也有同样的局限性。它的 AiServices 仅允许在非流式调用中返回自定义 POJO、列表和枚举。对于流式传输，返回类型必须是 TokenStream，它会随着文本到达而提供原始文本，并提供丰富的回调钩子，但不会在流式输出中提供类型化对象。
实际上，任何希望获得带有推理的结构化对象流的人，都必须像这篇 Baeldung 文章 那样自行构建该功能。 这意味着要选择一种以换行符分隔的格式（如 JSONL），将 token 缓冲到一行完整可用，将每一行解析为 JSON，并在此过程中过滤掉任何思考块。实现这一过程会占用实际项目的时间。
这正是 Embabel Agentic AI 发挥作用的地方。它原生提供了所需的功能，我们将在下文中进一步探讨。
3. 环境搭建
首先，让我们定义必要的 Embabel 依赖：
embabel-agent-starter-anthropic 依赖提供了使用 Embabel 调用 Anthropic 模型所需的核心类。在本教程的示例中，我们使用 Anthropic 模型 claude-sonnet-4-5。
此外，我们引入 embabel-agent-test-internal，以便在 @SpringBootTest 中使用 AgentTestApplication 测试支持类。
3.1. Embabel 中的提供商支持
请注意，不同模型提供商在将流式传输与思考、工具结合的方式上存在差异。Anthropic 的原生思考会生成显式的推理块，但其他提供商可能不会以相同方式呈现推理。
Embabel 支持十多个 LLM 提供商，Baeldung 此前曾发布过 使用 OpenAI 的示例。
3.2. ParkingRecommendation
在本文的所有测试中，我们将流式传输 ParkingRecommendation 对象：
[LOADING...]
该记录包含四个标量属性：停车场景（scenario）、所选方案成本（estimatedTotalCost）、摘要以及所选方案（枚举 Option）。
4. 流式传输单个对象
首先，我们展示如何在没有推理和工具调用的情况下流式传输单个对象。测试方法为 whenStreaming_thenReceivesParkingRecommendation()：
PARKING_PROMPT 请求单个停车推荐，私有方法 streamParkingRecommendations() 负责实际的对象流式传输：
这里：
- 我们通过调用 createObjectStream() 创建一个
Flux<ParkingRecommendation>。Flux是一种 Reactor 类型，表示随着时间推移到达的包含零个或多个元素的异步序列，而不是一次性全部到达。 在本例中，模型会生成一个 ParkingRecommendation 对象。 - PARKING_PROMPT 指示 LLM 从三个可用选项中找出最佳停车方案：路边停车、计时停车和车库停车。该提示指定了每个备选方案的约束条件和成本。
- 该管道使用 doOnNext() 订阅，以收集并记录每个对象。
- blockLast() 会阻塞调用线程，直到流完成。这在测试环境中是合适的，尽管生产代码会继续链接响应式操作符而不是阻塞。例如，我们可以直接将 Flux 返回给 Web 端点，由端点将响应流式传输给客户端。
我们可以像这样运行单元测试方法 whenStreaming_thenReceivesParkingRecommendation()：
流完成后，日志确认该流传递了一个完整的 ParkingRecommendation 对象：
该对象包含我们期望的所有属性：scenario、chosenOption、estimatedTotalCost 和 summary。
5. 流式传输多个对象
streamParkingRecommendations() 方法可以流式传输多个对象。为了演示这一点，我们使用 TIMED_PARKING_PROMPT 请求三个推荐方案：
在流式传输时：
- doOnNext() 会收集并记录到达的每个对象
- blockLast() 会等待流完成
我们可以像这样运行单元测试方法 whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario：
我们不会得到包含所有对象的单个批量响应，而是在收到每个 ParkingRecommendation 时，将它作为独立对象获取：
日志中每个对象创建的时间戳证实了完整对象是按顺序获取的，并且集合是增量填充的。
请注意日志中的条目“Unhandled event type: CONTENT_BLOCK_STOP”。它只是表示“此内容块（文本块、工具调用块等）已结束”。它不携带任何负载，只是说明我们收到了一个服务器端事件，但没有为它绑定处理器。
6. 带推理的流式传输
首先，我们必须通过为 LlmOptions 配置 token 预算来启用深度思考：
该预算控制模型在生成答案之前可以用于推理的 token 数量，并且必须低于 max_tokens（对于 claude-sonnet-4-5 为 8192）。
对象在私有方法 streamParkingRecommendationsWithThinking() 中流式传输。它使用 createObjectStreamWithThinking() 而非 createObjectStream()，返回类型也相应变化：
StreamingEvent 包装了类型化对象和推理片段，因为流会交错出现这两种事件。 doOnNext() 回调通过 event.isObject() 和 event.isThinking() 检查来区分它们，并将它们收集到不同的列表中：received 用于对象，reasoning 用于思考片段：
我们可以按如下方式运行单元测试 whenStreamingWithThinking_thenReceivesReasoningAndRecommendation()：
日志表明，推荐结果和推理片段在到达时即被处理。
请注意，每个推理块都表示单行内容，因为 Embabel 会将小片段聚合为一行。## 7. 使用工具与推理进行流式处理
最后，让我们将工具调用与流式处理和思考结合起来。对于阻塞式调用，Baeldung 文章 使用 Embabel Agentic AI 框架进行 LLM 工具调用推理 展示了工具调用如何受益于 LLM 推理。
该 API 与推理用例非常相似，不同之处在于工具注册 withToolObject(new ParkingTooling()) 和用于可观测性的日志检查器：
TOOL_PARKING_PROMPT 指示模型在给出三条推荐之前主动使用工具，而 streamParkingRecommendationsWithThinking() 方法则对推荐进行流式输出。
该管道使用了相同的 createObjectStreamWithThinking() 方法和相同的 event.isObject() / event.isThinking() 模式。关键区别在于工具调用的时机：模型首先调用工具，然后对结果进行推理，再输出对象，最后提供推理摘要。 由于 Spring AI 会在所有工具调用完成后启动新的流，因此推理块只会在工具调用之后发出。
我们可以像下面这样运行单元测试 whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning()：
与 whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() 类似，该流会同时传递一条推荐和一个推理块（在工具完成各自任务之后）。
8. 结论
在本文中，我们展示了如何使用 Embabel Fluent API 生成类型化 Java 对象的流，并透明地处理思考和工具结果。
在 Spring AI 和 LangChain4j 中，流式输出原始文本都很直接，但流式输出类型化对象并不是这两个框架开箱即用的功能。Embabel 在框架层面解决了这个问题，无需手动缓冲令牌、解析 NDJSON 或过滤推理块。 同样的模式既可以从单个对象扩展到集合，也可以从简单提示扩展到多工具推理链。对于既重视结构又重视响应性的应用来说，这消除了原本会落在每个构建智能体 AI 应用的团队身上的样板代码层。
文章代码可在 GitHub 上获取。