首页 > 开源 > :fire: 不只是 Token 流：Solon AI 4.1 的语义化聊天事件

:fire: 不只是 Token 流：Solon AI 4.1 的语义化聊天事件

OSChina资讯 2026-09-07 13:01 4 阅读 查看原文
大模型 流式输出 AI框架

LLM 流式输出看起来像是不断返回文本分片，但现代模型的一条连接中可能同时出现正文、思考过程、工具参数、引用、媒体、安全事件、用量快照、状态和错误。若把每一帧都包装成“部分 ChatResponse”，调用方很难区分过程数据和最终结果，也容易把供应商协议细节带进 UI、Agent 和网关。

Solon AI 4.1 将流式接口调整为：

4.1 之前：Flux<ChatResponse>
4.1：     Flux<ChatEvent>

其中：

  • ChatEvent 表示响应进行中的语义事件；
  • ChatResponse 表示已经聚合的结果；

call() 与 stream() 回答不同问题

同步调用直接取得完整结果：

ChatResponse response = chatModel.prompt("解释语义化流式输出").call();
String text = response.getText();

流式调用观察响应生命周期：

Flux<ChatEvent> events = chatModel
        .prompt("解释语义化流式输出")
        .stream();

只投影正文时，应按精确事件类型过滤：

Flux<String> text = chatModel.prompt(message).stream()
        .filter(event -> event.is(ChatEventType.TEXT_DELTA)
                && event.hasText())
        .map(ChatEvent::getText);

这里不能只用 isDelta()，因为增量事件还可能是思考、工具参数、媒体片段或拒答。getText() 也允许为空，因此映射前要先调用 hasText()。

九组事件构成稳定路由层

当前源码定义了 31 种事件，归入九个分组：

分组 语义 代表事件
LIFECYCLE 整个响应生命周期 RESPONSE_START、RESPONSE_END、ABORT
STEP 一轮模型调用 STEP_START、STEP_END
TEXT 用户可见正文 TEXT_START、TEXT_DELTA、TEXT_END
THINKING 思考相关输出 THINKING_START、THINKING_DELTA、THINKING_END
TOOL_CALL 客户端执行工具 TOOL_CALL_START、TOOL_CALL_ARGS_DELTA、TOOL_RESULT
SERVER_TOOL 供应商侧工具 SERVER_TOOL_START、SERVER_TOOL_RESULT
MEDIA 引用与媒体 CITATION、MEDIA_PARTIAL、MEDIA_DONE
SAFETY 拒答与内容过滤 REFUSAL_DELTA、CONTENT_FILTER
META 用量、错误、原始与自定义数据 USAGE、ERROR、RAW、CUSTOM

通用消费端可以先按分组路由，再在组内识别具体类型。这样新增具体事件时，不需要把整个消费者改写为封闭式枚举分支。

END 阶段不等于全流终止

TEXT_END、THINKING_END、TOOL_CALL_END 和 STEP_END 只关闭局部范围。当前 isTerminal() 只对以下三类返回 true：

RESPONSE_END
ABORT
ERROR

因此不能用 event.getPhase() == END 判断整个响应是否结束。ERROR 是终止事件，但它的阶段为 NONE；多个局部事件阶段为 END，却不会终止整个响应。

从 RESPONSE_END 获取最终聚合

同步提取：

ChatResponse response = chatModel.prompt(query).stream()
        .filter(event -> event.is(ChatEventType.RESPONSE_END))
        .map(ChatEvent::getResponse)
        .blockFirst();

响应式提取：

Mono<ChatResponse> response = chatModel.prompt(query).stream()
        .filter(event -> event.is(ChatEventType.RESPONSE_END))
        .map(ChatEvent::getResponse)
        .next();

不要直接对整个事件流调用 blockFirst()，因为第一个事件通常是 RESPONSE_START。前端可以自行拼接 TEXT_DELTA 做打字机效果，但完整消息、工具调用和用量应以 Solon AI 聚合的终态响应为准。

Normalizer 把供应商帧变成可消费协议

供应商流不一定有完整边界。ChatEventNormalizer 负责补齐和整理：

TEXT_START -> TEXT_DELTA* -> TEXT_END
THINKING_START -> THINKING_DELTA* -> THINKING_END
TOOL_CALL_START -> TOOL_CALL_ARGS_DELTA* -> TOOL_CALL_END

正文和思考采用较严格的块跟踪：

  • 裸增量可自动补开始事件；
  • 重复开始事件会被丢弃；
  • 没有对应开始的结束事件会被丢弃；
  • 正文与思考切换时会关闭前一个块；
  • 正常或错误收尾时补齐仍开放的块。

工具调用采取更宽松的策略。后续参数分片可能不再携带完整调用 ID，因此归一化器倾向于“缺边界时补齐”，而不是充当严格协议校验器。

业务端不应把某个 TOOL_CALL_ARGS_DELTA 当成完整 JSON。完整工具调用应从已完成步骤或最终响应中读取。

自动工具调用引入 Step

一次工具辅助回答通常包含多轮模型请求：

