Stream

查看 Markdown查看 Markdown

Stream 把消息从 Flow 实时传给读取方。Step 或 Client 写入消息。Client 按顺序读取，并能从上次停下的地方继续。Dex 只以 best-effort 方式保存 Stream 消息。

什么时候用 Stream

Stream 适合能改善用户体验、但不属于关键业务的数据：

• LLM reasoning summary、生成文本 chunk 和 agent progress
• 进度条、诊断事件与实时预览
• Step 仍在运行时展示的临时状态

不能丢失的数据，用 Channel 或 Attribute。

保证

Stream 用 durability 换取速度：

• Dex 把消息保存在 server 内存中。server 重启或 standby 接管时，消息会丢失。
• 每个 Stream 有以字节计的容量。容量满时，Dex 删除最旧的消息。读得慢的一方可能错过它们。
• Step 的写入立即发出，此时 Step 仍在运行。之后 Step 失败也不会撤回它。
• Stream 写入失败永远不会让 Step 失败。
• 读取方按 Dex 保存的顺序看到消息。

容量由同一 Flow type 和 Stream name 下的所有 Flow run 共享，不是每个 Flow ID 的限制。见 Write semantics、Capacity and trimming 和 Best-effort storage。

定义并写入 Stream

Stream 有稳定的名字、消息类型和容量。把它加到 Flow 的 persistence schema。

这个例子通过 buffered text writer 写入进度文本。Dex 在保存 Step 结果之前发出缓冲的文本。Step 不等待 server 保存文本。见 Buffer text chunks。

stream_flow.pyPython

12345678910111213141516171819202122

class RenderPreview(Step[str]):
    def __init__(self, progress: Stream[str]) -> None:
        self.progress = progress

    async def execute(self, context: AsyncContext, input: str) -> StepDecision:
        progress = self.progress.buffered_text(context)
        progress.write(f"Rendering preview for {input}")
        progress.write(f"Preview ready for {input}")
        return graceful_complete(f"Rendered {input}")


class StreamFlow(Flow[str]):
    progress = Stream("Progress", str, 10 * 1024 * 1024)

    def __init__(self) -> None:
        self.render_preview = RenderPreview(self.progress)

    def get_steps(self) -> StepList[str]:
        return StepList.start_step(self.render_preview)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.progress)

例子: examples/python/dex_examples/primitives/stream/stream_flow.py

读取 Stream

Client 有两种方式读取 Stream：

• ReadStream 向前读取，每次一条消息。每条消息带一个 resume token。把 token 传回去就能读下一条。没有新消息时，调用会等待。
• ListStreamMessages 从最新的消息向后分页。它立即返回。

见 Resumable reads 和 List messages。

Client 也可以写入 Stream，Flow 关闭后也可以。RPC handler 通过注入的 Client 写入。

参考

每个 SDK 参考的 Stream 页列出该语言的 API 名称。

Data primitive semantics 详细说明写入语义、buffered text、读取、列表、容量和存储。server 配置见 Streams。