Skip to content

赵月辉 / fastgpt-migrated

Go to a project
Commit 6527bebf by Archer Committed by GitHub
Options

fix(workflow): record loop and parallel item runtimes independently (#7312) 

* fix(workflow): record container item runtime independently

* docs: update v4.15.2 workflow timing fix

* test(workflow): update runtime summary assertions
parent f60c4591
Showing with 102 additions and 36 deletions
document/content/self-host/upgrading/4-15/4152.en.mdx
......@@ -24,6 +24,7 @@ description: 'FastGPT V4.15.2 Release Notes'
3. Custom chunk delimiters now reject a single `|` or consecutive `||` to prevent incorrect parsing of large numbers of chunks.
4. Improved automatic license purchase logic for WeCom edition customers after payment to prevent duplicate or missing license purchases.
5. Fixed empty Tag labels in the Plugin Marketplace.
6. Fixed runtime calculations for LoopRun iterations and ParallelRun tasks so each item reports its own elapsed time instead of summing the runtimes of its child steps.
## 🛠️ Code Improvements
......
document/content/self-host/upgrading/4-15/4152.mdx
......@@ -18,11 +18,13 @@ description: 'FastGPT V4.15.2 更新说明'
6. 同步模式下不显示注册用户按钮。
## 🐛 修复
1. 优化 CI 流程，通过 hashtag 固定 step 版本，规避 CI 供应链投毒攻击风险。
2. 移除 PPTX 解析依赖的高风险解压库，改为流式解压解析流程，规避恶意代码执行风险。
3. 自定义分块标识符拒绝传入单一"|"或连续"||"符号，避免错误解析大量 chunks。
4. 企微版本客户付款后自动购买 license 判断逻辑优化，避免重复购买/少购买的情况
5. 插件市场空 Tag 标签问题
6. 修复循环运行节点的迭代项和并行运行节点的任务项耗时计算错误，改为分别记录每个子项的实际运行时间，不再累加子节点耗时。
## 🛠️ 代码优化
......
document/data/doc-last-modified.json
......@@ -320,8 +320,8 @@
"content/self-host/upgrading/4-15/41507.mdx": "2026-06-30T17:31:43+08:00",
"content/self-host/upgrading/4-15/4151.en.mdx": "2026-07-07T21:14:28+08:00",
"content/self-host/upgrading/4-15/4151.mdx": "2026-07-07T21:14:28+08:00",
"content/self-host/upgrading/4-15/4152.en.mdx": "2026-07-08T22:37:19+08:00",
"content/self-host/upgrading/4-15/4152.mdx": "2026-07-08T22:37:19+08:00",
"content/self-host/upgrading/4-15/4152.en.mdx": "2026-07-14T14:06:21+08:00",
"content/self-host/upgrading/4-15/4152.mdx": "2026-07-14T14:06:21+08:00",
"content/self-host/upgrading/outdated/40.en.mdx": "2026-04-26T21:08:47+08:00",
"content/self-host/upgrading/outdated/40.mdx": "2026-04-26T21:08:47+08:00",
"content/self-host/upgrading/outdated/41.en.mdx": "2026-04-26T21:08:47+08:00",
......@@ -464,4 +464,4 @@
"content/self-host/upgrading/upgrade-intruction.mdx": "2026-04-26T21:08:47+08:00",
"content/toc.en.mdx": "2026-07-08T22:37:19+08:00",
"content/toc.mdx": "2026-07-06T22:53:13+08:00"
}
\ No newline at end of file
}
packages/global/core/workflow/runtime/type.ts
......@@ -168,7 +168,7 @@ export const DispatchNodeResponseSchema = z
.record(z.string(), z.any())
.optional()
.meta({ description: '模块名 i18n 插值参数' }),
runningTime: z.number().optional().meta({ description: '运行时间: 毫秒' }),
runningTime: z.number().optional().meta({ description: '运行时间: 秒' }),
query: z.string().optional().meta({ description: '查询语句' }),
textOutput: z.string().optional().meta({ description: '文本输出' }),
......
packages/service/core/workflow/dispatch/loopRun/runLoopRun.ts
......@@ -166,6 +166,7 @@ export const dispatchLoopRun = async (props: Props): Promise<Response> => {
const isResumeIteration = !!interactiveData && iteration === resumeIteration;
const loopRunNodeResponseId = props.nodeResponseParentId || node.nodeId;
const iterationResponseId = `${loopRunNodeResponseId}:iter:${iteration}`;
const iterationStartTime = Date.now();
if (isResumeIteration) {
isolatedNodes.forEach((n) => {
......@@ -208,7 +209,7 @@ export const dispatchLoopRun = async (props: Props): Promise<Response> => {
response
});
const iterationChildResponseCount = wrapperSummary.childResponseCount;
const iterationRunningTime = wrapperSummary.runningTime;
const iterationRunningTime = +((Date.now() - iterationStartTime) / 1000).toFixed(2);
assistantResponses.push(...response.assistantResponses);
const iterationTotalPoints = pushSubWorkflowUsage({
usagePush: props.usagePush,
......
packages/service/core/workflow/dispatch/parallelRun/runParallelRun.ts
......@@ -93,6 +93,8 @@ export const dispatchParallelRun = async (props: Props): Promise<Response> => {
maxRetryAttempts > 0
? `${taskResponseIdPrefix}_task_${index}_attempt_${attempt}`
: `${taskResponseIdPrefix}_task_${index}`;
const startTime = Date.now();
const getRunningTime = () => +((Date.now() - startTime) / 1000).toFixed(2);
try {
const taskVariableState = props.variableState.clone();
......@@ -108,6 +110,7 @@ export const dispatchParallelRun = async (props: Props): Promise<Response> => {
runtimeNodes: taskRuntimeNodes,
runtimeEdges: taskRuntimeEdges
});
const runningTime = getRunningTime();
// Push usage per attempt (resources were consumed regardless of success)
const attemptPoints = pushSubWorkflowUsage({
......@@ -132,6 +135,7 @@ export const dispatchParallelRun = async (props: Props): Promise<Response> => {
const attemptResult = {
...result,
taskResponseId,
runningTime,
totalPoints: attemptPoints
};
attemptResults.push(attemptResult);
......@@ -139,19 +143,26 @@ export const dispatchParallelRun = async (props: Props): Promise<Response> => {
return {
...result,
taskResponseId,
runningTime,
totalPoints: accumulatedPoints
};
}
// Non-retryable: interactive response will never succeed on retry
if (response.workflowInteractiveResponse)
return { ...result, taskResponseId, totalPoints: accumulatedPoints };
return { ...result, taskResponseId, runningTime, totalPoints: accumulatedPoints };
lastResult = { ...result, taskResponseId, totalPoints: accumulatedPoints };
lastResult = {
...result,
taskResponseId,
runningTime,
totalPoints: accumulatedPoints
};
} catch (err) {
const attemptResult = {
...parseTaskError(index, err),
taskResponseId,
runningTime: getRunningTime(),
totalPoints: 0
};
attemptResults.push(attemptResult);
......
packages/service/core/workflow/dispatch/parallelRun/service.ts
......@@ -111,6 +111,7 @@ export type ParallelTaskResult =
data: any;
response: DispatchFlowResponse;
totalPoints: number;
runningTime?: number;
taskResponseId?: string;
}
| {
......@@ -119,6 +120,7 @@ export type ParallelTaskResult =
error?: string;
response?: DispatchFlowResponse;
totalPoints: number;
runningTime?: number;
taskResponseId?: string;
};
......@@ -225,7 +227,6 @@ const buildParallelTaskWrapper = ({
const runtimeSummary = result.response
? getRuntimeNodeResponseSummary(result.response)
: undefined;
const runningTime = runtimeSummary?.runningTime || 0;
const taskNodeId = result.taskResponseId || `${parentNodeId}_task_${result.index}`;
return {
......@@ -234,7 +235,7 @@ const buildParallelTaskWrapper = ({
moduleType: FlowNodeTypeEnum.parallelRun,
moduleName: i18nT('workflow:parallel_task'),
moduleNameArgs: { index: result.index + 1 },
runningTime: Math.round(runningTime * 100) / 100,
runningTime: result.runningTime,
totalPoints: result.totalPoints,
loopInputValue: input,
loopOutputValue: result.success ? result.data : undefined,
......
packages/service/core/workflow/dispatch/type.ts
......@@ -51,8 +51,6 @@ export type RuntimeNodeResponseSummary = {
nestedEndOutput?: any;
/** pluginOutput 输出值。插件调用用它替代从完整 nodeResponse 列表里查 pluginOutput 节点。 */
pluginOutput?: Record<string, any>;
/** 子 workflow 节点运行时间总和。parallel/loopRun 虚拟包装节点展示用。 */
runningTime: number;
/** 子 workflow 顶层节点自身 totalPoints 总和。当前主要用于兜底统计。 */
totalPoints?: number;
/** 子 workflow 所有响应的积分总和，仅作为运行期费用聚合中间态，不写入 responseData。 */
......
packages/service/core/workflow/dispatch/utils/index.ts
......@@ -48,15 +48,14 @@ export const createRuntimeNodeResponseSummary = (): RuntimeNodeResponseSummary =
hasError: false,
hasLoopRunBreak: false,
hasToolStop: false,
hasNestedEnd: false,
runningTime: 0
hasNestedEnd: false
});
/**
* 增量更新父 workflow 运行控制需要的临时字段。
*
* 完整 nodeResponse 会由 writer 及时落库并释放；父节点只需要这些信号来判断
* nestedEnd 输出、错误、loop break、tool stop、完成节点、耗时和 child 统计。
* nestedEnd 输出、错误、loop break、tool stop、完成节点和 child 统计。
* 调用方每处理完一批 nodeResponse，就把当前 summary 和本批响应传进来，返回新的
* summary，避免重新保存或扫描完整 nodeResponse 列表。
*/
......@@ -78,7 +77,7 @@ export const summarizeRuntimeNodeResponses = (
.filter((parentId): parentId is string => !!parentId)
);
// 已进入 currentSummary 的 response id 不能再次计入统计，避免重复事件或分批更新导致
// runningTime/points/responseCount 被累加两次。
// points/responseCount 被累加两次。
const countedIds = new Set(initialSummary.responseIds);
const addResponseToSummary = (
......@@ -116,8 +115,6 @@ export const summarizeRuntimeNodeResponses = (
summary.pluginOutput = response.pluginOutput;
}
summary.runningTime += typeof response.runningTime === 'number' ? response.runningTime : 0;
const children = getChildrenResponses(response);
const hasConcreteChild =
children.length > 0 || (response.id ? responseIdsWithConcreteParent.has(response.id) : false);
......@@ -167,7 +164,6 @@ export const mergeRuntimeNodeResponseSummary = (
if (summary.pluginOutput !== undefined) {
merged.pluginOutput = summary.pluginOutput;
}
merged.runningTime += summary.runningTime;
merged.totalPoints = (merged.totalPoints || 0) + (summary.totalPoints || 0);
merged.childTotalPoints = (merged.childTotalPoints || 0) + (summary.childTotalPoints || 0);
merged.childResponseCount =
......
packages/service/test/core/workflow/dispatch/ai/toolcall/hooks/useToolNodeResponse.test.ts
......@@ -222,7 +222,6 @@ describe('useToolNodeResponse', () => {
responseIds: expect.arrayContaining(['interactive']),
finishedNodeIds: expect.arrayContaining(['interactive']),
hasError: false,
runningTime: 0.1,
childResponseCount: 2,
childTotalPoints: 0.2
})
......@@ -273,8 +272,7 @@ describe('useToolNodeResponse', () => {
responseIds: ['req_empty_compress'],
finishedNodeIds: ['req_empty_compress'],
childResponseCount: 1,
childTotalPoints: 0.1,
runningTime: 0.2
childTotalPoints: 0.1
})
})
);
......
packages/service/test/core/workflow/dispatch/ai/toolcall/toolCall.test.ts
......@@ -353,8 +353,7 @@ describe('runToolCall compression node responses', () => {
expect(result.runtimeNodeResponseSummary).toEqual(
expect.objectContaining({
responseIds: [],
finishedNodeIds: [],
runningTime: 0
finishedNodeIds: []
})
);
});
......
packages/service/test/core/workflow/dispatch/loopRun/runLoopRun.test.ts
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import {
FlowNodeOutputTypeEnum,
FlowNodeTypeEnum
......@@ -201,6 +201,10 @@ describe('runLoopRun (integration with mocked runWorkflow)', () => {
runWorkflowMock.mockReset();
});
afterEach(() => {
vi.restoreAllMocks();
});
it('array mode 数组正常跑完 → loopHistory 全 success, data 含最后一轮快照', async () => {
// Each iteration returns a clean response that writes a new value to chatNode
runWorkflowMock.mockImplementation((args: any) => {
......@@ -309,8 +313,6 @@ describe('runLoopRun (integration with mocked runWorkflow)', () => {
});
it('循环体内变量更新外部节点输出后，每轮成功结束会同步到父运行态', async () => {
let unchangedExternalNode: RuntimeNodeItemType | undefined;
let props: any;
let iteration = 0;
runWorkflowMock.mockImplementation((args: any) => {
......@@ -345,7 +347,7 @@ describe('runLoopRun (integration with mocked runWorkflow)', () => {
);
});
props = makeProps({
const props = makeProps({
[NodeInputKeyEnum.loopRunMode]: LoopRunModeEnum.array,
[NodeInputKeyEnum.loopRunInputArray]: ['first', 'second']
});
......@@ -394,7 +396,7 @@ describe('runLoopRun (integration with mocked runWorkflow)', () => {
}
]
});
unchangedExternalNode = props.runtimeNodes.find(
const unchangedExternalNode = props.runtimeNodes.find(
(node: RuntimeNodeItemType) => node.nodeId === 'unchangedExternal'
);
props.runtimeNodes.push({
......@@ -1027,6 +1029,33 @@ describe('runLoopRun (integration with mocked runWorkflow)', () => {
expect(nodeResponse.childTotalPoints).toBeUndefined();
});
it('每轮包装节点独立计时，不累加子节点 runningTime', async () => {
vi.spyOn(Date, 'now').mockReturnValueOnce(1000).mockReturnValue(2400);
const nodeResponseWriter = {
recordWithParent: vi.fn().mockResolvedValue([])
};
runWorkflowMock.mockResolvedValue(
makeDispatchFlowResponse({
nodeResponses: [
makeResponseItem('startNode', { runningTime: 5 }),
makeResponseItem('chatNode', { runningTime: 6 })
]
})
);
await dispatchLoopRun({
...makeProps({
[NodeInputKeyEnum.loopRunMode]: LoopRunModeEnum.array,
[NodeInputKeyEnum.loopRunInputArray]: ['a'],
[NodeInputKeyEnum.childrenNodeIdList]: ['startNode', 'chatNode']
}),
nodeResponseWriter,
nodeResponseParentId: 'loop-parent-response'
});
expect(nodeResponseWriter.recordWithParent.mock.calls[0][0][0].runningTime).toBe(1.4);
});
it('失败轮不内嵌 loopRunDetail，父响应保留错误和 child 统计', async () => {
let iter = 0;
runWorkflowMock.mockImplementation(() => {
......
packages/service/test/core/workflow/dispatch/parallelRun/runParallelRun.test.ts
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import {
FlowNodeOutputTypeEnum,
FlowNodeTypeEnum
......@@ -118,6 +118,10 @@ describe('dispatchParallelRun', () => {
runWorkflowMock.mockReset();
});
afterEach(() => {
vi.restoreAllMocks();
});
it('共用 nodeResponseWriter 时写入任务包装节点，父响应只保留轻量统计', async () => {
const nodeResponseWriter = {
recordWithParent: vi.fn().mockResolvedValue([])
......@@ -164,6 +168,34 @@ describe('dispatchParallelRun', () => {
expect(nodeResponse.parallelDetail).toBeUndefined();
});
it('任务包装节点独立计时，不累加子节点 runningTime', async () => {
vi.spyOn(Date, 'now').mockReturnValueOnce(1000).mockReturnValue(2250);
const nodeResponseWriter = {
recordWithParent: vi.fn().mockResolvedValue([])
};
runWorkflowMock.mockResolvedValue(
makeDispatchFlowResponse({
nodeResponses: [
makeResponseItem('chatNode', { runningTime: 10 }),
makeResponseItem('nestedEnd', {
moduleType: FlowNodeTypeEnum.nestedEnd,
runningTime: 20,
loopOutputValue: 'done'
})
]
})
);
await dispatchParallelRun(
makeProps({
nodeResponseWriter,
nodeResponseParentId: 'parallel-parent-response'
})
);
expect(nodeResponseWriter.recordWithParent.mock.calls[0][0][0].runningTime).toBe(1.25);
});
it('重试成功时保留失败 attempt 详情并写入最终成功 attempt', async () => {
const nodeResponseWriter = {
recordWithParent: vi.fn().mockResolvedValue([])
......@@ -313,7 +345,6 @@ describe('dispatchParallelRun', () => {
});
it('成功任务会把变量更新写到外部节点的 output 同步回父运行态', async () => {
let unchangedExternalNode: RuntimeNodeItemType | undefined;
runWorkflowMock.mockImplementation((args: any) => {
const externalNode = args.runtimeNodes.find((node: any) => node.nodeId === 'externalText');
externalNode.outputs[0].value = `updated-${args.nodeResponseParentId}`;
......@@ -373,7 +404,7 @@ describe('dispatchParallelRun', () => {
}
]
});
unchangedExternalNode = props.runtimeNodes.find(
const unchangedExternalNode = props.runtimeNodes.find(
(node: RuntimeNodeItemType) => node.nodeId === 'unchangedExternal'
);
......
packages/service/test/core/workflow/dispatch/parallelRun/service.test.ts
......@@ -489,12 +489,13 @@ describe('parallelRun/service', () => {
expect(responseDetails[0].childrenResponses).toBeUndefined();
});
it('wrapper.runningTime 为子节点 runningTime 之和（精确到百分位）', () => {
it('wrapper.runningTime 使用任务自身耗时，不累加子节点耗时', () => {
const withTimings: ParallelTaskResult = {
success: true,
index: 0,
data: 'ok',
totalPoints: 0,
runningTime: 1.23,
response: makeDispatchFlowResponse({
nodeResponses: [
{ id: 'a', runningTime: 0.33 } as any,
......@@ -504,7 +505,7 @@ describe('parallelRun/service', () => {
})
};
const { responseDetails } = agg([withTimings]);
expect(responseDetails[0].runningTime).toBe(1);
expect(responseDetails[0].runningTime).toBe(1.23);
});
it('全部成功 → fullResultsArray 每项 {success:true, message:"", data}', () => {
......
packages/service/test/core/workflow/dispatch/utils.test.ts
......@@ -1283,7 +1283,7 @@ describe('summarizeRuntimeNodeResponses', () => {
expect(nextSummary.responseIds).toEqual(['repeat', 'next']);
expect(nextSummary.finishedNodeIds).toEqual(['repeat-node', 'next-node']);
expect(nextSummary.runningTime).toBe(4);
expect(nextSummary).not.toHaveProperty('runningTime');
expect(nextSummary.childTotalPoints).toBe(6);
expect(nextSummary.childResponseCount).toBe(2);
});
......
projects/app/test/pages/api/support/mcp/server/toolList.test.ts
......@@ -229,7 +229,6 @@ describe('callMcpServerTool', () => {
hasLoopRunBreak: false,
hasToolStop: false,
hasNestedEnd: false,
runningTime: 0,
pluginOutput: {
result: 'plugin output value'
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment