From 200bda3ff27535cb7f29ddb32201852286edf432 Mon Sep 17 00:00:00 2001 From: Ethan-Zhang Date: Tue, 5 Aug 2025 12:28:54 +0800 Subject: [PATCH 1/2] =?UTF-8?q?Feat:=20=E5=A2=9E=E5=8A=A0=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E8=8A=82=E7=82=B9(TODO:=E5=B0=9A=E6=9C=AA=E8=81=94?= =?UTF-8?q?=E8=B0=83)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/scheduler/call/__init__.py | 5 +- apps/scheduler/call/loop/__init__.py | 0 apps/scheduler/call/loop/loop.py | 311 +++++++++++++++++++++++++++ apps/scheduler/call/loop/schema.py | 34 +++ apps/scheduler/executor/step.py | 2 +- apps/schemas/enum_var.py | 1 + 6 files changed, 351 insertions(+), 2 deletions(-) create mode 100644 apps/scheduler/call/loop/__init__.py create mode 100644 apps/scheduler/call/loop/loop.py create mode 100644 apps/scheduler/call/loop/schema.py diff --git a/apps/scheduler/call/__init__.py b/apps/scheduler/call/__init__.py index 5381c7eb..d7abec89 100644 --- a/apps/scheduler/call/__init__.py +++ b/apps/scheduler/call/__init__.py @@ -11,6 +11,8 @@ from apps.scheduler.call.reply.direct_reply import DirectReply from apps.scheduler.call.sql.sql import SQL from apps.scheduler.call.suggest.suggest import Suggestion from apps.scheduler.call.choice.choice import Choice +from apps.scheduler.call.loop.loop import Loop + # 只包含需要在编排界面展示的工具 __all__ = [ "API", @@ -22,5 +24,6 @@ __all__ = [ "SQL", "Graph", "Suggestion", - "Choice" + "Choice", + "Loop" ] diff --git a/apps/scheduler/call/loop/__init__.py b/apps/scheduler/call/loop/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/scheduler/call/loop/loop.py b/apps/scheduler/call/loop/loop.py new file mode 100644 index 00000000..cbd76a1e --- /dev/null +++ b/apps/scheduler/call/loop/loop.py @@ -0,0 +1,311 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""使用条件控制的循环节点""" + +import copy +import logging +import uuid +from collections.abc import AsyncGenerator +from typing import Any + +from pydantic import Field + +from apps.scheduler.call.core import CoreCall +from apps.scheduler.call.choice.condition_handler import ConditionHandler +from apps.scheduler.call.loop.schema import LoopInput, LoopOutput, LoopStopCondition +from apps.scheduler.pool.loader.flow import FlowLoader +from apps.scheduler.variable.integration import VariableIntegration +from apps.schemas.enum_var import CallOutputType, CallType +from apps.schemas.flow import Flow, Step, Edge +from apps.schemas.flow_topology import PositionItem +from apps.schemas.scheduler import ( + CallError, + CallInfo, + CallOutputChunk, + CallVars, +) + +logger = logging.getLogger(__name__) + + +class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): + """循环节点""" + + to_user: bool = Field(default=False) + variables: dict[str, Any] = Field(description="循环变量", default={}) + stop_condition: LoopStopCondition = Field(description="循环终止条件", default=LoopStopCondition()) + max_iteration: int = Field(description="最大循环次数", default=10, ge=1, le=100) + sub_flow_id: str = Field(description="子工作流ID", default="") + + @classmethod + def info(cls) -> CallInfo: + """返回Call的名称和描述""" + return CallInfo( + name="循环", + type=CallType.LOGIC, + description="直到循环终止条件达成或最大循环次数到达之前，子工作流将不断循环执行" + ) + + async def _process_stop_condition(self, call_vars: CallVars) -> tuple[bool, str]: + """处理停止条件 + + Args: + call_vars: Call变量 + + Returns: + tuple[bool, str]: (是否成功处理, 错误消息) + """ + try: + # 如果没有条件，默认为有效 + if not self.stop_condition.conditions: + return True, "" + + # 处理每个条件，复用choice.py的逻辑 + valid_conditions = [] + for condition_original in self.stop_condition.conditions: + condition = copy.deepcopy(condition_original) + + # 处理左值 + if condition.left.type and hasattr(condition.left, 'value'): + try: + resolved_left = await self._resolve_single_value(condition.left, call_vars) + condition.left = resolved_left + except Exception as e: + logger.warning(f"[Loop] 停止条件左值解析失败: {e}") + continue + + # 处理右值 + if condition.right.type and hasattr(condition.right, 'value'): + try: + resolved_right = await self._resolve_single_value(condition.right, call_vars) + condition.right = resolved_right + except Exception as e: + logger.warning(f"[Loop] 停止条件右值解析失败: {e}") + continue + + # 验证条件 + if condition.operate and ConditionHandler.check_value_type(condition.left) and ConditionHandler.check_value_type(condition.right): + valid_conditions.append(condition) + + # 更新有效条件 + self.stop_condition.conditions = valid_conditions + return True, "" + + except Exception as e: + return False, f"停止条件处理失败: {e}" + + def _check_stop_condition(self) -> bool: + """检查是否满足停止条件 + + Returns: + bool: 是否应该停止循环 + """ + try: + if not self.stop_condition.conditions: + return False + + results = [] + for condition in self.stop_condition.conditions: + result = ConditionHandler._judge_condition(condition) + results.append(result) + + if self.stop_condition.logic.value == "and": + return all(results) + elif self.stop_condition.logic.value == "or": + return any(results) + else: + return all(results) if results else False + + except Exception as e: + logger.error(f"[Loop] 检查停止条件失败: {e}") + return False + + def _create_default_sub_flow(self) -> Flow: + """创建默认的子flow配置（只包含开始和结束节点） + + Returns: + Flow: 默认的子flow配置 + """ + return Flow( + name="循环子流程", + description="循环节点的子工作流", + connectivity=False, + # 对于嵌入式显示，设置较小的默认焦点区域，方便在循环节点内部显示 + focus_point=PositionItem(x=400.0, y=200.0), # 更小的默认焦点区域 + debug=False, + steps={ + "start": Step( + node="Empty", + type="start", + name="开始", + description="开始节点", + # 针对嵌入式显示调整节点位置，使其更紧凑 + pos=PositionItem(x=50.0, y=100.0), + params={} + ), + "end": Step( + node="Empty", + type="end", + name="结束", + description="结束节点", + # 结束节点位置也相应调整，保持合理间距 + pos=PositionItem(x=350.0, y=100.0), + params={} + ) + }, + edges=[] + ) + + async def _create_sub_flow_if_not_exists(self, app_id: str, parent_flow_id: str) -> str: + """创建子flow如果不存在 + + Args: + app_id: 应用ID + parent_flow_id: 父flow ID + + Returns: + str: 子flow的ID + """ + if not self.sub_flow_id: + # 生成新的子flow ID，包含父flow信息以体现关系 + self.sub_flow_id = f"{parent_flow_id}_loop_{str(uuid.uuid4())[:8]}" + + # 检查子flow是否已存在 + flow_loader = FlowLoader() + existing_flow = await flow_loader.load(app_id, self.sub_flow_id) + + if not existing_flow: + # 创建默认子flow配置 + default_flow = self._create_default_sub_flow() + + try: + # 保存子flow到文件系统 + await flow_loader.save(app_id, self.sub_flow_id, default_flow) + logger.info(f"[Loop] 创建新的子flow: {self.sub_flow_id}") + except Exception as e: + logger.error(f"[Loop] 创建子flow失败: {e}") + raise CallError(message=f"创建子工作流失败: {e}", data={}) + else: + logger.info(f"[Loop] 使用现有子flow: {self.sub_flow_id}") + + return self.sub_flow_id + + async def _execute_sub_flow(self, iteration: int, call_vars: CallVars) -> dict[str, Any]: + """执行子工作流 + + Args: + iteration: 当前循环次数 + call_vars: Call变量 + + Returns: + dict[str, Any]: 子工作流的执行结果 + """ + logger.info(f"[Loop] 执行第 {iteration} 次循环，子flow ID: {self.sub_flow_id}") + + # TODO: 这里需要实现实际的子flow执行逻辑 + # 在实际的实现中，这里应该： + # 1. 创建一个新的FlowExecutor实例来执行子flow + # 2. 将当前的循环变量设置到子flow的执行上下文中 + # 3. 执行子flow并获取结果 + # 4. 从结果中提取更新后的变量 + + # 由于这需要依赖FlowExecutor和完整的执行环境， + # 现在先返回当前变量作为占位符 + # 在实际集成时，需要调用FlowExecutor来执行子flow + + try: + # 模拟变量更新（实际应该从子flow执行结果中获取） + updated_variables = copy.deepcopy(self.variables) + + # 添加循环相关的系统变量 + updated_variables.update({ + f"loop_iteration_{iteration}": iteration, + "current_iteration": iteration, + }) + + return updated_variables + + except Exception as e: + logger.error(f"[Loop] 执行子flow失败: {e}") + raise CallError(message=f"执行子工作流失败: {e}", data={}) + + async def _init(self, call_vars: CallVars) -> LoopInput: + """初始化Loop工具""" + # 处理停止条件 + success, error_msg = await self._process_stop_condition(call_vars) + if not success: + logger.warning(f"[Loop] 停止条件处理失败: {error_msg}") + + # 从call_vars中获取app_id和flow_id（如果可用） + app_id = getattr(call_vars.ids, 'app_id', 'default') + flow_id = getattr(call_vars.state, 'flow_id', 'default') + + # 创建子flow + sub_flow_id = await self._create_sub_flow_if_not_exists(app_id, flow_id) + + return LoopInput( + variables=self.variables, + stop_condition=self.stop_condition, + max_iteration=self.max_iteration, + sub_flow_id=sub_flow_id, + ) + + async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + """执行Loop工具""" + # 解析输入数据 + data = LoopInput(**input_data) + + try: + iteration_count = 0 + stop_reason = "" + current_variables = copy.deepcopy(data.variables) + + logger.info(f"[Loop] 开始循环执行，最大次数: {data.max_iteration}, 子flow: {data.sub_flow_id}") + + # 开始循环 + while iteration_count < data.max_iteration: + iteration_count += 1 + logger.info(f"[Loop] 开始第 {iteration_count} 次循环") + + # 检查停止条件（在执行前检查） + if self._check_stop_condition(): + stop_reason = "condition_met" + logger.info(f"[Loop] 满足停止条件，在第 {iteration_count - 1} 次循环后停止") + iteration_count -= 1 # 未实际执行这次循环 + break + + # 执行子工作流 + call_vars = CallVars() # 临时创建，实际需要传入正确的call_vars + result_variables = await self._execute_sub_flow(iteration_count, call_vars) + + # 更新循环变量 + current_variables.update(result_variables) + self.variables = current_variables + + # 再次检查停止条件（在执行后检查） + if self._check_stop_condition(): + stop_reason = "condition_met" + logger.info(f"[Loop] 满足停止条件，在第 {iteration_count} 次循环后停止") + break + + # 如果达到最大循环次数 + if iteration_count >= data.max_iteration and not stop_reason: + stop_reason = "max_iteration_reached" + logger.info(f"[Loop] 达到最大循环次数 {data.max_iteration}") + + # 返回结果 + output = LoopOutput( + iteration_count=iteration_count, + stop_reason=stop_reason, + variables=current_variables + ) + + logger.info(f"[Loop] 循环执行完成，执行次数: {iteration_count}, 停止原因: {stop_reason}") + + yield CallOutputChunk( + type=CallOutputType.DATA, + content=output.model_dump(exclude_none=True, by_alias=True), + ) + + except Exception as e: + logger.exception(f"[Loop] 循环执行失败: {e}") + raise CallError(message=f"循环执行失败：{e!s}", data={}) from e diff --git a/apps/scheduler/call/loop/schema.py b/apps/scheduler/call/loop/schema.py new file mode 100644 index 00000000..ba6cb10a --- /dev/null +++ b/apps/scheduler/call/loop/schema.py @@ -0,0 +1,34 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""循环节点的数据结构""" + +from typing import Any +import uuid + +from pydantic import BaseModel, Field + +from apps.scheduler.call.core import DataBase +from apps.scheduler.call.choice.schema import Condition, Logic + + +class LoopStopCondition(DataBase): + """循环停止条件，类似ChoiceBranch但无需branch_id和is_default""" + + logic: Logic = Field(description="逻辑运算符", default=Logic.AND) + conditions: list[Condition] = Field(description="条件列表", default=[]) + + +class LoopInput(DataBase): + """循环节点的输入""" + + variables: dict[str, Any] = Field(description="循环变量", default={}) + stop_condition: LoopStopCondition = Field(description="循环终止条件", default=LoopStopCondition()) + max_iteration: int = Field(description="最大循环次数", default=10, ge=1, le=100) + sub_flow_id: str = Field(description="子工作流ID", default="") + + +class LoopOutput(DataBase): + """循环节点的输出""" + + iteration_count: int = Field(description="实际执行的循环次数", default=0) + stop_reason: str = Field(description="停止原因", default="") # "max_iteration_reached" 或 "condition_met" + variables: dict[str, Any] = Field(description="循环后的变量状态", default={}) \ No newline at end of file diff --git a/apps/scheduler/executor/step.py b/apps/scheduler/executor/step.py index fe5e65d5..900b495c 100644 --- a/apps/scheduler/executor/step.py +++ b/apps/scheduler/executor/step.py @@ -120,7 +120,7 @@ class StepExecutor(BaseExecutor): params.update(self.step.step.params) # 对于需要扁平化处理的Call类型，将input_parameters中的内容提取到顶级 - if self._call_id in ["Choice", "DirectReply"] and "input_parameters" in params: + if self._call_id in ["Choice", "DirectReply", "Loop"] and "input_parameters" in params: # 提取input_parameters中的所有字段到顶级 input_params = params.get("input_parameters", {}) if isinstance(input_params, dict): diff --git a/apps/schemas/enum_var.py b/apps/schemas/enum_var.py index a8268ff1..ba18faaf 100644 --- a/apps/schemas/enum_var.py +++ b/apps/schemas/enum_var.py @@ -151,6 +151,7 @@ class SpecialCallType(str, Enum): START = "start" END = "end" CHOICE = "Choice" + LOOP = "Loop" class CommentType(str, Enum): -- Gitee From 4ab3b8389e26d4a2bd3579623432cfa068827722 Mon Sep 17 00:00:00 2001 From: Ethan-Zhang Date: Sat, 9 Aug 2025 11:34:48 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E8=8A=82=E7=82=B9&=E6=96=B0=E5=A2=9E=E5=8F=98?= =?UTF-8?q?=E9=87=8F=E8=B5=8B=E5=80=BC=E8=8A=82=E7=82=B9&=E5=8F=98?= =?UTF-8?q?=E9=87=8F=E5=BC=95=E7=94=A8debug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CONVERSATION_VARIABLE_FIX.md | 179 ------ apps/common/queue.py | 9 +- apps/routers/chat.py | 10 +- apps/routers/flow.py | 129 +++- apps/routers/variable.py | 6 + apps/scheduler/call/__init__.py | 4 +- apps/scheduler/call/choice/choice.py | 15 +- .../call/choice/condition_handler.py | 28 +- apps/scheduler/call/choice/schema.py | 4 +- apps/scheduler/call/core.py | 60 +- apps/scheduler/call/facts/facts.py | 27 +- apps/scheduler/call/facts/schema.py | 4 +- apps/scheduler/call/loop/loop.py | 486 +++++++++++++-- apps/scheduler/call/loop/schema.py | 4 +- .../call/variable_assign/__init__.py | 6 + apps/scheduler/call/variable_assign/schema.py | 61 ++ .../call/variable_assign/variable_assign.py | 535 +++++++++++++++++ apps/scheduler/executor/base.py | 18 +- apps/scheduler/executor/step.py | 55 +- apps/scheduler/pool/loader/app.py | 6 +- apps/scheduler/pool/loader/flow.py | 124 ++++ apps/scheduler/variable/integration.py | 14 +- apps/scheduler/variable/parser.py | 13 +- apps/schemas/enum_var.py | 1 + apps/schemas/subflow.py | 16 + apps/services/flow.py | 561 +++++++++++++++++- apps/services/flow_validate.py | 107 ++++ apps/services/node.py | 27 +- 28 files changed, 2191 insertions(+), 318 deletions(-) delete mode 100644 CONVERSATION_VARIABLE_FIX.md create mode 100644 apps/scheduler/call/variable_assign/__init__.py create mode 100644 apps/scheduler/call/variable_assign/schema.py create mode 100644 apps/scheduler/call/variable_assign/variable_assign.py create mode 100644 apps/schemas/subflow.py diff --git a/CONVERSATION_VARIABLE_FIX.md b/CONVERSATION_VARIABLE_FIX.md deleted file mode 100644 index 2a84ca94..00000000 --- a/CONVERSATION_VARIABLE_FIX.md +++ /dev/null @@ -1,179 +0,0 @@ -# 对话变量模板问题修复总结 - -## 🔍 **问题描述** - -用户报告创建对话级变量`test`成功后，无法通过API接口查询到： -``` -GET /api/variable/list?scope=conversation&flow_id=52e069c7-5556-42af-bdfc-63f4dc2dcd28 -``` - -## 🔧 **根本原因分析** - -经过分析发现，问题出现在我之前重构变量架构时的遗漏： - -### 1. **创建API工作正常** -- ✅ 对话变量正确存储到FlowVariablePool的`_conversation_templates`字典中 -- ✅ 数据库持久化成功 - -### 2. **查询API有缺陷** -- ❌ `pool_manager.py`中`list_variables_from_any_pool`方法只处理了`conversation_id`参数 -- ❌ 没有处理`scope=conversation&flow_id=xxx`的查询情况 -- ❌ `get_variable_from_any_pool`方法也有同样问题 - -### 3. **更新删除API有问题** -- ❌ FlowVariablePool的`update_variable`和`delete_variable`方法只在`_variables`字典中查找 -- ❌ 找不到存储在`_conversation_templates`字典中的对话变量模板 - -## 🛠️ **修复方案** - -### 1. **修复查询逻辑** - -#### `list_variables_from_any_pool`方法 -**修复前**： -```python -elif scope == VariableScope.CONVERSATION and conversation_id: - pool = await self.get_conversation_pool(conversation_id) - if pool: - return await pool.list_variables(include_system=False) - return [] -``` - -**修复后**： -```python -elif scope == VariableScope.CONVERSATION: - if conversation_id: - # 使用conversation_id查询对话变量实例 - pool = await self.get_conversation_pool(conversation_id) - if pool: - return await pool.list_variables(include_system=False) - elif flow_id: - # 使用flow_id查询对话变量模板 - flow_pool = await self.get_flow_pool(flow_id) - if flow_pool: - return await flow_pool.list_conversation_templates() - return [] -``` - -#### `get_variable_from_any_pool`方法 -类似的修复，支持通过`flow_id`查询对话变量模板。 - -### 2. **修复创建逻辑** - -#### 修改`create_variable`路由 -**修复前**： -```python -# 创建变量 -variable = await pool.add_variable(...) -``` - -**修复后**： -```python -# 根据作用域创建不同类型的变量 -if request.scope == VariableScope.CONVERSATION: - # 创建对话变量模板 - variable = await pool.add_conversation_template(...) -else: - # 创建其他类型的变量 - variable = await pool.add_variable(...) -``` - -### 3. **增强FlowVariablePool功能** - -为FlowVariablePool添加了重写的方法，支持多字典操作： - -#### `update_variable`方法 -- 在环境变量、系统变量模板、对话变量模板中按顺序查找 -- 找到变量后执行更新操作 -- 正确持久化到数据库 - -#### `delete_variable`方法 -- 支持删除存储在不同字典中的变量 -- 保留权限检查（系统变量模板不允许删除） - -#### `get_variable`方法 -- 统一的变量查找接口 -- 支持跨字典查找 - -## ✅ **修复验证** - -### 现在支持的完整工作流程： - -#### 1. **Flow级别操作**（变量模板管理） -```bash -# 创建对话变量模板 -POST /api/variable/create -{ - "name": "test", - "var_type": "string", - "scope": "conversation", - "value": "123", - "description": "321", - "flow_id": "52e069c7-5556-42af-bdfc-63f4dc2dcd28" -} - -# 查询对话变量模板 -GET /api/variable/list?scope=conversation&flow_id=52e069c7-5556-42af-bdfc-63f4dc2dcd28 - -# 更新对话变量模板 -PUT /api/variable/update?name=test&scope=conversation&flow_id=52e069c7-5556-42af-bdfc-63f4dc2dcd28 - -# 删除对话变量模板 -DELETE /api/variable/delete?name=test&scope=conversation&flow_id=52e069c7-5556-42af-bdfc-63f4dc2dcd28 -``` - -#### 2. **Conversation级别操作**（变量实例管理） -```bash -# 查询对话变量实例 -GET /api/variable/list?scope=conversation&conversation_id=conv123 - -# 更新对话变量实例值 -PUT /api/variable/update?name=test&scope=conversation&conversation_id=conv123 -``` - -## 🎯 **测试建议** - -### 立即测试 -现在可以重新测试原来失败的API调用： -```bash -curl "http://10.211.55.10:8002/api/variable/list?scope=conversation&flow_id=52e069c7-5556-42af-bdfc-63f4dc2dcd28" -``` - -### 完整测试流程 -1. **创建对话变量模板**（前端已测试成功） -2. **查询对话变量模板**（现在应该能查到） -3. **更新对话变量模板** -4. **删除对话变量模板** - -### 自动化测试 -运行测试脚本验证： -```bash -cd euler-copilot-framework -python test_conversation_variables.py -``` - -## 📊 **架构完整性验证** - -现在所有变量类型的查询都应该正常工作： - -### Flow级别查询 -- ✅ 系统变量模板：`GET /api/variable/list?scope=system&flow_id=xxx` -- ✅ 对话变量模板：`GET /api/variable/list?scope=conversation&flow_id=xxx` -- ✅ 环境变量：`GET /api/variable/list?scope=environment&flow_id=xxx` - -### Conversation级别查询 -- ✅ 系统变量实例：`GET /api/variable/list?scope=system&conversation_id=xxx` -- ✅ 对话变量实例：`GET /api/variable/list?scope=conversation&conversation_id=xxx` - -### User级别查询 -- ✅ 用户变量：`GET /api/variable/list?scope=user` - -## 🎉 **预期结果** - -修复后，你的前端应该能够： - -1. **成功创建对话变量模板**（已验证） -2. **成功查询对话变量模板**（修复的核心问题） -3. **成功更新对话变量模板** -4. **成功删除对话变量模板** - -所有操作都在Flow级别进行，符合你的设计需求：**Flow级别管理模板定义，Conversation级别操作实际数据**。 \ No newline at end of file diff --git a/apps/common/queue.py b/apps/common/queue.py index 5601c93a..13165c5d 100644 --- a/apps/common/queue.py +++ b/apps/common/queue.py @@ -36,7 +36,7 @@ class MessageQueue: self._heartbeat_task = asyncio.get_event_loop().create_task(self._heartbeat()) async def push_output(self, task: Task, event_type: str, data: dict[str, Any]) -> None: - """组装用于向用户（前端/Shell端）输出的消息""" + """组装用于向用户（前端/Shell端）输出的消息""" if event_type == EventType.DONE.value: await self._queue.put("[DONE]") return @@ -74,7 +74,12 @@ class MessageQueue: content=data, ) - await self._queue.put(json.dumps(message.model_dump(by_alias=True, exclude_none=True), ensure_ascii=False)) + try: + message_json = json.dumps(message.model_dump(by_alias=True, exclude_none=True), ensure_ascii=False) + await self._queue.put(message_json) + except Exception as e: + logger.error(f"[MessageQueue] 消息入队失败 - event_type: {event_type}, error: {e}") + raise async def get(self) -> AsyncGenerator[str, None]: """从Queue中获取消息；变为async generator""" diff --git a/apps/routers/chat.py b/apps/routers/chat.py index 589000be..cd7063a7 100644 --- a/apps/routers/chat.py +++ b/apps/routers/chat.py @@ -82,11 +82,17 @@ async def chat_generator(post_body: RequestData, user_sub: str, session_id: str) # 获取最终答案 task = scheduler.task - if not task.runtime.answer: - logger.error("[Chat] 答案为空") + # 🔑 修复：对于工作流调试模式或纯逻辑节点，允许答案为空 + is_flow_debug = post_body.app and post_body.app.flow_id + if not task.runtime.answer and not is_flow_debug: + logger.error("[Chat] 答案为空且非工作流调试模式") yield "data: [ERROR]\n\n" await Activity.remove_active(user_sub) return + elif not task.runtime.answer and is_flow_debug: + logger.info("[Chat] 工作流调试模式，答案为空是正常的（可能是纯逻辑节点）") + # 为工作流调试提供默认响应 + task.runtime.answer = "工作流执行完成" # 对结果进行敏感词检查 if await WordsCheck().check(task.runtime.answer) != 1: diff --git a/apps/routers/flow.py b/apps/routers/flow.py index 646213a8..256321ac 100644 --- a/apps/routers/flow.py +++ b/apps/routers/flow.py @@ -135,7 +135,7 @@ async def put_flow( put_body.flow = await FlowService.remove_excess_structure_from_flow(put_body.flow) await FlowService.validate_flow_illegal(put_body.flow) put_body.flow.connectivity = await FlowService.validate_flow_connectivity(put_body.flow) - result = await FlowManager.put_flow_by_app_and_flow_id(app_id, flow_id, put_body.flow) + result = await FlowManager.put_flow_by_app_and_flow_id(app_id, flow_id, put_body.flow, user_sub) if result is None: return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -180,6 +180,133 @@ async def put_flow( ) +@router.put( + "/subflow", + response_model=FlowStructurePutRsp, + responses={ + status.HTTP_400_BAD_REQUEST: {"model": ResponseData}, + status.HTTP_403_FORBIDDEN: {"model": ResponseData}, + status.HTTP_404_NOT_FOUND: {"model": ResponseData}, + status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": ResponseData}, + }, +) +async def put_subflow( + user_sub: Annotated[str, Depends(get_user)], + app_id: Annotated[str, Query(alias="appId")], + flow_id: Annotated[str, Query(alias="flowId")], + sub_flow_id: Annotated[str, Query(alias="subFlowId")], + put_body: Annotated[PutFlowReq, Body(...)], +) -> JSONResponse: + """修改子工作流拓扑结构""" + if not await AppManager.validate_app_belong_to_user(user_sub, app_id): + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content=FlowStructurePutRsp( + code=status.HTTP_403_FORBIDDEN, + message="用户没有权限访问该流", + result=FlowStructurePutMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + # 验证父工作流是否存在 + parent_flow = await FlowManager.get_flow_by_app_and_flow_id(app_id, flow_id) + if parent_flow is None: + return JSONResponse( + status_code=status.HTTP_404_NOT_FOUND, + content=FlowStructurePutRsp( + code=status.HTTP_404_NOT_FOUND, + message="父工作流不存在", + result=FlowStructurePutMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + # 子工作流特殊处理：移除多余结构，但不强制要求end节点 + put_body.flow = await FlowService.remove_excess_structure_from_flow(put_body.flow) + await FlowService.validate_subflow_illegal(put_body.flow) # 使用子工作流专用验证 + put_body.flow.connectivity = await FlowService.validate_subflow_connectivity(put_body.flow) # 子工作流连通性验证 + + # 保存子工作流到专用路径 + result = await FlowManager.put_subflow_by_app_flow_and_subflow_id( + app_id, flow_id, sub_flow_id, put_body.flow + ) + if result is None: + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=FlowStructurePutRsp( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="子工作流更新失败", + result=FlowStructurePutMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + # 获取更新后的子工作流 + subflow = await FlowManager.get_subflow_by_app_flow_and_subflow_id(app_id, flow_id, sub_flow_id) + if subflow is None: + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=FlowStructurePutRsp( + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + message="子工作流更新后获取失败", + result=FlowStructurePutMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + return JSONResponse( + status_code=status.HTTP_200_OK, + content=FlowStructurePutRsp( + code=status.HTTP_200_OK, + message="子工作流更新成功", + result=FlowStructurePutMsg(flow=subflow), + ).model_dump(exclude_none=True, by_alias=True), + ) + +@router.get( + "/subflow", + response_model=FlowStructureGetRsp, + responses={ + status.HTTP_403_FORBIDDEN: {"model": ResponseData}, + status.HTTP_404_NOT_FOUND: {"model": ResponseData}, + }, +) +async def get_subflow( + user_sub: Annotated[str, Depends(get_user)], + app_id: Annotated[str, Query(alias="appId")], + flow_id: Annotated[str, Query(alias="flowId")], + sub_flow_id: Annotated[str, Query(alias="subFlowId")], +) -> JSONResponse: + """获取子工作流拓扑结构""" + if not await AppManager.validate_app_belong_to_user(user_sub, app_id): + return JSONResponse( + status_code=status.HTTP_403_FORBIDDEN, + content=FlowStructureGetRsp( + code=status.HTTP_403_FORBIDDEN, + message="用户没有权限访问该流", + result=FlowStructureGetMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + # 获取子工作流 + result = await FlowManager.get_subflow_by_app_flow_and_subflow_id(app_id, flow_id, sub_flow_id) + if result is None: + return JSONResponse( + status_code=status.HTTP_404_NOT_FOUND, + content=FlowStructureGetRsp( + code=status.HTTP_404_NOT_FOUND, + message="子工作流不存在", + result=FlowStructureGetMsg(), + ).model_dump(exclude_none=True, by_alias=True), + ) + + return JSONResponse( + status_code=status.HTTP_200_OK, + content=FlowStructureGetRsp( + code=status.HTTP_200_OK, + message="子工作流获取成功", + result=FlowStructureGetMsg(flow=result), + ).model_dump(exclude_none=True, by_alias=True), + ) + + @router.delete( "", response_model=FlowStructureDeleteRsp, diff --git a/apps/routers/variable.py b/apps/routers/variable.py index 2e04e8ad..7f10bcff 100644 --- a/apps/routers/variable.py +++ b/apps/routers/variable.py @@ -594,6 +594,7 @@ async def list_variables( flow_id: Optional[str] = Query(default=None, description="流程ID（环境级和对话级变量必需）"), conversation_id: Optional[str] = Query(default=None, description="对话ID（系统级和对话级变量必需）"), current_step_id: Optional[str] = Query(default=None, description="当前步骤ID（用于获取前置节点变量）"), + exclude_pattern: Optional[str] = Query(default=None, description="排除模式：'step_id'排除包含.的变量名") ) -> VariableListResponse: """列出指定作用域的变量""" try: @@ -614,6 +615,11 @@ async def list_variables( ) variables.extend(predecessor_variables) + # 应用排除模式过滤 + if exclude_pattern == "step_id" and scope == VariableScope.CONVERSATION: + # 排除包含"."的变量名（即节点特定变量），只保留全局对话变量 + variables = [var for var in variables if "." not in var.name] + # 过滤权限并构建响应 filtered_variables = [] for variable in variables: diff --git a/apps/scheduler/call/__init__.py b/apps/scheduler/call/__init__.py index d7abec89..65d937e0 100644 --- a/apps/scheduler/call/__init__.py +++ b/apps/scheduler/call/__init__.py @@ -12,6 +12,7 @@ from apps.scheduler.call.sql.sql import SQL from apps.scheduler.call.suggest.suggest import Suggestion from apps.scheduler.call.choice.choice import Choice from apps.scheduler.call.loop.loop import Loop +from apps.scheduler.call.variable_assign.variable_assign import VariableAssign # 只包含需要在编排界面展示的工具 __all__ = [ @@ -25,5 +26,6 @@ __all__ = [ "Graph", "Suggestion", "Choice", - "Loop" + "Loop", + "VariableAssign" ] diff --git a/apps/scheduler/call/choice/choice.py b/apps/scheduler/call/choice/choice.py index ab7d581d..50485105 100644 --- a/apps/scheduler/call/choice/choice.py +++ b/apps/scheduler/call/choice/choice.py @@ -224,18 +224,18 @@ class Choice(CoreCall, input_model=ChoiceInput, output_model=ChoiceOutput): condition.right = _right # 检查运算符是否有效 - if condition.operate is None: + if condition.operator is None: return False, "条件缺少运算符" # 根据运算符确定期望的值类型 try: - expected_type = ConditionHandler.get_value_type_from_operate(condition.operate) + expected_type = ConditionHandler.get_value_type_from_operate(condition.operator) except Exception as e: - return False, f"不支持的运算符: {condition.operate}" + return False, f"不支持的运算符: {condition.operator}" # 检查类型是否与运算符匹配 if not expected_type == _left.type == _right.type: - return False, f"左值类型 {_left.type.value} 与运算符 {condition.operate} 不匹配" + return False, f"左值类型 {_left.type.value} 与运算符 {condition.operator} 不匹配" return True, "" @@ -303,16 +303,21 @@ class Choice(CoreCall, input_model=ChoiceInput, output_model=ChoiceOutput): async def _init(self, call_vars: CallVars) -> ChoiceInput: """初始化Choice工具""" + + prepared_choices = await self._prepare_message(call_vars) + return ChoiceInput( - choices=await self._prepare_message(call_vars), + choices=prepared_choices, ) async def _exec( self, input_data: dict[str, Any] ) -> AsyncGenerator[CallOutputChunk, None]: """执行Choice工具""" + # 解析输入数据 data = ChoiceInput(**input_data) + try: branch_id = ConditionHandler.handler(data.choices) yield CallOutputChunk( diff --git a/apps/scheduler/call/choice/condition_handler.py b/apps/scheduler/call/choice/condition_handler.py index 7048c2e9..c14b3c79 100644 --- a/apps/scheduler/call/choice/condition_handler.py +++ b/apps/scheduler/call/choice/condition_handler.py @@ -70,6 +70,10 @@ class ConditionHandler(BaseModel): return True if value.type == ValueType.BOOL and isinstance(value.value, bool): return True + # 添加对REFERENCE类型的处理 - 如果是REFERENCE类型但没有被解析，说明解析失败 + # 在这种情况下，我们将其视为字符串类型进行基本的兼容性检查 + if value.type == ValueType.REFERENCE and isinstance(value.value, str): + return True return False @staticmethod @@ -112,39 +116,39 @@ class ConditionHandler(BaseModel): 判断条件是否成立。 Args: - condition (Condition): 'left', 'operate', 'right' + condition (Condition): 'left', 'operator', 'right' Returns: bool """ left = condition.left - operate = condition.operate + operator = condition.operator right = condition.right # 根据操作符动态推断值类型 try: - value_type = ConditionHandler.get_value_type_from_operate(operate) + value_type = ConditionHandler.get_value_type_from_operate(operator) except Exception as e: - logger.error("无法推断操作符 %s 的值类型: %s", operate, e) - msg = f"无法推断操作符 {operate} 的值类型: {e}" + logger.error("无法推断操作符 %s 的值类型: %s", operator, e) + msg = f"无法推断操作符 {operator} 的值类型: {e}" raise ValueError(msg) result = None if value_type == ValueType.STRING: - result = ConditionHandler._judge_string_condition(left, operate, right) + result = ConditionHandler._judge_string_condition(left, operator, right) elif value_type == ValueType.NUMBER: - result = ConditionHandler._judge_number_condition(left, operate, right) + result = ConditionHandler._judge_number_condition(left, operator, right) elif value_type == ValueType.BOOL: - result = ConditionHandler._judge_bool_condition(left, operate, right) + result = ConditionHandler._judge_bool_condition(left, operator, right) elif value_type == ValueType.LIST: - result = ConditionHandler._judge_list_condition(left, operate, right) + result = ConditionHandler._judge_list_condition(left, operator, right) elif value_type == ValueType.DICT: - result = ConditionHandler._judge_dict_condition(left, operate, right) + result = ConditionHandler._judge_dict_condition(left, operator, right) else: - logger.error("不支持的数据类型: %s", value_type) - msg = f"不支持的数据类型: {value_type}" + msg = f"不支持的值类型: {value_type}" raise ValueError(msg) + return result @staticmethod diff --git a/apps/scheduler/call/choice/schema.py b/apps/scheduler/call/choice/schema.py index 6435b6a3..38dc617c 100644 --- a/apps/scheduler/call/choice/schema.py +++ b/apps/scheduler/call/choice/schema.py @@ -4,7 +4,7 @@ import uuid from enum import Enum -from pydantic import Field +from pydantic import BaseModel, Field from apps.schemas.parameters import ( NumberOperate, @@ -37,7 +37,7 @@ class Condition(DataBase): left: Value = Field(description="左值", default=Value()) right: Value = Field(description="右值", default=Value()) - operate: NumberOperate | StringOperate | ListOperate | BoolOperate | DictOperate | None = Field( + operator: NumberOperate | StringOperate | ListOperate | BoolOperate | DictOperate | None = Field( description="运算符", default=None) id: str = Field(description="条件ID", default_factory=lambda: str(uuid.uuid4())) diff --git a/apps/scheduler/call/core.py b/apps/scheduler/call/core.py index 3c53d91e..c1a65d16 100644 --- a/apps/scheduler/call/core.py +++ b/apps/scheduler/call/core.py @@ -255,7 +255,8 @@ class CoreCall(BaseModel): match.strip(), user_sub=call_vars.ids.user_sub, flow_id=call_vars.ids.flow_id, - conversation_id=call_vars.ids.conversation_id + conversation_id=call_vars.ids.conversation_id, + current_step_id=getattr(self, '_step_id', None) ) # 替换原始文本中的变量引用 resolved_text = resolved_text.replace(f'{{{{{match}}}}}', str(resolved_value)) @@ -266,36 +267,63 @@ class CoreCall(BaseModel): return resolved_text - async def _resolve_single_value(self, value: "Value", call_vars: CallVars) -> "Value": - """解析Value对象中的变量引用（{{...}} 语法） + async def _resolve_single_value(self, value, call_vars: CallVars): + """解析单个变量引用 Args: - value: Value对象 - call_vars: Call变量 + value: Value对象，包含type和value字段 + call_vars: Call变量上下文 Returns: - 解析后的Value对象 + Value: 解析后的Value对象，如果是引用类型则解析为具体值和类型 """ - # 运行时导入避免循环依赖 + # 🔑 将导入语句放在方法开头，避免作用域问题 + from apps.schemas.parameters import ValueType + from apps.scheduler.variable.type import VariableType as VarType from apps.scheduler.call.choice.schema import Value - - # 检查是否包含变量引用语法 - match = re.match(r'^\{\{.*?\}\}$', value.value) - if not value.type == ValueType.REFERENCE or not match: - return value + # 如果不是引用类型，直接返回 + if value.type != ValueType.REFERENCE: + return value + try: # 解析变量引用 resolved_value, resolved_type = await VariableIntegration.resolve_variable_reference( - match.group().strip(), + reference=value.value, user_sub=call_vars.ids.user_sub, flow_id=call_vars.ids.flow_id, - conversation_id=call_vars.ids.conversation_id + conversation_id=call_vars.ids.conversation_id, + current_step_id=getattr(self, '_step_id', None) ) + + # 🔑 关键修复：将VariableType转换为ValueType + # VariableType到ValueType的映射 + type_mapping = { + VarType.STRING: ValueType.STRING, + VarType.NUMBER: ValueType.NUMBER, + VarType.BOOLEAN: ValueType.BOOL, + VarType.OBJECT: ValueType.DICT, + VarType.ARRAY_STRING: ValueType.LIST, + VarType.ARRAY_NUMBER: ValueType.LIST, + VarType.ARRAY_OBJECT: ValueType.LIST, + VarType.ARRAY_ANY: ValueType.LIST, + VarType.ARRAY_FILE: ValueType.LIST, + VarType.ARRAY_BOOLEAN: ValueType.LIST, + VarType.ARRAY_SECRET: ValueType.LIST, + } + + # 转换类型 + if resolved_type in type_mapping: + converted_type = type_mapping[resolved_type] + else: + # 如果没有映射，默认为STRING + converted_type = ValueType.STRING + except Exception as e: - logger.warning(f"[CoreCall] 解析变量引用 '{match.group()}' 失败: {e}") + logger.warning(f"[CoreCall] 解析变量引用 '{value.value}' 失败: {e}") + return value - return Value(value=resolved_value, type=resolved_type) + return Value(value=resolved_value, type=converted_type) async def _set_input(self, executor: "StepExecutor") -> None: """获取Call的输入""" diff --git a/apps/scheduler/call/facts/facts.py b/apps/scheduler/call/facts/facts.py index 10241d85..33556a8d 100644 --- a/apps/scheduler/call/facts/facts.py +++ b/apps/scheduler/call/facts/facts.py @@ -1,6 +1,7 @@ # Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. """提取事实工具""" +import logging from collections.abc import AsyncGenerator from typing import TYPE_CHECKING, Any, Self @@ -84,18 +85,28 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): # 提取事实信息 facts_tpl = env.from_string(FACTS_PROMPT) facts_prompt = facts_tpl.render(conversation=data.message) - facts_obj: FactsGen = await self._json([ - {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": facts_prompt}, - ], FactsGen) # type: ignore[arg-type] + try: + facts_obj: FactsGen = await self._json([ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": facts_prompt}, + ], FactsGen) # type: ignore[arg-type] + except Exception as e: + # 如果 LLM 返回格式不正确，使用默认空列表 + logging.warning(f"[FactsCall] 事实提取失败，使用默认值: {e}") + facts_obj = FactsGen(facts=[]) # 更新用户画像 domain_tpl = env.from_string(DOMAIN_PROMPT) domain_prompt = domain_tpl.render(conversation=data.message) - domain_list: DomainGen = await self._json([ - {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": domain_prompt}, - ], DomainGen) # type: ignore[arg-type] + try: + domain_list: DomainGen = await self._json([ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": domain_prompt}, + ], DomainGen) # type: ignore[arg-type] + except Exception as e: + # 如果 LLM 返回格式不正确，使用默认空列表 + logging.warning(f"[FactsCall] 域名提取失败，使用默认值: {e}") + domain_list = DomainGen(keywords=[]) for domain in domain_list.keywords: await UserDomainManager.update_user_domain_by_user_sub_and_domain_name(data.user_sub, domain) diff --git a/apps/scheduler/call/facts/schema.py b/apps/scheduler/call/facts/schema.py index c16b94eb..e5842b1f 100644 --- a/apps/scheduler/call/facts/schema.py +++ b/apps/scheduler/call/facts/schema.py @@ -9,13 +9,13 @@ from apps.scheduler.call.core import DataBase class DomainGen(BaseModel): """生成的领域信息结果""" - keywords: list[str] = Field(description="关键词或标签列表，可以为空。") + keywords: list[str] = Field(default=[], description="关键词或标签列表，可以为空。") class FactsGen(BaseModel): """生成的提取事实结果""" - facts: list[str] = Field(description="从对话中提取的事实条目，可以为空。") + facts: list[str] = Field(default=[], description="从对话中提取的事实条目，可以为空。") class FactsInput(DataBase): diff --git a/apps/scheduler/call/loop/loop.py b/apps/scheduler/call/loop/loop.py index cbd76a1e..001aa6ff 100644 --- a/apps/scheduler/call/loop/loop.py +++ b/apps/scheduler/call/loop/loop.py @@ -4,11 +4,15 @@ import copy import logging import uuid +import asyncio from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, TYPE_CHECKING from pydantic import Field +if TYPE_CHECKING: + from apps.scheduler.executor.step import StepExecutor + from apps.scheduler.call.core import CoreCall from apps.scheduler.call.choice.condition_handler import ConditionHandler from apps.scheduler.call.loop.schema import LoopInput, LoopOutput, LoopStopCondition @@ -35,6 +39,15 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): stop_condition: LoopStopCondition = Field(description="循环终止条件", default=LoopStopCondition()) max_iteration: int = Field(description="最大循环次数", default=10, ge=1, le=100) sub_flow_id: str = Field(description="子工作流ID", default="") + + # 输出节流相关参数 + output_throttle_interval: float = Field(description="输出节流间隔(秒)", default=0.1) + progress_report_interval: int = Field(description="进度报告间隔(每N次循环)", default=1) + + # 保存传入的CallVars用于子工作流执行 + call_vars: CallVars | None = Field(default=None, exclude=True) + # 保存StepExecutor引用用于子工作流执行 + step_executor: Any = Field(default=None, exclude=True) @classmethod def info(cls) -> CallInfo: @@ -59,31 +72,63 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): if not self.stop_condition.conditions: return True, "" + # 保存原始引用信息的字典 + self._original_references = {} + # 处理每个条件，复用choice.py的逻辑 valid_conditions = [] - for condition_original in self.stop_condition.conditions: + for i, condition_original in enumerate(self.stop_condition.conditions): condition = copy.deepcopy(condition_original) # 处理左值 - if condition.left.type and hasattr(condition.left, 'value'): - try: - resolved_left = await self._resolve_single_value(condition.left, call_vars) - condition.left = resolved_left - except Exception as e: - logger.warning(f"[Loop] 停止条件左值解析失败: {e}") - continue + success, resolved_left, error_msg = await self._process_condition_value( + condition.left, call_vars, "左值") + if not success: + logger.warning(f"[Loop] 停止条件左值处理失败: {error_msg}") + continue + + # 🔑 保存原始的引用信息，用于后续重新解析 + condition_id = getattr(condition, 'id', f'condition_{i}') + if condition.left.type.value == 'reference': + self._original_references[f'{condition_id}_left'] = condition.left.value # 处理右值 - if condition.right.type and hasattr(condition.right, 'value'): - try: - resolved_right = await self._resolve_single_value(condition.right, call_vars) - condition.right = resolved_right - except Exception as e: - logger.warning(f"[Loop] 停止条件右值解析失败: {e}") - continue + success, resolved_right, error_msg = await self._process_condition_value( + condition.right, call_vars, "右值") + if not success: + logger.warning(f"[Loop] 停止条件右值处理失败: {error_msg}") + continue + + # 🔑 保存原始的引用信息，用于后续重新解析 + if condition.right.type.value == 'reference': + self._original_references[f'{condition_id}_right'] = condition.right.value + + # 更新条件对象中的解析后的值 + condition.left = resolved_left + condition.right = resolved_right + + # 检查运算符是否有效 + if condition.operator is None: + logger.warning(f"[Loop] 停止条件缺少运算符") + continue + + # 根据运算符确定期望的值类型 + try: + expected_type = ConditionHandler.get_value_type_from_operate(condition.operator) + except Exception as e: + logger.warning(f"[Loop] 不支持的运算符: {condition.operator}") + continue + + # 检查类型是否与运算符匹配 + if not expected_type == resolved_left.type == resolved_right.type: + logger.warning(f"[Loop] 左值类型 {resolved_left.type.value} 与运算符 {condition.operator} 不匹配") + continue # 验证条件 - if condition.operate and ConditionHandler.check_value_type(condition.left) and ConditionHandler.check_value_type(condition.right): + left_valid = ConditionHandler.check_value_type(condition.left) + right_valid = ConditionHandler.check_value_type(condition.right) + + if left_valid and right_valid: valid_conditions.append(condition) # 更新有效条件 @@ -93,27 +138,218 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): except Exception as e: return False, f"停止条件处理失败: {e}" - def _check_stop_condition(self) -> bool: + async def _process_condition_value(self, value, call_vars: CallVars, value_position: str) -> tuple[bool, Any, str]: + """处理条件值（左值或右值）- 参考Choice组件的实现 + + Args: + value: 需要处理的值对象 + call_vars: Call变量 + value_position: 值的位置描述（"左值"或"右值"） + + Returns: + tuple[bool, Value, str]: (是否成功, 处理后的Value对象，错误消息) + """ + # 运行时导入避免循环依赖 + from apps.scheduler.call.choice.schema import Value + from apps.schemas.parameters import ValueType + + # 处理reference类型 + if value.type == ValueType.REFERENCE: + try: + resolved_value = await self._resolve_single_value(value, call_vars) + except Exception as e: + return False, None, f"{value_position}引用解析失败: {e}" + else: + resolved_value = value + + # 对于非引用类型，先进行类型转换，然后验证值类型（与Choice组件保持一致） + try: + converted_value = self._convert_value_to_expected_type(resolved_value) + resolved_value = converted_value + except Exception as e: + return False, None, f"{value_position}类型转换失败: {e}" + + # 对于非引用类型，进行类型验证 + is_valid = ConditionHandler.check_value_type(resolved_value) + + if not is_valid: + return False, None, f"{value_position}类型不匹配: {resolved_value.value} 应为 {resolved_value.type}" + + return True, resolved_value, "" + + def _convert_value_to_expected_type(self, value) -> Any: + """根据期望的类型转换值 - 复制自Choice组件 + + Args: + value: 需要转换的Value对象 + + Returns: + Value: 转换后的Value对象 + + Raises: + ValueError: 当值无法转换为指定类型时 + """ + # 运行时导入避免循环依赖 + from apps.scheduler.call.choice.schema import Value + from apps.schemas.parameters import ValueType + + if value.value is None: + return value + + # 如果已经是正确的类型，直接返回 + if ConditionHandler.check_value_type(value): + return value + + # 创建新的Value对象进行转换 + converted_value = Value(type=value.type, value=value.value) + + try: + if value.type == ValueType.NUMBER: + # 转换为数字 + if isinstance(value.value, str): + if value.value.strip() == "": + converted_value.value = 0 + elif '.' in value.value or 'e' in value.value.lower(): + converted_value.value = float(value.value) + else: + converted_value.value = int(value.value) + elif isinstance(value.value, bool): + converted_value.value = int(value.value) + elif isinstance(value.value, (int, float)): + converted_value.value = value.value + else: + converted_value.value = float(value.value) + + elif value.type == ValueType.STRING: + # 转换为字符串 + if isinstance(value.value, (dict, list)): + import json + converted_value.value = json.dumps(value.value) + else: + converted_value.value = str(value.value) + + elif value.type == ValueType.BOOL: + # 转换为布尔值 + if isinstance(value.value, str): + lower_value = value.value.lower().strip() + if lower_value in ('true', '1', 'yes', 'on'): + converted_value.value = True + elif lower_value in ('false', '0', 'no', 'off'): + converted_value.value = False + else: + converted_value.value = bool(value.value) + else: + converted_value.value = bool(value.value) + + elif value.type == ValueType.LIST: + # 转换为列表 + if isinstance(value.value, str): + import json + try: + converted_value.value = json.loads(value.value) + if not isinstance(converted_value.value, list): + # 如果不是列表，尝试按逗号分割 + converted_value.value = [item.strip() for item in value.value.split(',') if item.strip()] + except json.JSONDecodeError: + # 按逗号分割 + converted_value.value = [item.strip() for item in value.value.split(',') if item.strip()] + elif isinstance(value.value, list): + converted_value.value = value.value + else: + converted_value.value = [value.value] + + elif value.type == ValueType.DICT: + # 转换为字典 + if isinstance(value.value, str): + import json + try: + converted_value.value = json.loads(value.value) + if not isinstance(converted_value.value, dict): + raise ValueError(f"解析后的值不是字典类型: {type(converted_value.value)}") + except json.JSONDecodeError as e: + raise ValueError(f"无法解析JSON字典: {e}") + elif isinstance(value.value, dict): + converted_value.value = value.value + else: + raise ValueError(f"无法将 {type(value.value)} 转换为字典") + + else: + # 对于其他类型，保持原值 + converted_value.value = value.value + + except (ValueError, TypeError, ImportError) as e: + raise ValueError(f"无法将值 '{value.value}' 转换为类型 '{value.type.value}': {str(e)}") + + return converted_value + + async def _check_stop_condition(self) -> bool: """检查是否满足停止条件 Returns: bool: 是否应该停止循环 """ try: + if not self.stop_condition.conditions: return False results = [] - for condition in self.stop_condition.conditions: - result = ConditionHandler._judge_condition(condition) - results.append(result) + for i, condition in enumerate(self.stop_condition.conditions): + # 🔑 关键修复：在每次评估时重新解析REFERENCE类型的值 + # 创建条件副本并重新解析引用 + eval_condition = copy.deepcopy(condition) + + condition_id = getattr(condition, 'id', f'condition_{i}') + + # 重新解析左值如果它是REFERENCE类型 + left_ref_key = f'{condition_id}_left' + if hasattr(self, '_original_references') and left_ref_key in self._original_references: + try: + # 运行时导入避免循环依赖 + from apps.scheduler.call.choice.schema import Value + from apps.schemas.parameters import ValueType + + # 使用原始引用创建新的Value对象进行解析 + original_ref_value = Value( + type=ValueType.REFERENCE, + value=self._original_references[left_ref_key] + ) + resolved_left = await self._resolve_single_value(original_ref_value, self.call_vars) + eval_condition.left = resolved_left + except Exception as e: + logger.error(f"[Loop] 左值重新解析失败: {e}") + # 重新解析右值如果它是REFERENCE类型 + right_ref_key = f'{condition_id}_right' + if hasattr(self, '_original_references') and right_ref_key in self._original_references: + try: + # 运行时导入避免循环依赖 + from apps.scheduler.call.choice.schema import Value + from apps.schemas.parameters import ValueType + + # 使用原始引用创建新的Value对象进行解析 + original_ref_value = Value( + type=ValueType.REFERENCE, + value=self._original_references[right_ref_key] + ) + resolved_right = await self._resolve_single_value(original_ref_value, self.call_vars) + eval_condition.right = resolved_right + except Exception as e: + logger.error(f"[Loop] 右值重新解析失败: {e}") + + # 评估条件 + result = ConditionHandler._judge_condition(eval_condition) + results.append(result) + + # 计算最终结果 if self.stop_condition.logic.value == "and": - return all(results) + final_result = all(results) elif self.stop_condition.logic.value == "or": - return any(results) + final_result = any(results) else: - return all(results) if results else False + final_result = all(results) if results else False + + return final_result except Exception as e: logger.error(f"[Loop] 检查停止条件失败: {e}") @@ -171,15 +407,15 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): # 检查子flow是否已存在 flow_loader = FlowLoader() - existing_flow = await flow_loader.load(app_id, self.sub_flow_id) + existing_flow = await flow_loader.load_subflow(app_id, parent_flow_id, self.sub_flow_id) if not existing_flow: # 创建默认子flow配置 default_flow = self._create_default_sub_flow() try: - # 保存子flow到文件系统 - await flow_loader.save(app_id, self.sub_flow_id, default_flow) + # 保存子flow到正确的层次化路径 + await flow_loader.save_subflow(app_id, parent_flow_id, self.sub_flow_id, default_flow) logger.info(f"[Loop] 创建新的子flow: {self.sub_flow_id}") except Exception as e: logger.error(f"[Loop] 创建子flow失败: {e}") @@ -189,47 +425,135 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): return self.sub_flow_id - async def _execute_sub_flow(self, iteration: int, call_vars: CallVars) -> dict[str, Any]: + async def _execute_sub_flow(self, iteration: int, call_vars: CallVars, step_executor=None) -> dict[str, Any]: """执行子工作流 Args: iteration: 当前循环次数 call_vars: Call变量 + step_executor: 父工作流的StepExecutor，用于访问消息队列 Returns: dict[str, Any]: 子工作流的执行结果 """ logger.info(f"[Loop] 执行第 {iteration} 次循环，子flow ID: {self.sub_flow_id}") - # TODO: 这里需要实现实际的子flow执行逻辑 - # 在实际的实现中，这里应该： - # 1. 创建一个新的FlowExecutor实例来执行子flow - # 2. 将当前的循环变量设置到子flow的执行上下文中 - # 3. 执行子flow并获取结果 - # 4. 从结果中提取更新后的变量 - - # 由于这需要依赖FlowExecutor和完整的执行环境， - # 现在先返回当前变量作为占位符 - # 在实际集成时，需要调用FlowExecutor来执行子flow - try: - # 模拟变量更新（实际应该从子flow执行结果中获取） + # 获取应用ID和父工作流ID + app_id = call_vars.ids.app_id + parent_flow_id = call_vars.ids.flow_id + + # 加载子工作流 + flow_loader = FlowLoader() + sub_flow = await flow_loader.load_subflow(app_id, parent_flow_id, self.sub_flow_id) + + if not sub_flow: + logger.warning(f"[Loop] 子工作流不存在: {self.sub_flow_id}") + # 返回当前变量，不做修改 + return copy.deepcopy(self.variables) + + logger.info(f"[Loop] 成功加载子工作流: {sub_flow.name}") + + # 开始真正执行子工作流 updated_variables = copy.deepcopy(self.variables) # 添加循环相关的系统变量 updated_variables.update({ f"loop_iteration_{iteration}": iteration, "current_iteration": iteration, + "sub_flow_executed": True, + "sub_flow_name": sub_flow.name, }) + # 执行子工作流中的每个步骤（除了start和end） + if sub_flow.steps and step_executor: + await self._execute_sub_flow_steps(sub_flow, step_executor, iteration, updated_variables) + + logger.info(f"[Loop] 第 {iteration} 次循环执行完成，处理了 {len(sub_flow.steps)} 个步骤") + return updated_variables except Exception as e: logger.error(f"[Loop] 执行子flow失败: {e}") - raise CallError(message=f"执行子工作流失败: {e}", data={}) + # 发生错误时，返回原变量避免中断循环 + logger.warning(f"[Loop] 由于错误，第 {iteration} 次循环返回原变量") + return copy.deepcopy(self.variables) + + async def _execute_sub_flow_steps(self, sub_flow, step_executor, iteration: int, variables: dict) -> None: + """执行子工作流中的具体步骤 + + Args: + sub_flow: 子工作流对象 + step_executor: 父工作流的StepExecutor + iteration: 当前循环次数 + variables: 当前变量状态 + """ + from apps.scheduler.executor.step import StepExecutor + from apps.schemas.task import StepQueueItem + + # 保存原始的循环节点状态，确保不被子步骤状态影响 + original_loop_step_id = step_executor.task.state.step_id + original_loop_step_name = step_executor.task.state.step_name + original_loop_status = step_executor.task.state.status + + # 筛选出需要执行的步骤（排除start和end） + executable_steps = [ + (step_id, step) for step_id, step in sub_flow.steps.items() + if step.type not in ['start', 'end'] + ] + + logger.info(f"[Loop] 子工作流中有 {len(executable_steps)} 个可执行步骤") + + for step_id, step in executable_steps: + try: + logger.info(f"[Loop] 执行子工作流步骤: {step.name} (ID: {step_id})") + + # 创建步骤队列项 + step_queue_item = StepQueueItem( + step_id=step_id, + step=step + ) + + # 🔑 关键修改：创建独立的任务状态副本，避免影响循环节点本身的状态 + import copy + sub_task = copy.deepcopy(step_executor.task) + + # 创建子步骤执行器，使用独立的任务状态 + sub_step_executor = StepExecutor( + task=sub_task, + msg_queue=step_executor.msg_queue, + background=step_executor.background, + question=step_executor.question, + step=step_queue_item + ) + + # 初始化并运行子步骤 + await sub_step_executor.init() + + # 🔑 为子工作流步骤设置独立的状态标识，不影响父循环节点 + sub_step_executor.task.state.step_id = f"loop_{iteration}_{step_id}" + sub_step_executor.task.state.step_name = f"[循环{iteration}] {step.name}" + + # 执行步骤 + await sub_step_executor.run() + + logger.info(f"[Loop] 子工作流步骤 {step.name} 执行完成") + + except Exception as e: + logger.error(f"[Loop] 子工作流步骤 {step.name} 执行失败: {e}") + # 继续执行其他步骤，不中断整个循环 + + # 🔑 关键修改：确保循环节点的状态完全恢复，不受子步骤影响 + step_executor.task.state.step_id = original_loop_step_id + step_executor.task.state.step_name = original_loop_step_name + step_executor.task.state.status = original_loop_status + logger.info(f"[Loop] 已完全恢复循环节点状态: {original_loop_step_name} (ID: {original_loop_step_id})") async def _init(self, call_vars: CallVars) -> LoopInput: """初始化Loop工具""" + # 保存call_vars以便后续使用 + self.call_vars = call_vars + # 处理停止条件 success, error_msg = await self._process_stop_condition(call_vars) if not success: @@ -237,7 +561,7 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): # 从call_vars中获取app_id和flow_id（如果可用） app_id = getattr(call_vars.ids, 'app_id', 'default') - flow_id = getattr(call_vars.state, 'flow_id', 'default') + flow_id = getattr(call_vars.ids, 'flow_id', 'default') # 创建子flow sub_flow_id = await self._create_sub_flow_if_not_exists(app_id, flow_id) @@ -248,6 +572,15 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): max_iteration=self.max_iteration, sub_flow_id=sub_flow_id, ) + + async def exec(self, executor: "StepExecutor", input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + """重写exec方法来保存executor引用""" + # 保存executor引用 + self.step_executor = executor + + # 调用父类的exec方法 + async for chunk in super().exec(executor, input_data): + yield chunk async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: """执行Loop工具""" @@ -258,6 +591,7 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): iteration_count = 0 stop_reason = "" current_variables = copy.deepcopy(data.variables) + last_output_time = 0.0 logger.info(f"[Loop] 开始循环执行，最大次数: {data.max_iteration}, 子flow: {data.sub_flow_id}") @@ -267,22 +601,64 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): logger.info(f"[Loop] 开始第 {iteration_count} 次循环") # 检查停止条件（在执行前检查） - if self._check_stop_condition(): + if await self._check_stop_condition(): stop_reason = "condition_met" logger.info(f"[Loop] 满足停止条件，在第 {iteration_count - 1} 次循环后停止") iteration_count -= 1 # 未实际执行这次循环 break # 执行子工作流 - call_vars = CallVars() # 临时创建，实际需要传入正确的call_vars - result_variables = await self._execute_sub_flow(iteration_count, call_vars) + result_variables = await self._execute_sub_flow(iteration_count, self.call_vars, self.step_executor) # 更新循环变量 current_variables.update(result_variables) self.variables = current_variables + # 检查是否需要输出进度信息（节流机制） + current_time = asyncio.get_event_loop().time() + should_output_progress = ( + iteration_count % self.progress_report_interval == 0 or + (current_time - last_output_time) >= self.output_throttle_interval + ) + + if should_output_progress and iteration_count < data.max_iteration: + # 🔑 发送循环进度事件，确保前端知道循环仍在进行 + if self.step_executor: + from apps.schemas.enum_var import EventType + progress_data = { + "iteration": iteration_count, + "total": data.max_iteration, + "status": "running", + "current_iteration": iteration_count, + "loop_step_name": self.step_executor.task.state.step_name + } + # 发送循环进度事件，保持循环节点本身的stepId和stepName + await self.step_executor.push_message("loop.progress", progress_data) + + # 输出中间进度信息 + progress_output = LoopOutput( + iteration_count=iteration_count, + stop_reason="", + current_iteration=iteration_count + ) + + yield CallOutputChunk( + type=CallOutputType.PROGRESS, + content={ + "iteration": iteration_count, + "total": data.max_iteration, + "status": "running", + "current_iteration": iteration_count + }, + ) + last_output_time = current_time + + # 添加短暂延迟，避免过于频繁的输出 + if self.output_throttle_interval > 0: + await asyncio.sleep(self.output_throttle_interval) + # 再次检查停止条件（在执行后检查） - if self._check_stop_condition(): + if await self._check_stop_condition(): stop_reason = "condition_met" logger.info(f"[Loop] 满足停止条件，在第 {iteration_count} 次循环后停止") break @@ -292,11 +668,25 @@ class Loop(CoreCall, input_model=LoopInput, output_model=LoopOutput): stop_reason = "max_iteration_reached" logger.info(f"[Loop] 达到最大循环次数 {data.max_iteration}") - # 返回结果 + # 🔑 关键修改：在循环完成时发送明确的完成事件 + if self.step_executor: + from apps.schemas.enum_var import EventType + completion_data = { + "iteration_count": iteration_count, + "stop_reason": stop_reason, + "current_iteration": iteration_count, + "status": "completed", + "loop_step_name": self.step_executor.task.state.step_name + } + # 发送循环完成事件，确保前端知道循环已结束 + await self.step_executor.push_message("loop.completed", completion_data) + logger.info(f"[Loop] 已发送循环完成事件: {self.step_executor.task.state.step_name}") + + # 返回最终结果 output = LoopOutput( iteration_count=iteration_count, stop_reason=stop_reason, - variables=current_variables + current_iteration=iteration_count ) logger.info(f"[Loop] 循环执行完成，执行次数: {iteration_count}, 停止原因: {stop_reason}") diff --git a/apps/scheduler/call/loop/schema.py b/apps/scheduler/call/loop/schema.py index ba6cb10a..0c3c2f68 100644 --- a/apps/scheduler/call/loop/schema.py +++ b/apps/scheduler/call/loop/schema.py @@ -20,7 +20,7 @@ class LoopStopCondition(DataBase): class LoopInput(DataBase): """循环节点的输入""" - variables: dict[str, Any] = Field(description="循环变量", default={}) + variables: dict[str, Any] = Field(description="循环变量，保存flow时会自动创建对话变量模板conversation.{step_id}.k=v供后续节点引用", default={}) stop_condition: LoopStopCondition = Field(description="循环终止条件", default=LoopStopCondition()) max_iteration: int = Field(description="最大循环次数", default=10, ge=1, le=100) sub_flow_id: str = Field(description="子工作流ID", default="") @@ -31,4 +31,4 @@ class LoopOutput(DataBase): iteration_count: int = Field(description="实际执行的循环次数", default=0) stop_reason: str = Field(description="停止原因", default="") # "max_iteration_reached" 或 "condition_met" - variables: dict[str, Any] = Field(description="循环后的变量状态", default={}) \ No newline at end of file + current_iteration: int = Field(description="当前循环次数", default=0) \ No newline at end of file diff --git a/apps/scheduler/call/variable_assign/__init__.py b/apps/scheduler/call/variable_assign/__init__.py new file mode 100644 index 00000000..0a7c7b9f --- /dev/null +++ b/apps/scheduler/call/variable_assign/__init__.py @@ -0,0 +1,6 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""变量赋值Call包""" + +from apps.scheduler.call.variable_assign.variable_assign import VariableAssign + +__all__ = ["VariableAssign"] diff --git a/apps/scheduler/call/variable_assign/schema.py b/apps/scheduler/call/variable_assign/schema.py new file mode 100644 index 00000000..b899cec2 --- /dev/null +++ b/apps/scheduler/call/variable_assign/schema.py @@ -0,0 +1,61 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""变量赋值Call的Schema定义""" + +from typing import Any, List, Optional +from enum import StrEnum + +from pydantic import Field + +from apps.scheduler.call.core import DataBase + + +class StringOperation(StrEnum): + """字符串类型变量操作枚举""" + OVERWRITE = "overwrite" # 覆盖 + CLEAR = "clear" # 清空 + + +class NumberOperation(StrEnum): + """数值类型变量操作枚举""" + OVERWRITE = "overwrite" # 覆盖 + ADD = "add" # 加法 + SUBTRACT = "subtract" # 减法 + MULTIPLY = "multiply" # 乘法 + DIVIDE = "divide" # 除法 + MODULO = "modulo" # 求余 + POWER = "power" # 乘幂 + SQRT = "sqrt" # 开方 + CLEAR = "clear" # 清空 + + +class ArrayOperation(StrEnum): + """数组类型变量操作枚举""" + OVERWRITE = "overwrite" # 覆盖 + CLEAR = "clear" # 清空 + APPEND = "append" # 追加单个元素 + EXTEND = "extend" # 扩展（追加多个元素） + POP_FIRST = "pop_first" # 移除首项 + POP_LAST = "pop_last" # 移除尾项 + + +class VariableOperation(DataBase): + """单个变量操作定义""" + variable_name: str = Field(description="要操作的变量名称") + operation: str = Field(description="操作类型") + value: Any = Field(description="操作值（某些操作不需要此参数）", default=None) + variable_type: Optional[str] = Field(description="变量类型", default=None) + is_value_reference: Optional[bool] = Field(description="值是否为变量引用", default=False) + + +class VariableAssignInput(DataBase): + """变量赋值输入参数""" + + operations: Optional[List[VariableOperation]] = Field( + description="变量操作列表", + default=None + ) + + +class VariableAssignOutput(DataBase): + """变量赋值输出结果（实际不使用）""" + pass diff --git a/apps/scheduler/call/variable_assign/variable_assign.py b/apps/scheduler/call/variable_assign/variable_assign.py new file mode 100644 index 00000000..583109a2 --- /dev/null +++ b/apps/scheduler/call/variable_assign/variable_assign.py @@ -0,0 +1,535 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""变量赋值Call""" + +import logging +import math +from collections.abc import AsyncGenerator +from typing import Any + +from apps.scheduler.call.core import CoreCall +from apps.scheduler.call.variable_assign.schema import ( + VariableAssignInput, + VariableAssignOutput, + VariableOperation, + StringOperation, + NumberOperation, + ArrayOperation, +) +from apps.scheduler.variable.type import VariableType +from apps.scheduler.variable.pool_manager import get_pool_manager +from apps.schemas.enum_var import CallType +from apps.schemas.scheduler import CallInfo, CallVars, CallOutputChunk, CallOutputType, CallError + + +logger = logging.getLogger(__name__) + + +class VariableAssign(CoreCall, input_model=VariableAssignInput, output_model=VariableAssignOutput): + """变量赋值Call""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._call_vars = None + + @classmethod + def info(cls) -> CallInfo: + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + return CallInfo( + name="变量赋值", + type=CallType.TRANSFORM, + description="对已有变量进行值的操作，支持字符串、数值和数组类型变量的多种操作" + ) + + async def _init(self, call_vars: CallVars) -> VariableAssignInput: + """ + 初始化Call + + :param CallVars call_vars: 由Executor传入的变量，包含当前运行信息 + :return: Call的输入 + :rtype: VariableAssignInput + """ + # 保存CallVars以便在_exec中使用 + self._call_vars = call_vars + + # 从实例属性中获取operations（已通过StepExecutor扁平化处理） + operations = getattr(self, 'operations', []) + + return VariableAssignInput( + operations=operations, + ) + + async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + """ + 执行变量赋值操作 + + :param dict[str, Any] input_data: 填充后的Call的最终输入 + """ + try: + # 获取CallVars从实例属性 + call_vars = getattr(self, '_call_vars', None) + if not call_vars: + raise CallError( + message="无法获取调用上下文", + data={} + ) + + results = [] # 记录执行结果 + + # 优先处理 operations 数组（从input_data中获取） + operations = input_data.get('operations', None) + logger.info(f"开始执行 {len(operations)} 个变量操作") + for i, operation in enumerate(operations): + try: + result = await self._execute_single_operation(operation, call_vars) + results.append({ + "index": i + 1, + "status": "success", + "variable_name": operation.get('variable_name') if isinstance(operation, dict) else operation.variable_name, + "operation": operation.get('operation') if isinstance(operation, dict) else operation.operation, + "result": result + }) + logger.debug(f"第 {i+1} 个操作执行成功") + except Exception as e: + error_info = { + "index": i + 1, + "status": "error", + "variable_name": operation.get('variable_name') if isinstance(operation, dict) else operation.variable_name, + "operation": operation.get('operation') if isinstance(operation, dict) else operation.operation, + "error": str(e) + } + results.append(error_info) + logger.error(f"第 {i+1} 个操作执行失败: {e}") + # 继续执行其他操作，不中断 + continue + + # 成功完成时 yield 结果 + yield CallOutputChunk( + type=CallOutputType.DATA, + content=VariableAssignOutput().model_dump(by_alias=True, exclude_none=True) + ) + + except Exception as e: + if isinstance(e, CallError): + raise # 重新抛出 CallError，让 StepExecutor 处理 + else: + logger.error(f"变量赋值操作失败: {e}") + raise CallError( + message=f"变量赋值操作失败: {e}", + data={} + ) from e + + async def _execute_single_operation(self, operation, call_vars) -> dict[str, Any]: + """ + 执行单个变量操作 + + :param operation: 变量操作对象或字典 + :param call_vars: 调用上下文 + :return: 操作结果 + """ + + try: + # 兼容处理：如果是字典则转换为属性访问 + if isinstance(operation, dict): + variable_name = operation.get('variable_name') + op_type = operation.get('operation') + value = operation.get('value') + variable_type = operation.get('variable_type') + is_value_reference = operation.get('is_value_reference', False) + else: + # 如果是 VariableOperation 对象 + variable_name = operation.variable_name + op_type = operation.operation + value = operation.value + variable_type = getattr(operation, 'variable_type', None) + is_value_reference = getattr(operation, 'is_value_reference', False) + + if not variable_name: + raise CallError( + message="变量名称不能为空", + data={} + ) + + if not op_type: + raise CallError( + message="操作类型不能为空", + data={} + ) + + # 如果值是变量引用，需要解析引用 + if is_value_reference and value: + value = await self._resolve_variable_reference(value, call_vars) + + # 执行变量操作 + await self._execute_variable_operation( + variable_name, op_type, value, call_vars + ) + + return { + "variable_name": variable_name, + "operation": op_type, + "value": value, + "success": True + } + + except Exception as e: + logger.error(f"执行单个变量操作失败: {e}") + raise + + async def _resolve_variable_reference(self, variable_reference: str, call_vars) -> Any: + """ + 解析变量引用，获取变量的实际值 + + :param variable_reference: 变量引用字符串 + :param call_vars: 调用上下文 + :return: 解析后的变量值 + """ + try: + # 使用统一的变量解析逻辑，支持更复杂的引用格式 + from apps.scheduler.variable.integration import VariableIntegration + + resolved_value, _ = await VariableIntegration.resolve_variable_reference( + reference=variable_reference, + user_sub=call_vars.ids.user_sub, + flow_id=call_vars.ids.flow_id, + conversation_id=call_vars.ids.conversation_id, + current_step_id=getattr(self, '_step_id', None) + ) + + return resolved_value + + except Exception as e: + logger.warning(f"变量引用 '{variable_reference}' 未找到，使用原始值") + return variable_reference + + async def _execute_variable_operation( + self, variable_name: str, operation: str, value: Any, call_vars: CallVars + ) -> None: + """ + 执行具体的变量操作 + + :param variable_name: 变量名称 + :param operation: 操作类型 + :param value: 操作值 + :param call_vars: 调用上下文 + :return: 操作结果 + """ + try: + # 解析变量名，支持 conversation.test 格式 + actual_variable_name, target_scope = self._parse_variable_name(variable_name) + + # 获取变量当前值和类型 + pool_manager = await get_pool_manager() + + old_variable = None + + # 根据解析出的作用域查找变量 + if target_scope == "conversation": + if call_vars.ids.conversation_id: + conversation_pool = await pool_manager.get_conversation_pool(call_vars.ids.conversation_id) + if conversation_pool: + old_variable = await conversation_pool.get_variable(actual_variable_name) + elif target_scope == "user": + if call_vars.ids.user_sub: + user_pool = await pool_manager.get_user_pool(call_vars.ids.user_sub) + if user_pool: + old_variable = await user_pool.get_variable(actual_variable_name) + elif target_scope == "environment" or target_scope == "system": + if call_vars.ids.flow_id: + flow_pool = await pool_manager.get_flow_pool(call_vars.ids.flow_id) + if flow_pool: + old_variable = await flow_pool.get_variable(actual_variable_name) + else: + # 没有明确作用域，按原有逻辑依次查找 + # 首先尝试从对话变量中获取 + if call_vars.ids.conversation_id: + conversation_pool = await pool_manager.get_conversation_pool(call_vars.ids.conversation_id) + if conversation_pool: + old_variable = await conversation_pool.get_variable(actual_variable_name) + + # 如果对话变量中没有，尝试从用户变量中获取 + if not old_variable and call_vars.ids.user_sub: + user_pool = await pool_manager.get_user_pool(call_vars.ids.user_sub) + if user_pool: + old_variable = await user_pool.get_variable(actual_variable_name) + + # 如果用户变量中也没有，尝试从环境变量中获取 + if not old_variable and call_vars.ids.flow_id: + flow_pool = await pool_manager.get_flow_pool(call_vars.ids.flow_id) + if flow_pool: + old_variable = await flow_pool.get_variable(actual_variable_name) + + if not old_variable: + raise CallError( + message=f"变量 '{variable_name}' 不存在", + data={"variable_name": variable_name, "actual_name": actual_variable_name, "scope": target_scope} + ) + + old_value = old_variable.value + var_type = old_variable.var_type + + # 根据变量类型执行不同的操作 + if var_type == VariableType.STRING: + new_value = await self._execute_string_operation(old_value, operation, value) + elif var_type == VariableType.NUMBER: + new_value = await self._execute_number_operation(old_value, operation, value) + elif var_type.is_array_type(): + new_value = await self._execute_array_operation(old_value, operation, value, var_type) + else: + raise CallError( + message=f"不支持的变量类型: {var_type}", + data={"variable_name": variable_name, "variable_type": str(var_type)} + ) + + # 更新变量值 + success = await self._update_variable_value( + actual_variable_name, new_value, call_vars, target_scope + ) + + if success: + logger.info(f"变量 '{variable_name}' 操作 '{operation}' 执行成功，从 {old_value} 更新为 {new_value}") + else: + raise CallError( + message=f"变量 '{variable_name}' 更新失败", + data={"variable_name": variable_name, "actual_name": actual_variable_name, "scope": target_scope} + ) + + except Exception as e: + if isinstance(e, CallError): + raise # 重新抛出 CallError + else: + logger.error(f"执行变量操作失败: {e}") + raise CallError( + message=f"执行变量操作失败: {e}", + data={"variable_name": variable_name, "operation": operation} + ) from e + + def _parse_variable_name(self, variable_name: str) -> tuple[str, str | None]: + """ + 解析变量名，支持 conversation.test、user.config 等格式 + + :param variable_name: 原始变量名 + :return: (实际变量名, 目标作用域) + """ + if "." not in variable_name: + return variable_name, None + + parts = variable_name.split(".", 1) + scope_part = parts[0].lower() + actual_name = parts[1] + + # 支持的作用域前缀 + valid_scopes = ["conversation", "user", "environment", "system"] + + if scope_part in valid_scopes: + return actual_name, scope_part + else: + # 如果不是有效的作用域前缀，就把整个字符串当作变量名 + return variable_name, None + + async def _execute_string_operation(self, old_value: Any, operation: str, value: Any) -> str: + """执行字符串类型变量操作""" + if operation == StringOperation.OVERWRITE: + return str(value) if value is not None else "" + elif operation == StringOperation.CLEAR: + return "" + else: + raise ValueError(f"字符串类型不支持操作: {operation}") + + async def _execute_number_operation(self, old_value: Any, operation: str, value: Any) -> float: + """执行数值类型变量操作""" + old_num = float(old_value) if old_value is not None else 0.0 + + if operation == NumberOperation.OVERWRITE: + return float(value) if value is not None else 0.0 + elif operation == NumberOperation.CLEAR: + return 0.0 + elif operation == NumberOperation.ADD: + return old_num + float(value) + elif operation == NumberOperation.SUBTRACT: + return old_num - float(value) + elif operation == NumberOperation.MULTIPLY: + return old_num * float(value) + elif operation == NumberOperation.DIVIDE: + if float(value) == 0: + raise ValueError("除数不能为零") + return old_num / float(value) + elif operation == NumberOperation.MODULO: + if float(value) == 0: + raise ValueError("模数不能为零") + return old_num % float(value) + elif operation == NumberOperation.POWER: + return math.pow(old_num, float(value)) + elif operation == NumberOperation.SQRT: + if old_num < 0: + raise ValueError("不能对负数开方") + return math.sqrt(old_num) + else: + raise ValueError(f"数值类型不支持操作: {operation}") + + async def _execute_array_operation(self, old_value: Any, operation: str, value: Any, var_type: VariableType) -> list: + """执行数组类型变量操作""" + old_array = old_value if isinstance(old_value, list) else [] + element_type = var_type.get_array_element_type() + + if operation == ArrayOperation.OVERWRITE: + if not isinstance(value, list): + raise ValueError("覆盖操作需要提供数组值") + return await self._validate_array_elements(value, element_type) + elif operation == ArrayOperation.CLEAR: + return [] + elif operation == ArrayOperation.APPEND: + await self._validate_single_element(value, element_type) + new_array = old_array.copy() + new_array.append(value) + return new_array + elif operation == ArrayOperation.EXTEND: + if not isinstance(value, list): + raise ValueError("扩展操作需要提供数组值") + validated_elements = await self._validate_array_elements(value, element_type) + new_array = old_array.copy() + new_array.extend(validated_elements) + return new_array + elif operation == ArrayOperation.POP_FIRST: + if not old_array: + raise ValueError("数组为空，无法移除首项") + new_array = old_array.copy() + new_array.pop(0) + return new_array + elif operation == ArrayOperation.POP_LAST: + if not old_array: + raise ValueError("数组为空，无法移除尾项") + new_array = old_array.copy() + new_array.pop() + return new_array + else: + raise ValueError(f"数组类型不支持操作: {operation}") + + async def _validate_single_element(self, element: Any, element_type: VariableType) -> None: + """验证单个数组元素的类型""" + if element_type == VariableType.STRING and not isinstance(element, str): + raise ValueError("字符串数组只能添加字符串元素") + elif element_type == VariableType.NUMBER and not isinstance(element, (int, float)): + raise ValueError("数值数组只能添加数值元素") + elif element_type == VariableType.FILE and not isinstance(element, str): + # 文件类型通常以字符串路径表示 + raise ValueError("文件数组只能添加文件路径字符串") + + async def _validate_array_elements(self, elements: list, element_type: VariableType) -> list: + """验证数组元素的类型一致性""" + for element in elements: + await self._validate_single_element(element, element_type) + return elements + + async def _update_variable_value(self, variable_name: str, new_value: Any, call_vars: CallVars, target_scope: str | None = None) -> bool: + """更新变量值""" + try: + pool_manager = await get_pool_manager() + + # 如果指定了目标作用域，优先在该作用域更新 + if target_scope == "conversation": + if call_vars.ids.conversation_id: + conversation_pool = await pool_manager.get_conversation_pool(call_vars.ids.conversation_id) + if conversation_pool: + try: + await conversation_pool.update_variable(variable_name, value=new_value) + logger.debug(f"对话变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"对话变量 {variable_name} 更新失败: {e}") + raise CallError( + message=f"对话变量 {variable_name} 更新失败: {e}", + data={"variable_name": variable_name, "scope": "conversation"} + ) from e + else: + raise CallError( + message="无法更新对话变量：缺少 conversation_id", + data={"variable_name": variable_name, "scope": "conversation"} + ) + elif target_scope == "user": + if call_vars.ids.user_sub: + user_pool = await pool_manager.get_user_pool(call_vars.ids.user_sub) + if user_pool: + try: + await user_pool.update_variable(variable_name, value=new_value) + logger.debug(f"用户变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"用户变量 {variable_name} 更新失败: {e}") + raise CallError( + message=f"用户变量 {variable_name} 更新失败: {e}", + data={"variable_name": variable_name, "scope": "user"} + ) from e + else: + raise CallError( + message="无法更新用户变量：缺少 user_sub", + data={"variable_name": variable_name, "scope": "user"} + ) + elif target_scope in ["environment", "system"]: + if call_vars.ids.flow_id: + flow_pool = await pool_manager.get_flow_pool(call_vars.ids.flow_id) + if flow_pool: + try: + await flow_pool.update_variable(variable_name, value=new_value) + logger.debug(f"环境变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"环境变量 {variable_name} 更新失败: {e}") + raise CallError( + message=f"环境变量 {variable_name} 更新失败: {e}", + data={"variable_name": variable_name, "scope": target_scope} + ) from e + else: + raise CallError( + message="无法更新环境变量：缺少 flow_id", + data={"variable_name": variable_name, "scope": target_scope} + ) + else: + # 没有指定作用域，按原有逻辑依次尝试更新 + # 首先尝试更新对话变量 + if call_vars.ids.conversation_id: + conversation_pool = await pool_manager.get_conversation_pool(call_vars.ids.conversation_id) + if conversation_pool: + try: + await conversation_pool.update_variable(variable_name, value=new_value) + logger.debug(f"对话变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"对话变量 {variable_name} 更新失败: {e}") + + # 如果对话变量更新失败，尝试更新用户变量 + if call_vars.ids.user_sub: + user_pool = await pool_manager.get_user_pool(call_vars.ids.user_sub) + if user_pool: + try: + await user_pool.update_variable(variable_name, value=new_value) + logger.debug(f"用户变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"用户变量 {variable_name} 更新失败: {e}") + + # 如果用户变量也更新失败，尝试更新环境变量 + if call_vars.ids.flow_id: + flow_pool = await pool_manager.get_flow_pool(call_vars.ids.flow_id) + if flow_pool: + try: + await flow_pool.update_variable(variable_name, value=new_value) + logger.debug(f"环境变量 {variable_name} 更新成功") + return True + except Exception as e: + logger.debug(f"环境变量 {variable_name} 更新失败: {e}") + + return False + except Exception as e: + if isinstance(e, CallError): + raise # 重新抛出 CallError + else: + logger.error(f"更新变量值失败: {e}") + raise CallError( + message=f"更新变量值失败: {e}", + data={"variable_name": variable_name, "scope": target_scope} + ) from e diff --git a/apps/scheduler/executor/base.py b/apps/scheduler/executor/base.py index cf2f4e68..7ae03b73 100644 --- a/apps/scheduler/executor/base.py +++ b/apps/scheduler/executor/base.py @@ -43,7 +43,7 @@ class BaseExecutor(BaseModel, ABC): :param event_type: 事件类型 :param data: 消息数据，如果是FLOW_START事件且data为None，则自动构建FlowStartContent - """ + """ if event_type == EventType.FLOW_START.value and isinstance(data, dict): data = FlowStartContent( question=self.question, @@ -59,11 +59,17 @@ class BaseExecutor(BaseModel, ABC): elif isinstance(data, str): data = TextAddContent(text=data).model_dump(exclude_none=True, by_alias=True) - await self.msg_queue.push_output( - self.task, - event_type=event_type, - data=data, # type: ignore[arg-type] - ) + logger.info(f"[BaseExecutor] 调用msg_queue.push_output - event_type: {event_type}") + try: + await self.msg_queue.push_output( + self.task, + event_type=event_type, + data=data, # type: ignore[arg-type] + ) + logger.info(f"[BaseExecutor] msg_queue.push_output调用成功 - event_type: {event_type}") + except Exception as e: + logger.error(f"[BaseExecutor] msg_queue.push_output调用失败 - event_type: {event_type}, error: {e}") + raise @abstractmethod async def run(self) -> None: diff --git a/apps/scheduler/executor/step.py b/apps/scheduler/executor/step.py index 900b495c..cfa41a91 100644 --- a/apps/scheduler/executor/step.py +++ b/apps/scheduler/executor/step.py @@ -120,7 +120,8 @@ class StepExecutor(BaseExecutor): params.update(self.step.step.params) # 对于需要扁平化处理的Call类型，将input_parameters中的内容提取到顶级 - if self._call_id in ["Choice", "DirectReply", "Loop"] and "input_parameters" in params: + # TODO Call中自带属性区分是否需要扁平化，避免逻辑判断频繁修改，或者修改Code逻辑为统一设计 + if self._call_id not in ["Code"] and "input_parameters" in params: # 提取input_parameters中的所有字段到顶级 input_params = params.get("input_parameters", {}) if isinstance(input_params, dict): @@ -129,7 +130,6 @@ class StepExecutor(BaseExecutor): params[key] = value # 移除input_parameters，避免重复 params.pop("input_parameters", None) - logger.info(f"[StepExecutor] 对 {self._call_id} 节点进行参数扁平化处理") try: self.obj = await call_cls.instance(self, self.node, **params) @@ -249,15 +249,52 @@ class StepExecutor(BaseExecutor): # 保存每个output_parameter到变量池，并进行类型验证 saved_count = 0 failed_params = [] + + # 特殊处理：如果是旧格式的JSON Schema结构（主要是Loop节点） + if (isinstance(output_parameters, dict) and + "type" in output_parameters and + "items" in output_parameters and + isinstance(output_parameters["items"], dict)): + + # 提取items中的真正参数配置 + output_parameters = output_parameters["items"] + + # 清理每个参数配置中的多余字段（如嵌套的items） + for param_name, param_config in output_parameters.items(): + if isinstance(param_config, dict) and "items" in param_config: + # 移除多余的items字段，保持参数配置的简洁性 + param_config.pop("items", None) + + logger.debug(f"[StepExecutor] 转换后的output_parameters: {output_parameters}") for param_name, param_config in output_parameters.items(): - try: + try: + # 检查param_config格式，确保它是字典 + if not isinstance(param_config, dict): + logger.warning(f"[StepExecutor] 输出参数 {param_name} 的配置不是字典格式: {param_config} (类型: {type(param_config)})") + # 如果不是字典，尝试转换为标准格式 + if isinstance(param_config, str): + param_config = {"type": param_config, "description": ""} + else: + param_config = {"type": "string", "description": "", "raw_config": str(param_config)} + # 获取参数值 param_value = self._extract_value_from_output_data(param_name, data_dict, param_config) if param_value is not None: # 获取期望的类型 - expected_type = param_config.get("type", "string") + raw_expected_type = param_config.get("type", "string") + + # 映射类型到变量系统支持的类型 + type_mapping = { + "integer": "number", # integer 映射到 number + "int": "number", # int 映射到 number + "float": "number", # float 映射到 number + "str": "string", # str 映射到 string + "bool": "boolean", # bool 映射到 boolean + "dict": "object", # dict 映射到 object + } + expected_type = type_mapping.get(raw_expected_type, raw_expected_type) # 进行类型验证 if not self._validate_output_value_type(param_value, expected_type): @@ -379,7 +416,6 @@ class StepExecutor(BaseExecutor): async def run(self) -> None: """运行单个步骤""" self.validate_flow_state(self.task) - logger.info("[StepExecutor] 运行步骤 %s", self.step.step.name) # 进行自动参数填充 await self._run_slot_filling() @@ -439,6 +475,9 @@ class StepExecutor(BaseExecutor): output_data=output_data, ) self.task.context.append(history.model_dump(exclude_none=True, by_alias=True)) - - # 推送输出 - await self.push_message(EventType.STEP_OUTPUT.value, output_data) + + try: + await self.push_message(EventType.STEP_OUTPUT.value, output_data) + except Exception as e: + logger.error(f"[StepExecutor] {self.step.step.name} - push_message调用失败: {e}") + raise \ No newline at end of file diff --git a/apps/scheduler/pool/loader/app.py b/apps/scheduler/pool/loader/app.py index a85395bf..678bbd37 100644 --- a/apps/scheduler/pool/loader/app.py +++ b/apps/scheduler/pool/loader/app.py @@ -49,9 +49,13 @@ class AppLoader: flow_ids = [app_flow.id for app_flow in metadata.flows] new_flows: list[AppFlow] = [] - async for flow_file in flow_path.rglob("*.yaml"): + + # 只处理主工作流文件（直接在flow目录下的.yaml文件，不包括子目录中的） + async for flow_file in flow_path.glob("*.yaml"): if flow_file.stem not in flow_ids: logger.warning("[AppLoader] 工作流 %s 不在元数据中", flow_file) + continue + flow = await flow_loader.load(app_id, flow_file.stem) if not flow: err = f"[AppLoader] 工作流 {flow_file} 加载失败" diff --git a/apps/scheduler/pool/loader/flow.py b/apps/scheduler/pool/loader/flow.py index 2b4fcdc7..47a08152 100644 --- a/apps/scheduler/pool/loader/flow.py +++ b/apps/scheduler/pool/loader/flow.py @@ -20,6 +20,7 @@ from apps.services.node import NodeManager from apps.common.lance import LanceDB from apps.common.mongo import MongoDB from apps.scheduler.util import yaml_enum_presenter, yaml_str_presenter +from apps.schemas.subflow import AppSubFlow logger = logging.getLogger(__name__) BASE_PATH = Path(Config().get_config().deploy.data_dir) / "semantics" / "app" @@ -350,3 +351,126 @@ class FlowLoader: if retry_count_insert >= max_retries_insert: logger.error(f"[FlowLoader] LanceDB插入flow达到最大重试次数，操作失败: {metadata.id}") raise RuntimeError(f"LanceDB插入flow失败，达到最大重试次数: {metadata.id}") + + async def save_subflow(self, app_id: str, flow_id: str, sub_flow_id: str, flow: Flow) -> None: + """保存子工作流到层次化路径""" + # 子工作流路径: {app_id}/flow/{flow_id}/subflow/{sub_flow_id}.yaml + subflow_path = BASE_PATH / app_id / "flow" / flow_id / "subflow" / f"{sub_flow_id}.yaml" + if not await subflow_path.parent.exists(): + await subflow_path.parent.mkdir(parents=True, exist_ok=True) + + flow_dict = flow.model_dump(by_alias=True, exclude_none=True) + async with aiofiles.open(subflow_path, mode="w", encoding="utf-8") as f: + yaml.add_representer(str, yaml_str_presenter) + yaml.add_representer(EdgeType, yaml_enum_presenter) + await f.write( + yaml.dump( + flow_dict, + allow_unicode=True, + sort_keys=False, + ), + ) + + # 更新数据库中的子工作流元数据 + await self._update_subflow_db( + app_id, + flow_id, + AppSubFlow( + id=sub_flow_id, + name=flow.name, + description=flow.description, + path=str(subflow_path), + debug=flow.debug, + ), + ) + + async def load_subflow(self, app_id: str, flow_id: str, sub_flow_id: str) -> Flow | None: + """加载子工作流""" + subflow_path = BASE_PATH / app_id / "flow" / flow_id / "subflow" / f"{sub_flow_id}.yaml" + + if not await subflow_path.exists(): + logger.warning("[FlowLoader] 子工作流文件不存在: %s", subflow_path) + return None + + try: + async with aiofiles.open(subflow_path, mode="r", encoding="utf-8") as f: + content = await f.read() + flow_dict = yaml.safe_load(content) + return Flow(**flow_dict) + except Exception: + logger.exception("[FlowLoader] 加载子工作流失败: %s", subflow_path) + return None + + async def delete_subflow(self, app_id: str, flow_id: str, sub_flow_id: str) -> bool: + """删除子工作流文件""" + subflow_path = BASE_PATH / app_id / "flow" / flow_id / "subflow" / f"{sub_flow_id}.yaml" + + if await subflow_path.exists(): + try: + await subflow_path.unlink() + logger.info("[FlowLoader] 成功删除子工作流文件：%s", subflow_path) + + # 从数据库中删除子工作流元数据 + await self._delete_subflow_db(app_id, flow_id, sub_flow_id) + return True + + except Exception: + logger.exception("[FlowLoader] 删除子工作流文件失败：%s", subflow_path) + return False + else: + logger.warning("[FlowLoader] 子工作流文件不存在：%s", subflow_path) + return True + + async def _update_subflow_db(self, app_id: str, flow_id: str, metadata: "AppSubFlow") -> None: + """更新数据库中的子工作流元数据""" + try: + app_collection = MongoDB().get_collection("app") + + # 查找应用 + app_record = await app_collection.find_one({"_id": app_id}) + if not app_record: + logger.error("[FlowLoader] 应用不存在: %s", app_id) + return + + # 确保子工作流元数据结构存在 + if "subflows" not in app_record: + app_record["subflows"] = {} + if flow_id not in app_record["subflows"]: + app_record["subflows"][flow_id] = [] + + # 更新或添加子工作流元数据 + subflows = app_record["subflows"][flow_id] + existing_index = None + for i, subflow in enumerate(subflows): + if subflow.get("id") == metadata.id: + existing_index = i + break + + subflow_data = metadata.model_dump(by_alias=True, exclude_none=True) + if existing_index is not None: + subflows[existing_index] = subflow_data + else: + subflows.append(subflow_data) + + # 保存到数据库 + await app_collection.update_one( + {"_id": app_id}, + {"$set": {f"subflows.{flow_id}": subflows}} + ) + + except Exception: + logger.exception("[FlowLoader] 更新子工作流数据库元数据失败") + + async def _delete_subflow_db(self, app_id: str, flow_id: str, sub_flow_id: str) -> None: + """从数据库中删除子工作流元数据""" + try: + app_collection = MongoDB().get_collection("app") + + # 从应用的子工作流列表中移除 + await app_collection.update_one( + {"_id": app_id}, + {"$pull": {f"subflows.{flow_id}": {"id": sub_flow_id}}} + ) + + except Exception: + logger.exception("[FlowLoader] 删除子工作流数据库元数据失败") diff --git a/apps/scheduler/variable/integration.py b/apps/scheduler/variable/integration.py index 3bfb1b8d..4b761f3b 100644 --- a/apps/scheduler/variable/integration.py +++ b/apps/scheduler/variable/integration.py @@ -39,7 +39,8 @@ class VariableIntegration: async def parse_call_input(input_data: Dict[str, Any], user_sub: str, flow_id: Optional[str] = None, - conversation_id: Optional[str] = None) -> Union[str, Dict, List]: + conversation_id: Optional[str] = None, + current_step_id: Optional[str] = None) -> Union[str, Dict, List]: """解析Call输入中的变量引用 Args: @@ -47,6 +48,7 @@ class VariableIntegration: user_sub: 用户ID flow_id: 流程ID conversation_id: 对话ID + current_step_id: 当前步骤ID，用于支持{{self.xxx}}语法 Returns: Dict[str, Any]: 解析后的输入数据 @@ -55,7 +57,8 @@ class VariableIntegration: parser = VariableParser( user_sub=user_sub, flow_id=flow_id, - conversation_id=conversation_id + conversation_id=conversation_id, + current_step_id=current_step_id ) # 递归解析JSON模板中的变量引用 @@ -73,7 +76,8 @@ class VariableIntegration: reference: str, user_sub: str, flow_id: Optional[str] = None, - conversation_id: Optional[str] = None + conversation_id: Optional[str] = None, + current_step_id: Optional[str] = None ) -> Tuple[Any, Any]: """解析单个变量引用 @@ -82,6 +86,7 @@ class VariableIntegration: user_sub: 用户ID flow_id: 流程ID conversation_id: 对话ID + current_step_id: 当前步骤ID，用于支持{{self.xxx}}语法 Returns: Tuple[Any, Any]: 解析后的变量值和变量类型 @@ -90,7 +95,8 @@ class VariableIntegration: parser = VariableParser( user_id=user_sub, flow_id=flow_id, - conversation_id=conversation_id + conversation_id=conversation_id, + current_step_id=current_step_id ) # 清理引用字符串（移除花括号） diff --git a/apps/scheduler/variable/parser.py b/apps/scheduler/variable/parser.py index 6c5f11df..765faa57 100644 --- a/apps/scheduler/variable/parser.py +++ b/apps/scheduler/variable/parser.py @@ -20,7 +20,8 @@ class VariableParser: user_id: Optional[str] = None, flow_id: Optional[str] = None, conversation_id: Optional[str] = None, - user_sub: Optional[str] = None): + user_sub: Optional[str] = None, + current_step_id: Optional[str] = None): """初始化变量解析器 Args: @@ -28,11 +29,13 @@ class VariableParser: flow_id: 流程ID conversation_id: 对话ID user_sub: 用户订阅ID (优先使用，用于未来鉴权等需求) + current_step_id: 当前步骤ID，用于支持{{self.xxx}}语法 """ # 优先使用 user_sub，如果没有则使用 user_id self.user_id = user_sub if user_sub is not None else user_id self.flow_id = flow_id self.conversation_id = conversation_id + self.current_step_id = current_step_id self._pool_manager = None async def _get_pool_manager(self): @@ -96,6 +99,14 @@ class VariableParser: scope_str, var_path = parts + # 处理特殊的self作用域 + if scope_str == "self": + if not self.current_step_id: + raise ValueError("使用{{self.xxx}}语法时，当前步骤ID不可用") + # 将self.xxx转换为conversation.current_step_id.xxx + scope_str = "conversation" + var_path = f"{self.current_step_id}.{var_path}" + # 确定作用域 scope_map = { "sys": VariableScope.SYSTEM, diff --git a/apps/schemas/enum_var.py b/apps/schemas/enum_var.py index ba18faaf..dfe2baae 100644 --- a/apps/schemas/enum_var.py +++ b/apps/schemas/enum_var.py @@ -137,6 +137,7 @@ class CallOutputType(str, Enum): TEXT = "text" DATA = "data" + PROGRESS = "progress" class SpecialCallType(str, Enum): diff --git a/apps/schemas/subflow.py b/apps/schemas/subflow.py new file mode 100644 index 00000000..c0bd785a --- /dev/null +++ b/apps/schemas/subflow.py @@ -0,0 +1,16 @@ +# Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. +"""子工作流相关的Schema定义""" + +from pydantic import BaseModel, Field + + +class AppSubFlow(BaseModel): + """应用的子工作流元数据""" + + id: str = Field(description="子工作流ID") + name: str = Field(description="子工作流名称") + description: str = Field(description="子工作流描述") + path: str = Field(description="子工作流文件路径") + debug: bool = Field(default=False, description="是否已调试") + parent_flow_id: str = Field(default="", description="父工作流ID") + created_at: float = Field(default=0, description="创建时间") \ No newline at end of file diff --git a/apps/services/flow.py b/apps/services/flow.py index 74268ab7..5b7321b8 100644 --- a/apps/services/flow.py +++ b/apps/services/flow.py @@ -2,6 +2,7 @@ """flow Manager""" import logging +from typing import Any from pymongo import ASCENDING @@ -84,9 +85,15 @@ class FlowManager: params_schema, output_schema = await NodeManager.get_node_params(node_pool_record["_id"]) try: # TODO: 由于现在没有动态表单，所以暂时使用Slot的create_empty_slot方法 + # 对于循环节点，输出参数已经是扁平化格式，不需要再次处理 + if node_pool_record["call_id"] == "Loop": + output_parameters = output_schema + else: + output_parameters = Slot(output_schema).extract_type_desc_from_schema() + parameters = { "input_parameters": Slot(params_schema).create_empty_slot(), - "output_parameters": Slot(output_schema).extract_type_desc_from_schema(), + "output_parameters": output_parameters, } except Exception: logger.exception("[FlowManager] generate_from_schema 失败") @@ -267,9 +274,16 @@ class FlowManager: _, output_parameters = await NodeManager.get_node_params(node_config.node) else: output_parameters = {} + + # 对于循环节点，输出参数已经是扁平化格式，不需要再次处理 + if hasattr(node_config, 'type') and node_config.type == "Loop": + processed_output_parameters = output_parameters + else: + processed_output_parameters = Slot(output_parameters).extract_type_desc_from_schema() + parameters = { "input_parameters": input_parameters, - "output_parameters": Slot(output_parameters).extract_type_desc_from_schema(), + "output_parameters": processed_output_parameters, } node_item = NodeItem( @@ -360,6 +374,7 @@ class FlowManager: app_id: str, flow_id: str, flow_item: FlowItem, + user_sub: str = "", ) -> FlowItem | None: """ 存储/更新flow的数据库数据和配置文件 @@ -388,8 +403,23 @@ class FlowManager: connectivity=flow_item.connectivity, debug=flow_item.debug, ) + + # 获取旧的flow配置以便比较节点变化 + flow_loader = FlowLoader() + old_flow_config = await flow_loader.load(app_id, flow_id) + + # 收集新配置中的所有步骤ID + new_step_ids = set() + for node_item in flow_item.nodes: params = node_item.parameters + new_step_ids.add(node_item.step_id) + + # 处理节点变量模板（支持多种节点类型） + if user_sub: + await FlowManager._process_node_variable_templates( + node_item, flow_id, user_sub + ) flow_config.steps[node_item.step_id] = Step( type=node_item.call_id, @@ -399,20 +429,47 @@ class FlowManager: pos=node_item.position, params=params, ) + + # 检查是否有节点被删除，如果有则清理相关变量 + if old_flow_config and user_sub: + old_step_ids = set(old_flow_config.steps.keys()) + deleted_step_ids = old_step_ids - new_step_ids + + if deleted_step_ids: + logger.info(f"[FlowManager] 检测到删除的节点: {deleted_step_ids}") + # 清理删除节点的相关变量 + await FlowManager._cleanup_deleted_node_variables( + deleted_step_ids, flow_id, user_sub + ) + for edge_item in flow_item.edges: - edge_from = edge_item.source_node - if edge_item.branch_id: - edge_from = edge_from + "." + edge_item.branch_id - edge_config = Edge( - id=edge_item.edge_id, - edge_from=edge_from, - edge_to=edge_item.target_node, - edge_type=EdgeType(edge_item.type) if edge_item.type else EdgeType.NORMAL, - ) - flow_config.edges.append(edge_config) - - flow_loader = FlowLoader() - old_flow_config = await flow_loader.load(app_id, flow_id) + try: + edge_from = edge_item.source_node + if edge_item.branch_id: + edge_from = edge_from + "." + edge_item.branch_id + + # 安全处理EdgeType + edge_type = EdgeType.NORMAL + if edge_item.type: + try: + edge_type = EdgeType(edge_item.type) + except ValueError: + logger.warning(f"[FlowManager] 无效的边类型: {edge_item.type}，使用默认值") + edge_type = EdgeType.NORMAL + + edge_config = Edge( + id=edge_item.edge_id, + edge_from=edge_from, + edge_to=edge_item.target_node, + edge_type=edge_type, + ) + flow_config.edges.append(edge_config) + logger.info(f"[FlowManager] 添加边: {edge_item.edge_id}, {edge_from} -> {edge_item.target_node}, type: {edge_type}") + except Exception as e: + logger.error(f"[FlowManager] 创建边失败: {edge_item.edge_id}, 错误: {e}") + continue + + logger.info(f"[FlowManager] 构建完成，flow_config.edges数量: {len(flow_config.edges)}") if old_flow_config is None: error_msg = f"[FlowManager] 流 {flow_id} 不存在；可能为新创建" @@ -477,3 +534,477 @@ class FlowManager: return False else: return True + + @staticmethod + async def _process_node_variable_templates( + node_item, + flow_id: str, + user_sub: str + ) -> None: + """处理节点变量模板，将其保存到Flow变量池供后续节点引用 + + 支持的节点类型和配置： + - Loop节点：从input_parameters.variables提取 + - 其他节点可以扩展支持 + + Args: + node_item: 节点项 + flow_id: 工作流ID + user_sub: 用户ID + """ + # 根据节点类型确定变量提取策略 + variables_config = await FlowManager._extract_node_variables_config(node_item) + if not variables_config: + return + + # 批量保存变量模板 + await FlowManager._save_node_variables_batch( + variables_config, node_item.step_id, flow_id, user_sub + ) + + @staticmethod + async def _extract_node_variables_config(node_item) -> dict[str, any] | None: + """从节点中提取变量配置 + + Args: + node_item: 节点项 + + Returns: + dict[str, any] | None: 变量配置字典，格式为{变量名: 变量配置} + """ + try: + if not node_item.parameters or not isinstance(node_item.parameters, dict): + return None + + # 根据节点类型提取变量 + if node_item.call_id == "Loop": + # Loop节点：从input_parameters.variables提取 + input_parameters = node_item.parameters.get("input_parameters", {}) + if isinstance(input_parameters, dict): + variables = input_parameters.get("variables", {}) + if isinstance(variables, dict) and variables: + return variables + + # 可以在这里添加其他节点类型的变量提取逻辑 + # elif node_item.call_id == "OtherNodeType": + # return FlowManager._extract_other_node_variables(node_item) + + return None + + except Exception as e: + logger.error(f"[FlowManager] 提取节点 {node_item.step_id} 的变量配置失败: {e}") + return None + + @staticmethod + async def _save_node_variables_batch( + variables_config: dict[str, any], + step_id: str, + flow_id: str, + user_sub: str + ) -> None: + """批量保存节点变量模板 + + Args: + variables_config: 变量配置字典 + step_id: 节点步骤ID + flow_id: 工作流ID + user_sub: 用户ID + """ + try: + + saved_count = 0 + failed_count = 0 + + for var_name, var_value in variables_config.items(): + try: + # 构造变量名：step_id.变量名 + full_var_name = f"{step_id}.{var_name}" + + # 解析变量配置 + var_type, actual_value, description = FlowManager._parse_variable_config( + var_value, var_name, step_id + ) + + # 保存变量模板 + success = await FlowManager._save_variable_template( + var_name=full_var_name, + value=actual_value, + var_type=var_type, + description=description, + user_sub=user_sub, + flow_id=flow_id + ) + + if success: + saved_count += 1 + logger.debug(f"[FlowManager] 已保存节点变量: conversation.{full_var_name} = {actual_value}") + else: + failed_count += 1 + logger.warning(f"[FlowManager] 保存节点变量失败: {full_var_name}") + + except Exception as e: + failed_count += 1 + logger.error(f"[FlowManager] 处理节点变量 {var_name} 失败: {e}") + + if saved_count > 0: + logger.info(f"[FlowManager] 节点 {step_id} 成功保存了 {saved_count} 个变量模板") + if failed_count > 0: + logger.warning(f"[FlowManager] 节点 {step_id} 有 {failed_count} 个变量保存失败") + + except Exception as e: + logger.error(f"[FlowManager] 批量保存节点 {step_id} 的变量失败: {e}") + + @staticmethod + def _parse_variable_config(var_value: any, var_name: str, step_id: str) -> tuple[str, any, str]: + """解析变量配置，返回类型、值和描述 + + Args: + var_value: 变量值配置 + var_name: 变量名 + step_id: 节点步骤ID + + Returns: + tuple[str, any, str]: (变量类型, 实际值, 描述) + """ + if isinstance(var_value, dict) and "type" in var_value: + # 复杂格式：{type: "string", value: "默认值", description: "描述"} + var_type = var_value.get("type", "string") + actual_value = var_value.get("value", "") + description = var_value.get("description", f"节点 {step_id} 的变量 {var_name}") + else: + # 简单格式：直接值，根据Python类型推断 + actual_value = var_value + description = f"节点 {step_id} 的变量 {var_name}" + + # 类型推断 + if isinstance(var_value, bool): + var_type = "boolean" + elif isinstance(var_value, (int, float)): + var_type = "number" + elif isinstance(var_value, (list, tuple)): + var_type = "array" + elif isinstance(var_value, dict): + var_type = "object" + else: + var_type = "string" + + return var_type, actual_value, description + + @staticmethod + async def _save_variable_template( + var_name: str, + value: Any, + var_type: str, + description: str, + user_sub: str, + flow_id: str + ) -> bool: + """保存节点变量模板到Flow级别的对话变量模板池 + + Args: + var_name: 变量名（格式：step_id.变量名） + value: 变量值 + var_type: 变量类型 + description: 变量描述 + user_sub: 用户ID + flow_id: 流程ID + + Returns: + bool: 是否保存成功 + """ + try: + # 导入必要的模块 + from apps.scheduler.variable.pool_manager import get_pool_manager + from apps.scheduler.variable.type import VariableType + + # 获取flow级别的变量池 + pool_manager = await get_pool_manager() + flow_pool = await pool_manager.get_flow_pool(flow_id) + + if not flow_pool: + logger.warning(f"[FlowManager] 无法获取Flow变量池: {flow_id}") + return False + + # 转换变量类型 + try: + var_type_enum = VariableType(var_type) + except ValueError: + var_type_enum = VariableType.STRING + logger.warning(f"[FlowManager] 未知的变量类型 {var_type}，使用默认类型 string") + + # 尝试更新对话变量模板，如果不存在则创建 + try: + # 检查是否已存在对话变量模板 + existing_template = await flow_pool.get_conversation_template(var_name) + if existing_template: + # 更新现有模板 - 使用通用的update_variable方法 + await flow_pool.update_variable(var_name, value=value, description=description) + logger.debug(f"[FlowManager] 节点变量模板已更新: {var_name} = {value}") + else: + # 创建新的对话变量模板 + await flow_pool.add_conversation_template( + name=var_name, + var_type=var_type_enum, + default_value=value, # 注意参数名是default_value，不是value + description=description, + created_by=user_sub or "system" + ) + logger.debug(f"[FlowManager] 节点变量模板已创建: {var_name} = {value}") + return True + except Exception as e: + logger.error(f"[FlowManager] 处理节点变量模板失败: {var_name}, 错误: {e}") + return False + + except Exception as e: + logger.error(f"[FlowManager] 保存节点变量模板失败: {var_name}, 错误: {e}") + return False + + @staticmethod + async def put_subflow_by_app_flow_and_subflow_id( + app_id: str, + flow_id: str, + sub_flow_id: str, + flow_item: FlowItem, + ) -> FlowItem | None: + """ + 存储/更新子工作流的数据库数据和配置文件 + + :param app_id: 应用的id + :param flow_id: 父工作流的id + :param sub_flow_id: 子工作流的id + :param flow_item: 子工作流的item + :return: 子工作流的item + """ + try: + app_collection = MongoDB().get_collection("app") + app_record = await app_collection.find_one({"_id": app_id}) + if app_record is None: + logger.error("[FlowManager] 应用 %s 不存在", app_id) + return None + except Exception: + logger.exception("[FlowManager] 获取应用失败") + return None + + try: + # 构建子工作流配置 + flow_config = Flow( + name=flow_item.name, + description=flow_item.description, + steps={}, + edges=[], + focus_point=flow_item.focus_point, + connectivity=flow_item.connectivity, + debug=flow_item.debug, + ) + + for node_item in flow_item.nodes: + params = node_item.parameters + flow_config.steps[node_item.step_id] = Step( + type=node_item.call_id, + node=node_item.node_id, + name=node_item.name, + description=node_item.description, + pos=node_item.position, + params=params, + ) + + for edge_item in flow_item.edges: + edge_from = edge_item.source_node + if edge_item.branch_id: + edge_from = edge_from + "." + edge_item.branch_id + + edge_config = Edge( + id=edge_item.edge_id, + edge_from=edge_from, + edge_to=edge_item.target_node, + edge_type=EdgeType.NORMAL, # 子工作流默认使用普通边 + ) + flow_config.edges.append(edge_config) + + # 使用子工作流专用的保存路径 + flow_loader = FlowLoader() + await flow_loader.save_subflow(app_id, flow_id, sub_flow_id, flow_config) + + return flow_item + + except Exception: + logger.exception("[FlowManager] 保存子工作流失败") + return None + + @staticmethod + async def get_subflow_by_app_flow_and_subflow_id( + app_id: str, + flow_id: str, + sub_flow_id: str, + ) -> FlowItem | None: + """ + 根据应用id、父工作流id和子工作流id获取子工作流 + + :param app_id: 应用的id + :param flow_id: 父工作流的id + :param sub_flow_id: 子工作流的id + :return: 子工作流的FlowItem + """ + try: + flow_loader = FlowLoader() + flow_config = await flow_loader.load_subflow(app_id, flow_id, sub_flow_id) + + if flow_config is None: + return None + + # 转换为FlowItem格式 + focus_point = flow_config.focus_point or PositionItem(x=0, y=0) + flow_item = FlowItem( + flowId=sub_flow_id, + name=flow_config.name, + description=flow_config.description, + enable=True, + editable=True, + nodes=[], + edges=[], + focusPoint=focus_point, + connectivity=flow_config.connectivity, + debug=flow_config.debug, + ) + + for node_id, node_config in flow_config.steps.items(): + # 参数处理逻辑与主工作流保持一致 + if node_config.type == "Code" or node_config.type == "DirectReply" or node_config.type == "Choice": + parameters = node_config.params # 直接使用保存的完整params + else: + # 其他节点：使用原有逻辑 + input_parameters = node_config.params.get("input_parameters") + if node_config.node not in ("Empty"): + try: + _, output_parameters = await NodeManager.get_node_params(node_config.node) + except Exception: + logger.exception("[FlowManager] 获取节点参数失败，使用空参数") + output_parameters = {} + else: + output_parameters = {} + + # 对于循环节点，输出参数已经是扁平化格式，不需要再次处理 + if hasattr(node_config, 'type') and node_config.type == "Loop": + processed_output_parameters = output_parameters + else: + processed_output_parameters = Slot(output_parameters).extract_type_desc_from_schema() + + parameters = { + "input_parameters": input_parameters, + "output_parameters": processed_output_parameters, + } + + node_item = NodeItem( + stepId=node_id, + serviceId="", # 子工作流节点默认serviceId为空 + nodeId=node_config.node, + name=node_config.name, + description=node_config.description, + enable=True, + editable=True, + callId=node_config.type, + parameters=parameters, + position=PositionItem(x=node_config.pos.x, y=node_config.pos.y), + ) + flow_item.nodes.append(node_item) + + for edge_config in flow_config.edges: + edge_from = edge_config.edge_from + branch_id = "" + tmp_list = edge_config.edge_from.split(".") + if len(tmp_list) == 0 or len(tmp_list) > 2: + logger.error("[FlowManager] 子工作流中边的格式错误") + continue + if len(tmp_list) == 2: + edge_from = tmp_list[0] + branch_id = tmp_list[1] + + edge_item = EdgeItem( + edgeId=edge_config.id, + sourceNode=edge_from, + targetNode=edge_config.edge_to, + type=edge_config.edge_type.value if edge_config.edge_type else EdgeType.NORMAL.value, + branchId=branch_id, + ) + flow_item.edges.append(edge_item) + + return flow_item + + except Exception: + logger.exception("[FlowManager] 获取子工作流失败") + return None + + @staticmethod + async def delete_subflow_by_app_flow_and_subflow_id( + app_id: str, + flow_id: str, + sub_flow_id: str, + ) -> bool: + """ + 删除子工作流 + + :param app_id: 应用的id + :param flow_id: 父工作流的id + :param sub_flow_id: 子工作流的id + :return: 是否删除成功 + """ + try: + flow_loader = FlowLoader() + return await flow_loader.delete_subflow(app_id, flow_id, sub_flow_id) + except Exception: + logger.exception("[FlowManager] 删除子工作流失败") + return False + + @staticmethod + async def _cleanup_deleted_node_variables( + deleted_step_ids: set[str], + flow_id: str, + user_sub: str + ) -> None: + """清理被删除节点的相关变量 + + Args: + deleted_step_ids: 被删除的步骤ID集合 + flow_id: 工作流ID + user_sub: 用户ID + """ + try: + from apps.scheduler.variable.pool_manager import get_pool_manager + + # 获取flow级别的变量池 + pool_manager = await get_pool_manager() + flow_pool = await pool_manager.get_flow_pool(flow_id) + + if not flow_pool: + logger.warning(f"[FlowManager] 无法获取Flow变量池: {flow_id}") + return + + # 获取所有对话变量模板 + conversation_variables = await flow_pool.list_conversation_templates() + + if not conversation_variables: + logger.info(f"[FlowManager] 没有找到对话变量需要清理") + return + + cleaned_count = 0 + + for variable in conversation_variables: + try: + # 检查变量名是否以被删除的步骤ID开头 + for deleted_step_id in deleted_step_ids: + if variable.name.startswith(f"{deleted_step_id}."): + # 删除该变量 + await flow_pool.delete_variable(variable.name) + cleaned_count += 1 + logger.info(f"[FlowManager] 已清理删除节点的变量: {variable.name}") + break + except Exception as e: + logger.error(f"[FlowManager] 清理变量 {variable.name} 失败: {e}") + + if cleaned_count > 0: + logger.info(f"[FlowManager] 总共清理了 {cleaned_count} 个被删除节点的变量") + else: + logger.info(f"[FlowManager] 没有找到需要清理的节点变量") + + except Exception as e: + logger.error(f"[FlowManager] 清理删除节点变量失败: {e}") diff --git a/apps/services/flow_validate.py b/apps/services/flow_validate.py index b343fb7b..3abf8c09 100644 --- a/apps/services/flow_validate.py +++ b/apps/services/flow_validate.py @@ -7,6 +7,7 @@ import logging from apps.exceptions import FlowBranchValidationError, FlowEdgeValidationError, FlowNodeValidationError from apps.schemas.enum_var import NodeType from apps.schemas.flow_topology import EdgeItem, FlowItem, NodeItem +from pydantic import ValidationError logger = logging.getLogger(__name__) @@ -265,3 +266,109 @@ class FlowService: # 检查是否能到达终止节点 return end_id in vis + + @classmethod + async def validate_subflow_illegal(cls, flow: FlowItem) -> None: + """ + 验证子工作流是否违法（子工作流专用验证，不强制要求end节点） + + :param flow: 子工作流 + :raises ValidationError: 验证失败 + """ + await cls._validate_flow_nodes(flow, is_subflow=True) + + @classmethod + async def validate_subflow_connectivity(cls, flow: FlowItem) -> bool: + """ + 验证子工作流连通性（子工作流专用，不要求连接到end节点） + + :param flow: 子工作流 + :return: 是否连通 + """ + if not flow.nodes: + return True + + # 构建图结构 + graph = {} + start_nodes = [] + + for node in flow.nodes: + graph[node.step_id] = [] + if node.call_id == 'start' or not any( + edge.target_node == node.step_id for edge in flow.edges + ): + start_nodes.append(node.step_id) + + for edge in flow.edges: + if edge.source_node in graph: + graph[edge.source_node].append(edge.target_node) + + # 检查从开始节点是否能到达所有其他节点 + if not start_nodes: + return len(flow.nodes) <= 1 # 如果没有开始节点且只有一个或零个节点，认为连通 + + visited = set() + + def dfs(node_id): + if node_id in visited: + return + visited.add(node_id) + for neighbor in graph.get(node_id, []): + dfs(neighbor) + + # 从所有开始节点开始遍历 + for start_node in start_nodes: + dfs(start_node) + + # 检查是否所有节点都被访问到 + all_node_ids = {node.step_id for node in flow.nodes} + return len(visited) == len(all_node_ids) + + @classmethod + async def _validate_flow_nodes(cls, flow: FlowItem, is_subflow: bool = False) -> None: + """ + 验证工作流节点（支持子工作流模式） + + :param flow: 工作流 + :param is_subflow: 是否为子工作流 + :raises ValidationError: 验证失败 + """ + if not flow.nodes: + raise ValidationError("工作流不能为空") + + start_count = 0 + end_count = 0 + + for node in flow.nodes: + if node.call_id == "start": + start_count += 1 + elif node.call_id == "end": + end_count += 1 + + # 主工作流必须有start和end节点 + if not is_subflow: + if start_count != 1: + raise ValidationError("工作流必须有且仅有一个start节点") + if end_count != 1: + raise ValidationError("工作流必须有且仅有一个end节点") + else: + # 子工作流可以没有end节点，但如果有start节点，应该只有一个 + if start_count > 1: + raise ValidationError("子工作流最多只能有一个start节点") + if end_count > 1: + raise ValidationError("子工作流最多只能有一个end节点") + + # 验证节点ID唯一性 + node_ids = [node.step_id for node in flow.nodes] + if len(node_ids) != len(set(node_ids)): + raise ValidationError("节点ID必须唯一") + + # 验证边引用的节点存在 + for edge in flow.edges: + source_exists = any(node.step_id == edge.source_node for node in flow.nodes) + target_exists = any(node.step_id == edge.target_node for node in flow.nodes) + + if not source_exists: + raise ValidationError(f"边引用的源节点不存在: {edge.source_node}") + if not target_exists: + raise ValidationError(f"边引用的目标节点不存在: {edge.target_node}") diff --git a/apps/services/node.py b/apps/services/node.py index bf48e71f..6166a9c8 100644 --- a/apps/services/node.py +++ b/apps/services/node.py @@ -97,10 +97,31 @@ class NodeManager: err = f"[NodeManager] Call {call_id} 不存在" logger.error(err) raise ValueError(err) + # 生成输出参数Schema + output_schema = call_class.output_model.model_json_schema( # type: ignore[attr-defined] + override=node_data.override_output if node_data.override_output else {}, + ) + + # 特殊处理：对于循环节点，直接返回扁平化的输出参数结构 + if call_id == "Loop": + # 直接使用正确的扁平化格式，避免依赖JSON Schema转换 + output_schema = { + "iteration_count": { + "type": "integer", + "description": "实际执行的循环次数" + }, + "stop_reason": { + "type": "string", + "description": "停止原因" + }, + "variables": { + "type": "object", + "description": "循环后的变量状态" + } + } + # 返回参数Schema return ( NodeManager.merge_params_schema(call_class.model_json_schema(), node_data.known_params or {}), - call_class.output_model.model_json_schema( # type: ignore[attr-defined] - override=node_data.override_output if node_data.override_output else {}, - ), + output_schema, ) -- Gitee