第 4 章 · 流式输出与实时响应

本章目标：理解流式的价值与原理，掌握同步/异步两种流式写法与事件类型，学会在流式过程中拿到最终指标。

4.1 为什么需要流式

非流式请求要等模型生成完全部内容才返回，长回答动辄几十秒，用户面对空白页面极易流失。流式（Streaming）让服务端边生成边推送：用户几乎立刻看到第一个字，整体体验从"等电梯"变成"看直播"。

技术上，流式基于 HTTP 分块传输：模型服务商把生成的 token 逐段推给 Agno，Agno 再把它们包装成事件对象交给你迭代。

4.2 同步流式：最基础的打字机效果

给 run() 加上 stream=True，返回值就从 RunOutput 变成了事件迭代器：

python

# stream_basic.py —— 终端打字机效果
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

# stream=True 时返回 RunOutputEvent 迭代器而非 RunOutput
stream = agent.run("写一段 200 字的 Python 学习建议", stream=True)

for chunk in stream:
    # 默认只流出 RunContent 事件（模型正文片段）
    print(chunk.content, end="", flush=True)
print()  # 结束换行

end="" + flush=True 是打字机效果的固定搭配：不换行、立刻刷新缓冲区。注意此时不能再访问 .content 拿全文——每个 chunk 只含一个片段，需要自己拼接。

4.3 事件类型：不只流出正文

默认只推送模型文本（RunContent 事件）。打开 stream_events=True 后，工具调用、推理步骤、记忆更新等所有事件都会流出，这是构建实时 UI 和调试的利器：

python

# stream_events.py —— 观察完整事件流
from agno.agent import Agent, RunEvent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
import os
from dotenv import load_dotenv

load_dotenv()
agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    tools=[DuckDuckGoTools()],
)

stream = agent.run("Agno 是什么框架？", stream=True, stream_events=True)

for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(chunk.content or "", end="", flush=True)      # 正文片段
    elif chunk.event == RunEvent.tool_call_started:
        print(f"\n[开始调用工具: {chunk.tool.tool_name}]")   # 工具开始
    elif chunk.event == RunEvent.tool_call_completed:
        print("[工具调用完成]")
print()

官方事件类型速查（常用部分）：

类别	事件	说明
核心	`run_started` / `run_content` / `run_completed` / `run_error`	运行生命周期
工具	`tool_call_started` / `tool_call_completed` / `tool_call_error`	工具调用三阶段
推理	`reasoning_started` / `reasoning_step` / `reasoning_completed`	思考过程
记忆	`memory_update_started` / `memory_update_completed`	记忆写入

常见坑

run_content 事件的 content 在某些片段（如工具调用间隙）可能为 None，打印前务必做空值保护，否则会打出 "None" 字样污染界面。

4.4 异步流式

异步框架（FastAPI、WebSocket 服务）中要用 arun(stream=True)，配合 async for 消费：

python

# stream_async.py —— 异步流式消费
import asyncio
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

async def main():
    # arun + stream=True 返回异步迭代器
    stream = agent.arun("用三点总结微服务优缺点", stream=True)
    async for chunk in stream:
        print(chunk.content or "", end="", flush=True)

asyncio.run(main())

在 FastAPI 中，把 async for 里收到的片段通过 StreamingResponse 或 WebSocket 推给前端，就实现了网页版打字机——这是第 17 章（结合 FastAPI）的实战预告。

4.5 流式结束后拿指标

流式过程中拿不到完整 metrics，但流结束后最后一个事件（或 RunCompleted 事件，开启 stream_events=True 时）会携带汇总信息：

python

# stream_metrics.py —— 流式 + 最终指标
from agno.agent import Agent, RunEvent
from agno.models.openai import OpenAIChat
import os
from dotenv import load_dotenv

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

full_text = ""
stream = agent.run("解释什么是幂等", stream=True, stream_events=True)

for chunk in stream:
    if chunk.event == RunEvent.run_content:
        full_text += chunk.content or ""
    elif chunk.event == RunEvent.run_completed:
        # RunCompleted 事件上带有完整的运行结果与指标
        m = chunk.metrics
        print("\n--- 本次运行指标 ---")
        print("输入 tokens:", m.input_tokens, "输出 tokens:", m.output_tokens)

print("全文长度:", len(full_text))

模式总结：正文从 run_content 拼接，汇总从 run_completed 读取。这个模式在计费、日志埋点场景是标准做法。

4.6 本章小结

流式把"等全部生成"变成"边生成边看"，显著改善长回答体验；
stream=True 后返回事件迭代器：同步用 for，异步用 async for；
默认只流正文；stream_events=True 才能看到工具/推理/记忆等全部事件；
run_content 的 content 可能为 None，打印必须做空值保护；
指标在 run_completed 事件上读取——正文拼接与指标采集分离。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. agent.run(prompt, stream=True) 的返回值类型是？

2. 想在流式过程中实时显示"正在调用搜索工具"，正确做法是？

3. 流式输出时打印 chunk.content 前为什么要做空值保护？

4. 关于流式场景下获取 token 用量，正确的说法是？

🛠️ 动手实践

把 4.2 的打字机效果改成"每收到一个片段就统计已输出字数"，结束时打印总字数与耗时。
用 stream_events=True 跑一个带搜索工具的问题，把所有事件按顺序写入 events.log 文件，观察事件出现的先后规律。
用 FastAPI 写一个 /chat 接口（可用 StreamingResponse），把异步流式片段推给 curl 客户端，验证网页级流式可行性。

会"边生成边看"之后，进入第 5 章：结构化输出 output_schema。

第 4 章 · 流式输出与实时响应 ​

4.1 为什么需要流式 ​

4.2 同步流式：最基础的打字机效果 ​

4.3 事件类型：不只流出正文 ​

4.4 异步流式 ​

4.5 流式结束后拿指标 ​

4.6 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 4 章 · 流式输出与实时响应

4.1 为什么需要流式

4.2 同步流式：最基础的打字机效果

4.3 事件类型：不只流出正文

4.4 异步流式

4.5 流式结束后拿指标

4.6 本章小结

🛠️ 动手实践