Skip to content

流式返回与事件类型

Agently 把模型响应统一成事件流,让你不用被不同 provider 的返回格式绑死。

适合什么时候读

  • 你已经知道 get_response() 的基本用法
  • 你要把流式事件接到 UI、日志或下一步逻辑
  • 你想分清 deltametatool_callsinstant 各自适合什么场景

你会学到什么

  • Agently 为什么先做事件归一化
  • 常见事件类型分别表示什么
  • 什么时候用文本流,什么时候用结构化流
  • 在工程落地里为什么默认优先 async generator

事件流是怎么来的

这张图是在正文里说明:instant 不是另一套孤立接口,而是建立在统一事件层上的结构化视图。

常见事件类型

  • delta:文本增量
  • reasoning_delta:推理增量
  • tool_calls:工具调用片段
  • meta:usage、finish_reason 等元信息
  • done:最终结果
  • extra:扩展字段
  • error:请求或解析错误
  • instant / streaming_parse:结构化流式事件

Async First 推荐

如果你在服务端、SSE、WebSocket、TriggerFlow 或并发请求场景里工作,推荐把 async generator 当成默认入口:

python
async for item in response.get_async_generator(type="instant"):
    print(item.path, item.value)

原因不是“语法更现代”,而是:

  • 避免同步桥接阻塞事件循环
  • 更自然地和异步服务框架集成
  • 更容易把流式结果继续接到 TriggerFlow 或 runtime stream

只消费文本流

python
for chunk in response.result.get_generator(type="delta"):
    print(chunk, end="", flush=True)

适合:

  • 打字机式 UI
  • 简单日志输出

过滤你关心的事件

python
for event, value in response.result.get_generator(
    type="specific",
    specific=["reasoning_delta", "tool_calls", "done"],
):
    print(event, value)

读取结构化流

python
for item in response.result.get_generator(type="instant"):
    print(item.path, item.delta, item.is_complete)

适合:

  • 输出结构已经明确
  • 你要边生成边更新某个字段或列表

在工程落地里,上面这段更推荐改成异步版本:

python
async for item in response.get_async_generator(type="instant"):
    if item.is_complete:
        print(item.path, item.value)

常见误区

  • 想做结构化流式展示,却仍然只消费 delta 文本。
  • 把所有事件都原样暴露给前端,而不先收敛成业务语义。
  • 还没固定 response,就开始在多个地方并行消费流式事件。

下一步去哪

  • agently-model-response
  • agently-output-control