加入与重新加入流

加入与重新加入功能允许您从正在运行的代理流中断开连接，而不停止代理，之后再重新连接。当客户端离开时，代理继续在服务器端执行，您可以从断开的确切位置继续处理流。

此功能需要 LangGraph Agent Server。请在本地使用 langgraph dev 运行您的代理，或将其部署到 LangSmith 以使用此模式。

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

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

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

核心概念

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

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

stream.stop() 与取消运行有根本区别。停止仅断开客户端连接。代理继续在服务器端处理。要实际取消代理的执行，您需要使用中断或取消机制。

设置 `useStream`

关键设置步骤是从 onCreated 回调中捕获 run_id，以便稍后重新加入。定义一个与代理状态模式匹配的 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：没有它，重新加入是不可能的。同时使用组件状态和持久化存储以提高弹性。
显示清晰的连接状态：用户应始终知道他们是正在接收实时更新还是查看快照。
在可见性变化时自动重新加入：使用页面可见性 API 在用户返回标签页时自动重新加入。
设置合理的超时时间：如果重新加入尝试耗时过长，则回退到获取线程历史记录。
清理已完成的运行：在代理完成时移除持久化的运行 ID，以避免过期的重新加入尝试。

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

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

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

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

核心概念

设置 `useStream`

使用可恢复选项提交

从流中断开连接

重新加入流

构建连接状态指示器

断开连接和重新加入控件

持久化运行 ID

错误处理

完整示例

最佳实践

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

​核心概念

​设置 useStream

​使用可恢复选项提交

​从流中断开连接

​重新加入流

​构建连接状态指示器

​断开连接和重新加入控件

​持久化运行 ID

​错误处理

​完整示例

​最佳实践

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

核心概念

设置 `useStream`

使用可恢复选项提交

从流中断开连接

重新加入流

构建连接状态指示器

断开连接和重新加入控件

持久化运行 ID

错误处理

完整示例

最佳实践