尧图网页设计 
从 SSE 流式原理到 LangChain 结构化输出打字机效果与 JSON 解析全方案实战现在的 AI 应用谁还没个打字机效果都不好意思上线。但说实话我看过太多项目把“流式输出”做成了摆设——前端拿到一堆碎文本直接拼上去后端一个yield扔出去就算完事等到要让模型输出结构化 JSON 的时候整个链路直接崩掉。今天这篇文章咱们就把这条链路上的每一个环节都拆开揉碎SSE 协议到底在传什么、EventSource 怎么接、LangChain 怎么配合流式输出、以及最关键的结构化 JSON 在流式场景下怎么解析才不会翻车。适合谁看如果你正在做 LLM 应用的落地开发或者你只是听说过 SSE 和 LangChain 但一直没啃下源码这篇文章能帮你省掉至少一周的试错时间。我会把前端、后端、协议层、解析层全部串起来讲拿真实生产环境的方案说事。1. 从轮询到推送SSE 流式原理与选型逻辑很多人一听“流式输出”就想到 WebSocket这其实是个误区。Server-Sent EventsSSE和 WebSocket 虽然都能做实时推送但它们的定位完全不同。SSE 是单向的、基于 HTTP 的服务器推送协议浏览器原生支持一行EventSource就能接住WebSocket 是双向全双工通信适合聊天室、游戏这种需要频繁双向交互的场景。选择 SSE 来做大模型流式输出的理由非常实际大模型生成文本本来就是单向的——模型只管往客户端推数据不需要客户端频繁往回发消息。你用一个 WebSocket 连接去做一件只需要单向通知的事等于开着卡车去送快递不是不行但成本和复杂度完全不成比例。而且 SSE 走的是普通 HTTP天然兼容各种网关、代理、负载均衡器不存在 WebSocket 那种“连接要升级、防火墙要放行、长连接容易断”的麻烦。1.1 SSE 的本质就是一段不断输出的 HTTP 响应SSE 这个协议本身没什么神秘感。它就是一个 HTTP 响应只不过Content-Type设为text/event-stream服务器端不关闭连接持续往客户端写数据。每一帧数据遵循固定的格式规范data: 这是一行内容 data: 这是第二行内容事件帧之间用空行分隔每条数据以data:前缀开头。如果一行放不下可以分成多行data:客户端会自动用换行符拼接起来。如果想要把某条数据单独拉到指定事件类型还可以加event:字段客户端监听对应事件名即可。后端在 Python 里实现 SSE 端点其实非常简单——用 FastAPI 的StreamingResponse配合生成器一行yield就是一次推送from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() def token_stream(): for token in [你好, , 世界, ]: yield fdata: {token}\n\n app.get(/api/chat/stream) async def chat_stream(): return StreamingResponse( token_stream(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no } )注意那个X-Accel-Buffering: no头这个是我在生产环境踩过坑才加上的。Nginx 默认会缓冲后端响应如果不关掉缓冲前端收到的不是逐字打字机效果而是等整个响应结束才一次性拿到全部数据——流式效果直接归零。遇到这种情况第一反应不要怀疑代码先查网关层是不是在“捣乱”。1.2 为什么 EventSource 比 fetch 流式读取更省心浏览器原生的 SSE 客户端是EventSource对象用法极简const source new EventSource(/api/chat/stream); source.onmessage (event) { const token event.data; // 把 token 追加到界面上 appendText(token); }; source.onerror (err) { // 处理断连逻辑 console.error(SSE 连接断开, err); };用 fetch 走流式读取也可以但要注意一个关键差异EventSource 自带自动重连机制连接断开后浏览器会按retry:字段指定的间隔自动重新建立连接。而用 fetch 的ReadableStream做手动解析一旦连接中断你得自己写重连逻辑还要处理各种边缘情况。EventSource 还有一个限制它只支持 GET 请求。如果你需要把用户的历史对话通过 POST 发给后端再开启流式响应要么把参数拼在 URL 上要么改用fetch加POST配合ReadableStream来手动解析 SSE 帧。后者更灵活但需要自己实现帧解析逻辑这个我们放到后面讲。2. 前端打字机效果实现不只是“把文字贴上去”打字机效果的实现难度不在于“渲染文字”而在于“渲染的节奏怎么和生成节奏对齐”。模型输出的 token 到达时间是随机的最快的几个 token 可能几十毫秒就到了最慢的可能要等好几秒如果无脑把每个到达的 token 直接 append 到 DOM页面会有一阵一阵的卡顿感——这其实是 render 频率超过了浏览器的刷新率属于典型的性能问题。2.1 节流渲染让每帧最多只动一次 DOM正确的做法是引入一个节流机制把 SSE 收到的 token 先塞进一个缓冲区然后通过requestAnimationFrame循环把缓冲内容定时写入 DOM。每帧最多渲染一次帧缓冲浏览器会自然地在空闲时刷新不会出现内容以肉眼可见的“一坨一坨”冲出来的情况。export function createTypewriter(element) { let buffer ; let rendering false; let canceled false; function flush() { if (canceled) return; element.textContent buffer; buffer ; rendering false; } function scheduleFlush() { if (rendering) return; rendering true; requestAnimationFrame(flush); } return { push(text) { buffer text; scheduleFlush(); }, done() { canceled true; flush(); } }; }这里最关键的是scheduleFlush里的防抖逻辑同一帧里不管 push 进来多少 token只会触发一次flush。这样即使后端一次塞给前端十几个 token界面也是平滑地滚动更新不会闪烁或跳动。除了渲染节流还需要处理 markdown 的渲染问题。如果直接用textContent把文本塞进去接完流之后要做一次marked或markdown-it解析把纯文本转成带样式的 HTML。这里有个细节要注意流式过程中不要实时做 markdown 转 HTML因为 markdown 语法往往是跨多个 token 的比如代码块的三个反引号可能分几次传输中途转换很容易解析出残次品。稳妥的做法是流式过程只展示纯文本流结束后整体转一次 markdown。2.2 断连重连与消息状态的精准控制打字机的“结束态”比“进行态”更容易被忽视。前端跟用户沟通的方式不能只看文字有没有打完还要知道后端是不是主动关闭了流。SSE 连接结束后浏览器会触发onerror但这个onerror在正常关闭和异常断连时都会触发区分两者需要靠后端发送一个特殊的结束标识。惯例的做法是后端在流式响应完所有 token 后额外发送一个data: [DONE]帧前端收到这个标志就正常关闭连接不再触发重连逻辑。如果是中途超时、网关断连、服务器异常前端收不到[DONE]就需要走自动重连同时更新界面上的“重试”提示。event: message data: 今天天气不错 data: [DONE]事件监听里判断event.data [DONE]后调用source.close()这是一个很容易被新手漏掉的操作——你不主动 closeEventSource 会一直按重试间隔尝试重新连接产生一堆无意义的请求。2.3 SSE 用 GET 也能带参数EventSource 的兼容处理刚才提到 EventSource 只支持 GET很多人会被卡在这一步。如果业务场景要求用 POST 传参最干净的方案是保留 EventSource 的自动重连能力把复杂参数放在服务端会话里前端只需要用 GET 请求一个临时会话标识// 先通过 POST 建立会话拿到 session_id const { session_id } await fetch(/api/chat/start, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages: history }) }).then(res res.json()); // 再开启 SSE 流式连接 const source new EventSource(/api/chat/stream?session_id${session_id});后端拿到 session_id 后从内存或 Redis 中取对应用户的请求参数再开始生成流。这样既绕过了 GET 的参数长度限制又保留了对断线重连的友好性算是我在多个项目里反复验证过的标准做法。3. LangChain 流式输出实战Stream 与回调的取舍LangChain 做流式输出有两个层面的接口一个是最外层的stream()方法另一个是底层的callbacks机制。很多教程只讲stream()但真正要把流式输出部署到生产环境的项目里你大概率需要用回调回调来拿全链路的 token。3.1 用.stream()拿到大模型 token 流stream()的用法非常简单直接迭代即可from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm ChatOpenAI( modelgpt-4o-mini, temperature0, streamingTrue ) for chunk in llm.stream([HumanMessage(content讲个冷笑话)]): if hasattr(chunk, content) and chunk.content: yield fdata: {chunk.content}\n\n注意一点llm.stream()在内部其实也是开启 streaming然后逐块 yield AIMessageChunk。streamingTrue这个参数不是必须的但加上能确保底层调用过程使用流式接口减少首 token 延迟。在 FastAPI 的生成器函数里逐块yield出去前端打字机效果就有了。3.2 用 callbacks 实现多路输出与日志采集生产环境里普遍存在的需求是同一个流式响应里你既要给最终用户看完整回答又要给运营看 token 用量、给调试者看中间推理过程。stream()只能让你拿到最终的 token 流拿不到“模型在中间到底走了哪几步”。这种情况就要用 LangChain 的AsyncCallbackHandlerfrom langchain_core.callbacks import AsyncCallbackHandler from langchain_core.agents import AgentFinish class TokenCollector(AsyncCallbackHandler): def __init__(self): self.token_buffer [] self.agent_logs [] async def on_llm_new_token(self, token: str, **kwargs): self.token_buffer.append(token) # 这里把 token 实时推送到 SSE async def on_agent_action(self, action, **kwargs): self.agent_logs.append(f思考: {action.log}) async def on_agent_finish(self, finish: AgentFinish, **kwargs): self.agent_logs.append(f完成: {finish.return_values})回调机制特别适合 Agent 场景用户在界面上看到的不只是最终回答还能看到“Agent 正在搜索资料”“Agent 正在调用计算器”这样的中间状态体验会非常像真的在看一个人思考过程在推进。3.3 LangChain 版本差异的坑接口更新比模型还快LangChain 的接口变化非常频繁尤其是 LangChain v0.1 到 v0.2 再到 v0.3 这个跨度里很多老教程里的代码已经跑不起来了。比如langchain.chains.LLMChain在较新版本里被降级为 Legacy官方更推荐直接用RunnableSequence或LangGraph。如果你照着网上的旧教程抄代码很可能遇到DeprecationWarning甚至直接报错。我的建议是项目初始就锁定一个 LangChain 版本不要用latest。比如langchain0.2.x配langchain-openai的版本组合功能相对稳定。升级某个核心包时最好全量跑一遍链路的流式测试用例因为回调签名、流式输出的 chunk 类型、事件触发顺序都可能被改动影响。4. 结构化输出与 JSON 解析流式场景下最大的坑标题里说的“结构化输出”指的是让 LLM 稳定地返回一段符合 JSON Schema 的数据而不是一段格式可疑的文本。在这个基础上叠加流式传输问题立刻复杂了一个量级你收到的不是一个完整 JSON而是被拆成几千个碎片的 JSON 字符串。如何从碎片流中稳定地解析出结构化数据是这门手艺的真正核心。4.1 LangChain 的with_structured_output把格式控制交给模型LangChain 提供了非常顺手的结构化输出能力。核心方法是with_structured_output()配合 Pydantic 模型定义输出格式from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class WeatherReport(BaseModel): city: str Field(description城市名称) date: str Field(description日期格式 YYYY-MM-DD) temperature: float Field(description气温摄氏度) advice: str Field(description出行建议) llm ChatOpenAI(modelgpt-4o, temperature0) structured_llm llm.with_structured_output(WeatherReport) result structured_llm.invoke(明天北京天气怎么样) print(result.city, result.temperature) # 返回的是 Pydantic 对象不是原始字符串底层原理是LangChain 会把 Pydantic 模型定义转换成 JSON Schema通过 tool calling函数调用机制让模型选择一个合法的结构来响应。只要模型支持工具调用这种做法的稳定性比“把人话转 JSON”高一个档次。result是WeatherReport实例直接.city就能取到字段。但这里有个容易忽略的问题with_structured_output()默认不做流式输出配合invoke()使用时会一次性返回完整结果。如果你想在流式过程中拿到结构化的片段需要另外想办法。4.2 流式场景下的 JSON 断帧问题看似玄学实则线性当你开启 SSE 流式输出时模型生成的 JSON 会被拆成无数个 token 一帧帧传过来。前端的event.data可能长这样{ci ty: 北京 , tempe rature: 23 .5}这不是 JSON 坏了这只是 JSON 被“截”了。你没法直接JSON.parse这种半截数据。粗暴的做法是攒着等全部数据到齐再解析——这样最稳但同时失去了流式输出的意义另一种不想放弃流式的做法是用“增量 JSON 解析器”。增量 JSON 解析的核心思路是维护一个不断累积的字符串缓冲区每次收到新 token 就尝试解析一次缓冲区内容解析成功就更新 UI解析失败就继续等。由于 JSON 对象在结构上存在不完整状态单纯靠JSON.parse的 try-catch 无法支持增量所以需要判断“当前文本是否是一个合法 JSON 的前缀”。我常用的做法是加入一个“文本平衡校验”function hasBalancedBrackets(str) { SSE流式原理到LangChain结构化输出：打字机效果与JSON解析全方案
从 SSE 流式原理到 LangChain 结构化输出打字机效果与 JSON 解析全方案实战现在的 AI 应用谁还没个打字机效果都不好意思上线。但说实话我看过太多项目把“流式输出”做成了摆设——前端拿到一堆碎文本直接拼上去后端一个yield扔出去就算完事等到要让模型输出结构化 JSON 的时候整个链路直接崩掉。今天这篇文章咱们就把这条链路上的每一个环节都拆开揉碎SSE 协议到底在传什么、EventSource 怎么接、LangChain 怎么配合流式输出、以及最关键的结构化 JSON 在流式场景下怎么解析才不会翻车。适合谁看如果你正在做 LLM 应用的落地开发或者你只是听说过 SSE 和 LangChain 但一直没啃下源码这篇文章能帮你省掉至少一周的试错时间。我会把前端、后端、协议层、解析层全部串起来讲拿真实生产环境的方案说事。1. 从轮询到推送SSE 流式原理与选型逻辑很多人一听“流式输出”就想到 WebSocket这其实是个误区。Server-Sent EventsSSE和 WebSocket 虽然都能做实时推送但它们的定位完全不同。SSE 是单向的、基于 HTTP 的服务器推送协议浏览器原生支持一行EventSource就能接住WebSocket 是双向全双工通信适合聊天室、游戏这种需要频繁双向交互的场景。选择 SSE 来做大模型流式输出的理由非常实际大模型生成文本本来就是单向的——模型只管往客户端推数据不需要客户端频繁往回发消息。你用一个 WebSocket 连接去做一件只需要单向通知的事等于开着卡车去送快递不是不行但成本和复杂度完全不成比例。而且 SSE 走的是普通 HTTP天然兼容各种网关、代理、负载均衡器不存在 WebSocket 那种“连接要升级、防火墙要放行、长连接容易断”的麻烦。1.1 SSE 的本质就是一段不断输出的 HTTP 响应SSE 这个协议本身没什么神秘感。它就是一个 HTTP 响应只不过Content-Type设为text/event-stream服务器端不关闭连接持续往客户端写数据。每一帧数据遵循固定的格式规范data: 这是一行内容 data: 这是第二行内容事件帧之间用空行分隔每条数据以data:前缀开头。如果一行放不下可以分成多行data:客户端会自动用换行符拼接起来。如果想要把某条数据单独拉到指定事件类型还可以加event:字段客户端监听对应事件名即可。后端在 Python 里实现 SSE 端点其实非常简单——用 FastAPI 的StreamingResponse配合生成器一行yield就是一次推送from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() def token_stream(): for token in [你好, , 世界, ]: yield fdata: {token}\n\n app.get(/api/chat/stream) async def chat_stream(): return StreamingResponse( token_stream(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no } )注意那个X-Accel-Buffering: no头这个是我在生产环境踩过坑才加上的。Nginx 默认会缓冲后端响应如果不关掉缓冲前端收到的不是逐字打字机效果而是等整个响应结束才一次性拿到全部数据——流式效果直接归零。遇到这种情况第一反应不要怀疑代码先查网关层是不是在“捣乱”。1.2 为什么 EventSource 比 fetch 流式读取更省心浏览器原生的 SSE 客户端是EventSource对象用法极简const source new EventSource(/api/chat/stream); source.onmessage (event) { const token event.data; // 把 token 追加到界面上 appendText(token); }; source.onerror (err) { // 处理断连逻辑 console.error(SSE 连接断开, err); };用 fetch 走流式读取也可以但要注意一个关键差异EventSource 自带自动重连机制连接断开后浏览器会按retry:字段指定的间隔自动重新建立连接。而用 fetch 的ReadableStream做手动解析一旦连接中断你得自己写重连逻辑还要处理各种边缘情况。EventSource 还有一个限制它只支持 GET 请求。如果你需要把用户的历史对话通过 POST 发给后端再开启流式响应要么把参数拼在 URL 上要么改用fetch加POST配合ReadableStream来手动解析 SSE 帧。后者更灵活但需要自己实现帧解析逻辑这个我们放到后面讲。2. 前端打字机效果实现不只是“把文字贴上去”打字机效果的实现难度不在于“渲染文字”而在于“渲染的节奏怎么和生成节奏对齐”。模型输出的 token 到达时间是随机的最快的几个 token 可能几十毫秒就到了最慢的可能要等好几秒如果无脑把每个到达的 token 直接 append 到 DOM页面会有一阵一阵的卡顿感——这其实是 render 频率超过了浏览器的刷新率属于典型的性能问题。2.1 节流渲染让每帧最多只动一次 DOM正确的做法是引入一个节流机制把 SSE 收到的 token 先塞进一个缓冲区然后通过requestAnimationFrame循环把缓冲内容定时写入 DOM。每帧最多渲染一次帧缓冲浏览器会自然地在空闲时刷新不会出现内容以肉眼可见的“一坨一坨”冲出来的情况。export function createTypewriter(element) { let buffer ; let rendering false; let canceled false; function flush() { if (canceled) return; element.textContent buffer; buffer ; rendering false; } function scheduleFlush() { if (rendering) return; rendering true; requestAnimationFrame(flush); } return { push(text) { buffer text; scheduleFlush(); }, done() { canceled true; flush(); } }; }这里最关键的是scheduleFlush里的防抖逻辑同一帧里不管 push 进来多少 token只会触发一次flush。这样即使后端一次塞给前端十几个 token界面也是平滑地滚动更新不会闪烁或跳动。除了渲染节流还需要处理 markdown 的渲染问题。如果直接用textContent把文本塞进去接完流之后要做一次marked或markdown-it解析把纯文本转成带样式的 HTML。这里有个细节要注意流式过程中不要实时做 markdown 转 HTML因为 markdown 语法往往是跨多个 token 的比如代码块的三个反引号可能分几次传输中途转换很容易解析出残次品。稳妥的做法是流式过程只展示纯文本流结束后整体转一次 markdown。2.2 断连重连与消息状态的精准控制打字机的“结束态”比“进行态”更容易被忽视。前端跟用户沟通的方式不能只看文字有没有打完还要知道后端是不是主动关闭了流。SSE 连接结束后浏览器会触发onerror但这个onerror在正常关闭和异常断连时都会触发区分两者需要靠后端发送一个特殊的结束标识。惯例的做法是后端在流式响应完所有 token 后额外发送一个data: [DONE]帧前端收到这个标志就正常关闭连接不再触发重连逻辑。如果是中途超时、网关断连、服务器异常前端收不到[DONE]就需要走自动重连同时更新界面上的“重试”提示。event: message data: 今天天气不错 data: [DONE]事件监听里判断event.data [DONE]后调用source.close()这是一个很容易被新手漏掉的操作——你不主动 closeEventSource 会一直按重试间隔尝试重新连接产生一堆无意义的请求。2.3 SSE 用 GET 也能带参数EventSource 的兼容处理刚才提到 EventSource 只支持 GET很多人会被卡在这一步。如果业务场景要求用 POST 传参最干净的方案是保留 EventSource 的自动重连能力把复杂参数放在服务端会话里前端只需要用 GET 请求一个临时会话标识// 先通过 POST 建立会话拿到 session_id const { session_id } await fetch(/api/chat/start, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages: history }) }).then(res res.json()); // 再开启 SSE 流式连接 const source new EventSource(/api/chat/stream?session_id${session_id});后端拿到 session_id 后从内存或 Redis 中取对应用户的请求参数再开始生成流。这样既绕过了 GET 的参数长度限制又保留了对断线重连的友好性算是我在多个项目里反复验证过的标准做法。3. LangChain 流式输出实战Stream 与回调的取舍LangChain 做流式输出有两个层面的接口一个是最外层的stream()方法另一个是底层的callbacks机制。很多教程只讲stream()但真正要把流式输出部署到生产环境的项目里你大概率需要用回调回调来拿全链路的 token。3.1 用.stream()拿到大模型 token 流stream()的用法非常简单直接迭代即可from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm ChatOpenAI( modelgpt-4o-mini, temperature0, streamingTrue ) for chunk in llm.stream([HumanMessage(content讲个冷笑话)]): if hasattr(chunk, content) and chunk.content: yield fdata: {chunk.content}\n\n注意一点llm.stream()在内部其实也是开启 streaming然后逐块 yield AIMessageChunk。streamingTrue这个参数不是必须的但加上能确保底层调用过程使用流式接口减少首 token 延迟。在 FastAPI 的生成器函数里逐块yield出去前端打字机效果就有了。3.2 用 callbacks 实现多路输出与日志采集生产环境里普遍存在的需求是同一个流式响应里你既要给最终用户看完整回答又要给运营看 token 用量、给调试者看中间推理过程。stream()只能让你拿到最终的 token 流拿不到“模型在中间到底走了哪几步”。这种情况就要用 LangChain 的AsyncCallbackHandlerfrom langchain_core.callbacks import AsyncCallbackHandler from langchain_core.agents import AgentFinish class TokenCollector(AsyncCallbackHandler): def __init__(self): self.token_buffer [] self.agent_logs [] async def on_llm_new_token(self, token: str, **kwargs): self.token_buffer.append(token) # 这里把 token 实时推送到 SSE async def on_agent_action(self, action, **kwargs): self.agent_logs.append(f思考: {action.log}) async def on_agent_finish(self, finish: AgentFinish, **kwargs): self.agent_logs.append(f完成: {finish.return_values})回调机制特别适合 Agent 场景用户在界面上看到的不只是最终回答还能看到“Agent 正在搜索资料”“Agent 正在调用计算器”这样的中间状态体验会非常像真的在看一个人思考过程在推进。3.3 LangChain 版本差异的坑接口更新比模型还快LangChain 的接口变化非常频繁尤其是 LangChain v0.1 到 v0.2 再到 v0.3 这个跨度里很多老教程里的代码已经跑不起来了。比如langchain.chains.LLMChain在较新版本里被降级为 Legacy官方更推荐直接用RunnableSequence或LangGraph。如果你照着网上的旧教程抄代码很可能遇到DeprecationWarning甚至直接报错。我的建议是项目初始就锁定一个 LangChain 版本不要用latest。比如langchain0.2.x配langchain-openai的版本组合功能相对稳定。升级某个核心包时最好全量跑一遍链路的流式测试用例因为回调签名、流式输出的 chunk 类型、事件触发顺序都可能被改动影响。4. 结构化输出与 JSON 解析流式场景下最大的坑标题里说的“结构化输出”指的是让 LLM 稳定地返回一段符合 JSON Schema 的数据而不是一段格式可疑的文本。在这个基础上叠加流式传输问题立刻复杂了一个量级你收到的不是一个完整 JSON而是被拆成几千个碎片的 JSON 字符串。如何从碎片流中稳定地解析出结构化数据是这门手艺的真正核心。4.1 LangChain 的with_structured_output把格式控制交给模型LangChain 提供了非常顺手的结构化输出能力。核心方法是with_structured_output()配合 Pydantic 模型定义输出格式from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class WeatherReport(BaseModel): city: str Field(description城市名称) date: str Field(description日期格式 YYYY-MM-DD) temperature: float Field(description气温摄氏度) advice: str Field(description出行建议) llm ChatOpenAI(modelgpt-4o, temperature0) structured_llm llm.with_structured_output(WeatherReport) result structured_llm.invoke(明天北京天气怎么样) print(result.city, result.temperature) # 返回的是 Pydantic 对象不是原始字符串底层原理是LangChain 会把 Pydantic 模型定义转换成 JSON Schema通过 tool calling函数调用机制让模型选择一个合法的结构来响应。只要模型支持工具调用这种做法的稳定性比“把人话转 JSON”高一个档次。result是WeatherReport实例直接.city就能取到字段。但这里有个容易忽略的问题with_structured_output()默认不做流式输出配合invoke()使用时会一次性返回完整结果。如果你想在流式过程中拿到结构化的片段需要另外想办法。4.2 流式场景下的 JSON 断帧问题看似玄学实则线性当你开启 SSE 流式输出时模型生成的 JSON 会被拆成无数个 token 一帧帧传过来。前端的event.data可能长这样{ci ty: 北京 , tempe rature: 23 .5}这不是 JSON 坏了这只是 JSON 被“截”了。你没法直接JSON.parse这种半截数据。粗暴的做法是攒着等全部数据到齐再解析——这样最稳但同时失去了流式输出的意义另一种不想放弃流式的做法是用“增量 JSON 解析器”。增量 JSON 解析的核心思路是维护一个不断累积的字符串缓冲区每次收到新 token 就尝试解析一次缓冲区内容解析成功就更新 UI解析失败就继续等。由于 JSON 对象在结构上存在不完整状态单纯靠JSON.parse的 try-catch 无法支持增量所以需要判断“当前文本是否是一个合法 JSON 的前缀”。我常用的做法是加入一个“文本平衡校验”function hasBalancedBrackets(str) {