消息队列

消息队列允许用户快速连续发送多条消息，而无需等待代理完成当前消息的处理。每条消息在服务器端入队并按顺序处理，使您能够完全查看和控制待处理队列。

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

为什么需要消息队列？

在典型的聊天界面中，用户必须等待代理完成响应后才能发送另一条消息。这在以下几种场景中会造成不便：

批量提问：用户希望一次性提出五个相关问题，而不是等待每个答案
后续链：在代理仍在工作时提交澄清或额外上下文
自动化测试序列：以编程方式发送一系列提示以验证代理行为
数据录入工作流：连续输入结构化输入以进行处理

消息队列通过立即接受所有提交并按顺序处理来解决这个问题。

工作原理

在底层，LangGraph 使用 multitaskStrategy: "enqueue" 来管理并发提交。当代理正在处理消息时提交新消息，它会被添加到服务器端队列中。当前运行完成后，下一个排队的消息会自动被拾取。 useStream 钩子暴露了一个 queue 属性，提供对挂起消息的实时可见性：

属性	类型	描述
`queue.entries`	`QueueEntry[]`	所有待处理队列条目的数组
`queue.size`	`number`	当前队列中的条目数量
`queue.cancel(id)`	`(id: string) => Promise<void>`	通过 ID 取消特定的队列条目
`queue.clear()`	`() => Promise<void>`	取消所有队列条目

每个 QueueEntry 对象包含：

字段	类型	描述
`id`	`string`	此队列条目的唯一标识符
`values`	`object`	提交的输入值（包括消息）
`options`	`object`	随提交传递的任何附加选项
`createdAt`	`string`	条目创建时的 ISO 时间戳

设置 `useStream`

定义一个与代理状态模式匹配的 TypeScript 接口，并将其作为类型参数传递给 useStream，以实现对状态值的类型安全访问。在下面的示例中，请将 typeof myAgent 替换为您的接口名称：

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

interface AgentState {
  messages: BaseMessage[];
}

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

function Chat() {
  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "message_queue",
  });

  const handleSubmit = (text: string) => {
    stream.submit({
      messages: [{ type: "human", content: text }],
    });
  };

  // 访问队列状态
  const pendingCount = stream.queue.size;
  const entries = stream.queue.entries;

  return (
    <div>
      <MessageList messages={stream.messages} />
      {pendingCount > 0 && <QueueList entries={entries} queue={stream.queue} />}
      <ChatInput onSubmit={handleSubmit} />
    </div>
  );
}

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

const stream = useStream<typeof myAgent>({
  apiUrl: "http://localhost:2024",
  assistantId: "message_queue",
});

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

// 在脚本中通过 .value 访问队列状态
const pendingCount = computed(() => stream.queue.value.size);
const entries = computed(() => stream.queue.value.entries);
</script>

<template>
  <div>
    <MessageList :messages="stream.messages" />
    <QueueList v-if="stream.queue.size > 0" :entries="stream.queue.entries" :queue="stream.queue" />
    <ChatInput @submit="handleSubmit" />
  </div>
</template>

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

  const { messages, submit, queue } = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "message_queue",
  });

  function handleSubmit(text: string) {
    submit({
      messages: [{ type: "human", content: text }],
    });
  }
</script>

<div>
  <MessageList messages={$messages} />
  {#if $queue.size > 0}
    <QueueList entries={$queue.entries} queue={$queue} />
  {/if}
  <ChatInput on:submit={(e) => handleSubmit(e.detail)} />
</div>

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

@Component({
  selector: "app-chat",
  template: `
    <message-list [messages]="stream.messages()" />
    @if (stream.queue().size > 0) {
      <queue-list [entries]="stream.queue().entries" [queue]="stream.queue()" />
    }
    <chat-input (onSubmit)="handleSubmit($event)" />
  `,
})
export class ChatComponent {
  stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "message_queue",
  });

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

显示队列

构建一个 QueueList 组件，显示每条待处理消息并带有取消按钮。这使用户能够查看等待中的内容，并能够移除不再需要的项目。

function QueueList({ entries, queue }) {
  return (
    <div className="queue-panel">
      <div className="queue-header">
        <span>排队消息 ({entries.length})</span>
        <button onClick={() => queue.clear()}>全部清除</button>
      </div>
      <ul className="queue-entries">
        {entries.map((entry) => {
          const text = entry.values?.messages?.[0]?.content ?? "未知";
          return (
            <li key={entry.id} className="queue-entry">
              <span className="queue-text">{text}</span>
              <span className="queue-time">
                {new Date(entry.createdAt).toLocaleTimeString()}
              </span>
              <button
                className="queue-cancel"
                onClick={() => queue.cancel(entry.id)}
              >
                取消
              </button>
            </li>
          );
        })}
      </ul>
    </div>
  );
}

