加入和重新加入流

加入和重新加入功能允许你在不停止智能体的情况下断开正在运行的流，然后稍后重新连接。智能体在客户端离开期间继续在服务器端执行，你可以从中断处精确恢复流。

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 网络之间移动的移动用户可以无缝恢复
页面导航：用户离开聊天页面后返回，不会丢失进度
移动端后台运行：被操作系统挂起的应用在前台恢复时可以重新加入流
长时间运行的任务：智能体执行多分钟操作（研究、代码生成、数据分析）时用户无需保持页面打开
多设备交接：在手机上开始对话，在桌面上重新加入

核心概念

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

方法/选项	用途
`stream.stop()`	在不停止智能体的情况下断开客户端
`stream.joinStream(runId)`	通过运行 ID 重新连接到现有流
`onDisconnect: "continue"`	提交选项，告诉服务器在客户端断开后继续运行
`streamResumable: true`	提交选项，使流可以稍后被重新加入

stream.stop() 与取消运行有本质区别。停止仅断开客户端。智能体继续在服务器端处理。要真正取消智能体的执行，你需要使用中断或取消机制。

设置 `useStream`

关键的设置步骤是从 onCreated 回调中捕获 run_id，以便稍后重新加入。定义一个与你的智能体状态 schema 匹配的 TypeScript 接口，并将其作为类型参数传递给 useStream，以获得类型安全的状态值访问。在以下示例中，将 typeof myAgent 替换为你的接口名称：

import type { BaseMessage } from "@langchain/core/messages";

interface AgentState {
  messages: BaseMessage[];
}

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

function Chat() {
  const [savedRunId, setSavedRunId] = useState<string | null>(null);

  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    onCreated(run) {
      setSavedRunId(run.run_id);
    },
  });

  const isConnected = stream.isLoading;

  return (
    <div>
      <ConnectionStatus connected={isConnected} />
      <MessageList messages={stream.messages} />
      <ChatControls
        stream={stream}
        savedRunId={savedRunId}
        isConnected={isConnected}
      />
    </div>
  );
}

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

const savedRunId = ref<string | null>(null);

const stream = useStream<typeof myAgent>({
  apiUrl: "http://localhost:2024",
  assistantId: "join_rejoin",
  onCreated(run) {
    savedRunId.value = run.run_id;
  },
});

const isConnected = computed(() => stream.isLoading.value);
</script>

<template>
  <div>
    <ConnectionStatus :connected="isConnected" />
    <MessageList :messages="stream.messages" />
    <ChatControls
      :stream="stream"
      :savedRunId="savedRunId"
      :isConnected="isConnected"
    />
  </div>
</template>

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

  let savedRunId: string | null = null;

  const { messages, isLoading, submit, stop, joinStream } = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    onCreated(run) {
      savedRunId = run.run_id;
    },
  });
</script>

<div>
  <ConnectionStatus connected={$isLoading} />
  <MessageList messages={$messages} />
  <ChatControls
    {savedRunId}
    isConnected={$isLoading}
    on:disconnect={() => stop()}
    on:rejoin={() => joinStream(savedRunId)}
  />
</div>

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

@Component({
  selector: "app-chat",
  template: `
    <connection-status [connected]="stream.isLoading()" />
    <message-list [messages]="stream.messages()" />
    <chat-controls
      [stream]="stream"
      [savedRunId]="savedRunId()"
      [isConnected]="stream.isLoading()"
    />
  `,
})
export class ChatComponent {
  savedRunId = signal<string | null>(null);

  stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    onCreated: (run) => {
      this.savedRunId.set(run.run_id);
    },
  });
}

使用可恢复选项提交

提交消息时，传递 onDisconnect: "continue" 和 streamResumable: true 以启用加入/重新加入流程：

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

选项	默认值	描述
`onDisconnect`	`"cancel"`	客户端断开连接时的行为。`"continue"` 保持智能体运行；`"cancel"` 停止它。
`streamResumable`	`false`	为 `true` 时，服务器保留流状态以便客户端稍后重新加入。

始终同时使用这两个选项。设置 onDisconnect: "continue" 但不设置 streamResumable: true 意味着智能体继续运行但你无法重新加入流查看输出。

断开流连接

调用 stream.stop() 断开客户端。智能体继续在服务器端处理。

stream.stop();

调用 stop() 后：

stream.isLoading 变为 false
消息列表保留到断开点为止收到的所有消息
智能体继续在服务器上运行
在重新加入之前不会收到新消息

重新加入流

使用保存的运行 ID 调用 stream.joinStream(runId) 重新连接：

stream.joinStream(savedRunId);

重新加入后：

stream.isLoading 再次变为 true
断开期间生成的所有消息都会被传递
新的流式消息实时恢复
如果智能体已经完成，你会立即收到最终状态

构建连接状态指示器

视觉指示器帮助用户了解他们是否正在从智能体接收实时更新。

function ConnectionStatus({ connected }: { connected: boolean }) {
  return (
    <div className="connection-status">
      <span
        className={`status-dot ${connected ? "connected" : "disconnected"}`}
      />
      <span className="status-text">
        {connected ? "已连接" : "已断开"}
      </span>
    </div>
  );
}

