AI Agents · 课时

流式响应(SSE)

通过服务器推送事件逐个令牌地传输 LLM 输出,从而实现响应迅速的界面和进度提示。

第 3 / 4 课13 个步骤

流式响应(SSE) 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

为什么要流式传输

不使用流式传输时,您必须等到完整响应生成后才能显示任何内容。假设一个包含 50 个令牌的响应需要 5 秒,用户就要在 5 秒内什么都看不到。

流式传输会在令牌到达时立即显示它们——总耗时不变,但用户体验会感觉即时完成。

服务器发送事件(SSE)

OpenAI 和 Anthropic 都通过 SSE 进行流式传输。SSE 是一种单向 HTTP 协议,服务器会以 data: {...}\n\n 格式推送事件。

大多数软件开发工具包都会通过迭代器接口隐藏 SSE 的细节。

使用 OpenAI 进行流式传输

设置 stream=True,然后迭代响应:

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=messages,
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end='', flush=True)

收集完整响应

累积增量内容即可获得最终字符串:

buf = []
for chunk in stream:
    delta = chunk.choices[0].delta.content or ''
    buf.append(delta)
    print(delta, end='', flush=True)

full_text = ''.join(buf)
messages.append({'role': 'assistant', 'content': full_text})

使用 Anthropic 进行流式传输

模式相同,但接口略有不同:

with client.messages.stream(
    model='claude-sonnet-4-5',
    max_tokens=1024,
    messages=messages,
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)

    final = stream.get_final_message()
    print('\ndone, tokens:', final.usage.output_tokens)

流式传输工具调用

工具调用也可以进行流式传输。使用 OpenAI 时,函数名称和参数会分段到达:

for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.tool_calls:
        for tc in delta.tool_calls:
            # tc.function.name and tc.function.arguments arrive as partial strings
            pass

背压与取消

如果用户关闭页面,请中止流式传输,以免继续消耗令牌:

try:
    for chunk in stream:
        if request_was_cancelled():
            stream.close()
            break
finally:
    stream.close()

通过 Web 服务器进行流式传输

对于 FastAPI,请使用 StreamingResponse:

from fastapi.responses import StreamingResponse

def token_generator():
    for chunk in stream:
        delta = chunk.choices[0].delta.content or ''
        yield f'data: {delta}\n\n'

return StreamingResponse(token_generator(), media_type='text/event-stream')

按句子缓冲

对于文本转语音或分块显示,请一直缓冲到句子结束:

buf = ''
for chunk in stream:
    buf += chunk.choices[0].delta.content or ''
    while '. ' in buf:
        sentence, buf = buf.split('. ', 1)
        speak(sentence + '.')

解析流式 JSON

对于 JSON 输出,您无法在流式传输过程中间解析。可以选择:

  • 缓冲完整输出,然后一次性解析
  • 使用流式 JSON 解析器(ijson),在字段完成时逐个输出

延迟指标:TTFT 和 TPS

  • TTFT——首个令牌耗时(用户感知的响应速度)
  • TPS——每秒令牌数(吞吐量)

流式传输优化的是 TTFT,而不是总耗时。为了获得良好的聊天用户体验,目标应是让 TTFT 小于 500 毫秒。

为什么要流式传输

流式传输的主要好处是什么?

回顾

流式传输改善的是用户体验,而不是速度。任何需要与人实时交互的智能体都应实现流式传输。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码,获得全天候 AI 导师的即时帮助,并在网页或应用中继续学习。

课程
60
课程
239

常见问题解答

「流式响应(SSE)」课时是免费的吗?

是的 — 「流式响应(SSE)」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「流式响应(SSE)」这节课中我会学到什么?

通过服务器推送事件逐个令牌地传输 LLM 输出,从而实现响应迅速的界面和进度提示。 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗?

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 3 节课,共 4 节。

「流式响应(SSE)」课时需要多长时间?

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗?

能。每节 AI Agents 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 调用 OpenAI API:chat.completions
  2. 调用 Anthropic API:messages
  3. 流式响应(SSE)
  4. 成本意识:令牌计数与预算
← 返回 AI Agents