HelloAgents 流式输出SSE教程让智能体响应实时流式返回的完整实现【免费下载链接】HelloAgentsA agent framework based on the tutorial hello-agents项目地址: https://gitcode.com/gh_mirrors/he/HelloAgentsHelloAgents 是一个生产级多智能体框架其中的流式输出Streaming / SSE能力可以让智能体的回复像打字机一样实时逐字返回不再让用户干等 5~10 秒的白屏。本教程带你从零理解 SSE 协议到用 HelloAgents 的arun_stream()一行代码开启流式模式再结合 FastAPI 浏览器 EventSource 完成一个完整的前后端实时聊天方案。为什么需要流式输出普通模式下LLM 必须生成完整个回答才会一次性返回用户体验就是转圈等待模式首字延迟总延迟用户体验非流式5-10s5-10s 干等流式输出200-500ms5-10s⚡ 实时逐字显示流式输出把首字延迟从秒级降到百毫秒级是聊天机器人、代码分析、报告生成类应用的标准体验。什么是 SSE它和 WebSocket 有什么区别SSEServer-Sent Events是一种基于 HTTP 的服务端单向推送协议格式简单浏览器原生支持event: llm_chunk data: {type: llm_chunk, data: {chunk: 你好}}对比项SSEWebSocket通信方向单向服务端→客户端双向断线重连✅ 浏览器自动重连❌ 需手动实现实现复杂度低较高适用场景LLM 流式输出、进度推送实时协作、游戏 智能体响应属于服务端单向吐字场景SSE 是最简单直接的选择HelloAgents 的流式模块就是标准 SSE 实现。HelloAgents 流式输出的核心9 种事件类型HelloAgents 的流式能力实现在 hello_agents/core/streaming.py核心数据结构是StreamEvent。一次智能体执行会依次产生以下事件事件类型含义关键字段AGENT_START智能体开始运行inputAGENT_FINISH智能体执行完成result, durationSTEP_START/STEP_FINISH推理步骤开始/结束step, max_stepsTOOL_CALL_START/TOOL_CALL_FINISH工具调用开始/结束tool_name, resultLLM_CHUNKLLM 生成的文本块content, deltaTHINKING思考过程contentERROR发生错误error_type, message每个事件都能通过to_sse()方法一键转成标准 SSE 报文见 to_sse 实现。这意味着后端只需一个 for 循环就能把智能体内部所有动态思考、工具调用、吐字全部透传给前端。快速开始三行代码开启流式输出所有 AgentSimpleAgent、ReActAgent 等都提供arun_stream()异步流式方法import asyncio from hello_agents import ReActAgent, HelloAgentsLLM async def main(): agent ReActAgent(assistant, HelloAgentsLLM()) # 流式执行LLM 生成一个 token 就立即返回 async for event in agent.arun_stream(分析项目结构): if event.type llm_chunk: print(event.data[chunk], end, flushTrue) # 打字机效果 asyncio.run(main())以 ReActAgent.arun_stream 为例它会在每一步推理中实时 yield步骤开始 → LLM 文本块 → 工具调用开始/结束 → 最终答案终端里就能看到边思考边输出的效果。构建 SSE 服务端FastAPI 最小示例用 FastAPI 把事件流转成 HTTP 流式响应只需一个生成器 StreamingResponsefrom fastapi import FastAPI from fastapi.responses import StreamingResponse from hello_agents import ReActAgent, HelloAgentsLLM app FastAPI() agent ReActAgent(assistant, HelloAgentsLLM()) app.post(/chat/stream) async def chat_stream(message: str): async def event_generator(): async for event in agent.arun_stream(message): yield event.to_sse() # 转成 SSE 格式 return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no} )完整可运行版本含 CORS、计算器工具、4 种 Agent 切换见 examples/fastapi_sse_server.py启动方式uvicorn examples.fastapi_sse_server:app --reload前端客户端EventSource 实现打字机效果浏览器原生EventSourceAPI 即可消费 SSE 流按事件名分别监听script const eventSource new EventSource(/chat/stream?message${message}); // LLM 文本块 → 逐字拼接形成打字机效果 eventSource.addEventListener(llm_chunk, (e) { output.innerHTML JSON.parse(e.data).data.chunk; }); // 工具调用 → 显示正在调用工具…状态 eventSource.addEventListener(tool_call_start, (e) { output.innerHTML ${JSON.parse(e.data).data.tool_name} ...; }); // 结束 → 关闭连接 eventSource.addEventListener(agent_finish, () eventSource.close()); /script仓库里自带一个完整的前端 Demo 页面 examples/sse_client.html可直接打开配合上面的服务端体验流式相关的单测参考 tests/test_llm_streaming.py。生产环境最佳实践实践做法作用错误处理捕获异常后发送ERROR事件而非断开连接前端可优雅展示错误信息超时控制用asyncio.wait_for包裹事件流防止长请求挂死心跳保活每 30 秒发送: heartbeat\n\n注释行防止 Nginx/代理超时断连事件过滤只转发LLM_CHUNK/TOOL_CALL_*降低前端带宽与渲染压力反缓冲响应头加X-Accel-Buffering: no禁用 Nginx 缓冲保证逐字推送⚠️ 注意X-Accel-Buffering: no是关键配置——很多流式输出变成一次性返回的坑都出在反向代理把响应缓冲了。常见问题 FAQQ如何取消流式请求前端调用eventSource.close()即可断开连接。Q连接中断怎么办EventSource具备自动重连机制监听onerror即可感知中断eventSource.onerror () console.log(连接中断自动重连...);Q流式输出有延迟吗几乎没有LLM 生成 token → 立即发送网络传输 10ms浏览器渲染 5ms首字 200-500ms 内可见。相关资源官方流式输出文档docs/streaming-sse-guide.md流式事件源码hello_agents/core/streaming.pySSE 服务端示例examples/fastapi_sse_server.py前端 SSE 客户端examples/sse_client.html异步 Agent 指南docs/async-agent-guide.md可观测性追踪流式执行docs/observability-guide.md掌握arun_stream()to_sse()这两个入口你就拥有了 HelloAgents 完整的实时流式响应能力——剩下的只是把创意变成产品 。【免费下载链接】HelloAgentsA agent framework based on the tutorial hello-agents项目地址: https://gitcode.com/gh_mirrors/he/HelloAgents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