RESPONSE_START
  STEP_START (0)
    TOOL_CALL_START
    TOOL_CALL_ARGS_DELTA ...
    TOOL_CALL_END
    TOOL_RESULT
  STEP_END (0)
  STEP_START (1)
    TEXT_START
    TEXT_DELTA ...
    TEXT_END
  STEP_END (1)
RESPONSE_END

整个过程只有一个响应生命周期，每轮模型调用对应一个 step。当前实现从 step 0 开始递增。

正常完成的 STEP_END 携带该步终态快照和该步用量；RESPONSE_END 携带整个成功响应的聚合结果和跨步骤总用量。这使监控系统既能分析单轮调用，也能分析完整工具链路。

用量聚合有两个层次

同一步中的供应商 usage 通常是累计快照，不能对每帧直接相加；不同 step 是独立模型请求，需要跨步累计。

步内：保留或合并供应商累计快照
步间：累加正常完成步骤的 usage

因此：

  • STEP_END.getUsage() 是当前步骤用量；
  • RESPONSE_END.getUsage() 是整个成功响应的累计用量；
  • 单个 USAGE 事件不等于“把所有字段再次加入全局计数器”。

ERROR、ABORT 与 cancel 是三件事

ERROR 与 Reactor onError

进入主要响应式错误路径的失败，会尝试先发 ERROR 事件，再触发 Reactor onError：

  • ERROR 用于携带语义上下文，例如此前已完成步骤的结果或累计用量；
  • onError 用于 retryWhen、onErrorResume、超时和降级。

它们描述的是同一次失败。ERROR.getResponse() 也不保证包含当前失败步骤已经流出的所有文本；首步很早失败时，它可能为空。

自定义过滤器可以过滤掉属于 META 的 ERROR，因此无论是否处理错误事件，都应保留 Reactor 错误消费者。

ABORT

ABORT 是上游语义事件，不等于订阅方取消。

Reactor cancel

take(...) 等操作符可能主动取消订阅。取消后不能期待额外的 TEXT_END、STEP_END、ABORT 或 RESPONSE_END。资源清理应放在 doFinally 等 Reactor 生命周期钩子中。

事件过滤只控制投递

默认过滤器会拒绝 HEARTBEAT 和 RAW。诊断或网关场景可以显式开启全部事件：

chatModel.prompt(query)
        .eventFilter(ChatEventFilter.all())
        .stream();

也可在默认策略上开放 RAW：

ChatEventFilter filter = ChatEventFilter.DEFAULT.or(
        ChatEventFilter.of(ChatEventType.RAW));

当前源码还有一个细节：运行时会对非空自定义过滤器增加保护，强制保留 LIFECYCLE 和 STEP。所以 of(TEXT_DELTA) 并不意味着订阅方只会收到正文增量。稳妥做法是：

  • eventFilter 控制粗粒度投递；
  • 下游再次按精确事件类型做业务投影；
  • 只有协议诊断或透传确实需要时才用 all()。

过滤发生在内部归一化与聚合之后，因此订阅方看不到某个事件，不会反向破坏 Solon AI 的终态聚合。

自定义方言只负责协议翻译

方言的流式解析入口为：

void parseResponseJson(ChatStreamContext ctx, String respJson);

正文、思考和客户端工具调用通常进入 accumulator，由核心统一产生标准事件与聚合结果；引用、状态、供应商侧工具、媒体等扩展语义可以直接发射事件。

同一份语义负载不要同时写入 accumulator 又直接 emit，否则可能造成重复增量和重复聚合。

迁移清单

从 4.1 之前的流式代码迁移时：

  • 将 Flux<ChatResponse> 改为 Flux<ChatEvent>；
  • 只用 TEXT_DELTA + hasText() 渲染正文；
  • 将思考与正文分开路由；
  • 不要把 isDelta() 当成正文判断；
  • 先过滤 RESPONSE_END，再用 blockFirst() 或 next() 获取终态；
  • 不要从单个工具参数分片解析完整调用；
  • 区分 step usage 与 response usage；
  • 即使处理 ERROR 事件，也保留 Reactor onError；
  • 不把 cancel 当作 ABORT；
  • 自定义方言迁移到 parseResponseJson(ChatStreamContext, String)；
  • 后续升级时重新核对 4.1 预览 API。

测试核验

本次对源码检出的三个确定性核心测试类执行了验证：

ChatEventNormalizerTest       20
ChatEventFilterTest            6
ChatStreamSessionUsageTest     9
--------------------------------
合计                          35

结果：

Tests run: 35, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

这些测试覆盖边界归一化、过滤组合和跨步骤用量聚合，但不能证明所有供应商、所有任意事件序列、所有并发情况或整个负载对象图的深度不可变性。不同供应商也不一定都会产生思考、引用、媒体、安全、服务端工具和用量事件。

总结

Solon AI 4.1 的关键变化不是“多了 31 个枚举”，而是形成了清晰分层：

供应商 SSE / JSON 帧
        ↓
方言解析
        ↓
语义累积与事件发射
        ↓
边界归一化
        ↓
UI、Agent、监控和协议适配

Token 流回答“下一段字符是什么”，语义事件流回答“下一件发生的事是什么”。当系统开始组合思考、工具、引用、安全事件和多轮模型调用时，后一个问题更适合支撑可维护的工程架构。

 

相关推荐