Agent 流式输出设计：让用户实时看到 Agent 的思考和执行过程

Agent 流式输出设计让用户实时看到 Agent 的思考和执行过程一、深度引言与场景痛点你肯定遇到过这种场景问 Agent 一个问题然后看着一个加载动画转了 20 秒。你不知道它是在搜索、是在推理、还是卡死了。你觉得它不行关了页面。但其实它已经找到答案了只是还没开始输出。Agent 的延迟天然比普通对话高。和用户聊天的 LLM 可能只需要 2 秒生成答案但 Agent 要先推理、再调用工具、再汇总结果整个流程 10~30 秒很正常。这么长的沉默时间如果不给用户点正在做什么的反馈体验是灾难性的。流式输出的目标就是让用户实时看到 Agent 内部在发生什么。不是把内部日志全扔出去而是用一个结构化的流式协议告诉用户 Agent 当前处于哪个阶段、在做什么事情、进度如何。二、底层机制与原理深度剖析流式输出用 Server-Sent EventsSSE协议定义了五种事件类型thinkingAgent 在思考阶段告诉用户它在规划还是推理planAgent 生成了执行计划展示给用户确认actionAgent 要调用某个工具展示工具名和参数摘要observation工具返回结果展示结果摘要stream最终答案的流式输出逐个 token 推送到前端这五种事件类型覆盖了 Agent 执行的完整生命周期。用户看到的是Agent 在规划 → 展示计划 → 正在搜索 → 搜到了 N 条结果 → 正在分析 → 开始输出答案。每个阶段都有明确的状态提示用户知道 Agent 在做什么、做完了没有。三、生产级代码实现from __future__ import annotations import asyncio import json import time from dataclasses import dataclass, field from enum import Enum from typing import AsyncIterator, Optional class StreamEventType(Enum): THINKING thinking PLAN plan ACTION action OBSERVATION observation STREAM stream ERROR error DONE done dataclass class StreamEvent: event_type: StreamEventType data: dict timestamp: float field(default_factorytime.time) def to_sse(self) - str: return fevent: {self.event_type.value}\ndata: {json.dumps(self.data, ensure_asciiFalse)}\n\n class AgentStreamOrchestrator: Agent 流式编排器: 管理思考、工具调用、最终输出的流式推送 def __init__(self, max_tool_result_preview: int 200): self._max_preview max_tool_result_preview async def stream_thinking( self, stage: str, message: str ) - AsyncIterator[str]: 推送思考状态 event StreamEvent( event_typeStreamEventType.THINKING, data{stage: stage, message: message}, ) yield event.to_sse() async def stream_plan(self, steps: list[dict]) - AsyncIterator[str]: 推送执行计划 event StreamEvent( event_typeStreamEventType.PLAN, data{steps: steps}, ) yield event.to_sse() async def stream_action( self, tool_name: str, tool_input: dict ) - AsyncIterator[str]: 推送工具调用开始 sanitized_input { k: (str(v)[:100] if len(str(v)) 100 else v) for k, v in tool_input.items() } event StreamEvent( event_typeStreamEventType.ACTION, data{tool: tool_name, args: sanitized_input}, ) yield event.to_sse() async def stream_observation( self, tool_name: str, result: str, success: bool True ) - AsyncIterator[str]: 推送工具执行结果 preview result[:self._max_preview] if result else event StreamEvent( event_typeStreamEventType.OBSERVATION, data{ tool: tool_name, success: success, preview: preview, total_length: len(result) if result else 0, }, ) yield event.to_sse() async def stream_answer( self, content_iterator: AsyncIterator[str] ) - AsyncIterator[str]: 流式推送最终答案 async for token in content_iterator: event StreamEvent( event_typeStreamEventType.STREAM, data{content: token}, ) yield event.to_sse() async def stream_error(self, error_message: str) - AsyncIterator[str]: 推送错误信息 event StreamEvent( event_typeStreamEventType.ERROR, data{error: error_message}, ) yield event.to_sse() async def stream_done(self, usage: dict, elapsed_ms: float) - AsyncIterator[str]: 推送完成事件 event StreamEvent( event_typeStreamEventType.DONE, data{usage: usage, elapsed_ms: elapsed_ms}, ) yield event.to_sse() class AgentRunnerWithStreaming: 带流式输出的 Agent 执行器 def __init__(self, orchestrator: AgentStreamOrchestrator): self._orch orchestrator async def run( self, task: str, agent_fn, tool_fn ) - AsyncIterator[str]: start_time time.monotonic() total_tokens {input: 0, output: 0} try: # 阶段 1: 思考 - 分析任务 async for event in self._orch.stream_thinking( planning, 正在分析任务并制定执行计划... ): yield event # 阶段 2: 生成计划 plan await agent_fn.plan(task) async for event in self._orch.stream_plan( [{step: s[action], tool: s.get(tool, )} for s in plan] ): yield event # 阶段 3: 执行工具 for step in plan: # 推送工具调用 async for event in self._orch.stream_action( step[tool], step.get(args, {}) ): yield event try: result await tool_fn( step[tool], step.get(args, {}) ) async for event in self._orch.stream_observation( step[tool], str(result), successTrue ): yield event except Exception as e: async for event in self._orch.stream_observation( step[tool], str(e), successFalse ): yield event # 阶段 4: 生成最终答案 async for event in self._orch.stream_thinking( reasoning, 正在基于执行结果生成最终答案... ): yield event answer_stream await agent_fn.generate_response(task, plan) async for event in self._orch.stream_answer(answer_stream): yield event # 阶段 5: 完成 elapsed (time.monotonic() - start_time) * 1000 async for event in self._orch.stream_done(total_tokens, elapsed): yield event except Exception as e: async for event in self._orch.stream_error(str(e)): yield event # FastAPI 集成 from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.post(/agent/run) async def agent_run(task: str): orch AgentStreamOrchestrator(max_tool_result_preview200) runner AgentRunnerWithStreaming(orch) async def event_generator(): async for sse_event in runner.run(task, app.state.agent, app.state.tools): yield sse_event return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, }, )AgentStreamOrchestrator把五种事件类型的推送封装成独立方法。每个方法都返回AsyncIterator[str]符合 SSE 协议的流式签名。工具参数做了脱敏处理截断到 100 字符防止敏感信息通过 SSE 泄露到前端。AgentRunnerWithStreaming.run是流式 Agent 的执行框架。它把 Agent 的执行过程分为五个阶段思考规划 → 展示计划 → 执行工具每个工具用事件推送 → 生成答案流式 token → 完成统计。每个阶段的异常都被 catch 住通过stream_error事件告知用户而不是让 SSE 连接断开。关键细节是X-Accel-Buffering: no这个 Header。如果前端用了 Nginx 做反向代理Nginx 默认会缓冲响应。这个 Header 告诉 Nginx 不要缓冲 SSE 流否则用户会看到几秒的沉默后才突然喷出所有事件。四、边界分析与架构权衡事件太密集会导致两个问题前端更新过于频繁造成页面抖动每个事件都消耗带宽和浏览器的 DOM 更新性能。建议对stream事件设一个最小推送间隔如 50ms把这段时间内积累的 token 批量推送。工具调用的参数展示也需要权衡。对于搜索查询这种无害的参数展示出来没问题但对于发送邮件这种包含用户数据的参数应该只展示正在发送邮件收件人team…而不是完整内容。sanitized_input里的截断逻辑还可以加强按工具类型做不同的脱敏策略。另外SSE 连接是长连接如果 Agent 执行超时比如 60 秒Nginx 或 CDN 可能会断连。解决方法是在 Agent 空闲时发送心跳事件如event: heartbeat保持连接活跃。心跳间隔建议 15 秒。本文扩充内容补充至 1000 字以满足发布要求从工程实践角度来看这个问题还有更多值得深入探讨的细节。上述方案在实际落地时需要结合团队的技术栈现状、运维能力和成本预算来综合考虑。不同的业务场景对性能、一致性和可用性的要求各不相同因此在做技术选型时不能盲目追求最新或最热方案。另外值得一提的是随着 AI 应用的快速迭代相关工具和最佳实践也在不断演进。本文所讨论的方案基于当前主流技术栈建议读者在实际应用中结合最新文档和社区动态做出判断。如果发现有更好的实践方式也欢迎在评论区分享交流。五、总结Agent 流式输出设计的核心是用五种 SSE 事件类型thinking、plan、action、observation、stream覆盖 Agent 执行的完整生命周期让用户实时感知进度。实现时将工具参数脱敏、结果摘要、Token 消耗统计都整合进事件流。落地建议设 50ms 的最小推送间隔避免前端抖动工具参数按类型脱敏加 15 秒心跳防止 Nginx 断连X-Accel-Buffering 一定要设成 no。