使用绿色/红色圆点为指示器设置样式：

.status-dot {
  width: 8px;
  height: 8px;
  border-radius: 50%;
  display: inline-block;
  margin-right: 6px;
}

.status-dot.connected {
  background-color: #22c55e;
  box-shadow: 0 0 4px #22c55e;
}

.status-dot.disconnected {
  background-color: #ef4444;
  box-shadow: 0 0 4px #ef4444;
}

断开和重新加入控件

提供明确的按钮用于断开和重新加入，让用户拥有完全控制权：

function ChatControls({ stream, savedRunId, isConnected }) {
  const [input, setInput] = useState("");

  const handleSend = () => {
    if (!input.trim()) return;
    stream.submit(
      { messages: [{ type: "human", content: input.trim() }] },
      { onDisconnect: "continue", streamResumable: true }
    );
    setInput("");
  };

  return (
    <div className="controls">
      <div className="input-row">
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="输入消息..."
          onKeyDown={(e) => e.key === "Enter" && handleSend()}
        />
        <button onClick={handleSend}>发送</button>
      </div>

      <div className="stream-controls">
        {isConnected ? (
          <button onClick={() => stream.stop()} className="disconnect-btn">
            断开连接
          </button>
        ) : (
          savedRunId && (
            <button
              onClick={() => stream.joinStream(savedRunId)}
              className="rejoin-btn"
            >
              重新加入流
            </button>
          )
        )}
      </div>
    </div>
  );
}

持久化运行 ID

对于跨会话重新加入（例如用户关闭浏览器后返回），将运行 ID 持久化到存储中：

const stream = useStream<typeof myAgent>({
  apiUrl: "http://localhost:2024",
  assistantId: "join_rejoin",
  onCreated(run) {
    localStorage.setItem("activeRunId", run.run_id);
  },
});

// 页面加载时，检查是否有活动运行
const existingRunId = localStorage.getItem("activeRunId");
if (existingRunId) {
  stream.joinStream(existingRunId);
}

当运行完成时应清理持久化的运行 ID。监听流完成并移除已存储的 ID，以避免尝试重新加入已完成的运行。

错误处理

如果运行已过期、被删除或服务器已重启，重新加入可能会失败。优雅地处理这些情况：

try {
  stream.joinStream(savedRunId);
} catch (error) {
  console.error("重新加入流失败:", error);
  // 清除过期的运行 ID 并通知用户
  setSavedRunId(null);
  localStorage.removeItem("activeRunId");
}

完整示例

function JoinRejoinChat() {
  const [savedRunId, setSavedRunId] = useState<string | null>(null);
  const [input, setInput] = useState("");

  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "join_rejoin",
    onCreated(run) {
      setSavedRunId(run.run_id);
    },
  });

  const isConnected = stream.isLoading;

  const handleSend = () => {
    if (!input.trim()) return;
    stream.submit(
      { messages: [{ type: "human", content: input.trim() }] },
      { onDisconnect: "continue", streamResumable: true }
    );
    setInput("");
  };

  return (
    <div className="chat-container">
      <header>
        <h2>加入和重新加入演示</h2>
        <ConnectionStatus connected={isConnected} />
      </header>

      <div className="messages">
        {stream.messages.map((msg, i) => (
          <MessageBubble key={i} message={msg} />
        ))}
      </div>

      <div className="controls">
        <form onSubmit={(e) => { e.preventDefault(); handleSend(); }}>
          <input
            value={input}
            onChange={(e) => setInput(e.target.value)}
            placeholder="输入消息..."
          />
          <button type="submit">发送</button>
        </form>

        <div className="stream-actions">
          {isConnected ? (
            <button onClick={() => stream.stop()}>
              断开连接
            </button>
          ) : (
            savedRunId && (
              <button onClick={() => stream.joinStream(savedRunId)}>
                重新加入流
              </button>
            )
          )}
        </div>
      </div>
    </div>
  );
}

最佳实践

始终保存运行 ID：没有它就无法重新加入。同时使用组件状态和持久存储以增强鲁棒性。
显示清晰的连接状态：用户应始终知道他们是在接收实时更新还是在查看快照。
在可见性变化时自动重新加入：使用 Page Visibility API 在用户返回标签页时自动重新加入。
设置合理的超时：如果重新加入尝试耗时太长，回退到获取线程历史。
清理已完成的运行：智能体完成时移除持久化的运行 ID，以避免过期的重新加入尝试。

将这些文档连接到 Claude、VSCode 等工具，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交 issue。

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

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

核心概念

设置 `useStream`

使用可恢复选项提交

断开流连接

重新加入流

构建连接状态指示器

断开和重新加入控件

持久化运行 ID

错误处理

完整示例

最佳实践

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

​核心概念

​设置 useStream

​使用可恢复选项提交

​断开流连接

​重新加入流

​构建连接状态指示器

​断开和重新加入控件

​持久化运行 ID

​错误处理

​完整示例

​最佳实践

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

核心概念

设置 `useStream`

使用可恢复选项提交

断开流连接

重新加入流

构建连接状态指示器

断开和重新加入控件

持久化运行 ID

错误处理

完整示例

最佳实践