workflow-stop-design

1. Redis 状态管理方案

1.1 状态键设计

Redis Key 结构:

// Key 格式: agent_runtime_stopping:{appId}:{chatId}
const WORKFLOW_STATUS_PREFIX = 'agent_runtime_stopping';

type WorkflowStatusKey = `${typeof WORKFLOW_STATUS_PREFIX}:${string}:${string}`;

// 示例: agent_runtime_stopping:app_123456:chat_789012

状态值设计:

• 存在键 (值为 1): 工作流应该停止
• 不存在键: 工作流正常运行
• 设计简化: 不使用状态枚举,仅通过键的存在与否判断

参数类型定义:

type WorkflowStatusParams = {
  appId: string;
  chatId: string;
};

1.2 状态生命周期管理

状态转换流程:

正常运行(无键) → 停止中(键存在) → 完成(删除键)

TTL 设置:

• 停止标志 TTL: 60 秒
• 原因: 避免因意外情况导致的键泄漏
• 正常情况下会在工作流完成时主动删除
• 工作流完成后: 直接删除 Redis 键
• 原因: 不需要保留终态,减少 Redis 内存占用

1.3 核心函数说明

1. setAgentRuntimeStop

• 功能: 设置停止标志
• 参数: { appId, chatId }
• 实现: 使用 SETEX 命令,设置键值为 1,TTL 60 秒

2. shouldWorkflowStop

• 功能: 检查工作流是否应该停止
• 参数: { appId, chatId }
• 返回: Promise<boolean> - true=应该停止, false=继续运行
• 实现: GET 命令获取键值,存在则返回 true

3. delAgentRuntimeStopSign

• 功能: 删除停止标志
• 参数: { appId, chatId }
• 实现: DEL 命令删除键

4. waitForWorkflowComplete

• 功能: 等待工作流完成(停止标志被删除)
• 参数: { appId, chatId, timeout?, pollInterval? }
• 实现: 轮询检查停止标志是否被删除,超时返回

1.4 边界情况处理

1. Redis 操作失败

• 错误处理: 所有 Redis 操作都包含 .catch() 错误处理
• 降级策略:
• shouldWorkflowStop: 出错时返回 false (认为不需要停止,继续运行)
• delAgentRuntimeStopSign: 出错时记录错误日志,但不影响主流程
• 设计原因: Redis 异常不应阻塞工作流运行,降级到继续执行策略

2. TTL 自动清理

• TTL 设置: 60 秒
• 清理时机: Redis 自动清理过期键
• 设计原因:
• 避免因异常情况导致的 Redis 键泄漏
• 自动清理减少手动维护成本
• 60 秒足够大多数工作流完成停止操作

3. stop 接口等待超时

• 超时时间: 5 秒
• 超时策略: waitForWorkflowComplete 在 5 秒内轮询检查停止标志是否被删除
• 超时处理: 5 秒后直接返回,不影响工作流继续执行
• 设计原因:
• 避免前端长时间等待
• 5 秒足够大多数节点完成当前操作
• 用户体验优先,超时后前端可选择重试或放弃

4. 并发停止请求

• 处理方式: 多次调用 setAgentRuntimeStop 是安全的,Redis SETEX 是幂等操作
• 设计原因: 避免用户多次点击停止按钮导致的问题

2. Redis 工具函数实现

位置: packages/service/core/workflow/dispatch/workflowStatus.ts

import { addLog } from '../../../common/system/log';
import { getGlobalRedisConnection } from '../../../common/redis/index';
import { delay } from '@fastgpt/global/common/system/utils';

const WORKFLOW_STATUS_PREFIX = 'agent_runtime_stopping';
const TTL = 60; // 60秒

export const StopStatus = 'STOPPING';

export type WorkflowStatusParams = {
  appId: string;
  chatId: string;
};

// 获取工作流状态键
export const getRuntimeStatusKey = (params: WorkflowStatusParams): string => {
  return `${WORKFLOW_STATUS_PREFIX}:${params.appId}:${params.chatId}`;
};

// 设置停止标志
export const setAgentRuntimeStop = async (params: WorkflowStatusParams): Promise<void> => {
  const redis = getGlobalRedisConnection();
  const key = getRuntimeStatusKey(params);
  await redis.setex(key, TTL, 1);
};

// 删除停止标志
export const delAgentRuntimeStopSign = async (params: WorkflowStatusParams): Promise<void> => {
  const redis = getGlobalRedisConnection();
  const key = getRuntimeStatusKey(params);
  await redis.del(key).catch((err) => {
    addLog.error(`[Agent Runtime Stop] Delete stop sign error`, err);
  });
};

// 检查工作流是否应该停止
export const shouldWorkflowStop = (params: WorkflowStatusParams): Promise<boolean> => {
  const redis = getGlobalRedisConnection();
  const key = getRuntimeStatusKey(params);
  return redis
    .get(key)
    .then((res) => !!res)
    .catch(() => false);
};

