1. 从“等待”到“流淌”理解AI调用的两种范式最近在折腾AI应用开发特别是涉及到与大模型交互时一个绕不开的核心问题就是如何获取模型的响应是让用户干等着直到模型“憋”出一个完整的答案还是让答案像溪流一样一个字一个字地“流淌”出来这背后对应的就是标题里提到的两种调用方式阻塞式invoke和流式stream。这不仅仅是技术选型更是产品体验的分水岭。想象一下你问ChatGPT一个复杂问题如果页面一直转圈圈十几秒后突然“啪”一下弹出整段答案你可能会怀疑是不是卡死了。而如果答案是一个词一个词地出现即使总耗时一样你的感知却是“它正在为我思考”体验流畅得多。对于开发者而言理解这两种模式的底层机制、适用场景以及背后的坑是构建现代AI应用的基本功。无论是集成OpenAI API、使用Spring AI这类框架还是对接本地部署的大模型这个选择都会直接影响你的应用架构和用户体验。2. 阻塞式调用简单直接的“一问一答”阻塞式调用顾名思义就是调用方发起请求后线程会被“阻塞”住一直等待被调用方这里是AI模型处理完毕并返回完整结果后才能继续执行后续逻辑。这很像我们打电话问客服一个问题在客服没有给出完整答复前我们只能拿着听筒等待。2.1 技术实现与核心流程在代码层面一个典型的阻塞式调用看起来非常直观。以使用OpenAI的官方Node.js SDK为例import OpenAI from openai; const openai new OpenAI({ apiKey: your-api-key }); async function getBlockingResponse() { console.log(开始请求线程即将阻塞...); const completion await openai.chat.completions.create({ model: gpt-4, messages: [{ role: user, content: 请用300字介绍量子计算。 }], stream: false, // 明确关闭流式使用阻塞式 }); console.log(收到完整响应线程继续执行。); console.log(回答内容, completion.choices[0].message.content); return completion; } getBlockingResponse();在这个例子中await关键字就是“阻塞”的体现。执行到这一行时JavaScript的异步函数会暂停事件循环可以去处理其他任务但getBlockingResponse这个函数的后续逻辑打印日志和返回结果必须等待API调用完全结束。对于服务器端如果使用同步框架或在协程中处理这个请求线程/协程在等待期间无法处理其他任何请求。其核心流程可以概括为客户端打包请求将用户输入、模型参数等序列化为JSON通过HTTP POST发送。服务端排队与计算请求进入模型服务的队列轮到时加载模型并进行前向推理计算。对于大模型生成数百个token可能需要数秒到数十秒。完整生成与返回模型生成完整的回答文本后服务端将其打包成JSON响应。网络传输整个完整的响应体通过网络传输回客户端。客户端处理客户端收到完整响应后解析JSON获取content字段呈现给用户。2.2 适用场景与优缺点分析阻塞式调用并非过时的技术它在很多场景下依然是首选。优点逻辑简单代码易于编写、理解和调试。请求-响应模式与传统Web开发完全一致。状态完整一次性获得最终、确定性的结果便于进行后续处理如存入数据库、进行格式化或触发其他业务逻辑。资源管理清晰对于客户端一个请求对应一个响应连接生命周期明确。缺点用户体验差用户需要等待整个生成过程结束才能看到任何内容对于长文本生成等待时间难以忍受。超时风险高网络连接需要保持长时间稳定。如果生成耗时超过HTTP客户端或服务器的超时设置常见为30秒或60秒连接会中断导致整个请求失败用户什么也得不到。服务器资源占用长在服务器端处理该请求的线程或连接在等待期间被占用在高并发场景下会迅速耗尽连接池影响系统吞吐量。无法中途干预一旦请求发出无法中途取消或修改生成方向除非服务端支持特殊令牌但主流API一般不提供。注意很多初学者容易混淆“异步非阻塞”和“流式”。上述Node.js例子使用了async/await它是异步的但就本次API调用而言它仍然是阻塞式的因为我们必须等待这个特定的Promise完成即收到完整响应才能继续。流式是一种更特殊的非阻塞形式它允许在请求未完成时就开始处理部分结果。那么什么时候该用阻塞式生成内容极短时例如只是让模型进行简单的分类、情感分析或生成一个短语耗时在1-2秒内流式的收益不大。后端批量处理任务在离线或后台任务中需要处理成千上万个提示词然后统一存储结果此时稳定性和完整性比实时性更重要。与不支持流式的旧系统集成下游系统只能接收完整数据包。3. 流式调用实时交互的“文字雨”流式调用彻底改变了交互模式。它利用HTTP分块传输编码或WebSocket等协议在服务器端模型生成第一个token词元后就立即将其发送给客户端后续生成的token也持续不断地“流”过来。客户端可以边接收边渲染实现打字机效果。3.1 技术实现从SSE到WebSocket目前主流AI API如OpenAI、Anthropic的流式响应大多基于服务器发送事件技术。SSE是一种简单的HTTP协议允许服务器主动向客户端推送数据。它保持一个长连接服务器可以多次发送事件流。对于流式AI响应每个生成的token或一小段文本就作为一个事件发送。// 前端使用EventSource接收SSE流 const eventSource new EventSource(/api/chat-stream); eventSource.onmessage (event) { const data JSON.parse(event.data); // 假设data结构为 { content: ..., done: false } if (!data.done) { // 将data.content追加到页面上的聊天区域 document.getElementById(chat-output).innerHTML data.content; } else { eventSource.close(); console.log(流式传输结束); } }; eventSource.onerror (error) { console.error(EventSource failed:, error); eventSource.close(); };服务端以Node.js Express为例需要设置正确的响应头并以流的方式调用AI API并转发数据块app.get(/api/chat-stream, async (req, res) { // 设置SSE必需的响应头 res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); try { const stream await openai.chat.completions.create({ model: gpt-4, messages: [{ role: user, content: req.query.prompt }], stream: true, // 关键开启流式 }); for await (const chunk of stream) { const content chunk.choices[0]?.delta?.content || ; // 将数据封装为SSE格式发送 res.write(data: ${JSON.stringify({ content: content, done: false })}\n\n); } // 发送结束标志 res.write(data: ${JSON.stringify({ done: true })}\n\n); res.end(); } catch (error) { res.write(data: ${JSON.stringify({ error: error.message })}\n\n); res.end(); } });对于更复杂的双向实时交互如需要随时发送用户中途输入WebSocket是更好的选择但实现复杂度也更高。Spring AI等框架对这两种方式都提供了抽象支持。3.2 流式传输的底层细节与数据格式理解数据格式对于处理流式响应至关重要。以OpenAI API为例当设置stream: true后返回的不是一个JSON对象而是一个流其中包含一系列以data:开头的SSE事件。每个数据块chunk的格式大致如下data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1234567890,model:gpt-4,choices:[{index:0,delta:{content:你好},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1234567890,model:gpt-4,choices:[{index:0,delta:{content:},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1234567890,model:gpt-4,choices:[{index:0,delta:{content:世界},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1234567890,model:gpt-4,choices:[{index:0,delta:{},finish_reason:stop}]} data: [DONE]关键字段解析delta: 包含本次数据块带来的增量内容。对于文本通常在delta.content中。delta也可能包含role只在第一个块出现或其他字段。finish_reason: 通常为null直到最后一个内容块其值变为stop正常结束、length达到token限制或content_filter内容被过滤等。当finish_reason非空时delta通常为空对象。[DONE]: 一个特殊的事件标识整个流已完全结束。客户端需要做的就是拼接所有delta.content直到遇到finish_reason或[DONE]事件。3.3 流式调用的巨大优势与隐藏挑战优势是显而易见的极致用户体验实现“打字机”效果让用户感知延迟大幅降低交互感强。首字响应时间快用户通常在几百毫秒到一两秒内就能看到第一个词即使生成全文需要20秒用户也不会感到焦虑。更容错的网络即使中间出现短暂网络波动连接断开用户也已经看到了部分内容。并且可以通过重连机制如指定snapshot_id如果API支持从断点恢复而阻塞式调用一旦超时则前功尽弃。支持中途取消客户端可以在任何时候关闭连接服务器端理论上可以尽管并非所有服务都实现中止昂贵的模型计算节省资源。然而流式引入的复杂性不容小觑客户端状态管理复杂你需要维护一个不断增长的文本缓冲区并实时更新UI。在复杂的前端框架中这可能需要考虑响应式更新的性能。错误处理更棘手流可能在任意时刻中断。你需要监听错误事件并决定是重试、提示用户还是静默失败。错误信息可能在一个单独的数据块中传递而不是传统的HTTP错误状态码。资源泄露风险必须确保流被正确关闭。无论是用户离开页面还是组件卸载都要调用EventSource.close()或关闭WebSocket连接否则会导致内存泄漏和服务器资源浪费。数据完整性校验缺失流式传输通常没有对整个响应内容的完整性校验如MD5。虽然概率极低但存在网络传输中某个数据包损坏或丢失的风险导致最终拼接的文本有误。对后端代理和网关的要求更高如果你的应用前面有Nginx、API网关或CDN它们必须正确配置以支持长连接和分块传输并且超时时间要设置得足够长。4. 实战踩坑流式调用中的那些“暗礁”在实际项目中从阻塞式切换到流式绝非改个参数那么简单。下面是我在多个项目中总结的常见问题和解决方案。4.1 连接过早断开超时与缓冲区的博弈这是流式调用中最常见的问题。错误信息常常是stream disconnected before completion: transport error或upstream chat completions stream ended。根因分析代理服务器超时Nginx等反向代理默认的proxy_read_timeout可能只有60秒。如果一个复杂的思考过程导致模型超过60秒没有输出下一个tokenNginx就会主动断开连接。应用服务器框架超时Node.js的Express、Python的FastAPI等框架其底层HTTP服务器也有默认的超时设置。客户端超时浏览器或HTTP客户端库如Axios设置的超时时间过短。服务端主动关闭如果模型生成过程中遇到内部错误或者你的账户额度用尽如错误信息you have no credits remainingAI服务提供商也会主动断开流。解决方案逐层调整超时Nginx: 在对应的location块中显著增加超时时间。location /api/chat-stream { proxy_pass http://your_backend; proxy_http_version 1.1; proxy_set_header Connection ; proxy_set_header X-Real-IP $remote_addr; chunked_transfer_encoding off; # 对于某些代理可能需要关闭 proxy_buffering off; # 关键关闭代理缓冲让数据立即转发 proxy_cache off; proxy_read_timeout 300s; # 根据需求调整例如300秒 proxy_send_timeout 300s; }* **应用框架**以Node.js Express为例需要调整底层服务器的超时。 javascript const server app.listen(port); server.setTimeout(10 * 60 * 1000); // 设置为10分钟 * **客户端**使用支持流式且可配置超时的库。对于浏览器EventSource它本身不支持设置超时但可以通过定期发送“心跳”数据包来保持连接活跃并监听错误进行重连。关闭代理缓冲proxy_buffering off;这条指令至关重要。如果开启缓冲Nginx会尝试收集一定量的数据再发给客户端这违背了流式“实时”的初衷也可能导致客户端长时间收不到数据而认为连接已死。4.2 流式数据解析与拼接的陷阱即使连接稳定处理数据流本身也有坑。问题1数据块不完整或粘连网络传输可能将一个或多个SSE事件合并到一个TCP包中或者一个事件被拆分成多个包。你不能假设每次收到的数据刚好是一个完整的data: {...}\n\n。解决方案实现一个简单的状态机或使用成熟的解析器。let buffer ; function handleStreamData(chunk) { buffer chunk; let boundary; while ((boundary buffer.indexOf(\n\n)) ! -1) { const event buffer.substring(0, boundary); buffer buffer.substring(boundary 2); if (event.startsWith(data: )) { const eventData event.substring(6); // 去掉data: if (eventData.trim() [DONE]) { console.log(流结束); return; } try { const parsed JSON.parse(eventData); // 处理parsed数据... } catch (e) { console.error(解析JSON失败:, e, 原始数据:, eventData); } } } } // 在收到数据时调用 handleStreamData(receivedText)问题2delta对象可能为空或包含其他字段除了content第一个数据块的delta可能包含role最后一个数据块的delta为空而finish_reason有值。你的代码需要健壮地处理这些情况。// 在处理每个chunk.choices[0].delta时 const delta chunk.choices[0].delta; if (delta.role) { // 这是第一个块可能包含角色信息通常可以忽略或用于UI } if (delta.content ! undefined delta.content ! null) { accumulatedContent delta.content; // 更新UI... } if (chunk.choices[0].finish_reason) { // 生成结束进行清理工作 console.log(生成结束原因:, chunk.choices[0].finish_reason); }4.3 前端渲染性能与用户体验优化当答案快速涌出时频繁更新DOM可能导致页面卡顿。优化方案使用文档片段或虚拟DOM不要每次收到一个词就直接操作innerHTML。可以累积一小段例如每100毫秒或每收到5个token再更新一次。“打字机”效果实现如果想实现逐字打印可以用setInterval或requestAnimationFrame从缓冲区中逐个取出字符渲染而不是依赖网络流的速度。这样即使网络瞬间送来一大段文字也能以平稳的速度显示。处理中间思考过程一些模型如GPT-4在流式输出中可能会先输出“让我思考一下...”之类的内部推理词。你可能需要在客户端过滤掉这些内容或者将它们以特殊样式如斜体、灰色显示并在最终答案出现前将其删除以提供更干净的体验。4.4 结合Spring AI等框架的实践如果你在使用Spring AI这类高层框架它封装了与不同AI提供商OpenAI, Azure, Ollama等的流式交互。通常你需要返回一个FluxReactive Streams的发布者对象。RestController public class ChatController { GetMapping(/ai/stream) public FluxString streamChat(RequestParam String message) { Prompt prompt new Prompt(new UserMessage(message)); ChatResponse response chatClient.stream(prompt); // 返回FluxChatResponse return response .map(chatResponse - chatResponse.getResults().get(0).getOutput().getContent()) .doOnNext(content - System.out.println(收到流式内容: content)); } }框架下的坑响应头自动设置Spring WebFlux通常能自动处理好text/event-stream的响应头。但如果你的项目混用了Spring MVC可能需要手动配置。背压处理Flux支持背压即客户端可以控制接收速度。但在AI流式场景下服务器推送速度取决于模型生成速度通常不需要复杂背压逻辑但理解这个概念有助于调试。错误信号传递如果AI服务端出错错误如何通过Flux传递到客户端你需要确保Flux管道中有onErrorResume或类似的错误处理操作符将错误转换为客户端能理解的SSE事件格式而不是直接导致连接断裂。5. 架构思考何时选择阻塞何时选择流式技术选型没有银弹。在做决定时可以从以下几个维度评估交互性质强交互场景对话、创作辅助、代码实时提示无脑选择流式。用户体验提升是质的飞跃。弱交互或异步场景内容摘要、批量翻译、数据清洗阻塞式可能更简单。如果任务耗时很长甚至应该采用异步任务队列提交任务 - 立即返回任务ID - 客户端轮询或通过WebSocket获取结果而不是长连接阻塞或流式。生成内容长度短文本 50 token两者差异不大。流式省下的等待时间可能只有几百毫秒但引入了复杂度。中长文本 100 token流式优势开始显现。长文本 500 token流式几乎是必选项。客户端环境现代浏览器/原生App可以很好地支持EventSource或WebSocket。命令行工具流式输出可以模拟终端打字效果体验很好。服务器对服务器调用如果下游服务需要完整结果才能处理则用阻塞式。如果下游服务也能处理流可以考虑管道化的流式传输减少中间环节的延迟和内存占用。复杂度与成本开发与测试成本流式开发的复杂度是阻塞式的数倍调试也更困难。基础设施成本流式需要服务端和客户端都维护长连接对负载均衡器、连接数有更高要求。虽然单个连接资源消耗不大但海量并发时仍需评估。监控与运维流式连接的健康度、中断率、平均持续时间等都需要新的监控指标。一个折中的混合策略对于某些应用可以采用“快速模式完整模式”。例如先使用流式快速返回一个初步答案或大纲同时在后台用阻塞式生成一个更详细、润色过的版本。当流式传输结束后询问用户是否需要查看“增强版”如果需要则从后台获取已生成好的完整版本。这既保证了首响速度又提供了高质量的结果。从我个人的项目经验来看对于面向最终用户的AI功能只要条件允许我都会优先考虑实现流式接口。尽管初期会多花一些时间解决连接、解析和渲染的问题但它所带来的用户体验优势是决定性的。尤其是在今天当用户已经习惯了ChatGPT那种流畅的对话感受后一个需要漫长等待的应用会立刻显得笨拙和过时。把玩转“流”视为现代AI应用开发的必修课绝对不为过。