> ❝
>
> 把 Agent 当普通 API 调，用户盯着转圈圈等上十几秒，体验上和“程序崩了”没什么两样。

## **1. 「能跑通」和「能用」之间，隔着一条河**

我见过有些团队把 Agent demo 跑通之后就觉得万事大吉，结果一上线，用户点完按钮，屏幕中央一个转圈圈转了二十秒，然后“啪”地弹出一大段答案。技术上讲，这没毛病；体验上讲，用户大概率已经在第五秒的时候以为网络挂了，顺手关掉了页面。

问题出在哪？普通 API 是“请求—响应”一次性交付，Agent 是**多步、多回合、带工具调用**的长链路。一个稍微像样的 Agent，至少会经历“理解意图 → 规划步骤 → 调工具 A → 调工具 B → 把结果喂回模型 → 综合生成答案”这条链路。每一步都要等一次 LLM 推理（首字延迟几百毫秒到几秒不等），再加上工具本身的耗时（查数据库、调内部服务、跑检索），端到端轻松冲到十秒甚至几十秒。

这就是**感知延迟**和**真实延迟**的错位。流式交互解决的正是一件事：在不缩短总生成时间的前提下，把“用户第一次看到内容”的时间（Time to First Token，TTFT）从“整段生成完”压缩到“几百毫秒”。业界经验里，开流式之后 TTFT 通常落在 100–500 毫秒量级，而关掉流式，TTFT 就等于整段生成时间。总时间没变，但用户觉得快了一截，因为他在“边读边等”，而不是“干瞪眼”。

> ❝
>
> 踩坑提醒 🕳 我刚做 Agent 产品时，第一版就是“等所有步骤跑完再一次性返回”。用户测试时普遍反馈“卡死了”，其实是他在第八秒就放弃了。后来加了流式，同样的链路、同样的耗时，满意度直接上来了。

**实战经验**：流式不是锦上添花，是 Agent 从“能演示”到“能交付”的分水岭。只要你的 Agent 端到端超过三秒，就默认应该流式，没有例外。

## **2. 流式交互的三个层次**

流式不是简单地“把字一个个吐出来”。按信息粒度，它分三层，越往下越细、实现成本也越高。

**第一层：Token 流式**。只把模型最终那一段文本做逐字输出。这是成本最低、收益最高的一层，几乎所有聊天类产品都做了。它解决的是“最终答案”的等待感。

**第二层：步骤 / 事件流式**。把 Agent 的中间过程也流出来——它现在在调哪个工具、工具返回了什么、下一步要干嘛。这一层把“黑盒”变成了“玻璃盒”，用户能看见 Agent 在忙，而不是对着空白怀疑人生。

**第三层：动作流式**。更细：工具调用的参数还在“边生成边拼”，结构化输出（JSON）也是边生成边解析、边渲染。这一层最贴近“Agent 真正在思考”的体感，但实现最复杂，因为你要处理“半截 JSON”这种本来不合法的东西。

| 层次             | 流出内容             | 主要协议             | 用户感知    | 实现难度 |
| :--------------- | :------------------- | :------------------- | :---------- | :------- |
| 第一层 Token     | 最终文本逐字         | SSE 文本增量         | “AI 在打字” | 低       |
| 第二层 步骤/事件 | 工具调用、进度、日志 | 自定义事件流         | “AI 在干活” | 中       |
| 第三层 动作      | 工具参数、结构化字段 | 增量 JSON / 协议前缀 | “AI 在思考” | 高       |

> ❝
>
> 不要一上来就把三层全做了。先把第一层做扎实——它的投入产出比最高，能解决八成的“等得慌”问题。第二层在工具多的 Agent 上才值得做，第三层只在“结构化结果需要即时预览”的场景（比如 Agent 帮你填一张表、画一张图）才有必要。

## **3. 底层机制：SSE 与增量协议**

Agent 流式几乎都跑在 **SSE（Server-Sent Events）** 上。原因很朴素：流式是单向的“服务器推给客户端”，而 SSE 正好是单向的、基于一条长连接 HTTP 的纯文本协议，比 WebSocket 轻得多。浏览器原生提供 `EventSource`，连重连都帮你做好了。