/**
 * 等待工作流完成(停止标志被删除)
 * @param params 工作流参数
 * @param timeout 超时时间(毫秒),默认5秒
 * @param pollInterval 轮询间隔(毫秒),默认50毫秒
 */
export const waitForWorkflowComplete = async ({
  appId,
  chatId,
  timeout = 5000,
  pollInterval = 50
}: {
  appId: string;
  chatId: string;
  timeout?: number;
  pollInterval?: number;
}) => {
  const startTime = Date.now();

  while (Date.now() - startTime < timeout) {
    const sign = await shouldWorkflowStop({ appId, chatId });

    // 如果停止标志已被删除,说明工作流已完成
    if (!sign) {
      return;
    }

    // 等待下一次轮询
    await delay(pollInterval);
  }

  // 超时后直接返回
  return;
};

测试用例位置: test/cases/service/core/app/workflow/workflowStatus.test.ts

import { describe, test, expect, beforeEach } from 'vitest';
import {
  setAgentRuntimeStop,
  delAgentRuntimeStopSign,
  shouldWorkflowStop,
  waitForWorkflowComplete
} from '@fastgpt/service/core/workflow/dispatch/workflowStatus';

describe('Workflow Status Redis Functions', () => {
  const testAppId = 'test_app_123';
  const testChatId = 'test_chat_456';

  beforeEach(async () => {
    // 清理测试数据
    await delAgentRuntimeStopSign({ appId: testAppId, chatId: testChatId });
  });

  test('should set stopping sign', async () => {
    await setAgentRuntimeStop({ appId: testAppId, chatId: testChatId });
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(true);
  });

  test('should return false for non-existent status', async () => {
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(false);
  });

  test('should return false after deleting stop sign', async () => {
    await setAgentRuntimeStop({ appId: testAppId, chatId: testChatId });
    await delAgentRuntimeStopSign({ appId: testAppId, chatId: testChatId });
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(false);
  });

  test('should wait for workflow completion', async () => {
    // 设置初始停止标志
    await setAgentRuntimeStop({ appId: testAppId, chatId: testChatId });

    // 模拟异步完成(删除停止标志)
    setTimeout(async () => {
      await delAgentRuntimeStopSign({ appId: testAppId, chatId: testChatId });
    }, 500);

    // 等待完成
    await waitForWorkflowComplete({
      appId: testAppId,
      chatId: testChatId,
      timeout: 2000
    });

    // 验证停止标志已被删除
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(false);
  });

  test('should timeout when waiting too long', async () => {
    await setAgentRuntimeStop({ appId: testAppId, chatId: testChatId });

    // 等待超时(不删除标志)
    await waitForWorkflowComplete({
      appId: testAppId,
      chatId: testChatId,
      timeout: 100
    });

    // 验证停止标志仍然存在
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(true);
  });

  test('should handle concurrent stop sign operations', async () => {
    // 并发设置停止标志
    await Promise.all([
      setAgentRuntimeStop({ appId: testAppId, chatId: testChatId }),
      setAgentRuntimeStop({ appId: testAppId, chatId: testChatId })
    ]);

    // 停止标志应该存在
    const shouldStop = await shouldWorkflowStop({ appId: testAppId, chatId: testChatId });
    expect(shouldStop).toBe(true);
  });
});

3. 工作流停止检测机制改造

3.1 修改位置

文件: packages/service/core/workflow/dispatch/index.ts

3.2 工作流启动时的停止检测机制

改造点 1: 停止检测逻辑 (行 196-216)

使用内存变量 + 定时轮询 Redis 的方式:

import { delAgentRuntimeStopSign, shouldWorkflowStop } from './workflowStatus';

// 初始化停止检测
let stopping = false;
const checkIsStopping = (): boolean => {
  if (apiVersion === 'v2') {
    return stopping;
  }
  if (apiVersion === 'v1') {
    if (!res) return false;
    return res.closed || !!res.errored;
  }
  return false;
};

// v2 版本: 启动定时器定期检查 Redis
const checkStoppingTimer =
  apiVersion === 'v2'
    ? setInterval(async () => {
        stopping = await shouldWorkflowStop({
          appId: runningAppInfo.id,
          chatId
        });
      }, 100)
    : undefined;

设计要点:

• v2 版本使用内存变量 stopping + 100ms 定时器轮询 Redis
• v1 版本仍使用原有的 res.closed/res.errored 检测
• 轮询频率 100ms,平衡性能和响应速度

改造点 2: 工作流完成后清理 (行 232-249)

return runWorkflow({
  ...data,
  checkIsStopping,  // 传递检测函数
  query,
  histories,
  // ... 其他参数
}).finally(async () => {
  // 清理定时器
  if (streamCheckTimer) {
    clearInterval(streamCheckTimer);
  }
  if (checkStoppingTimer) {
    clearInterval(checkStoppingTimer);