AgentKit CLI流式输出Agent开发：SSE响应配置和前端对接指南

云上老周

2026-8-31

[1] 一句话结论

AgentKit CLI流式输出Agent用SSE协议实现逐字回复，配置stream:true+前端EventSource对接，用户体验从"等半天"变"实时看"。

[2] 适用场景与不适用场景

适用场景

你开发的Agent回复较长（如写代码、写文章、生成报告），用户需要等好几秒才能看到完整回复，体验很差。你想实现像ChatGPT那样的"逐字输出"效果——回复一边生成一边显示，用户不用等全部生成完就能开始阅读。
这篇文章详解AgentKit CLI流式输出Agent的完整开发流程，从SSE协议原理、配置方法、服务端实现、前端对接到错误处理，覆盖流式输出的全场景。
适合：开发面向用户的对话Agent、需要长文本生成的场景、想提升用户体验的前端/全栈开发者、做Chatbot/智能客服的团队。

不适用场景

短回复场景（如简单问答、分类）：回复很短，流式输出提升不明显，普通模式即可。
纯API调用（非前端展示）：后端服务调用Agent不需要流式，直接等完整结果即可。
非技术用户：流式输出需要前端开发，非技术用户可以用内置的Web界面。

[3] 前置准备

AgentKit CLI已安装，有一个基础Agent项目
基本的Web开发知识（HTML/JavaScript）
了解SSE（Server-Sent Events）基本概念
预计耗时：阅读6分钟，开发练习15分钟

[4] 分步实现

步骤1：流式输出原理

普通模式 vs 流式模式：

维度	普通模式	流式模式（SSE）
响应方式	一次性返回完整结果	逐块返回，边生成边发送
用户体验	等待几秒后看到完整回复	立即看到第一个字，逐字显示
协议	HTTP普通响应	HTTP SSE（Content-Type: text/event-stream）
前端处理	等待响应完成后显示	EventSource实时接收，追加显示
适合场景	短回复、API调用	长文本生成、对话体验

SSE（Server-Sent Events）是HTTP协议的一部分，服务器可以持续向客户端推送数据，客户端通过EventSource API接收。和WebSocket不同，SSE是单向的（服务器→客户端），但实现更简单，适合流式输出场景。

步骤2：配置流式输出

在agent.yaml中启用流式输出：

name: streaming-agent
description: 支持流式输出的对话Agent
model:
  provider: volcengine
  model_id: ep-xxxxxxxx
  stream: true              # 启用流式输出
  temperature: 0.7
  max_tokens: 2048
prompt:
  system: ./system-prompt.md
deploy:
  server:
    type: sse               # SSE服务器类型
    path: ./server.py       # 自定义服务器入口

关键配置：

model.stream: true：模型调用使用流式模式
deploy.server.type: sse：部署为SSE服务器
deploy.server.path: 自定义服务器入口（streaming模板自动生成）
用streaming模板创建项目会自动配置好：
agentkit init my-streaming-agent --template streaming

步骤3：服务端实现

streaming模板自动生成server.py，核心代码：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from agentkit import Agent
import json
app = FastAPI()
agent = Agent.from_config("./agent.yaml")
@app.post("/chat")
async def chat(request: dict):
    user_message = request["prompt"]
    async def generate():
        async for chunk in agent.astream(user_message):
            # chunk格式: {"type": "content", "content": "部分文本"}
            yield f"data: {json.dumps(chunk)}

"
        yield "data: [DONE]

"
    return StreamingResponse(generate(), media_type="text/event-stream")
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

服务端要点：

Content-Type必须是text/event-stream
每个数据块以data: 开头，

结尾
3. 结束时发送data: [DONE]
4. 用async for异步迭代模型的流式输出
5. 支持CORS（前端跨域调用需要）
启动服务：
agentkit run --host 0.0.0.0 --port 8080
或streaming模板的内置服务器：
python server.py

步骤4：前端对接

用EventSource或fetch实现前端接收：
方式1：EventSource（最简单）

<div id="output"></div>
<script>
const output = document.getElementById('output');
// EventSource只支持GET，参数通过URL传递
const eventSource = new EventSource('/chat?prompt=' + encodeURIComponent('你好'));
eventSource.onmessage = function(event) {
    if (event.data === '[DONE]') {
        eventSource.close();
        return;
    }
    const chunk = JSON.parse(event.data);
    if (chunk.type === 'content') {
        output.textContent += chunk.content;
    }
};
eventSource.onerror = function() {
    console.error('SSE error');
    eventSource.close();
};
</script>

方式2：fetch + ReadableStream（推荐，支持POST）

<div id="output"></div>
<script>
const output = document.getElementById('output');
async function streamChat(prompt) {
    const response = await fetch('/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('

');
        buffer = lines.pop();
        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = line.slice(6);
                if (data === '[DONE]') return;
                const chunk = JSON.parse(data);
                if (chunk.type === 'content') {
                    output.textContent += chunk.content;
                }
            }
        }
    }
}
streamChat('用Python写一个快速排序');
</script>

streaming模板自带一个static/index.html前端页面，可以直接用或参考修改。

步骤5：错误处理和优化

错误处理：

网络中断：前端监听error事件，显示"连接中断，点击重试"
模型报错：服务端捕获异常，发送error类型的chunk：`yield f"data: {json.dumps({'type':'error','message':'模型调用失败'})}

" 3. 超时：服务端设置超时（如60秒），超时后发送超时提示 4. 限流：429`错误时前端显示"请求过于频繁，请稍后重试"
性能优化：