各家大模型厂商的流式接口，本质上都是在 SSE 里塞“增量块”（delta），由你边收边拼。

**OpenAI** 的做法是：请求里带 `stream=True`，返回的是一连串 `chat.completion.chunk`，每个块带 `choices[0].delta`。文本增量在 `delta.content` 里；工具调用则出现在 `delta.tool_calls` 数组里，而且有个容易踩的坑——**只有第一个增量块带 `id` 和 `function.name`，后续块里这两个字段是 `null`，只剩 `function.arguments` 的碎片**。你必须按 `index` 把这些碎片拼起来，等流结束（`finish_reason` 变成 `tool_calls`）再一次性 `json.loads`。流的结尾是一行字面量 `data: [DONE]`。

```
import json
from openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "巴黎今天天气怎么样？"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string"}},
                "required": ["location"],
            },
        },
    }],
    stream=True,
)

# 按 index 累积工具调用的参数碎片，绝不能中途解析
tool_args = {}
for chunk in stream:
    for tc in chunk.choices[0].delta.tool_calls or []:
        buf = tool_args.setdefault(tc.index, {"name": "", "arguments": ""})
        if tc.id:
            buf["name"] = tc.function.name  # 仅首块携带
        buf["arguments"] += tc.function.arguments or ""
# 流结束后才解析完整 JSON
args = json.loads(tool_args[0]["arguments"])
print(buf["name"], args)
```

**Anthropic** 的流式走一套命名事件：`message_start` → 若干个 `content_block_start` / `content_block_delta` / `content_block_stop` → `message_delta` → `message_stop`。工具调用对应的是 `tool_use` 类型的 content block，它的参数以 `content_block_delta` 事件里 `delta.type == "input_json_delta"` 的 `partial_json` 字符串碎片流出。官方文档明确说“当前模型一次只吐一个完整的键值对”，所以碎片粒度大致在 JSON 属性这一级。同样，要等 `content_block_stop` 之后再拼起来解析。

> ❝
>
> 踩坑提醒 🕳 千万别在收到每个 `partial_json` 碎片时就 `json.loads`。那玩意儿在中途根本不是合法 JSON，一定会抛异常。缓冲区攒到“块结束”事件再解析，这是铁律。

## **4. 把「黑盒等待」变成「过程可见」**

第一层 Token 流解决的是“最终答案”的等待感，但 Agent 真正的耗时大头在**工具调用**。用户看不到工具在跑，依然会觉得“卡了”。所以第二层——把过程流出来——价值极大。

做法是在 Agent 执行循环里埋“事件发射器”，每个有意义的节点发一个结构化事件：

```
def run_agent(query, emit):
    emit({"type": "thinking", "text": "先拆解一下你的需求…"})
    emit({"type": "tool_call", "name": "search_kb", "args": {"q": query}})
    result = search_kb(query)            # 真实工具调用
    emit({"type": "tool_result", "name": "search_kb", "summary": f"命中 {len(result)} 条"})
    emit({"type": "thinking", "text": "开始综合答案…"})
    for token in llm_stream(query, result):
        emit({"type": "token", "text": token})
    emit({"type": "done"})
```

前端把这些事件渲染成一条“活动流”：🔍 正在检索知识库… ✓ 命中 12 条 → ✍️ 正在组织答案…。死时间变成了信任建立的时间。

> ❝
>
> 实战经验：LangGraph 专门给这一层留了 `custom` 流式模式——在节点里调 `get_stream_writer()` 就能往流里塞任意自定义事件，不用去改状态结构。如果你的 Agent 跑在 LangGraph 上，进度上报直接用这个，最省事。

**粒度是关键**。事件太碎（每次循环变量都发一条）会刷屏、干扰阅读；太粗（只在开头和结尾发两条）又回到了黑盒。一个实用的经验法则：**一个工具调用 = 一个进度点**，外加“开始思考”“开始综合”这种少量里程碑即可。

## **5. 可打断的 Agent：人在回路里随时插话**

