消息队列

在 Agent 顺序处理的同时，将多条消息加入队列并进行管理

消息队列允许用户快速连续发送多条消息，而无需等待 Agent 完成对当前消息的处理。每条消息都会被立即接收，加入到当前活动线程的队列中，并按顺序依次处理，从而让你对所有待处理工作拥有完全的可见性与控制力。

NOTE

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

为什么需要消息队列？

在典型的聊天界面中，用户必须等待 Agent 响应完毕后才能发送下一条消息。这在以下几种场景中会带来体验阻碍：

批量提问：用户希望一次性提出五个相关问题，而不是等待每个问题的答案逐个返回
跟进链条（Follow-up chains）：在 Agent 仍在工作时提交澄清说明或额外的上下文信息
自动化测试序列：通过编程方式发送一系列提示词以验证 Agent 的行为
数据录入工作流：接连不断地输入结构化数据以供处理

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

这是一种 Agent 用户体验（UX）原语，而不仅是表面的聊天功能。SDK 将队列作为流控制器（stream controller）的一部分进行跟踪，因此你的 UI 可以在当前运行（run）继续进行的同时展示待处理工作、取消过期的请求，并保持输入框处于可用激活状态。

工作原理

当你想让某次提交在当前正在运行的请求之后排队等待时，传入 multitaskStrategy: "enqueue"。在 Agent 处于处理过程中时，排队的提交会被添加到当前活动线程的队列中。一旦当前运行完成，下一条排队的消息就会被自动派发。

使用对应框架的配套队列辅助工具读取队列状态：

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

每个 SubmissionQueueEntry 对象包含：

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

配置 `useStream`

将 useStream 连接到你的 Agent，然后将其与适用于你的框架的提交队列辅助函数搭配使用。在运行进行中调用 stream.submit() 发送消息；在需要在当前活动请求之后排队等待的提交上，传入 multitaskStrategy: "enqueue"。读取 queue.entries 和 queue.size 来渲染待处理工作，并使用 queue.cancel() 或 queue.clear() 在条目开始处理之前将其移除。

NOTE

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

ReactVueSvelteAngular

tsx

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

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

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

  const pendingCount = queue.size;
  const entries = queue.entries;

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

vue

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

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

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

const pendingCount = computed(() => queue.size.value);
const entries = computed(() => queue.entries.value);
`&lt;/script&gt;`

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

svelte

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

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

  function handleSubmit(text: string) {
    stream.submit({
      messages: [{ type: "human", content: text }],
    });
  }
`&lt;/script&gt;`

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

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

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

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

显示队列

构建一个 QueueList 组件，用于显示每条排队中的消息并提供取消按钮。这使用户能够清晰看到正在等待的内容，并有能力移除他们不再需要的条目。

tsx

function QueueList({ entries, queue }) {
  return (
    <div className="queue-panel">
      <div className="queue-header">
        <span>Queued messages ({entries.length})</span>
        <button onClick={() => queue.clear()}>Clear all</button>
      </div>
      <ul className="queue-entries">
        {entries.map((entry) => {
          const text = entry.values?.messages?.at(-1)?.content ?? "Pending...";
          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)}
              >
                Cancel
              </button>
            </li>
          );
        })}
      </ul>
    </div>
  );
}

TIP

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

取消排队的消息

你拥有两个层级的取消操作：

取消单个条目

通过 ID 从队列中移除指定消息。Agent 将跳过该消息并继续处理下一个条目。

await queue.cancel(entryId);

清空整个队列

一次性移除所有待处理消息。在用户切换上下文或想要重新开始时非常有用。

await queue.clear();

NOTE

取消队列条目仅对尚未开始处理的消息生效。如果 Agent 已经在处理某条消息，从队列中取消它将不起作用。请使用 stream.stop() 来中断当前运行。

使用 `onCreated` 链式提交后续消息

onCreated 回调在创建新的运行（run）时触发，为你提供了一个以编程方式提交后续消息的钩子。这对于构建多步骤工作流非常有用，其中下一个问题取决于上一个提交已被成功接收。

stream.submit(
  { messages: [{ type: "human", content: "What is quantum computing?" }] },
  {
    onCreated(run) {
      console.log("Run created:", run.runId);
      // 链式追加后续消息
      stream.submit({
        messages: [{ type: "human", content: "Give me a simple analogy." }],
      });
    },
  }
);

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

开始新线程

当用户想要开始全新对话时，更新你传递给流的响应式 threadId。传入 null 将清除当前的线程绑定；下一次提交将创建一个新的线程。

ReactVueSvelteAngular

tsx

function NewThreadButton() {
  const [threadId, setThreadId] = useState<string | null>(null);
  const stream = useStream<typeof myAgent>({ threadId, onThreadId: setThreadId });

  return (
    <button onClick={() => setThreadId(null)}>
      New conversation
    </button>
  );
}

vue

<script setup lang="ts">
const threadId = ref<string | null>(null);
const stream = useStream<typeof myAgent>({
  threadId,
  onThreadId: (id) => (threadId.value = id),
});
`&lt;/script&gt;`

<template>
  <button @click="threadId = null">New conversation</button>
</template>

svelte

<script lang="ts">
  let threadId = $state<string | null>(null);
  const stream = useStream<typeof myAgent>({
    threadId: () => threadId,
    onThreadId: (id) => (threadId = id),
  });
`&lt;/script&gt;`

<button onclick={() => (threadId = null)}>New conversation</button>

threadId = signal<string | null>(null);
stream = injectStream<typeof myAgent>({
  threadId: this.threadId,
  onThreadId: (id) => this.threadId.set(id),
});

// 在模板中：
// <button (click)="threadId.set(null)">New conversation</button>

最佳实践

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

消息队列 ​

为什么需要消息队列？ ​

工作原理 ​

配置 useStream ​

显示队列 ​

取消排队的消息 ​

取消单个条目 ​

清空整个队列 ​

使用 onCreated 链式提交后续消息 ​

开始新线程 ​

最佳实践 ​

消息队列

为什么需要消息队列？

工作原理

配置 `useStream`

显示队列

取消排队的消息

取消单个条目

清空整个队列

使用 `onCreated` 链式提交后续消息

开始新线程

最佳实践