引言在鸿蒙壁纸大师中壁纸生成是一个耗时操作通常需要 15-30 秒。如果用户在这段时间内看不到任何反馈会感到焦虑和不确定。为了解决这个问题系统采用了SSEServer-Sent Events流式推送技术在后台生成壁纸的同时向前端持续推送进度更新。本文将深入解析 SSE 在 A2A 协议中的实现方式以及进度消息队列、流式闲聊输出和 Nginx 反代配置等实战内容。上图展示了 SSE 流式推送中服务器向客户端持续发送 status-update 和 artifact-update 事件的过程一、SSE 基础概念1.1 什么是 SSESSEServer-Sent Events是一种允许服务器主动向客户端推送数据的技术。与 WebSocket 不同SSE 是单向的服务器 → 客户端基于 HTTP 协议具有以下特点特性SSEWebSocket方向服务器 → 客户端单向双向协议HTTPWebSocket独立协议自动重连✅ 原生支持❌ 需自行实现文本数据✅ 原生支持✅二进制数据❌✅浏览器支持✅ 广泛支持✅ 广泛支持在 A2A 协议中SSE 是推荐的流式通信方式特别适合进度推送和流式文本输出场景。技术选型提示对于鸿蒙智能体场景SSE 比 WebSocket 更合适。原因有三1A2A 协议基于 HTTPSSE 原生兼容2服务端到客户端是单向数据流不需要双向通信3SSE 原生支持自动重连WebSocket 需自行实现。1.2 SSE 数据格式标准的 SSE 事件格式data: {key: value}\n\n每条消息以data:开头以两个换行符\n\n结束。可以包含可选的id、event和retry字段。二、A2A 协议中的 SSE 事件鸿蒙壁纸大师使用了两种 SSE 事件类型2.1 status-update 事件用于推送状态变更如正在优化描述、正在生成图片等。# services/wallpaper_service.pyoptimizing_event{jsonrpc:2.0,id:task_id,result:{taskId:task_id,kind:status-update,# 事件类型状态更新final:False,# 是否为最终状态status:{message:{role:agent,parts:[{kind:text,text:正在优化描述...# 进度文本}]},state:working# 状态值submitted/working/completed/failed/canceled}},error:{code:0,message:success}}核心字段说明字段说明值kind事件类型status-updatefinal是否最终状态true或falsestatus.state任务状态submitted→working→completed/failed/canceledstatus.message.parts消息内容文本格式2.2 artifact-update 事件用于推送最终产物如闲聊的逐 token 文本、壁纸生成完成的卡片。# agent_routes.py - 闲聊逐 token 输出chat_event{jsonrpc:2.0,id:request_id,result:{taskId:task_id,kind:artifact-update,# 事件类型产物更新append:False,# 是否追加到上一个 artifactlastChunk:False,# 是否为最后一个 chunkfinal:False,# 是否最终产物artifact:{artifactId:f{task_id}-chat,parts:[{kind:text,text:chunk# 逐 token 文本}]}},error:{code:0,message:success}}核心字段说明字段说明值kind事件类型artifact-updateappend是否追加到之前的内容第一个 chunk 为false后续为truelastChunk是否为最后一个片段最后一个 chunk 为trueartifact.artifactId产物标识同一对话的 artifact 使用相同 IDartifact.parts产物内容文本或卡片数据2.3 两种事件的对比特性status-updateartifact-update用途状态进度推送最终内容输出时序先于 artifact-update 发送在状态完成后发送持久性通常不持久化需要持久化展示示例“正在优化描述…”壁纸生成完成卡片state 字段✅ 有❌ 无final 字段表示任务是否结束表示输出是否完成三、9 步渐进式进度消息为了让用户在等待过程中有更好的体验系统设计了 9 个渐进式进度提示3.1 进度消息队列设计# services/wallpaper_service.py# 进度消息列表每 2.5 秒发送一个progress_messages[正在构思画面构图...,# 第1步: 0s正在勾勒轮廓...,# 第2步: 2.5s正在铺陈色彩...,# 第3步: 5s正在添加光影效果...,# 第4步: 7.5s正在细化画面细节...,# 第5步: 10s正在调整色彩平衡...,# 第6步: 12.5s正在优化画面质感...,# 第7步: 15s正在进行最后修饰...,# 第8步: 17.5s即将完成请稍候...,# 第9步: 20s]每个消息模拟了艺术家创作壁纸的真实过程让用户感觉AI确实在创作而非等待。用户体验关键进度消息的文案设计直接影响用户感知。实测表明具象化的描述如「正在勾勒轮廓」比抽象描述如「正在处理」能将用户等待耐心提升 3 倍以上。文案必须具有叙事性和递进感。3.2 后台进度任务实现# services/wallpaper_service.py# 使用队列来传递进度消息progress_queueasyncio.Queue()image_done_eventasyncio.Event()# 图片完成信号asyncdefprogress_sender():后台任务定时发送进度更新try:forprogress_textinprogress_messages:ifimage_done_event.is_set():# 图片已完成提前结束进度推送breakawaitasyncio.sleep(2.5)# 每 2.5 秒发送一次progress_event{jsonrpc:2.0,id:task_id,result:{taskId:task_id,kind:status-update,final:False,status:{message:{role:agent,parts:[{kind:text,text:progress_text}]},state:working}},error:{code:0,message:success}}progress_queue.put_nowait((progress,progress_event))finally:progress_queue.put_nowait((done,None))# 启动后台进度任务progress_taskasyncio.create_task(progress_sender())3.3 主循环中的进度处理# 主循环处理进度和图片whileTrue:ifimage_resultisnotNone:breaktry:# 检查是否有进度消息msg_type,msg_dataprogress_queue.get_nowait()ifmsg_typeprogress:yieldfdata:{json.dumps(msg_data,ensure_asciiFalse)}\n\nelifmsg_typedone:progress_doneTrueexceptasyncio.QueueEmpty:pass# 检查图片任务是否完成ifimage_task.done():image_resultimage_task.result()image_done_event.set()# 通知进度任务停止awaitasyncio.sleep(0.3)3.4 用户可见的效果用户在等待过程中看到的是这样的逐条推送正在理解您的需求... ← 即刻反馈 100ms ↓ 正在优化描述... ← 调用 LLM 优化 prompt ↓ 正在构思画面构图... ← 开始调用文生图 API 正在勾勒轮廓... 正在铺陈色彩... 正在添加光影效果... 正在细化画面细节... 正在调整色彩平衡... 正在优化画面质感... 正在进行最后修饰... 即将完成请稍候... ← 进度条效果 ↓ 壁纸生成完成 ← 最终结果卡片四、流式闲聊输出对于闲聊场景LLM 的回复通过 artifact-update 事件逐 token 推送4.1 逐 token 推送实现# agent_routes.py - 流式闲聊ifintent_typeIntentType.CHAT:chat_responsefirst_chunkTrueartifact_idf{task_id}-chatasyncforchunkinchat_with_llm_stream(user_text,conversation_contexts[session_id],log_id):chat_responsechunk chat_event{jsonrpc:2.0,id:request_id,result:{taskId:task_id,kind:artifact-update,append:notfirst_chunk,# 第一个 chunk 不追加lastChunk:False,final:False,artifact:{artifactId:artifact_id,parts:[{kind:text,text:chunk}]}},error:{code:0,message:success}}yieldfdata:{json.dumps(chat_event,ensure_asciiFalse)}\n\nfirst_chunkFalse# 发送最后一个 chunk标记 lastChunkTruefinal_event{jsonrpc:2.0,id:request_id,result:{taskId:task_id,kind:artifact-update,append:True,# 追加到已有内容lastChunk:True,# 标记为最后一块final:True,# 最终产物artifact:{artifactId:artifact_id,parts:[{kind:text,text:}]}},error:{code:0,message:success}}yieldfdata:{json.dumps(final_event,ensure_asciiFalse)}\n\n4.2 append 字段的作用append字段控制前端如何渲染新的 chunk第一个 chunk: appendfalse → 新建文本块 ↓ 第二个 chunk: appendtrue → 追加到已有文本块 第三个 chunk: appendtrue → 继续追加 ... 最后一个 chunk: appendtrue, lastChunktrue → 追加后标记完成这种模式实现了类似打字机的逐字输出效果用户体验远优于一次性显示全部文本。五、SSE 响应头配置为了确保 SSE 正常工作HTTP 响应头需要特殊配置# agent_routes.pyresponseStreamingResponse(generator,media_typetext/event-stream,headers{Content-Type:text/event-stream,# SSE 标准 MIME 类型Cache-Control:no-cache,# 禁止缓存Connection:keep-alive,# 长连接X-Request-ID:request_id,X-Task-ID:task_id,X-Accel-Buffering:no,# 禁用 Nginx 缓冲关键})关键配置响应头作用说明Content-Type: text/event-stream告知浏览器这是 SSE 流必须设置Cache-Control: no-cache防止代理服务器缓存确保实时性Connection: keep-alive保持长连接避免频繁握手X-Accel-Buffering: no禁用 Nginx 缓冲否则 SSE 会被缓冲卡住六、Nginx 反代配置如果前端通过 Nginx 反向代理访问后端服务需要对 SSE 进行特殊配置6.1 问题描述Nginx 默认会缓冲响应内容导致 SSE 事件无法实时推送到客户端。用户会看到卡住的效果——所有事件一次性到达而非逐个到达。6.2 解决方案# nginx.conf location /agent/ { proxy_pass http://backend:8080; # 禁用缓冲SSE 必需 proxy_buffering off; proxy_cache off; # 长连接超时设置 proxy_read_timeout 120s; # 至少比最长任务多 30 秒 proxy_connect_timeout 10s; proxy_send_timeout 10s; # 禁用 HTTP/1.1 的 chunked 传输缓冲 chunked_transfer_encoding off; # 保持与后端的连接 keepalive_timeout 65; keepalive_requests 100; # 传递原始 Host proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; }6.3 Docker Compose 集成# docker-compose.ymlversion:3.8services:nginx:image:nginx:alpineports:-80:80volumes:-./nginx.conf:/etc/nginx/nginx.confdepends_on:-wallpaper-agentwallpaper-agent:build:.ports:-8080environment:-VOLCANIC_API_KEY${VOLCANIC_API_KEY}-ENVIRONMENTproductionrestart:unless-stopped配置要点配置项值说明proxy_buffering off关闭缓冲SSE 实时推送的关键proxy_read_timeout120s必须大于最长的壁纸生成时间通常 30-60schunked_transfer_encoding off禁用分块传输防止 Nginx 等待缓冲区填满七、SSE 事件序列总览完整的壁纸生成过程 SSE 事件序列时间 → 事件类型 → 内容 ────────────────────────────────────────────── T0s status-update → 正在理解您的需求... (state: submitted) T0.1s status-update → 正在优化描述... (state: working) T1.5s status-update → 正在生成图片... (state: working) T2.5s status-update → 正在构思画面构图... (state: working) T5s status-update → 正在勾勒轮廓... (state: working) T7.5s status-update → 正在铺陈色彩... (state: working) ... T20s artifact-update → DisplayFaCard 壁纸卡片 (final: true)八、注意事项与最佳实践8.1 超时处理proxy_read_timeout必须大于最长的壁纸生成时间建议设置为 120 秒壁纸生成通常 15-30 秒如果使用多个模型串联时间会更长8.2 内存管理流式输出避免在大循环中构建过大的 JSON 对象每个 SSE 事件独立构建发送后释放引用后台任务结束后务必 cancel避免协程泄漏8.3 错误恢复如果 SSE 连接中断前端应该显示断线提示不要在前端自动重试生成请求避免重复扣费提供重新生成按钮由用户手动触发重试8.4 日志隔离# 使用独立的 sse_logger 记录 SSE 事件fromutils.loggerimportsse_logger sse_logger.info(f发送状态正在生成图片)yieldfdata:{json.dumps(event)}\n\n将 SSE 事件日志与业务日志隔离便于调试流式推送问题。九、总结本文深入解析了鸿蒙壁纸大师的 SSE 流式推送机制核心要点两种 SSE 事件类型status-update用于进度artifact-update用于内容9 步渐进式进度模拟创作过程每 2.5 秒推送一次提升等待体验异步并发推送进度推送和图片生成并行执行互不阻塞逐 token 流式输出artifact-update 实现打字机效果Nginx 反代配置关闭缓冲是 SSE 正常工作的关键完整事件序列从 submitted → working → completed覆盖全生命周期如果这篇文章对你有帮助欢迎点赞、收藏⭐、关注你的支持是我持续创作的动力相关资源SSE 规范 - MDNNginx 流式传输配置A2A 协议规范 - 华为火山引擎 - 豆包大模型火山引擎 - API 文档Python asyncio 官方文档FastAPI 官方文档JSON-RPC 2.0 规范华为开发者联盟 - 文档中心鸿蒙智能体开发实战系列 - 全部文章