传统请求—响应模式下，用户发完就只能干等。但 Agent 一跑十几秒、还带着“调用外部工具”“发邮件”“改数据库”这种有副作用的动作，用户中途想喊停、想改主意、想确认“你真要删库？”——这些都是刚需。

可打断性有三档做法：

**第一档：执行前确认（pre-tool confirmation）**。凡是带副作用的工具（发邮件、写数据库、调支付），先暂停，把“我要执行 XXX，参数是 YYY”推给用户，确认了再真的跑。这其实就是 HITL（人在回路）的最小实践，和权限模型是一体两面。

**第二档：中途改向（mid-stream edit）**。用户在 Agent 跑到一半时插一句话“不对，换个思路”，Agent 能接管当前状态、调整计划继续跑。LangGraph 的 `interrupt()` 配合 checkpointer 能做到在任意节点挂起、注入新指令再恢复。

**第三档：取消（cancellation token）**。直接中止，释放资源。前端把“停止”按钮接到后端的取消信号上，LLM 流和工具调用都能被中断。

> ❝
>
> 踩坑提醒 🕳 打断后最容易翻车的是**状态不一致**。尤其是已经执行过副作用的工具——你都把钱转出去了，用户再喊停也撤不回来。所以我的铁律是：把工具分成“只读查询”和“写操作”两类，写操作一律前置确认，且尽量做成幂等（见下一节）。只读工具随便打断，问题不大。

**实战经验**：可打断不是“加个停止按钮”就完事，它逼着你重新审视 Agent 的动作边界。一个打断安全的 Agent，必然是动作边界清晰、副作用受控的 Agent。

## **6. 结构化输出的流式化：边生成边解析**

很多 Agent 最终要吐出的是结构化结果——JSON、表格、一张填好的表单。如果等模型把整段 JSON 憋完再返回，前面十几秒又是空白。第三层流式就是解决这个问题：让结构化结果也“长”出来。

Vercel AI SDK 在这里做了件很贴心的事：它会在内部把工具调用的参数碎片**先拼好、再作为完整的 tool-call 部分 emit 出去**，保证“半成品 JSON 永远不会进到你的应用代码里”。它自有的 DataStream 协议用换行分隔、带类型前缀，前端按前缀分流即可——文本是 `0:`、自定义数据走 `2:`、UI 元数据走 `8:`，而工具调用则是一组带类型前缀的事件（`b:` 开始、`c:` 参数增量、`9:` 完整调用、`a:` 调用结果）。

如果你是自己手搓，常见两种策略：

**骨架先行（skeleton-first）**：先发一个带好字段名、值为空的 JSON 骨架，再随着模型生成把对应字段填上。前端能立刻画出“表头”，字段一个个亮起来，体验非常顺。

**逐字段流式**：模型按字段顺序生成，每完成一个字段就推一次增量。适合表单、配置这类场景。

> ❝
>
> 踩坑提醒 🕳 不管哪种策略，客户端都别拿“还没长完的 JSON”直接 `JSON.parse`。要么用带容错的增量解析器，要么就只解析已经 `done` 的字段。骨架先行的好处是：即使流中断，用户至少看到了结构，不至于白屏。

## **7. 推理模型的“思考”也要流：别让用户对着空白等**

在用 o1、DeepSeek-R1、Claude 扩展思考这类**推理模型**时，还有一个常被忽略的体验坑：模型会先产出一大段“思考过程”，再给答案。如果这段思考是“先憋着、最后一次性抖出来”，用户面对的就是一段比答案还长的空白。

正确做法是把**思考通道也流式化**。Anthropic 的扩展思考（extended thinking）在流式下会以 `content_block_delta` 里 `delta.type == "thinking_delta"` 的方式逐片流出；Vercel AI SDK 的 DataStream 协议里，推理内容对应的类型前缀是 `g:`（reasoning part），前端可单独渲染成一个可折叠的“思考中”面板。OpenAI 的 Responses API 也把推理过程作为 reasoning item 流式返回，并可开 reasoning summaries 做轻量展示。

