加入和重新加入允许您断开与正在运行的智能体流的连接而不会停止智能体，之后再重新连接。智能体在客户端离开时继续在服务器端执行，您可以完全从离开的位置继续接收流。

为什么需要加入和重新加入？

传统的流式 API 将客户端和服务器紧密耦合：如果客户端断开连接，流就会丢失。加入和重新加入打破了这种耦合，实现了几个重要的模式：

• 网络中断：移动用户在基站或 Wi-Fi 网络之间移动时可以无缝恢复
• 页面导航：用户从聊天页面离开后稍后返回不会丢失进度
• 移动后台运行：被操作系统挂起的应用可以在切换到前台时重新加入流
• 长时间运行的任务：智能体执行耗时数分钟的操作（研究、代码生成、数据分析），用户无需保持页面打开
• 多设备切换：在手机上开始对话，在桌面设备上重新加入

核心概念

The join/rejoin pattern involves three key mechanisms:

设置 useStream

关键设置步骤是持久化 threadId。当组件使用相同的线程 ID 重新挂载时， 流会附加到线程的当前状态和任何 进行中的运行。

function Chat() {
  const [connected, setConnected] = useState(true);
  const [mountKey, setMountKey] = useState(0);
  const [threadId, setThreadId] = useState<string | null>(
    () => sessionStorage.getItem("activeThreadId"),
  );

  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    threadId,
    onThreadId(id) {
      setThreadId(id);
      if (id) sessionStorage.setItem("activeThreadId", id);
    },
  });

  const disconnect = useCallback(() => {
    void stream.disconnect();
    setConnected(false);
  }, [stream]);

  const rejoin = useCallback(() => {
    setMountKey((key) => key + 1);
    setConnected(true);
  }, []);

  return (





  );
}

<script setup lang="ts">



const connected = ref(true);
const mountKey = ref(0);
const threadId = ref<string | null>(sessionStorage.getItem("activeThreadId"));

const stream = useStream<typeof myAgent>({
  apiUrl: "http://localhost:2024",
  assistantId: "join_rejoin",
  threadId,
  onThreadId(id) {
    threadId.value = id;
    if (id) sessionStorage.setItem("activeThreadId", id);
  },
});

function disconnect() {
  void stream.disconnect();
  connected.value = false;
}

function rejoin() {
  mountKey.value += 1;
  connected.value = true;
}
</script>

<template>





</template>

<script lang="ts">


  let connected = $state(true);
  let mountKey = $state(0);
  let threadId = $state<string | null>(sessionStorage.getItem("activeThreadId"));

  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    threadId: () => threadId,
    onThreadId(id) {
      threadId = id;
      if (id) sessionStorage.setItem("activeThreadId", id);
    },
  });

  function disconnect() {
    void stream.disconnect();
    connected = false;
  }

  function rejoin() {
    mountKey += 1;
    connected = true;
  }
</script>




  

@Component({
  selector: "app-chat",
  template: `
    <connection-status [connected]="connected()" />
    <message-list [messages]="stream.messages()" />
    <chat-controls
      [stream]="stream"
      [threadId]="threadId()"
      [connected]="connected()"
      (disconnect)="disconnect()"
      (rejoin)="rejoin()"
    />
  `,
})

  threadId = signal<string | null>(sessionStorage.getItem("activeThreadId"));
  connected = signal(true);
  mountKey = signal(0);

  stream = injectStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    threadId: this.threadId,
    onThreadId: (id) => {
      this.threadId.set(id);
      if (id) sessionStorage.setItem("activeThreadId", id);
    },
  });

  disconnect() {
    void this.stream.disconnect();
    this.connected.set(false);
  }

  rejoin() {
    this.mountKey.update((key) => key + 1);
    this.connected.set(true);
  }
}

提交消息

正常提交消息。线程 ID 绑定允许后续重新挂载 重新连接到同一对话：

stream.submit({ messages: [{ type: "human", content: text }] });

断开流连接

调用 stream.disconnect() 在不取消运行的情况下离开流。智能体继续在服务器端处理。

await stream.disconnect();
// equivalent to: await stream.stop({ cancel: false })

Do **不要** 在此处使用 stream.stop() ——默认情况下它会取消服务器上的运行。

调用后 disconnect(): - stream.isLoading 变为 false - 您自己的 connected 标志也应变为 false - 消息列表保留断开点之前接收到的所有消息 - 代理在服务器上继续运行 - 重新加入前不会收到新消息

重新加入流

使用保存的线程ID重新挂载流消费者以重新连接。在React中， 演示触发一个 mountKey; 在其他框架中，使用等效的重新挂载或 条件渲染模式：

setMountKey((key) => key + 1);
setConnected(true);

重新加入后： - connected 变为 true - 断开连接期间生成的任何消息都会被传递 - 新的流消息实时恢复 - 如果代理仍在运行， stream.isLoading 变为 true; 如果它已 经完成，您会立即收到最终状态

最佳实践

• - **使用 disconnect() for join/rejoin, stop() 来取消**：离开或后台运行应用时应调用 stream.disconnect()。面向用户的"停止"或"取消"按钮应调用 stream.stop() (or client.runs.cancel).
• 始终保存线程ID：没有它就无法重新加入。使用组件状态和持久存储来提高弹性。
• 显示清晰的连接状态：用户应始终知道他们是在接收实时更新还是查看快照。
• 可见性变化时自动重新加入：使用页面可见性API在用户返回标签页时自动重新加入。
• 设置合理的超时时间：如果重新加入尝试花费时间过长，请回退到获取线程历史记录。
• 清理过期的线程：当用户重新开始或后端报告线程不可用时，删除持久化的线程ID。