加入与重新加入流

断开连接并重新连接到正在运行的 Agent 流

加入与重新加入（Join and rejoin）让你可以断开与正在运行的 Agent 流的连接而无需停止 Agent，然后在稍后重新连接。当客户端离开时，Agent 会在服务端继续执行，而你可以在返回时精确地从离开时的位置继续接收流。

NOTE

该特性需要 LangGraph Agent Server。请使用 langgraph dev 在本地运行你的 Agent，或将其部署到 LangSmith 来使用此模式。

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

传统的流式传输 API 将客户端与服务端紧密耦合在一起：如果客户端断开连接，流就会丢失。加入与重新加入打破了这种耦合，从而支持了几种重要的模式：

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

核心概念

加入/重新加入模式涉及三个关键机制：

方法 / 选项	用途
`threadId`	将流绑定到你想要观察的 LangGraph 线程
`onThreadId`	持久化新创建的线程 ID，以便重新挂载时可以重新连接
`stream.disconnect()`	在客户端离开流，同时 Agent 在服务端继续运行
使用相同的 `threadId` 重新挂载	重新连接到该线程正在进行中的任务（in-flight work）

NOTE

加入/重新加入使用的是 stream.disconnect()，而不是 stream.stop()。 默认情况下，stream.stop() 会取消活跃的运行：它会断开客户端连接并取消服务端的运行。对于加入/重新加入，请调用 stream.disconnect()（即 stop({ cancel: false }) 的别名），以便 Agent 在你离开期间继续处理。

若要从应用代码中显式取消执行，请使用 stream.stop() 或 client.runs.cancel。

配置 `useStream`

关键的配置步骤是持久化 threadId。当组件使用相同的线程 ID 重新挂载时，流将连接到该线程的当前状态以及任何正在进行中的运行。

NOTE

代码示例使用 useStream<typeof myAgent> 来实现类型安全的流状态。请参阅针对 Python 或 JavaScript 后端的类型推断。

ReactVueSvelteAngular

tsx

import { useStream } from "@langchain/react";
import { useCallback, useState } from "react";

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 (
    <div key={mountKey}>
      <ConnectionStatus connected={connected} />
      <MessageList messages={stream.messages} />
      <ChatControls
        stream={stream}
        threadId={threadId}
        connected={connected}
        onDisconnect={disconnect}
        onRejoin={rejoin}
      />
    </div>
  );
}

vue

<script setup lang="ts">
import { useStream } from "@langchain/vue";
import { ref } from "vue";

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;
}
`&lt;/script&gt;`

<template>
  <div :key="mountKey">
    <ConnectionStatus :connected="connected" />
    <MessageList :messages="stream.messages" />
    <ChatControls
      :stream="stream"
      :threadId="threadId"
      :connected="connected"
      @disconnect="disconnect"
      @rejoin="rejoin"
    />
  </div>
</template>

svelte

<script lang="ts">
  import { useStream } from "@langchain/svelte";

  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;
  }
`&lt;/script&gt;`

<div key={mountKey}>
  <ConnectionStatus {connected} />
  <MessageList messages={stream.messages} />
  <ChatControls
    {threadId}
    {connected}
    onDisconnect={disconnect}
    onRejoin={rejoin}
  />
</div>

import { Component, signal } from "@angular/core";
import { injectStream } from "@langchain/angular";

@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()"
    />
  `,
})
export class ChatComponent {
  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() 可以在不取消运行的情况下离开流。Agent 会在服务端继续处理。

await stream.disconnect();
// 等价于: await stream.stop({ cancel: false })

此处切勿使用 stream.stop()——默认情况下它会取消服务端的运行。

调用 disconnect() 之后：

stream.isLoading 变为 false
你自己的 connected 标志也应该变为 false
消息列表保留截至断开连接点所接收到的所有消息
Agent 在服务端继续运行
在重新加入之前不会接收到新消息

重新加入流

使用保存的线程 ID 重新挂载流消费者以重新连接。在 React 中，示例通过递增 mountKey 来实现；在其他框架中，请使用等效的重新挂载或条件渲染模式：

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

重新加入之后：

connected 变为 true
断开连接期间生成的任何消息都将被送达
新的流式消息实时恢复
如果 Agent 仍在运行，stream.isLoading 变为 true；如果它已经完成，你将立即收到最终状态

最佳实践

在加入/重新加入时使用 disconnect()，在取消时使用 stop()：离开页面或将应用置于后台时应调用 stream.disconnect()。面向用户的“停止”或“取消”按钮应调用 stream.stop()（或 client.runs.cancel）。
务必保存线程 ID：没有它，重新加入将无从谈起。同时使用组件状态和持久化存储以增强弹性。
清晰展示连接状态：用户应始终清楚自己是在接收实时更新还是在查看快照。
在页面可见性变化时自动重新加入：使用 Page Visibility API 在用户返回标签页时自动重新加入。
设置合理的超时时间：如果重新加入尝试耗时过长，应回退到直接获取线程历史记录。
清理过期线程：当用户重新开始或后端报告该线程不可用时，移除已持久化的线程 ID。

加入与重新加入流 ​

为什么需要加入与重新加入？ ​

核心概念 ​

配置 useStream ​

提交消息 ​

断开与流的连接 ​

重新加入流 ​

最佳实践 ​

加入与重新加入流

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

核心概念

配置 `useStream`

提交消息

断开与流的连接

重新加入流

最佳实践