![img](https://developer.qcloudimg.com/http-save/yehe-5687454/2b17885d2743124ff89a5b7b89bc9e47.png)

> ❝
>
> 思考流要给用户“可控的透明”。默认折叠、可一键展开，且明确标注“这是模型的内部推演、不代表最终结论”。既消除空白焦虑，又避免用户把思考过程误读成事实。

## **8. 生产环境的三道坎：重连、背压、幂等**

Demo 阶段流式很美好，上了生产才有三道真实的坎。

**第一道：断线重连**。SSE 跑在 HTTP 长连接上，连接说断就断（代理超时、负载均衡、网络抖动）。好在浏览器的 `EventSource` 会自动重连，而且重连请求里会带上 `Last-Event-ID` 头——前提是你每条事件都发了 `id` 字段。服务端拿到 `Last-Event-ID`，从游标存储里把“客户端错过那段”重放一遍，流就无缝续上了。顺带提醒几个生产必加的响应头：`Cache-Control: no-cache`（别让中间层缓存流）、`X-Accel-Buffering: no`（关掉 nginx 的响应缓冲，否则你会看到“攒一批再吐”而不是真流式）。浏览器默认重连间隔约 3000 毫秒，而且**不会自动做指数退避**，想控制节奏就发 `retry:` 字段。

**第二道：背压（backpressure）**。如果客户端消费慢（比如前端卡顿、移动网络差），而你还在拼命 `write`，服务端的发送缓冲区会一路涨，最后内存爆掉。处理方式要么“丢弃非关键事件”（进度日志可以丢，最终答案不能丢），要么“暂停上游”等客户端跟上。Vercel 官方文档也把 backpressure handling 列为流式优化的要点之一。

**第三道：幂等（idempotency）**。流式连接断了会被重连，重连意味着同一次用户意图可能被“重试”。如果你的工具是有副作用的写操作，重试一次就多写一次——灾难。给每个工具调用带上**幂等键（idempotency key）**，服务端对同一键只生效一次，这是流式 Agent 上生产的硬门槛。

> ❝
>
> 这三道坎里，幂等最容易被忽视，也最致命。我建议在 Agent 的工具层统一加一层“幂等中间件”，而不是让每个工具自己记得去判重。

## **9. 框架怎么做的：LangGraph / Vercel AI SDK / Anthropic**

到了落地层，主流框架已经把流式封装得相当完善，不用从零手搓 SSE。

**LangGraph** 提供一组流式模式（stream mode）：`values` 在每个步骤后吐出完整状态快照；`updates` 只吐这一步的增量；`messages` 把图里任意 LLM 调用的 token 按“（token, 元数据）”流出来，适合做打字机效果；`custom` 让你在节点里用 `get_stream_writer()` 发任意自定义事件（进度、中间结果都行）；另外还有 `checkpoints`、`tasks`、`debug`。新版（v2）把这些统一成了 `StreamPart` 字典，字段是 `type` / `ns` / `data`，多个模式还能用列表组合着开。

**Vercel AI SDK** 的 `streamText()` 是核心原语，调一下 `result.toDataStreamResponse()` 就把流变成前端能直接吃的 HTTP 响应，SSE 头、重连、分块全包了；前端用 `useChat` 这个 hook 管理消息历史、流式更新和加载态。它最大的价值是“厂商无关”——换 OpenAI、Anthropic 还是 Gemini，只改一行 model 构造。

**Anthropic** 官方 SDK 直接暴露那套命名 SSE 事件（`message_start`、`content_block_delta` 等），想精细控制工具流式（比如实时显示正在拼的参数）就走原始事件，想省事就用 `client.messages.stream()` 的高层封装。

| 框架 / 厂商   | 流式入口                                  | 特色能力                      | 适合场景               |
| :------------ | :---------------------------------------- | :---------------------------- | :--------------------- |
| LangGraph     | `stream()` + stream mode                  | `custom` 自定义事件、状态快照 | 复杂多步图、要进度上报 |
| Vercel AI SDK | `streamText()` + `toDataStreamResponse()` | 厂商无关、`useChat` 客户端    | JS/TS 全栈、快速搭聊天 |
| Anthropic SDK | `messages.stream()`                       | 原始命名事件、工具参数增量    | 要精细控制工具流式     |

## **10. 一个能跑的流式 Agent Demo**

下面这个 Demo 只用 Python 标准库，起一个 SSE 服务，模拟一个“会思考、会调工具、会逐字回答”的 Agent。把工具换成你自己的真实实现，它就是一个最小可用的流式 Agent 骨架。

完整文件见 `streaming_agent_demo.py`，默认直接打印模拟事件流（方便你本地验证逻辑），加 `--serve` 参数则启动 HTTP SSE 服务，用 `curl -N http://127.0.0.1:8000/stream?q=你好` 就能看到流式输出。

核心思路就三步：事件发射器把每一步包成字典；Agent 循环依次发 `thinking` / `tool_call` / `tool_result` / `token`；SSE 处理器把这些字典按 `data: {json}\n\n` 的格式往外推。下面截取最关键的 SSE 推送部分：

```
def sse_pack(event: dict) -> str:
    return f"id: {event['seq']}\nevent: agent\ndata: {json.dumps(event, ensure_ascii=False)}\n\n"

class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if not self.path.startswith("/stream"):
            self.send_error(404); return
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("X-Accel-Buffering", "no")
        self.end_headers()
        q = parse_qs(urlparse(self.path).query).get("q", ["你好"])[0]
        for ev in run_agent(q):           # run_agent 是生成器，逐条 yield 事件
            try:
                self.wfile.write(sse_pack(ev).encode("utf-8")); self.wfile.flush()
            except (BrokenPipeError, ConnectionResetError):
                break                     # 客户端断开，安全退出，不报错
```

注意末尾那个 `except`——客户端中途关掉页面时，服务端继续 `write` 会抛 `BrokenPipeError`，不捕获的话日志里全是红字。这是流式服务上线必处理的细节。

## **11. 写在最后**

流式交互这件事，技术上不神秘，工程上很琐碎，但它是 Agent 产品观感的分水岭。我把它浓缩成一张上线前自查清单：

- **TTFT**：最终答案有没有在几百毫秒内开始露头？
- **过程可见**：工具调用、进度有没有流出来，而不是一片空白？
- **思考可见**：推理模型的思考过程有没有单独、可控地流出来？
- **可打断**：用户能不能中途停、中途改、危险动作能不能前置确认？
- **结构化流式**：JSON / 表单有没有骨架先行、边生成边渲染？
- **重连**：断线后靠 `Last-Event-ID` 续上了吗？`X-Accel-Buffering` 关了吗？
- **背压与幂等**：客户端慢的时候服务端没崩吧？重试不会重复写吧？

把这几条都过一遍，你的 Agent 就不再是“能跑通的 demo”，而是“用户愿意一直用的产品”。这行干久了你会发现，Agent 的护城河从来不在模型多聪明，而在这些让用户“感觉顺畅”的工程细节里。

把 Agent 当普通 API 调，用户盯着转圈圈等上十几秒，体验上和“程序崩了”没什么两样。

AI Agent让人等得想砸键盘？流式交互与实时体验的工程实战

本文深入解析Agent流式交互的工程实践，指出“能跑通”不等于“能用”。重点阐述三层流式（Token/步骤/动作）、SSE底层机制、可打断设计、结构化输出流式化及生产级重连/背压/幂等方案，并对比LangGraph、Vercel AI SDK等框架实现，助力Agent从Demo迈向可用产品。

人工智能

后端

架构设计

流式交互是Agent从Demo到可交付产品的分水岭，通过SSE将TTFT压缩至毫秒级，实现Token、步骤、动作三层流式输出。关键策略包括：工具调用过程可视化、可打断机制、结构化输出骨架先行、推理思考流式化，并解决重连、背压、幂等三大生产难题。

负载均衡

代码解释

中间件

数据库

服务器

LangGraph

WebSocket

DeepSeek

Python

Agent

JSON

2026上云采购 | AI焕新·智启新局

lexiang

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

AI Agent让人等得想砸键盘？流式交互与实时体验的工程实战-腾讯云开发者社区-腾讯云

AI Agent让人等得想砸键盘？流式交互与实时体验的工程实战

AI Agent让人等得想砸键盘？流式交互与实时体验的工程实战

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