ChatGPT/FastGPT
1
0
Fork 0
mirror of https://github.com/labring/FastGPT.git synced 2026-05-05 01:02:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: workflow batch repeat run (#6186)

Browse Source 
* stop design doc

* remove invalid doc

* perf: auto fit

* fix: icon

* perf: icon

* perf: icon

* perf: icon

* perf: icon

* perf: variable disabled input ui

* fix: workflow batch run

* fix: tsc
This commit is contained in:
Archer
2026-01-05 12:05:13 +08:00
committed by GitHub
parent 1e68a5fec0
commit 88ed97bc9d
12 changed files with 140 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/service/core/workflow/dispatch/index.ts
+13 -12
View File
@@ -387,7 +387,7 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
this.processActiveNode();
}
// Process next active node
private processActiveNode() {
private async processActiveNode() {
// Finish
if (this.activeRunQueue.size === 0 && this.runningNodeCount === 0) {
if (isDebugMode) {
@@ -414,6 +414,8 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
return;
}
// Thread avoidance
await surrenderProcess();
const nodeId = this.activeRunQueue.keys().next().value;
const node = nodeId ? this.runtimeNodesMap.get(nodeId) : undefined;
@@ -443,7 +445,9 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
this.skipNodeQueue.set(node.nodeId, { node, skippedNodeIdList: concatSkippedNodeIdList });
}
private processSkipNodes() {
private async processSkipNodes() {
// Thread avoidance
await surrenderProcess();
// 取一个 node，并且从队列里删除
const skipItem = this.skipNodeQueue.values().next().value;
if (skipItem) {
@@ -867,9 +871,6 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
return;
}
// Thread avoidance
await surrenderProcess();
addLog.debug(`Run node`, { maxRunTimes: data.maxRunTimes, appId: data.runningAppInfo.id });
// Get node run status by edges
@@ -881,6 +882,13 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
const nodeRunResult = await (async () => {
if (status === 'run') {
// All source edges status to waiting
runtimeEdges.forEach((item) => {
if (item.target === node.nodeId) {
item.status = 'waiting';
}
});
const blanceCheckResult = await this.checkTeamBlance();
if (blanceCheckResult) {
return {
@@ -890,13 +898,6 @@ export const runWorkflow = async (data: RunWorkflowProps): Promise<DispatchFlowR
};
}
// All source edges status to waiting
runtimeEdges.forEach((item) => {
if (item.target === node.nodeId) {
item.status = 'waiting';
}
});
addLog.debug(`[dispatchWorkFlow] nodeRunWithActive: ${node.name}`);
return this.nodeRunWithActive(node);
}