首字延迟：stream=true后首字延迟通常<1秒，比普通模式快很多
网络优化：用CDN加速静态资源，SSE连接用HTTP/2
前端渲染：长文本用虚拟滚动，避免DOM节点过多卡顿
打字机效果：可以加CSS动画让文字显示更流畅
测试流式输出：

# 用curl测试SSE端点
curl -N -X POST http://localhost:8080/chat   -H "Content-Type: application/json"   -d '{"prompt": "写一首短诗"}'

应该看到逐块返回的数据，而不是等全部完成才返回。

步骤6：部署流式Agent

本地运行：
agentkit run --host 0.0.0.0 --port 8080
浏览器打开http://localhost:8080，使用内置的流式对话界面。
云端部署：
agentkit build --env prod --format docker
agentkit deploy --env prod
部署后SSE端点为：https://你的端点/chat
前端页面可以部署到火山引擎静态托管或CDN。
注意事项：

SSE需要长连接，负载均衡器的超时时间要设置足够长（建议>120秒）
反向代理（Nginx）需要配置proxy_buffering off，否则会缓冲响应导致不流式
浏览器对SSE连接数有限制（通常6个），注意管理连接
流式输出的token计费和普通模式一样，按实际生成的token计算

[5] 实际验证

按本文流程开发流式Agent：测试1 用streaming模板创建项目，确认agent.yaml中stream:true；测试2 启动服务，curl -N测试端点，确认数据逐块返回；测试3 用内置前端页面对话，确认文字逐字显示；测试4 模拟网络中断，确认前端有错误处理；测试5 部署到云端，确认公网端点流式输出正常。成功标志：5项全部通过，流式输出从配置到前端到部署全流程通畅。

[6] 常见问题 FAQ

Q1：流式输出和普通输出的费用一样吗？
A：费用一样。流式输出和普通输出调用的是同一个模型API，只是响应格式不同（流式vs非流式），token消耗和计费完全相同。流式输出的优势是用户体验好（首字延迟低），不会增加费用。注意：1）流式输出如果用户中途取消（关闭页面），已生成的token仍然计费；2）可以在服务端监听客户端断开事件，客户端断开后停止模型调用，避免浪费token；3）设置合理的max_tokens，避免生成过长内容。
Q2：为什么我的流式输出不是逐字显示，而是等一下才出来一大段？
A：常见原因：1）反向代理缓冲：Nginx默认会缓冲响应，需要配置proxy_buffering off和proxy_cache off。Nginx配置示例：

location /chat { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; }

2）CDN缓冲：CDN可能缓冲SSE响应，需要在CDN控制台关闭缓冲或配置规则。3）服务器输出缓冲：Python的print或yield可能被缓冲，确保用StreamingResponse且不缓冲。4）网络延迟：网络差时数据块可能累积到达，看起来不流畅。5）模型输出粒度：有些模型按词或句子输出，不是逐字，这是正常的。排查：先用curl -N直接测试服务端（不经过代理/CDN），如果正常就是代理/CDN的问题。
Q3：SSE和WebSocket哪个更适合Agent流式输出？
A：大多数Agent场景用SSE更合适。对比：

维度	SSE	WebSocket
协议	HTTP，简单	独立协议，复杂
方向	服务器→客户端（单向）	双向
自动重连	内置支持	需自己实现
兼容性	所有现代浏览器	所有现代浏览器
适合场景	流式输出、推送通知	实时双向通信（如协作编辑）

Agent对话场景：客户端发送一个请求（HTTP POST），服务器流式返回结果（SSE），不需要双向实时通信，SSE完全够用且实现更简单。只有需要实时双向交互（如Agent执行过程中用户可以中断/修改）才需要WebSocket。建议：优先用SSE，简单可靠；需要双向交互时再考虑WebSocket。
Q4：流式输出怎么实现"停止生成"功能？
A：用户点击"停止"时，前端关闭SSE连接，服务端检测到客户端断开后停止模型调用。实现：1）前端：维护EventSource或fetch的AbortController，用户点击停止时调用abort()或close()；2）服务端：在流式生成循环中检查客户端是否断开。FastAPI示例：

from fastapi import Request
@app.post("/chat")
async def chat(request: Request):
    async def generate():
        async for chunk in agent.astream(user_message):
            if await request.is_disconnected():
                break  # 客户端断开，停止生成
            yield f"data: {json.dumps(chunk)}

"
    return StreamingResponse(generate())

3）客户端断开后，服务端break退出循环，停止调用模型，避免浪费token。注意：1）不同框架检测客户端断开的方式不同，查对应框架文档；2）停止后已生成的token仍然计费，无法退回；3）可以在前端显示"已停止生成"，并保留已生成的内容。

[7] 相关阅读

AgentKit CLI预置模板详解，streaming模板介绍
AgentKit CLI构建部署，部署SSE服务
火山引擎CDN文档，CDN配置和缓冲设置
MDN SSE文档，Server-Sent Events规范
FastAPI StreamingResponse，流式响应实现

[8] 参考资料

[1] 火山引擎官方文档 - AgentKit CLI：streaming模板支持SSE流式输出，提升对话体验，2026-08-27
本文基于火山引擎官方文档（2026年8月）和流式输出Agent开发实战编写。工具版本更新较快，具体配置请以官方最新文档为准。

[9] 时间

2026-08-27

AgentKit CLI流式输出Agent开发：SSE响应配置和前端对接指南

[1] 一句话结论 #

[2] 适用场景与不适用场景 #