跳转到内容
Flue

输入关键词开始搜索文档。

事件

最后更新 查看 Markdown 源文件

Flue Agent SDK（@flue/sdk）以两种词汇、两个抽象级别传递会话数据：

运行时的底层活动事件联合类型（FlueEvent）不是 SDK 的一部分：没有 FlueClient 方法传递它。进程内消费者使用 @flue/runtime 的 observe() 订阅它。

每种词汇在拥有它的页面上记录，上面已链接并在下面逐导出映射。本页仅添加 SDK 自身拥有的内容：FlueEventStream 迭代接口和 SDK 消费者看到的偏移量和重发语义。

一个最小消费者从观察中读取消息和部分：

import { createFlueClient } from '@flue/sdk';

const client = createFlueClient({ url: 'https://app.example.com/api/support/thread-42' });
const observation = client.observe();

observation.subscribe(() => {
  const { conversation } = observation.getSnapshot();
  for (const message of conversation?.messages ?? []) {
    if (message.display !== 'visible') continue;
    for (const part of message.parts) {
      if (part.type === 'text') renderText(message.id, part.text, part.state);
    }
  }
});

导出类型

  • FlueConversationSnapshot、FlueConversationState、FlueConversationMessage、FlueConversationPart、FlueConversationSettlement — 物化会话，在 FlueClient 上与 history() 和 observe() 一起逐字段记录。
  • ConversationStreamChunk — updates 视图块联合类型，在流协议参考上记录。由 wait() 的 onEvent 原样传递；不是稳定的应用 API — 应用代码应通过 observe() 消费物化状态。
  • PromptUsage — 聚合的 token 和成本用量，由会话结算携带；格式与运行时同名导出相同（由线路一致性类型测试固定）。

FlueEventStream

interface FlueEventStream<T = ConversationStreamChunk> extends AsyncIterable<T> {
  cancel(reason?: unknown): void;
  readonly offset: string;
}

由 Durable Streams 连接支持的事件异步可迭代对象，具有自动重连、基于偏移量的回放和实时尾随。SDK 在内部构造这些（wait() 消费一个 FlueEventStream<ConversationStreamChunk>）；该类型被导出以便第一方展示者可以为自己的流定义类型。for await...of 是消费接口；跳出循环会清理底层连接。

  • cancel(reason?) — 取消流并中止底层连接。迭代随后结束（done: true）而非抛出。
  • offset — 恢复检查点。它在每个已传递批次上推进：只有当该批次中的每个事件都已产出时才移动到批次的下一个偏移量，因此从检查点恢复永远不会跳过未传递的事件 — 最多重发检查点获取时正在传输的批次（至少一次）。

每个流式值在产出前经过调用方提供的验证器；抛出的验证器对流是终止性的（连接被取消，后续的 next() 调用重新抛出）。SDK 自己的流根据物化会话协议验证（参见 SDK 错误中的 ConversationStreamError）。

FlueStreamOptions

interface FlueStreamOptions {
  offset?: string;
  live?: LiveMode;
  signal?: AbortSignal;
  backoffOptions?: BackoffOptions;
}

事件流读取的选项。LiveMode 和 BackoffOptions 从 @durable-streams/client 重新导出。

  • offset — 起始偏移量。默认为 '-1'（完整历史）。
  • live — 实时尾随模式：boolean | 'long-poll' | 'sse'。默认为 true（long-poll）。false 读取到当前末尾并完成。
  • signal — 中止流；迭代结束而不抛出。
  • backoffOptions — 连接尝试的重试行为（initialDelay、maxDelay、multiplier 和回调，按 @durable-streams/client）。

偏移量和重发

每次会话读取都锚定在持久流偏移量上 — 一个不透明的字符串检查点。偏移量在 FlueConversationSnapshot、AgentSendResult、AgentConversationObservationSnapshot 和 FlueEventStream 上暴露。将它们视为不透明：如果必须，可以比较相等性，但永远不要解析或进行算术运算。块的 position 值不是偏移量 — 它们标识和排序项目但不能用作恢复点。线路级偏移量格式和协调头部在流协议参考中指定。

传递是至少一次的：当连接在批次传输中断开时，传输从批次前的偏移量重连并重放正在传输的批次。SDK 的消费者各自吸收这一点：

  • observe() 按块 position 去重，并在重连时重填全新快照而非增量恢复；参见 observe()。
  • wait() 仅监视其提交的终止 submission-settled 块，该块在重发下是幂等的。然而其 onEvent 回调接收原始流：它可能在重连后观察到同一个块多次，且它接收从接收偏移量开始的会话的每个块，不仅仅是属于被等待提交的块。如果区别重要，按 position 去重；对于维护的 UI 状态建议使用 observe()。

两种实时模式（'long-poll' 和 'sse'）携带相同的块并具有相同的保证；SSE 用连接生命周期换取更低的逐 token 延迟。