python执行json工作流输出每一步的结果示例代码-python代码-BFW代码库

python执行json工作流输出每一步的结果示例代码

代码语言：python

所属分类：其他

代码描述：python执行json工作流输出每一步的结果示例代码

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

import json
import requests

class WorkflowExecutor:
    def __init__(self, workflow_json):
        self.workflow = workflow_json
        self.node_outputs = {}
        self.executed_nodes = set()  # 跟踪已执行的节点

    def execute_node(self, node):
        node_type = node['type']
        node_config = node['config']
        node_id = str(node['id'])
        
        # 如果节点已执行过，直接返回结果
        if node_id in self.node_outputs:
            return self.node_outputs[node_id]
        
        if node_type == 'input':
            return node_config.get('defaultValue')
            
        elif node_type == 'process':
            input_data = self.get_input_data(node)
            # 创建一个本地变量来存储处理函数
            local_vars = {'input_data': input_data}
            # 执行处理函数
            exec(node_config['function'], globals(), local_vars)
            # 调用处理函数并返回结果
            return local_vars['process'](input_data)
            

        elif node_type == 'condition':
            input_data = self.get_input_data(node)
            try:
                # 尝试将输入转换为数字，如果可能的话
                if isinstance(input_data, str) and input_data.replace('.', '', 1).isdigit():
                    if input_data.isdigit():
                        numeric_input = int(input_data)
                    else:
                        numeric_input = float(input_data)
                    condition_result = eval(node_config['condition'].replace('input', 'numeric_input'))
                else:
                    condition_result = eval(node_config['condition'], {'input': input_data})
                
                # 返回一个元组，包含条件结果和原始输入数据
                return (condition_re.........完整代码请登录后点击上方下载按钮下载查看