用户点了停止，Agent怎么安全收尾

上一篇讲了同一会话的互斥：一轮运行持有会话锁，收尾后才释放。用户主动点“停止”时，还要解决一个问题：停止请求可能落到另一个服务节点，怎样让正在运行的Agent停下来？

以比特严选的一次实测为例，用户问：

我上周买的AirPods Pro左耳突然没声音了，放回充电盒也没用，你帮我看看是不是坏了

Agent查完订单和排查手册，开始输出回答。用户读到“先说结论：从你的描述看，左耳‘突然完全无声 + 放回充电盒也没用’，这类”时，点了停止。页面上的半截回复要保留，下一轮模型也要知道刚才说到了哪里。

这涉及两份数据：

数据	用来做什么
消息表t_agent_message	保存回复正文、思考内容和工具调用记录，供页面展示和历史回放
框架状态t_agent_state	保存Agent的会话状态，其中的上下文会在下一轮调用模型时使用

一条回复里的回答文本、思考内容、工具调用记录等部分，在项目里分别用“块”表示。后文的“保存消息”指写入消息表，“保存状态”指保存框架状态。

整条停止流程是：前端发送停止请求，执行节点收到通知后中断Agent、保存已有内容，再通过原来的SSE连接通知页面结束。工具已经执行的业务操作是否还能撤回，后面用提交工单的例子说明。

本文实测来自本地实例，智能体是比特严选，模型是deepseek-flash。测试脚本分别在回答输出中、检索工具执行中、用户确认提交工单后触发停止，结果放在机制讲解之后对照。

停止请求怎么找到正在运行的那一轮

1. 前端发停止请求，但不断开连接

用户点停止后，前端不中断正在读取的SSE连接，只记下cancelRequested，再调用POST /api/ragent/agent/v1/stop?taskId=…。taskId是这一轮开始时META事件带回来的，后端用它标识一次运行。

agentChatStore.ts中的实现如下：

cancelGeneration: () => {
  const { isStreaming, streamTaskId, cancelRequested } = get();
  if (!isStreaming || cancelRequested) return;
  // 不中断 fetch：后端落库部分内容后回发 cancel + done 完成收尾
  set({ cancelRequested: true });
  if (streamTaskId) {
    requestStop(streamTaskId, get().currentViewKey);
  }
},

连接保留着，是为了接收后端的CANCEL事件。已保存回复时，这个事件会带回消息ID，前端据此更新页面上的临时回复，并将它标记为已停止。如果直接断开连接，前端就接不到这次保存的结果，需要重新加载会话才能看到库里的记录。

cancelRequested记录本轮已经发起停止，避免重复点击。如果点击时还没收到META、拿不到taskId，前端先保留这个标记，收到META后补发请求；请求失败则清除标记，让用户可以重试。

2. 停止请求可能落在另一个节点

停止接口调到StreamTaskManager.cancelByUser()。接收停止请求的节点不一定在运行这一轮，所以它不直接动手，而是通过Redis通知所有节点：

public void cancelByUser(String taskId) {
    String requester = UserContext.requireUser().getUserId();
    RBucket<String> owner = redissonClient.getBucket(ownerKey(taskId));
    String ownerUserId = owner.get();
    if (StrUtil.isNotBlank(ownerUserId) && !ownerUserId.equals(requester)) {
        // 省略日志
        // 属主不匹配时返回统一提示
        throw new ClientException("任务不存在或已结束");
    }
    // Redis 里查不到用户 ID：多半是这一轮已经结束，也可能执行节点还没登记
    // 照样写标记、发广播，带上点停止的用户，等执行节点登记时再核对
    publishCancel(taskId, requester);
}

private void publishCancel(String taskId, String requester) {
    // 先设置 Redis 标记，再发布消息
    RBucket<String> bucket = redissonClient.getBucket(cancelKey(taskId));
    bucket.set(requester, CANCEL_TTL);
    // 省略注释
    redissonClient.getTopic(CANCEL_TOPIC).publish(taskId + PAYLOAD_SEPARATOR + requester);
}

一轮开始时，执行节点把所属用户的ID写进Redis的ragent:stream:owner:{taskId}。停止接口查到属主后，与当前登录用户比对，不匹配就拒绝。随后通过Redis做两件事：

• 写取消标记：Redis里的一个普通键ragent:stream:cancel:{taskId}，值是点停止的用户ID，保留30分钟。
• 发一条广播：用Redis的发布订阅，往频道ragent:stream:cancel发一条消息，内容是taskId|用户 ID。消息不存进Redis，只推给此刻订阅了这个频道的节点。

