当前位置：首页 > news >正文

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

news 2026/7/16 19:30:49

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

一、深度引言与场景痛点

你肯定遇到过这种场景：问 Agent 一个问题，然后看着一个加载动画转了 20 秒。你不知道它是在搜索、是在推理、还是卡死了。你觉得它不行，关了页面。但其实它已经找到答案了，只是还没开始输出。

Agent 的延迟天然比普通对话高。和用户聊天的 LLM 可能只需要 2 秒生成答案，但 Agent 要先推理、再调用工具、再汇总结果，整个流程 10~30 秒很正常。这么长的沉默时间，如果不给用户点"正在做什么"的反馈，体验是灾难性的。

流式输出的目标就是让用户实时看到 Agent 内部在发生什么。不是把内部日志全扔出去，而是用一个结构化的流式协议，告诉用户 Agent 当前处于哪个阶段、在做什么事情、进度如何。

二、底层机制与原理深度剖析

流式输出用 Server-Sent Events（SSE）协议，定义了五种事件类型：

thinking：Agent 在思考阶段，告诉用户它在规划还是推理
plan：Agent 生成了执行计划，展示给用户确认
action：Agent 要调用某个工具，展示工具名和参数摘要
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_factory=time.time) def to_sse(self) -> str: return f"event: {self.event_type.value}\ndata: {json.dumps(self.data, ensure_ascii=False)}\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_type=StreamEventType.THINKING, data={"stage": stage, "message": message}, ) yield event.to_sse() async def stream_plan(self, steps: list[dict]) -> AsyncIterator[str]: """推送执行计划""" event = StreamEvent( event_type=StreamEventType.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_type=StreamEventType.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_type=StreamEventType.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_type=StreamEventType.STREAM, data={"content": token}, ) yield event.to_sse() async def stream_error(self, error_message: str) -> AsyncIterator[str]: """推送错误信息""" event = StreamEvent( event_type=StreamEventType.ERROR, data={"error": error_message}, ) yield event.to_sse() async def stream_done(self, usage: dict, elapsed_ms: float) -> AsyncIterator[str]: """推送完成事件""" event = StreamEvent( event_type=StreamEventType.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), success=True ): yield event except Exception as e: async for event in self._orch.stream_observation( step["tool"], str(e), success=False ): 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_preview=200) 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_type="text/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 流，否则用户会看到几秒的沉默后才突然喷出所有事件。