LangChain 中文文档

外观

加入与重新加入（join and rejoin）让你能够在不停下智能体的情况下断开与正在运行的智能体流的连接，之后再重新连接。当客户端离开时，智能体会继续在服务器端执行，你可以从上次离开的确切位置继续接收流。

import { PatternEmbed } from "/snippets/pattern-embed.jsx"

INFO

This feature requires the LangGraph Agent Server. Run your agent locally with langgraph dev or deploy it to LangSmith to use this pattern.

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

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

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

核心概念 ​

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

方法 / 选项用途
threadId将流绑定到你想观察的 LangGraph 线程
onThreadId持久化新创建的线程 ID，以便重新挂载时可以重新连接
stream.disconnect()在客户端离开流，同时智能体继续在服务器端运行
使用相同的 threadId 重新挂载重新附加到该线程的在途工作

INFO

加入/重新加入使用 stream.disconnect()，而不是 stream.stop()。 默认情况下，stream.stop() 会取消正在进行的运行：它会断开客户端的连接并取消服务器上的运行。对于加入/重新加入，请调用 stream.disconnect()（stop({ cancel: false }) 的别名），这样在你离开时智能体会继续处理。

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

设置 useStream ​

关键的设置步骤是持久化 threadId。当组件以相同的线程 ID 重新挂载时，流会附加到该线程的当前状态以及任何在途运行。

INFO

The code examples use useStream<typeof myAgent> for type-safe stream state. See Type inference for Python or JavaScript backends.

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

<template>
    <ConnectionStatus :connected="connected" />
    <MessageList :messages="stream.messages" />
    <ChatControls
      :stream="stream"
      :threadId="threadId"
      :connected="connected"
      @disconnect="disconnect"
      @rejoin="rejoin"
    />
</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;
  }
</script>

  <ConnectionStatus {connected} />
  <MessageList messages={stream.messages} />
  <ChatControls
    {threadId}
    {connected}
    onDisconnect={disconnect}
    onRejoin={rejoin}
  />
ts
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 绑定正是让后续的重新挂载能够重新连接到同一对话的原因：

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

断开与流的连接 ​

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

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

这里不要使用 stream.stop() —— 默认情况下它会在服务器上取消运行。

调用 disconnect() 之后：

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

重新加入流 ​

使用保存的线程 ID 重新挂载流消费组件以重新连接。在 React 中，示例会递增一个 mountKey；在其他框架中，使用等价的重新挂载或条件渲染模式：

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

重新加入之后：

  • connected 变为 true
  • 断开期间生成的所有消息都会被投递
  • 新的流式消息会实时恢复
  • 如果智能体仍在运行，stream.isLoading 变为 true；如果它已经完成，你会立即收到最终状态

最佳实践 ​

  • 加入/重新加入使用 disconnect()，取消则使用 stop()：离开页面或让应用进入后台时应该调用 stream.disconnect()。面向用户的"停止"或"取消"按钮应该调用 stream.stop()（或 client.runs.cancel）。
  • 始终保存线程 ID：没有它，重新加入就不可能实现。同时使用组件状态与持久化存储以增强韧性。
  • 展示清晰的连接状态：用户应该始终知道他们是在接收实时更新，还是在查看快照。
  • 在可见性变化时自动重新加入：使用 Page Visibility API，在用户返回该标签页时自动重新加入。
  • 设置合理的超时：如果重新加入尝试耗时过长，则改为回退到获取线程历史。
  • 清理过期的线程：当用户重新开始或后端报告线程不可用时，移除已持久化的线程 ID。