广播用于通知已经登记的任务。取消标记则补上登记之前的情况：如果停止请求先到、任务还没登记，接口仍会写标记、发广播；执行节点登记任务时再核对标记里的用户ID，属于本轮用户就执行取消，不再启动Agent。

广播发布后，停止接口就返回成功，不等待消息保存和资源释放。前端继续通过原SSE连接接收最终结果。

3. 执行节点先中断再收尾

每个节点启动时都订阅了这个频道，收到广播后调用cancelLocal()。只有运行这一轮的节点在本机内存里登记过这个taskId，其他节点查不到，直接返回。停止请求恰好落在执行节点上时也一样：停止接口只发广播，不直接调用cancelLocal()，本机同样等广播到了再处理，两种情况走同一条路径。

查到之后的后半段如下：

// 点停止的用户和这一轮的用户对不上，就不取消
if (isRequesterDenied(taskInfo, requester)) {
    // 省略日志
    return;
}

// 只有第一次取消能把 false 改成 true，后续重复取消直接返回
if (!taskInfo.cancelled.compareAndSet(false, true)) {
    return;
}

if (taskInfo.cancelAction != null) {
    taskInfo.cancelAction.run();
}

// 在取消时执行收尾回调，保存已累积的内容
if (taskInfo.finalizer != null) {
    taskInfo.finalizer.run();
}

cancelAction和finalizer是一轮启动时登记的两个动作：前者调用AgentRunHandle.interruptUpstream()，通知框架中断并等待事件流结束，超时则取消订阅；后者调用AgentStreamEventBridge.finishCancelledStream()，处理消息保存和结束通知。AgentStreamEventBridge下文简称事件桥，负责把Agent的事件转换成发给页面的SSE事件，并累积本轮的消息块。

cancelLocal()是被Redis广播触发的，执行它的是执行节点上Redisson收到广播的那条回调线程。两个动作在这条线程上依次执行，interruptUpstream()不返回，finishCancelledStream()就不会开始，所以收尾一定发生在中断动作返回之后。

SSE连接超时或断开时，服务端通过本机的cancel(taskId)进入同一套中断和收尾逻辑，不经过Redis广播。

执行节点怎样停止Agent

执行节点收到取消广播时，Agent还在运行。Redisson回调线程负责通知它停止，运行Agent的框架线程负责检查中断标志并结束当前执行。先看线程分工，再看它们之间怎样通知和等待。

1. 一轮运行用到哪些线程

下图展示一次回答输出期间停止的过程，各条线程分别做了什么。停止请求线程可能在任意节点上，画在执行节点框外：

请求线程通过订阅事件流启动Agent：agent.streamEvents()返回一个Reactor的Flux，调用subscribe()后开始执行。请求线程先执行开头的同步部分，遇到框架配置的调度点后，把后续工作交给线程池，再返回。此时SSE连接仍然保留，Agent会继续运行，事件桥接收事件并向前端发送数据。

一轮运行没有从头到尾固定使用的专属线程。AgentScope 2.0.2通过subscribeOn(Schedulers.boundedElastic())，把读取模型响应、执行工具、保存状态等工作调度到线程池中，前后两步可能由不同线程执行。下文的“框架线程”指正在执行这些工作的池线程。

按职责区分，这里有四类线程：

线程角色	线程名称示例	负责什么
对话请求线程	http-nio-*	Tomcat线程，发送META、登记任务、订阅事件流；交出后续工作后返回，不等待整轮结束
框架线程	boundedElastic-*	来自Reactor的共享线程池，供AgentScope和项目代码调度使用；读取模型响应、执行工具、保存框架状态，并调用事件桥处理事件
停止请求线程	http-nio-*	Tomcat线程，处理另一个HTTP请求：核对归属、写取消标记、发布广播，然后返回
Redisson回调线程	redisson-*	Redisson客户端的回调线程。执行节点收到取消广播后，在这条线程上调用cancelLocal()，先执行中断动作，再执行取消收尾

对话请求线程和停止请求线程都由各自所在节点的Tomcat请求线程池提供，同一节点上共用请求线程池。这里按处理的请求分别列出。

下文的onComplete()是Agent事件流正常结束时调用的事件桥方法。它与浏览器SSE连接的完成回调是两回事：Agent停下来之后，项目还需要保存消息，再通过SSE把结果发给页面。

解锁付费内容，👉戳