---
title: "Stream"
description: "Stream 把消息从 Flow 实时传给读取方。Step 或 Client 写入消息。Client 按顺序读取，并能从上次停下的地方继续。Dex 只以 best-effort 方式保存 Stream 消息。"
canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/primitives/stream/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/primitives/stream.mdx"
locale: "zh-Hans"
---

# Stream

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

## 什么时候用 Stream

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

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

不能丢失的数据，用 [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md) 或 [Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md)。

<a id="语义"></a>

<a id="写入语义"></a>

<a id="近似容量"></a>

<a id="best-effort-storage"></a>

## 保证

Stream 用 durability 换取速度：

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

容量由同一 Flow type 和 Stream name 下的所有 Flow run 共享，不是每个 Flow ID 的限制。见 [Write semantics](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-write-semantics)、[Capacity and trimming](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-capacity) 和 [Best-effort storage](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-storage)。

<a id="定义并注册-stream"></a>

## 定义并写入 Stream

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

<a id="buffer-text-chunk"></a>

这个例子通过 buffered text writer 写入进度文本。Dex 在保存 Step 结果之前发出缓冲的文本。Step 不等待 server 保存文本。见 [Buffer text chunks](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-buffered-text)。

**Python**

```python
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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/stream/stream_flow.py)

**Go**

```go
var Progress = dex.DefineStream[string]("Progress", 10<<20)

type StreamFlow struct {
	dex.FlowDefaults
}

func (*StreamFlow) GetSteps() []dex.StepDef {
	return []dex.StepDef{dex.DefineStartStep[string](renderPreview{})}
}

func (*StreamFlow) GetPersistenceSchema() dex.PersistenceSchema {
	return dex.PersistenceSchema{Streams: []dex.StreamDef{Progress}}
}

type renderPreview struct {
	dex.StepDefaultsNoWaitFor[string]
}

func (renderPreview) Execute(ctx dex.Context, input string) (*dex.StepDecision, error) {
	progress, err := dex.NewBufferedTextStream(ctx, Progress)
	if err != nil {
		return nil, err
	}
	if err := progress.Write("Rendering preview for " + input); err != nil {
		return nil, err
	}
	if err := progress.Write("Preview ready for " + input); err != nil {
		return nil, err
	}
	return dex.GracefulComplete("Rendered " + input), nil
}
```

例子: [examples/go/primitives/stream/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/stream/workflow.go)

**Java**

```java
public final class StreamFlow implements Flow<String> {
    public final Stream<String> progress =
            Stream.define("Progress", String.class, 10L * 1024L * 1024L);
    private final RenderPreview renderPreview = new RenderPreview();

    @Override
    public StepList<String> getSteps() {
        return StepList.startStep(renderPreview);
    }

    @Override
    public PersistenceSchema getPersistenceSchema() {
        return PersistenceSchema.of(progress);
    }

    final class RenderPreview implements Step<String> {
        @Override
        public Class<String> getInputType() {
            return String.class;
        }

        @Override
        public StepDecision execute(final Context context, final String input) {
            final BufferedTextStream writer = BufferedTextStream.create(context, progress);
            writer.write("Rendering preview for " + input);
            writer.write("Preview ready for " + input);
            return StepDecision.gracefulComplete("Rendered " + input);
        }
    }
}
```

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/stream/StreamFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/stream/StreamFlow.java)

**TypeScript**

```typescript
export const progress = new Stream("Progress", stringCodec, 10 * 1024 * 1024);

class RenderPreview implements Step<string> {
  public readonly inputCodec = stringCodec;

  public getStepType(): string {
    return "RenderPreview";
  }

  public async execute(context: Context, input: string): Promise<StepDecision> {
    const writer = progress.bufferedText(context, { flushIntervalMs: 500 });
    writer.write(`Rendering preview for ${input}`);
    writer.write(`Preview ready for ${input}`);
    return gracefulComplete(`Rendered ${input}`);
  }
}

const renderPreview = new RenderPreview();

export class StreamFlow implements Flow<string> {
  public getFlowType(): string {
    return "StreamFlow";
  }

  public getSteps() {
    return StepList.startStep(renderPreview);
  }

  public getPersistenceSchema(): PersistenceSchema {
    return { streams: [progress] };
  }
}
```

例子: [examples/typescript/src/primitives/stream/stream-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/stream/stream-flow.ts)

**Rust**

```rust
pub static PROGRESS: LazyLock<Stream<String>> =
    LazyLock::new(|| Stream::new("Progress", 10 * 1024 * 1024));

#[derive(Default)]
pub struct StreamFlow {
    render_preview: RenderPreview,
}

impl Flow for StreamFlow {
    type StartInput = String;

    fn steps(&self) -> StepList<'_, Self::StartInput> {
        StepList::start(&self.render_preview)
    }

    fn persistence(&self) -> PersistenceSchema {
        PersistenceSchema::new().stream(&PROGRESS)
    }
}

#[derive(Default)]
struct RenderPreview;

impl Step for RenderPreview {
    type Input = String;

    fn execute(
        &self,
        context: &mut Context,
        input: Self::Input,
    ) -> HandlerResult<StepDecision> {
        let progress = PROGRESS.buffered_text_with_options(
            context,
            BufferedTextStreamOptions::new(Duration::from_millis(500), 16 * 1024),
        )?;
        progress.write(format!("Rendering preview for {input}"))?;
        progress.write(format!("Preview ready for {input}"))?;
        Ok(StepDecision::graceful_complete(format!(
            "Rendered {input}"
        )))
    }
}
```

例子: [examples/rust/src/primitives/stream/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/stream/flow.rs)

## 读取 Stream

Client 有两种方式读取 Stream：

- <a id="可续读"></a>**ReadStream** 向前读取，每次一条消息。每条消息带一个 resume token。把 token 传回去就能读下一条。没有新消息时，调用会等待。
- <a id="列出消息"></a>**ListStreamMessages** 从最新的消息向后分页。它立即返回。

见 [Resumable reads](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-resumable-reads) 和 [List messages](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#stream-listing)。

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

## 参考

每个 SDK 参考的 [Stream](https://docs.superdurable.io/zh-Hans/reference/python/stream.md) 页列出该语言的 API 名称。

[Data primitive semantics](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#streams) 详细说明写入语义、buffered text、读取、列表、容量和存储。server 配置见 [Streams](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#streams)。