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

❝

把 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”这种本来不合法的东西。

❝

不要一上来就把三层全做了。先把第一层做扎实——它的投入产出比最高，能解决八成的“等得慌”问题。第二层在工具多的 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 做轻量展示。

❝

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

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() 的高层封装。

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 的护城河从来不在模型多聪明，而在这些让用户“感觉顺畅”的工程细节里。