""" 子流程节点 """ import logging import time from typing import Any, Dict, List, Optional from sqlalchemy import select from ..base import BaseNode, NodeContext, NodeResult from ..registry import NodeRegistry logger = logging.getLogger(__name__) @NodeRegistry.register class SubflowNode(BaseNode): """ 子流程节点 调用另一个工作流作为子流程执行 """ node_type = 'subflow' node_name = '子流程' node_category = 'flow_control' node_icon = 'workflow' node_description = '调用另一个工作流作为子流程' inputs = [ { 'name': 'input', 'type': 'any', 'description': '输入数据', }, ] outputs = [ { 'name': 'subflow_result', 'type': 'object', 'description': '子流程执行结果', }, ] def execute(self, context: NodeContext) -> NodeResult: """同步执行子流程（不推荐，建议使用异步版本）""" return NodeResult( success=False, error='子流程节点需要异步执行，请使用 execute_async 方法', ) async def execute_async(self, context: NodeContext) -> NodeResult: """异步执行子流程""" try: # 获取配置 subflow_id = self.config.get('subflow_id', '') if not subflow_id: return NodeResult(success=False, error='未配置子流程') # 解析变量引用 subflow_id = context.resolve_template(subflow_id) var_pass_mode = self.config.get('var_pass_mode', 'all') result_pass_mode = self.config.get('result_pass_mode', 'all') result_variable = self.config.get('result_variable', 'subflow_result') max_depth = self.config.get('max_depth', 5) # 检查嵌套深度 current_depth = context.metadata.get('subflow_depth', 0) if current_depth >= max_depth: return NodeResult( success=False, error=f'子流程嵌套深度超过限制 ({max_depth})', ) # 加载子工作流 subflow = await self._load_subflow(context.db_session, subflow_id) if not subflow: return NodeResult(success=False, error=f'子流程不存在: {subflow_id}') # 检查循环依赖 parent_workflows = context.metadata.get('parent_workflows', []) if subflow_id in parent_workflows: return NodeResult( success=False, error=f'检测到循环依赖: 子流程 {subflow_id} 已在调用链中', ) # 准备子流程输入变量 subflow_inputs = self._prepare_inputs(context, var_pass_mode) # 检查是否有恢复状态（从等待中恢复） subflow_state = context.metadata.pop('_subflow_state', None) # 执行子流程 start_time = time.time() if subflow_state: # 从等待状态恢复执行 logger.info(f'[子流程] 从等待状态恢复执行，当前节点: {subflow_state.get("current_node_id")}') subflow_result = await self._resume_subflow( subflow_state, subflow_inputs, context, current_depth + 1, parent_workflows + [subflow_id], ) else: # 正常执行子流程 subflow_result = await self._execute_subflow( subflow, subflow_inputs, context, current_depth + 1, parent_workflows + [subflow_id], ) elapsed_time = int((time.time() - start_time) * 1000) # 检查是否需要转发交互式节点 forward_interactive = self.config.get('forward_interactive', False) if subflow_result.get('waiting_for_input') and forward_interactive: # 转发等待状态到主流程 return self._forward_waiting_state(subflow_result, context, elapsed_time) elif subflow_result.get('waiting_for_input') and not forward_interactive: # 不转发交互式节点，报错 waiting_node = subflow_result.get('logs', [{}])[-1].get('node_label', '未知') return NodeResult( success=False, error=f'子流程中的交互式节点 "{waiting_node}" 需要用户输入，但未开启"转发交互式节点"选项', elapsed_time=elapsed_time, ) # 处理结果 return self._process_result( subflow_result, context, result_pass_mode, result_variable, elapsed_time, ) except Exception as e: logger.exception(f'子流程执行失败: {e}') return NodeResult(success=False, error=f'子流程执行失败: {str(e)}') async def _load_subflow(self, db_session, subflow_id: str): """加载子工作流""" if not db_session: return None from ai_platform.models import AIWorkflow result = await db_session.execute( select(AIWorkflow).where( AIWorkflow.id == subflow_id, AIWorkflow.is_deleted == False ) ) return result.scalar_one_or_none() def _prepare_inputs(self, context: NodeContext, mode: str) -> Dict[str, Any]: """准备子流程输入变量""" if mode == 'none': return {} elif mode == 'all': # 传递所有变量（排除内部变量） return { k: v for k, v in context.variables.items() if not k.startswith('_') } elif mode == 'selected': # 使用新的变量映射结构 var_mappings = self.config.get('var_mappings', []) result = {} for mapping in var_mappings: name = mapping.get('name', '') # 子流程中的变量名 value = mapping.get('value', '') # 主流程中的变量值（可能是模板） if not name: continue # 解析变量值，保留原始数据类型 resolved_value = self._resolve_variable_value(value, context) result[name] = resolved_value return result return {} def _resolve_variable_value(self, value: str, context: NodeContext) -> Any: """ 解析变量值，保留原始数据类型 支持格式： - {{variable_name}} - 直接变量引用 - {{NodeID.key}} - 节点输出引用 """ import re if not isinstance(value, str): return value value = value.strip() # 检查是否是纯变量引用 {{...}} if value.startswith('{{') and value.endswith('}}'): # 检查是否只有一个变量引用（没有其他文本） pattern = r'^\{\{([^}]+)\}\}$' match = re.match(pattern, value) if match: var_ref = match.group(1).strip() # 尝试解析 NodeID.key 格式 if '.' in var_ref: parts = var_ref.split('.', 1) node_id = parts[0] var_key = parts[1] # 先尝试从节点输出命名空间获取 node_outputs = context.variables.get(f'_node_{node_id}') if isinstance(node_outputs, dict) and var_key in node_outputs: return node_outputs[var_key] # 回退：尝试直接从变量获取 node_data = context.variables.get(node_id) if isinstance(node_data, dict) and var_key in node_data: return node_data[var_key] # 直接变量引用 if var_ref in context.variables: return context.variables[var_ref] # 如果包含变量引用但不是纯变量引用，使用 resolve_template（会转换为字符串） if '{{' in value and '}}' in value: return context.resolve_template(value) # 普通字符串值 return value def _get_child_nodes(self, node_map: Dict[str, Any], parent_node_id: str) -> List[str]: """ 获取循环节点的子节点列表 子节点通过 parentNode 属性关联到循环节点 注意：parentNode 可能在节点顶层或 data 中 """ child_nodes = [] for node_id, node_config in node_map.items(): # 检查顶层 parentNode 属性（Vue Flow 标准格式） parent_node = node_config.get('parentNode') # 也检查 data 中的 parentNode（兼容其他格式） if not parent_node: parent_node = node_config.get('data', {}).get('parentNode') if parent_node == parent_node_id: child_nodes.append(node_id) return child_nodes async def _execute_loop_in_subflow( self, node_map: Dict[str, Any], edge_map: Dict[str, List[str]], parallel_edge_map: Dict[str, Dict[str, str]], loop_node_id: str, loop_config: Dict[str, Any], context: NodeContext, logs: List[Dict], subflow_id: str, resume_node_id: str = None, ) -> Dict[str, Any]: """ 在子流程中执行循环体 Args: node_map: 节点映射 edge_map: 边映射 parallel_edge_map: 并行边映射 loop_node_id: 循环节点 ID loop_config: 循环配置 context: 执行上下文 logs: 日志列表 subflow_id: 子流程 ID resume_node_id: 恢复执行的节点 ID（用于从暂停状态恢复） Returns: 循环执行结果 """ loop_mode = context.variables.get('_loop_mode', 'for_each') max_iterations = loop_config.get('max_iterations', 100) item_var_name = loop_config.get('item_variable_name', 'item') index_var_name = loop_config.get('index_variable_name', 'index') # 获取循环体子节点 child_node_ids = self._get_child_nodes(node_map, loop_node_id) logger.info(f'[子流程循环] 通过 parentNode 获取子节点: {child_node_ids}') if not child_node_ids: child_node_ids = loop_config.get('childNodes', []) logger.info(f'[子流程循环] 通过 childNodes 配置获取子节点: {child_node_ids}') if not child_node_ids: logger.warning(f'[子流程循环] 循环节点 {loop_node_id} 没有子节点，node_map keys: {list(node_map.keys())}') return {'success': True, 'total_tokens': 0} # 构建子节点的边映射 child_edge_map = {} for source_id in child_node_ids: targets = edge_map.get(source_id, []) child_targets = [t for t in targets if t in child_node_ids] if child_targets: child_edge_map[source_id] = child_targets # 找到循环体的起始节点 nodes_with_incoming = set() for targets in child_edge_map.values(): nodes_with_incoming.update(targets) start_nodes = [n for n in child_node_ids if n not in nodes_with_incoming] if not start_nodes: start_nodes = [child_node_ids[0]] loop_body_start_id = start_nodes[0] # 从当前循环索引开始 iteration = context.variables.get('_loop_index', 0) total_tokens = 0 logger.info(f'[子流程循环] 开始执行循环，mode={loop_mode}, max_iterations={max_iterations}, start_iteration={iteration}') logger.info(f'[子流程循环] child_node_ids={child_node_ids}, loop_body_start_id={loop_body_start_id}') logger.info(f'[子流程循环] child_edge_map={child_edge_map}') logger.info(f'[子流程循环] _loop_items={context.variables.get("_loop_items", [])}, _loop_index={context.variables.get("_loop_index")}') while iteration < max_iterations: # 检查是否继续循环 if loop_mode == 'for_each': items = context.variables.get('_loop_items', []) current_index = context.variables.get('_loop_index', 0) logger.info(f'[子流程循环] 迭代 {iteration}: items 长度={len(items)}, current_index={current_index}') if current_index >= len(items): logger.info(f'[子流程循环] 退出循环: current_index({current_index}) >= len(items)({len(items)})') break # 设置当前循环项和索引 current_item = items[current_index] context.variables[item_var_name] = current_item context.variables[index_var_name] = current_index context.variables['_current_item'] = current_item context.variables['_current_index'] = current_index context.variables[f'_node_{loop_node_id}'] = { item_var_name: current_item, index_var_name: current_index, 'total': len(items), } elif loop_mode == 'while': from ai_platform.nodes.builtin.loop_node import LoopNode loop_node_instance = LoopNode(loop_config) if not loop_node_instance._evaluate_condition(context): break context.variables[index_var_name] = iteration # 执行循环体 # 如果是恢复执行，从指定节点开始；否则从循环体起始节点开始 current_id = resume_node_id if resume_node_id and resume_node_id in child_node_ids else loop_body_start_id iteration_output = None executed_nodes = set() # 如果是恢复执行，记录日志 if resume_node_id and resume_node_id in child_node_ids: logger.info(f'[子流程循环] 从节点 {resume_node_id} 恢复执行（迭代 {iteration}）') # 恢复后只执行一次，然后清除 resume_node_id resume_node_id = None while current_id and current_id in child_node_ids: if current_id in executed_nodes: break executed_nodes.add(current_id) node_config = node_map.get(current_id) if not node_config: break node_type = node_config.get('type', '') node_label = node_config.get('data', {}).get('label', current_id) # 创建节点实例 node_instance = NodeRegistry.create_instance(node_type, node_config.get('data', {})) if not node_instance: next_nodes = child_edge_map.get(current_id, []) current_id = next_nodes[0] if next_nodes else None continue # 执行节点 start_time = time.time() result = await node_instance.execute_async(context) elapsed = int((time.time() - start_time) * 1000) # 记录日志 log_entry = { 'node_id': current_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed' if result.success else 'failed', 'output': result.output, 'error': result.error, 'elapsed_time': elapsed, 'tokens_used': result.tokens_used, 'loop_iteration': iteration, } if result.events: log_entry['events'] = result.events logs.append(log_entry) total_tokens += result.tokens_used # 检查是否需要等待用户输入 if result.waiting_for_input: return { 'success': True, 'waiting_for_input': True, 'waiting_config': result.waiting_config, 'outputs': context.variables, 'logs': logs, 'total_tokens': total_tokens, 'subflow_state': { 'current_node_id': current_id, 'node_map': {k: v for k, v in node_map.items()}, 'edge_map': edge_map, 'parallel_edge_map': parallel_edge_map, 'subflow_id': subflow_id, 'sent_message_nodes': [], 'loop_state': { 'loop_node_id': loop_node_id, 'iteration': iteration, 'current_index': context.variables.get('_loop_index', 0), }, }, } if not result.success: return { 'success': False, 'error': f'子流程循环节点 {node_label} 执行失败: {result.error}', 'outputs': context.variables, 'logs': logs, 'total_tokens': total_tokens, } # 更新上下文 context.variables.update(result.output_variables) context.previous_output = result.output iteration_output = result.output context.variables[f'_node_{current_id}'] = { 'output': result.output, **result.output_variables, } # 下一个节点 if result.next_node_id and result.next_node_id in child_node_ids: current_id = result.next_node_id else: next_nodes = child_edge_map.get(current_id, []) current_id = next_nodes[0] if next_nodes else None # 收集本次迭代结果 loop_results = context.variables.get('_loop_results', []) loop_results.append(iteration_output) context.variables['_loop_results'] = loop_results # 更新循环索引 context.variables['_loop_index'] = context.variables.get('_loop_index', 0) + 1 iteration += 1 # 清理循环变量 context.variables.pop('_loop_items', None) context.variables.pop('_loop_index', None) context.variables.pop('_loop_total', None) context.variables.pop('_loop_mode', None) context.variables.pop('_current_item', None) context.variables.pop('_current_index', None) return {'success': True, 'total_tokens': total_tokens} async def _execute_subflow( self, subflow, inputs: Dict[str, Any], parent_context: NodeContext, depth: int, parent_workflows: List[str], ) -> Dict[str, Any]: """执行子流程""" from ai_platform.nodes import NodeRegistry # 使用已发布版本，如果没有则使用草稿版本 definition = subflow.published_definition or subflow.definition nodes = definition.get('nodes', []) edges = definition.get('edges', []) if not nodes: return {'success': False, 'error': '子流程没有节点', 'logs': []} # 构建节点映射和边映射 node_map = {node['id']: node for node in nodes} edge_map = {} parallel_edge_map = {} for edge in edges: source = edge.get('source', '') target = edge.get('target', '') source_handle = edge.get('sourceHandle', '') if source not in edge_map: edge_map[source] = [] edge_map[source].append(target) if source_handle: if source not in parallel_edge_map: parallel_edge_map[source] = {} parallel_edge_map[source][source_handle] = target # 找到开始节点 start_node = None for node in nodes: if node.get('type') == 'start': start_node = node break if not start_node: return {'success': False, 'error': '子流程没有开始节点', 'logs': []} # 创建子流程上下文 subflow_context = NodeContext( workflow_run_id=f"{parent_context.workflow_run_id}_sub_{subflow.id}", variables=inputs.copy(), user_input=inputs.get('user_input', ''), user_id=parent_context.user_id, conversation_history=parent_context.conversation_history.copy(), db_session=parent_context.db_session, metadata={ 'subflow_depth': depth, 'parent_workflows': parent_workflows, 'parent_run_id': parent_context.workflow_run_id, }, ) # 执行节点序列 logs = [] total_tokens = 0 current_node_id = start_node['id'] while current_node_id: node_config = node_map.get(current_node_id) if not node_config: break node_type = node_config.get('type', '') node_label = node_config.get('data', {}).get('label', current_node_id) # 结束节点 if node_type == 'end': node_instance = NodeRegistry.create_instance(node_type, node_config.get('data', {})) if node_instance: result = node_instance.execute(subflow_context) logs.append({ 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed', 'output': result.output, }) break # 创建并执行节点 node_instance = NodeRegistry.create_instance(node_type, node_config.get('data', {})) if not node_instance: logger.warning(f'[子流程] 未知节点类型: {node_type}') next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None continue # 执行节点 start_time = time.time() result = await node_instance.execute_async(subflow_context) elapsed = int((time.time() - start_time) * 1000) # 循环节点：执行循环体 if node_type == 'loop': loop_node_id = current_node_id loop_config = node_config.get('data', {}) # 记录循环节点初始化日志 logs.append({ 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed' if result.success else 'failed', 'output': result.output, 'elapsed_time': elapsed, }) if result.success: # 更新上下文变量（循环节点会设置 _loop_items, _loop_index 等） subflow_context.variables.update(result.output_variables) logger.info(f'[子流程] 循环节点初始化完成，_loop_items 长度={len(subflow_context.variables.get("_loop_items", []))}, _loop_index={subflow_context.variables.get("_loop_index")}') # 执行循环体 loop_result = await self._execute_loop_in_subflow( node_map, edge_map, parallel_edge_map, loop_node_id, loop_config, subflow_context, logs, subflow.id ) total_tokens += loop_result.get('total_tokens', 0) # 检查循环是否因等待用户输入而暂停 if loop_result.get('waiting_for_input'): return loop_result # 检查循环是否失败 if not loop_result.get('success', True): return loop_result # 获取循环结果 loop_results = subflow_context.variables.get('_loop_results', []) output_var = loop_config.get('output_variable', 'loop_results') subflow_context.variables[output_var] = loop_results # 循环完成后从循环节点的输出边继续 next_nodes = edge_map.get(loop_node_id, []) # 过滤掉子节点 child_node_ids = self._get_child_nodes(node_map, loop_node_id) next_nodes = [n for n in next_nodes if n not in child_node_ids] current_node_id = next_nodes[0] if next_nodes else None continue # 检查是否需要等待用户输入（交互式节点） if result.waiting_for_input: # 记录等待状态的日志 log_entry = { 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'waiting', 'output': result.output, 'elapsed_time': elapsed, } logs.append(log_entry) # 返回等待状态，包含子流程执行状态以便恢复 return { 'success': True, 'waiting_for_input': True, 'waiting_config': result.waiting_config, 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, # 保存子流程状态以便恢复 'subflow_state': { 'current_node_id': current_node_id, 'node_map': {k: v for k, v in node_map.items()}, 'edge_map': edge_map, 'parallel_edge_map': parallel_edge_map, 'subflow_id': subflow.id, 'sent_message_nodes': [], # 初始化已发送消息节点列表 }, } # 记录日志 log_entry = { 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed' if result.success else 'failed', 'output': result.output, 'error': result.error, 'elapsed_time': elapsed, 'tokens_used': result.tokens_used, } # 收集消息事件（用于转发到主流程） if result.events: log_entry['events'] = result.events logs.append(log_entry) total_tokens += result.tokens_used if not result.success: return { 'success': False, 'error': f'子流程节点 {node_label} 执行失败: {result.error}', 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, } # 更新上下文 subflow_context.variables.update(result.output_variables) subflow_context.previous_output = result.output # 保存节点输出到命名空间 subflow_context.variables[f'_node_{current_node_id}'] = { 'output': result.output, **result.output_variables, } # 确定下一个节点 if result.next_node_id: if result.next_node_id in parallel_edge_map.get(current_node_id, {}): current_node_id = parallel_edge_map[current_node_id][result.next_node_id] elif result.next_node_id in node_map: current_node_id = result.next_node_id else: next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None else: next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None return { 'success': True, 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, } def _process_result( self, subflow_result: Dict[str, Any], context: NodeContext, result_pass_mode: str, result_variable: str, elapsed_time: int, ) -> NodeResult: """处理子流程结果""" if not subflow_result.get('success'): return NodeResult( success=False, error=subflow_result.get('error', '子流程执行失败'), elapsed_time=elapsed_time, ) output_vars = {} subflow_outputs = subflow_result.get('outputs', {}) if result_pass_mode == 'all': # 传递所有输出变量（排除内部变量） output_vars = { k: v for k, v in subflow_outputs.items() if not k.startswith('_') } elif result_pass_mode == 'selected': # 传递选定的变量 result_vars = self.config.get('result_vars', []) for var in result_vars: if var in subflow_outputs: output_vars[var] = subflow_outputs[var] # 将完整结果存储到指定变量 output_vars[result_variable] = { 'success': True, 'outputs': {k: v for k, v in subflow_outputs.items() if not k.startswith('_')}, 'logs': subflow_result.get('logs', []), 'total_tokens': subflow_result.get('total_tokens', 0), } # 收集子流程的消息事件（如果配置了显示子流程消息） # 只收集尚未发送的消息（避免重复发送） events = [] show_subflow_messages = self.config.get('show_subflow_messages', False) if show_subflow_messages: subflow_name = self.config.get('subflow_name', '子流程') # 获取已发送的消息节点ID列表（从子流程状态中） subflow_state = subflow_result.get('subflow_state', {}) sent_message_nodes = subflow_state.get('sent_message_nodes', []) for log in subflow_result.get('logs', []): node_id = log.get('node_id') log_events = log.get('events', []) for event in log_events: if event.get('type') == 'message': # 只发送尚未发送过的消息 if node_id not in sent_message_nodes: events.append({ 'type': 'message', 'content': event.get('content', ''), 'message_type': event.get('message_type', 'text'), 'from_subflow': True, 'subflow_name': subflow_name, }) return NodeResult( success=True, output=f"子流程执行成功，共 {len(subflow_result.get('logs', []))} 个节点", output_variables=output_vars, tokens_used=subflow_result.get('total_tokens', 0), elapsed_time=elapsed_time, events=events if events else None, ) def _forward_waiting_state( self, subflow_result: Dict[str, Any], context: NodeContext, elapsed_time: int, ) -> NodeResult: """转发子流程的等待状态到主流程""" subflow_name = self.config.get('subflow_name', '子流程') waiting_config = subflow_result.get('waiting_config', {}) subflow_state = subflow_result.get('subflow_state', {}) # 收集之前节点的消息事件（在等待节点之前的消息应该先显示） # 但只收集尚未发送的消息（避免重复发送） events = [] show_subflow_messages = self.config.get('show_subflow_messages', False) # 获取已发送的消息节点ID列表（从子流程状态中），复制一份避免修改原始列表 sent_message_nodes = list(subflow_state.get('sent_message_nodes', [])) if show_subflow_messages: for log in subflow_result.get('logs', []): # 跳过当前等待节点的日志 if log.get('status') == 'waiting': continue node_id = log.get('node_id') log_events = log.get('events', []) for event in log_events: if event.get('type') == 'message': # 只发送尚未发送过的消息 if node_id not in sent_message_nodes: events.append({ 'type': 'message', 'content': event.get('content', ''), 'message_type': event.get('message_type', 'text'), 'from_subflow': True, 'subflow_name': subflow_name, }) sent_message_nodes.append(node_id) # 在等待配置中添加子流程标识 forwarded_config = { **waiting_config, 'from_subflow': True, 'subflow_name': subflow_name, # 保存子流程状态以便恢复 '_subflow_state': { **subflow_state, 'outputs': subflow_result.get('outputs', {}), 'logs': subflow_result.get('logs', []), 'total_tokens': subflow_result.get('total_tokens', 0), 'sent_message_nodes': sent_message_nodes, # 始终保存，即使为空列表 }, } return NodeResult( success=True, output=f"子流程等待用户输入", waiting_for_input=True, waiting_config=forwarded_config, elapsed_time=elapsed_time, events=events if events else None, ) async def _resume_subflow( self, subflow_state: Dict[str, Any], inputs: Dict[str, Any], parent_context: NodeContext, depth: int, parent_workflows: List[str], ) -> Dict[str, Any]: """从等待状态恢复子流程执行""" # 恢复子流程状态 current_node_id = subflow_state.get('current_node_id') node_map = subflow_state.get('node_map', {}) edge_map = subflow_state.get('edge_map', {}) parallel_edge_map = subflow_state.get('parallel_edge_map', {}) saved_outputs = subflow_state.get('outputs', {}) saved_logs = subflow_state.get('logs', []) saved_tokens = subflow_state.get('total_tokens', 0) loop_state = subflow_state.get('loop_state') # 循环状态 subflow_id = subflow_state.get('subflow_id', '') # 创建子流程上下文，恢复之前的变量 subflow_context = NodeContext( workflow_run_id=f"{parent_context.workflow_run_id}_sub_resume", variables={**saved_outputs, **inputs}, user_input=parent_context.user_input, user_id=parent_context.user_id, conversation_history=parent_context.conversation_history.copy(), db_session=parent_context.db_session, metadata={ 'subflow_depth': depth, 'parent_workflows': parent_workflows, 'parent_run_id': parent_context.workflow_run_id, }, ) # 设置用户输入（交互式节点会读取这个值） subflow_context.variables['__user_input__'] = parent_context.variables.get('__user_input__') logs = saved_logs.copy() total_tokens = saved_tokens # 检查是否从循环中恢复 if loop_state: loop_node_id = loop_state.get('loop_node_id') loop_node_config = node_map.get(loop_node_id, {}) loop_config = loop_node_config.get('data', {}) # 继续执行循环，从暂停的节点恢复 loop_result = await self._execute_loop_in_subflow( node_map, edge_map, parallel_edge_map, loop_node_id, loop_config, subflow_context, logs, subflow_id, resume_node_id=current_node_id # 从暂停的节点继续执行 ) total_tokens += loop_result.get('total_tokens', 0) # 检查循环是否因等待用户输入而暂停 if loop_result.get('waiting_for_input'): return loop_result # 检查循环是否失败 if not loop_result.get('success', True): return loop_result # 获取循环结果 loop_results = subflow_context.variables.get('_loop_results', []) output_var = loop_config.get('output_variable', 'loop_results') subflow_context.variables[output_var] = loop_results # 循环完成后从循环节点的输出边继续 next_nodes = edge_map.get(loop_node_id, []) child_node_ids = self._get_child_nodes(node_map, loop_node_id) next_nodes = [n for n in next_nodes if n not in child_node_ids] current_node_id = next_nodes[0] if next_nodes else None # 清除 __user_input__ if '__user_input__' in subflow_context.variables: del subflow_context.variables['__user_input__'] # 从当前节点继续执行（先重新执行当前等待的节点） while current_node_id: node_config = node_map.get(current_node_id) if not node_config: break node_type = node_config.get('type', '') node_label = node_config.get('data', {}).get('label', current_node_id) # 结束节点 if node_type == 'end': node_instance = NodeRegistry.create_instance(node_type, node_config.get('data', {})) if node_instance: result = node_instance.execute(subflow_context) logs.append({ 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed', 'output': result.output, }) break # 创建并执行节点 node_instance = NodeRegistry.create_instance(node_type, node_config.get('data', {})) if not node_instance: logger.warning(f'[子流程恢复] 未知节点类型: {node_type}') next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None continue # 执行节点 start_time = time.time() result = await node_instance.execute_async(subflow_context) elapsed = int((time.time() - start_time) * 1000) # 循环节点：执行循环体 if node_type == 'loop': loop_node_id = current_node_id loop_config = node_config.get('data', {}) # 记录循环节点初始化日志 logs.append({ 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed' if result.success else 'failed', 'output': result.output, 'elapsed_time': elapsed, }) if result.success: # 更新上下文变量（循环节点会设置 _loop_items, _loop_index 等） subflow_context.variables.update(result.output_variables) # 执行循环体 loop_result = await self._execute_loop_in_subflow( node_map, edge_map, parallel_edge_map, loop_node_id, loop_config, subflow_context, logs, subflow_id ) total_tokens += loop_result.get('total_tokens', 0) # 检查循环是否因等待用户输入而暂停 if loop_result.get('waiting_for_input'): return loop_result # 检查循环是否失败 if not loop_result.get('success', True): return loop_result # 获取循环结果 loop_results = subflow_context.variables.get('_loop_results', []) output_var = loop_config.get('output_variable', 'loop_results') subflow_context.variables[output_var] = loop_results # 循环完成后从循环节点的输出边继续 next_nodes = edge_map.get(loop_node_id, []) child_node_ids = self._get_child_nodes(node_map, loop_node_id) next_nodes = [n for n in next_nodes if n not in child_node_ids] current_node_id = next_nodes[0] if next_nodes else None continue # 检查是否再次需要等待用户输入 if result.waiting_for_input: log_entry = { 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'waiting', 'output': result.output, 'elapsed_time': elapsed, } logs.append(log_entry) return { 'success': True, 'waiting_for_input': True, 'waiting_config': result.waiting_config, 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, 'subflow_state': { 'current_node_id': current_node_id, 'node_map': node_map, 'edge_map': edge_map, 'parallel_edge_map': parallel_edge_map, 'sent_message_nodes': subflow_state.get('sent_message_nodes', []), }, } # 更新最后一个等待状态的日志为完成 if logs and logs[-1].get('status') == 'waiting' and logs[-1].get('node_id') == current_node_id: logs[-1]['status'] = 'completed' if result.success else 'failed' logs[-1]['output'] = result.output logs[-1]['elapsed_time'] = elapsed else: # 记录新日志 log_entry = { 'node_id': current_node_id, 'node_type': node_type, 'node_label': node_label, 'status': 'completed' if result.success else 'failed', 'output': result.output, 'error': result.error, 'elapsed_time': elapsed, 'tokens_used': result.tokens_used, } if result.events: log_entry['events'] = result.events logs.append(log_entry) total_tokens += result.tokens_used if not result.success: return { 'success': False, 'error': f'子流程节点 {node_label} 执行失败: {result.error}', 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, } # 更新上下文 subflow_context.variables.update(result.output_variables) subflow_context.previous_output = result.output # 保存节点输出到命名空间 subflow_context.variables[f'_node_{current_node_id}'] = { 'output': result.output, **result.output_variables, } # 清除 __user_input__，避免下一个交互式节点读取到上一个节点的输入 if '__user_input__' in subflow_context.variables: del subflow_context.variables['__user_input__'] # 确定下一个节点 if result.next_node_id: if result.next_node_id in parallel_edge_map.get(current_node_id, {}): current_node_id = parallel_edge_map[current_node_id][result.next_node_id] elif result.next_node_id in node_map: current_node_id = result.next_node_id else: next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None else: next_nodes = edge_map.get(current_node_id, []) current_node_id = next_nodes[0] if next_nodes else None return { 'success': True, 'outputs': subflow_context.variables, 'logs': logs, 'total_tokens': total_tokens, 'subflow_state': { 'sent_message_nodes': subflow_state.get('sent_message_nodes', []), }, } @classmethod def get_config_schema(cls) -> Dict[str, Any]: """获取配置 Schema""" return { 'type': 'object', 'properties': { 'subflow_id': { 'type': 'string', 'title': '子流程ID', 'description': '要调用的子流程工作流ID', }, 'subflow_name': { 'type': 'string', 'title': '子流程名称', 'description': '子流程显示名称', }, 'var_pass_mode': { 'type': 'string', 'title': '变量传递模式', 'enum': ['all', 'selected', 'none'], 'default': 'all', }, 'selected_vars': { 'type': 'array', 'title': '选择传递的变量', 'items': {'type': 'string'}, }, 'var_mapping': { 'type': 'object', 'title': '变量映射', 'description': '子流程变量名 -> 主流程变量名', }, 'result_pass_mode': { 'type': 'string', 'title': '结果传递模式', 'enum': ['all', 'selected', 'none'], 'default': 'all', }, 'result_vars': { 'type': 'array', 'title': '选择传递的结果变量', 'items': {'type': 'string'}, }, 'result_variable': { 'type': 'string', 'title': '结果变量名', 'default': 'subflow_result', }, 'max_depth': { 'type': 'integer', 'title': '最大嵌套深度', 'default': 5, 'minimum': 1, 'maximum': 10, }, 'show_subflow_messages': { 'type': 'boolean', 'title': '显示子流程消息', 'description': '开启后，子流程中的消息节点会在主流程对话中显示', 'default': False, }, 'forward_interactive': { 'type': 'boolean', 'title': '转发交互式节点', 'description': '开启后，子流程中的问答、选项、确认等节点会在主流程中显示', 'default': False, }, }, 'required': ['subflow_id'], }