Spring AI 流式输出时的 JSON 截断与增量结构化补全

Spring AI 流式输出时的 JSON 截断与增量结构化补全大模型在处理结构化数据提取或复杂表单生成时流式输出Streaming Output能极大降低前端用户的首字等待延迟TTFT。然而在基于 Spring AI 或底层 Reactor Flux 对接 LLM 流式流Server-Sent Events的过程中一个常见的痛点是大模型下发的 JSON 片段在传输过程中是逐 Token 输出的。前端或后端消费方若想在流式传输阶段就实时渲染已经生成的字段必须面对“不完整 JSON 字符串”的反序列化失败问题。如果在中间管道中直接调用 Jackson 的ObjectMapper.readValue()不可避免会抛出JsonParseException: Unexpected end-of-input。等待整个流全部结束再一次性解析又完全失去了流式体验的交互优势。这里我们将深入探讨在 Spring AI 流式调用场景下如何通过增量状态机与括号栈修复算法实现流式 JSON 的实时截断修复与安全结构化消费。流式 JSON 的结构痛点与异常现场当要求模型输出 JSON 格式时如提取用户信息、报表数据模型在生成过程中会经历如下典型的 Token 切片Chunk 1: {name: 张三, scores: [85, 9 Chunk 2: 2, 78], profile: {age: 3 Chunk 3: 0, city: 杭州}}在 Chunk 1 到达时内容为{name: 张三, scores: [85, 9。此时数组scores缺少闭合方括号]。整个根对象缺少闭合花括号}。最后一个数字9尚未完成后续可能是92或9。如果在后端中间流处理中希望将中间态推送给下游 WebSocket 或实时计算逻辑解析器会直接报错崩溃com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input in numeric value at [Source: (String){name: 张三, scores: [85, 9; line: 1, column: 31] at com.fasterxml.jackson.core.base.ParserMinimalBase._reportInvalidEOF(ParserMinimalBase.java:160) at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._finishNumberMinus(ReaderBasedJsonParser.java:1520)业务诉求很明确我们需要一个低延迟、轻量级的增量修复器Incremental JSON Repairer能够在任意 Token 截断点将不完整的 JSON 串临时补齐为合法可解析的 JSON 对象同时标记未稳定字段。增量补全状态机设计修复截断 JSON 的核心是符号栈Bracket Stack与词法状态分析。通过单遍扫描当前缓冲区字符串跟踪以下状态字符串状态inString是否处于未闭合的双引号内部并处理转义字符\。符号嵌套栈stack记录当前处于{还是[的层级。键值分隔符colonPending识别是否出现key:但值还未开始输出的情况。悬挂标识与不完整字面量如截断在trutrue、falfalse、nulnull或未完成的数字上。状态机扫描与动态补全逻辑[原始缓冲区字符流] │ ▼ [词法状态扫描] ─── 处于未闭合字符串? ──► 自动追加闭合双引号 │ ▼ [检查末尾悬挂符号] ── 处于 key: 悬挂? ──► 补齐占位值 null │ ▼ [遍历符号栈] ── 依据栈内剩余 [ / { ──► 倒序追加 ] 和 } │ ▼ [输出合法临时 JSON] ──► Jackson 安全反序列化核心实现流式 JSON 增量补全器下面给出一个高性能的 Java 实现。该类在每次接收到新 Chunk 时能够以极低开销输出当前阶段的最优修复 JSON 字符串。package com.example.ai.stream; import java.util.ArrayDeque; import java.util.Deque; public class IncrementalJsonRepairer { public static String repair(String partialJson) { if (partialJson null || partialJson.trim().isEmpty()) { return {}; } String input partialJson.trim(); StringBuilder sb new StringBuilder(input); DequeCharacter stack new ArrayDeque(); boolean inString false; boolean escape false; for (int i 0; i input.length(); i) { char c input.charAt(i); if (escape) { escape false; continue; } if (c \\) { if (inString) { escape true; } continue; } if (c ) { inString !inString; continue; } if (!inString) { if (c { || c [) { stack.push(c); } else if (c }) { if (!stack.isEmpty() stack.peek() {) { stack.pop(); } } else if (c ]) { if (!stack.isEmpty() stack.peek() [) { stack.pop(); } } } } // 1. 如果截断在字符串内部先补全双引号 if (inString) { sb.append(\); } // 2. 清理末尾处于悬挂状态的分隔符 String current sb.toString().trim(); while (current.endsWith(:) || current.endsWith(,)) { current current.substring(0, current.length() - 1).trim(); // 如果去掉冒号后变成了未完成的 key再次尝试平衡 if (current.endsWith(\)) { // 保留为合法的 key: null 结构 current current : null; break; } } sb new StringBuilder(current); // 3. 处理可能截断的字面量 (true/false/null) trimIncompleteLiterals(sb); // 4. 根据当前栈深度倒序闭合括号 while (!stack.isEmpty()) { char openBracket stack.pop(); if (openBracket {) { sb.append(}); } else if (openBracket [) { sb.append(]); } } return sb.toString(); } private static void trimIncompleteLiterals(StringBuilder sb) { String text sb.toString(); String[] literals {true, false, null}; for (String literal : literals) { for (int len 1; len literal.length(); len) { String sub literal.substring(0, len); if (text.endsWith( sub) || text.endsWith(: sub) || text.endsWith(, sub)) { int idx text.lastIndexOf(sub); sb.delete(idx, sb.length()); sb.append(null); return; } } } } }结合 Spring AI ChatClient 的管道集成在 Spring AI 中ChatClient.prompt().stream().chatResponse()返回的是FluxChatResponse。我们可以在 Reactor 响应式流的操作链中注入增量修复器将原本离散的字符片段转换为实时可解析的 DTO 中间态对象。package com.example.ai.controller; import com.example.ai.stream.IncrementalJsonRepairer; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.ai.chat.client.ChatClient; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux; import java.util.concurrent.atomic.AtomicReference; RestController public class StreamStructureController { private final ChatClient chatClient; private final ObjectMapper objectMapper; public StreamStructureController(ChatClient.Builder builder, ObjectMapper objectMapper) { this.chatClient builder.build(); this.objectMapper objectMapper; } GetMapping(value /api/v1/stream/profile, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxJsonNode streamProfileExtraction(RequestParam String userPrompt) { StringBuilder buffer new StringBuilder(); AtomicReferenceString lastValidJson new AtomicReference({}); return chatClient.prompt() .user(请提取以下内容的用户画像严格以 JSON 格式输出\n userPrompt) .stream() .content() .map(chunk - { buffer.append(chunk); String raw buffer.toString(); // 清理 Markdown 代码块包裹符如 json ... String cleanJson extractJsonContent(raw); return IncrementalJsonRepairer.repair(cleanJson); }) .filter(repairedJson - !repairedJson.equals(lastValidJson.get())) .map(repairedJson - { try { JsonNode node objectMapper.readTree(repairedJson); lastValidJson.set(repairedJson); return node; } catch (Exception e) { // 遇到边缘未闭合场景返回上一有效帧 try { return objectMapper.readTree(lastValidJson.get()); } catch (Exception ex) { return objectMapper.createObjectNode(); } } }); } private String extractJsonContent(String raw) { String trimmed raw.trim(); if (trimmed.startsWith(json)) { trimmed trimmed.substring(7); } else if (trimmed.startsWith()) { trimmed trimmed.substring(3); } if (trimmed.endsWith()) { trimmed trimmed.substring(0, trimmed.length() - 3); } return trimmed.trim(); } }生产落地踩坑与防御机制数值截断抖动当模型正在输出数字12345时前序帧可能是12、123、1234。在前端做图表或实时统计渲染时数值会发生连续跃变。对于关键数值字段建议在前端或 DTO 层增加校验标记或由后端在反序列化时判定字段是否处于当前最后一个 Key避免引发前端视觉闪烁。大对象内存开销随着输出内容增加buffer.append()的字符串变长每一帧执行repair和objectMapper.readTree()会产生较多中间临时对象。在高并发网关场景下可以引入**帧节流Throttle/Debounce**机制限制每 50ms~100ms 触发一次反序列化解析既能保证视觉流畅度又能节省 70% 以上的 GC 开销。Markdown 围栏与前置思考文本部分推理模型如带有think标签或前置 Markdown 描述的模型会在输出 JSON 之前输出一段解释性文本。在进入修复器之前必须通过状态过滤剔除前置非 JSON 区域仅截取首个{之后的内容输入修复管道。