显示每条排队消息的前几个字符作为预览，以便用户无需阅读完整消息即可快速识别要取消的项目。

取消排队消息

您有两个级别的取消操作：

取消单个条目

通过 ID 从队列中移除特定消息。代理将跳过它并处理下一个条目。

await queue.cancel(entryId);

清除整个队列

一次性移除所有待处理消息。当用户更改上下文或想要重新开始时很有用。

await queue.clear();

取消队列条目仅影响尚未开始处理的消息。如果代理已经在处理某条消息，从队列中取消它不会产生任何效果。请使用 stream.stop() 来中断当前运行。

使用 `onCreated` 链接后续提交

onCreated 回调在创建新运行时触发，为您提供一个钩子以编程方式提交后续消息。这对于构建多步骤工作流很有用，其中下一个问题取决于前一个提交已被接受。

stream.submit(
  { messages: [{ type: "human", content: "什么是量子计算？" }] },
  {
    onCreated(run) {
      console.log("运行已创建:", run.run_id);
      // 链接一个后续消息
      stream.submit({
        messages: [{ type: "human", content: "给我一个简单的类比。" }],
      });
    },
  }
);

此模式自然地填充队列。第一条消息立即开始处理，后续消息则在其后排队。

开始新线程

当用户想要开始新的对话时，使用 switchThread(null) 创建一个新线程。这会清除当前的消息历史记录和队列。

function NewThreadButton() {
  const stream = useStream<typeof myAgent>({ /* ... */ });

  return (
    <button onClick={() => stream.switchThread(null)}>
      新对话
    </button>
  );
}

<script setup lang="ts">
const stream = useStream<typeof myAgent>({ /* ... */ });
</script>

<template>
  <button @click="stream.switchThread(null)">新对话</button>
</template>

<script lang="ts">
  const { switchThread } = useStream<typeof myAgent>({ /* ... */ });
</script>

<button on:click={() => switchThread(null)}>新对话</button>

stream = useStream<typeof myAgent>({ /* ... */ });

// 在模板中：
// <button (click)="stream.switchThread(null)">新对话</button>

完整示例

将所有内容整合在一起，这是一个带有队列管理的完整聊天组件：

function QueueChat() {
  const stream = useStream<typeof myAgent>({
    apiUrl: "http://localhost:2024",
    assistantId: "message_queue",
  });

  const [input, setInput] = useState("");

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

  return (
    <div className="chat-container">
      <header>
        <h2>队列聊天</h2>
        <button onClick={() => stream.switchThread(null)}>新线程</button>
      </header>

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

      {stream.queue.size > 0 && (
        <div className="queue-panel">
          <strong>排队中 ({stream.queue.size})</strong>
          <button onClick={() => stream.queue.clear()}>全部清除</button>
          {stream.queue.entries.map((entry) => (
            <div key={entry.id} className="queue-item">
              <span>{entry.values?.messages?.[0]?.content}</span>
              <button onClick={() => stream.queue.cancel(entry.id)}>×</button>
            </div>
          ))}
        </div>
      )}

      <form onSubmit={(e) => { e.preventDefault(); handleSubmit(); }}>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="输入消息（您可以发送多条！）"
        />
        <button type="submit">发送</button>
      </form>
    </div>
  );
}

最佳实践

限制队列大小：虽然客户端对队列大小没有硬性限制，但请注意过大的队列会降低用户体验。考虑在队列超过合理阈值（例如 10 个项目）时显示警告。
显示队列位置：为每个排队项目编号，以便用户了解处理顺序。
保持输入焦点：提交后保持输入框聚焦，以便用户可以立即输入下一条消息。
动画过渡：当项目开始处理时，平滑地将它们从队列面板移动到消息列表中。
优雅地处理错误：如果排队消息失败，请显示错误而不阻塞后续队列条目。
对快速提交进行去抖动：对于自动化或编程提交，在消息之间添加一个小延迟，以避免使服务器不堪重负。

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

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

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Deploy with LangSmith

为什么需要消息队列？

工作原理

设置 `useStream`

显示队列

取消排队消息

取消单个条目

清除整个队列

使用 `onCreated` 链接后续提交

开始新线程

完整示例

最佳实践

​为什么需要消息队列？

​工作原理

​设置 useStream

​显示队列

​取消排队消息

​取消单个条目

​清除整个队列

​使用 onCreated 链接后续提交

​开始新线程

​完整示例

​最佳实践

为什么需要消息队列？

工作原理

设置 `useStream`

显示队列

取消排队消息

取消单个条目

清除整个队列

使用 `onCreated` 链接后续提交

开始新线程

完整示例

最佳实践