企业官网建设流程全解析

Function Calling 工具调用的流式输出（Streaming）：基于 SSE 的实时进度回显

在大模型智能体（Agent）处理复杂多步任务（如：搜索全网知识库、调用代码解释器运行数据分析、生成报表并导出）时，整个后台工具链的执行通常需要持续5 到 15 秒。

如果系统采用传统的“黑盒等待模式”：

• 用户在前端界面点击发送后，整个聊天窗口陷入长达 15 秒的死寂白屏或仅有一个旋转的 Loading 菊花图；
• 用户的心理焦虑感会随时间急剧攀升，误以为系统已经卡死或报错，频繁刷新页面导致重复提交；
• 产品体验极其糟糕！

现代顶级 AI 应用（如 ChatGPT、Claude Artifacts、Cursor）的标配体验是：通过 Server-Sent Events（SSE）流式协议，将后台工具调用的“思考状态、调用哪个工具、入参是什么、执行进度百分比、工具产出快照”实时、打字机式地逐步推送到前端界面！

今天我们深入拆解基于Go / Python + SSE（Server-Sent Events）标准协议 + 统一 Event Frame 结构的生产级工具调用实时进度回显方案。

一、流式工具调用 SSE 事件帧驱动模型全景图

sequenceDiagram participant UserUI as 前端 Web / 终端界面 (EventSource 监听) participant Gateway as Go / Python SSE 网关 participant Agent as 大模型推理 & Tool 执行引擎 UserUI->>Gateway: 1. 发起流式请求: POST /api/v1/chat/stream Gateway-->>UserUI: 2. 建立 SSE 长连接: HTTP 200 OK (Content-Type: text/event-stream) Agent->>Gateway: 3. 决定调用工具: query_database(sql="SELECT...") Gateway-->>UserUI: 4. 推送事件 event: tool_start -> data: {"tool": "DBQuery", "step": 1} Note over UserUI: 前端高亮显示: "正在查询数据库数据..." (折叠卡片) Agent->>Gateway: 5. 工具执行产生中间进度 (完成 50%) Gateway-->>UserUI: 6. 推送事件 event: tool_progress -> data: {"progress": 50, "msg": "已扫描 5000 条记录..."} Agent->>Gateway: 7. 工具执行完毕，大模型开始流式输出最终分析 Gateway-->>UserUI: 8. 推送事件 event: text_delta -> data: {"delta": "根据上述财务数据..."} Gateway-->>UserUI: 9. 推送结束事件 event: done -> data: [DONE]

二、生产级 SSE Event Frame 协议规范与结构定义

标准的 SSE 流式帧必须遵循event: <name>\ndata: <json>\n\n规范：

from pydantic import BaseModel, Field from typing import Optional, Any, Literal # 标准化的 SSE 统一通信帧结构 class StreamEventFrame(BaseModel): event_type: Literal[ "thought_delta", # 大模型思考过程/思维链 Token "tool_call_start", # 工具调用开始（包含工具名与参数快照） "tool_progress", # 工具内部执行中间进度 (0~100) "tool_call_done", # 工具执行成功并返回数据 "text_delta", # 最终回答的文本 Token 流 "error", # 异常告警 "done" # 全流程结束标识 [DONE] ] payload: Any = None step_id: Optional[str] = None timestamp: float = Field(default_factory=time.time) def to_sse_format(self) -> str: """格式化为合规的 SSE 文本行""" json_data = json.dumps(self.model_dump(), ensure_ascii=False) return f"event: {self.event_type}\ndata: {json_data}\n\n"

三、生产级 Python / FastAPI 工具流式事件发射器实现

import json import time from typing import AsyncGenerator from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() async def generate_agent_stream_pipeline(user_prompt: str) -> AsyncGenerator[str, None]: """ 核心流式生成器：将思考、工具调用与结果逐步发射至前端 """ print(f"[*] 接收到用户流式请求: {user_prompt}") # 1. 发射思考过程 Token yield StreamEventFrame( event_type="thought_delta", payload="正在分析用户意图，检测到需要提取 2026 年 Q3 销售数据..." ).to_sse_format() time.sleep(0.3) # 2. 发射工具调用开始事件 yield StreamEventFrame( event_type="tool_call_start", payload={ "tool_name": "execute_bi_query", "arguments": {"quarter": "2026Q3", "region": "East_China"} }, step_id="step_01" ).to_sse_format() time.sleep(0.5) # 3. 发射工具内部执行进度 (模拟耗时 1.5 秒的 SQL 查询) for p in [25, 60, 95]: yield StreamEventFrame( event_type="tool_progress", payload={"progress": p, "message": f"正在聚合华东大区节点数据... ({p}%)"}, step_id="step_01" ).to_sse_format() time.sleep(0.4) # 4. 发射工具执行完毕事件 (带产出结果预览) yield StreamEventFrame( event_type="tool_call_done", payload={"summary": "成功获取 12,500 条订单记录，总销售额 4,850 万元。"}, step_id="step_01" ).to_sse_format() time.sleep(0.3) # 5. 发射最终大模型回复的打字机流 final_text = "根据最新的商业智能数据分析，2026年第三季度华东大区总销售额达到 4,850 万元，同比增长 18.5%，表现极其稳健。" for char in final_text: yield StreamEventFrame( event_type="text_delta", payload=char ).to_sse_format() time.sleep(0.03) # 模拟 30ms 一个字符的极度丝滑打字机体验 # 6. 发射结束帧 yield StreamEventFrame(event_type="done", payload="[DONE]").to_sse_format() @app.post("/api/v1/chat/stream") async def chat_stream_endpoint(request_body: dict): prompt = request_body.get("prompt", "") return StreamingResponse( generate_agent_stream_pipeline(prompt), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲区，保证逐字节秒级透传！ } )

四、生产治理防坑三大黄金铁律

1. Nginx 反向代理必须配置proxy_buffering off;与X-Accel-Buffering: no：
   • 默认情况下，Nginx 会等待内部缓冲区凑满 4KB 之后才向前端刷盘数据；
   • 如果不关闭缓冲，流式效果会被 Nginx 彻底吞噬，前端依然会卡死数秒后一次性收到一坨数据！
2. 保持定时心跳保活（Ping Frame）：在长耗时工具执行期间，若超过 15 秒没有任何数据产出，后台必须每隔 5 秒发射一个轻量注释帧（如: keep-alive\n\n），防止中间云厂商负载均衡器（ALB/SLB）静默掐断空闲 TCP 连接；
3. 前端防抖与 Markdown 增量渲染：前端在接收text_delta时，使用虚拟 DOM 增量更新，防止高频字符推送导致浏览器渲染卡顿。

把 SSE 流式工具回显做透，复杂的后台 Agent 多步协作才能转化为用户眼前极具科技感与确定性的极致交互体验。