UFO Galaxy TaskStarLine 深度解析用智能依赖边编织任务星座 DAG【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO本篇技术指南围绕 UFO 项目 Galaxy 框架中的核心组件TaskStarLine展开深入讲解其在任务星座Task Constellation中如何以有向依赖边edge连接两个 TaskStar支撑条件逻辑、仅成功执行与自定义条件求值等依赖关系。读完本文你将掌握 TaskStarLine 的四种依赖类型、生命周期状态机、条件求值引擎、序列化方案及其与 TaskConstellation 的集成方式并能在构建多智能体工作流时正确设计无环依赖图。TaskStarLine 是什么任务星座 DAG 的“边”在 Galaxy 的任务编排体系Constellation V2中TaskStar是原子执行单元顶点而TaskStarLine则负责连接这些顶点形成一张有向无环图DAG。每条 TaskStarLine 定义了一个从任务 $t_i$ 到任务 $t_j$ 的有向依赖关系并携带依赖类型、自然语言条件描述与可编程的条件求值器。其形式化定义为$$ e_{i \rightarrow j} (\text{from_task}_i, \text{to_task}_j, \text{type}, \text{description}) $$任务 $t_j$ 必须等到 $t_i$ 上的条件依据依赖类型而定被满足后才能开始执行。例如在“checkout → build → test → deploy”流水线中deploy 依赖于 test而 test 又依赖于 build每一条依赖都是一条 TaskStarLine。从源码看TaskStarLine 直接实现了 IDependency 接口——该接口规定了source_task_id、target_task_id、dependency_type与is_satisfied(completed_tasks)四个抽象成员。这意味着任何遵守IDependency契约的组件都能与 TaskStarLine 互换使用从而保证了框架的接口隔离原则与可扩展性。核心属性与状态追踪核心属性属性类型说明line_idstr唯一标识未提供时自动生成 UUID源码见 task_star_line.pyfrom_task_idstr前置任务源IDto_task_idstr依赖任务目标IDdependency_typeDependencyType依赖关系类型condition_descriptionstr条件的自然语言描述调试与可读性友好condition_evaluatorCallable求值条件是否满足的函数接收前置任务结果、返回boolmetadataDict[str, Any]依赖的附加元数据兼容性别名source_task_id与target_task_id分别是from_task_id与to_task_id的只读别名用于满足IDependency接口兼容task_star_line.py。状态追踪属性属性类型说明is_satisfiedbool依赖条件当前是否满足last_evaluation_resultbool最近一次条件求值的结果last_evaluation_timedatetime最近一次条件求值的时间戳created_atdatetime依赖创建时间戳updated_atdatetime最近修改时间戳所有状态追踪属性均为只读由 TaskStarLine 内部方法自动维护创建时created_at与updated_at初始化为同一 UTC 时间戳每次求值、修改或手动置位时自动刷新task_star_line.py。时间戳统一使用datetime.now(timezone.utc)保证跨设备协调时的时区一致性。四种依赖类型依赖类型由 DependencyType 枚举 定义其序列化值为小写字符串unconditional、conditional、success_only、completion_only。1. 无条件依赖UNCONDITIONAL任务 $t_j$总是等待 $t_i$ 完成无论成功还是失败。适用场景顺序流水线阶段任何任务完成后的资源清理日志或通知任务。# Task B always runs after Task A completes dep TaskStarLine.create_unconditional( from_task_idtask_a, to_task_idtask_b, descriptionB runs after A regardless of outcome )在求值引擎中UNCONDITIONAL直接短路返回Truetask_star_line.py即只要执行到求值环节就视为满足——它的语义是“执行权交给前置任务完成这个事实本身”。2. 仅成功依赖SUCCESS_ONLY任务 $t_j$仅当$t_i$ 成功完成时才继续执行。适用场景构建流水线仅当构建成功才部署多步数据处理条件工作流分支。# Task B only runs if Task A succeeds dep TaskStarLine.create_success_only( from_task_idbuild_task, to_task_iddeploy_task, descriptionDeploy only if build succeeds )注意成功的判定标准是前置任务返回结果不为Nonetask_star_line.py。这一约定与 TaskStar 的执行结果语义保持一致——返回None即视为失败。源码实现为result prerequisite_result is not None是四种类型中最直观的判定。3. 仅完成依赖COMPLETION_ONLY任务 $t_j$ 在 $t_i$完成时即继续无论成功或失败。适用场景清理任务通知任务审计日志。# Task B runs after Task A finishes, regardless of outcome dep TaskStarLine( from_task_idmain_task, to_task_idcleanup_task, dependency_typeDependencyType.COMPLETION_ONLY, condition_descriptionCleanup runs regardless of main task outcome )从语义上COMPLETION_ONLY与UNCONDITIONAL在求值结果上等价都返回True区别在于设计意图与文档化表达前者强调“无论结果如何都要执行”通常用于收尾类任务后者强调“排队等待前置完成”。选择类型时建议按业务语义区分便于后续阅读与调试。4. 条件依赖CONDITIONAL任务 $t_j$ 依据用户自定义条件对 $t_i$ 结果的求值结果决定是否继续。适用场景错误处理分支基于结果的路由性能驱动的优化路径。# Define custom condition evaluator def check_coverage_threshold(result): Run next task only if test coverage 80% if result and isinstance(result, dict): coverage result.get(coverage_percent, 0) return coverage 80 return False # Create conditional dependency dep TaskStarLine.create_conditional( from_task_idtest_task, to_task_idquality_gate_task, condition_descriptionProceed if test coverage 80%, condition_evaluatorcheck_coverage_threshold )注意若为CONDITIONAL类型但未提供condition_evaluator则默认回退为SUCCESS_ONLY行为即检查结果是否为None。该回退逻辑在源码中清晰可见task_star_line.py。依赖生命周期依赖对象从创建到被消费遵循如下状态机从实现角度可以这样理解该状态机Created对应构造完成且_is_satisfiedFalse当前置任务运行完毕星座编排器调用evaluate_condition(prerequisite_result)进入Evaluating求值结果写入_last_evaluation_result并同步到_is_satisfied随后进入Satisfied或Unsatisfied。mark_satisfied()可视为人工强制从任意状态跳到Satisfied。实战用法创建依赖以下代码演示了四种依赖的完整创建方式覆盖从工厂方法到手动构造的全部路径from galaxy.constellation import TaskStarLine from galaxy.constellation.enums import DependencyType # 1. Unconditional dependency dep1 TaskStarLine.create_unconditional( from_task_idcheckout_code, to_task_idbuild_project, descriptionBuild after checkout ) # 2. Success-only dependency dep2 TaskStarLine.create_success_only( from_task_idbuild_project, to_task_iddeploy_staging, descriptionDeploy only if build succeeds ) # 3. Conditional dependency with custom logic def check_test_results(result): return result.get(tests_passed, 0) result.get(total_tests, 0) dep3 TaskStarLine.create_conditional( from_task_idrun_tests, to_task_iddeploy_production, condition_descriptionDeploy to production only if all tests pass, condition_evaluatorcheck_test_results ) # 4. Manual construction dep4 TaskStarLine( from_task_idtask_a, to_task_idtask_b, dependency_typeDependencyType.COMPLETION_ONLY, condition_descriptionTask B runs after Task A completes, metadata{priority: high, category: cleanup} )三个工厂方法create_unconditional、create_success_only、create_conditional均为类方法classmethod其签名与默认描述在 task_star_line.py 中有精确定义。手动构造则提供了最大自由度可直接传入line_id与metadata。核心操作条件求值# Evaluate condition with prerequisite result prerequisite_result { status: success, coverage_percent: 85, tests_passed: 120, total_tests: 120 } is_satisfied dep.evaluate_condition(prerequisite_result) if is_satisfied: print(✅ Dependency satisfied, dependent task can run) print(fEvaluated at: {dep.last_evaluation_time}) else: print(❌ Dependency not satisfied, dependent task blocked) # Check evaluation history print(fLast result: {dep.last_evaluation_result})evaluate_condition是依赖引擎的心脏其实现task_star_line.py依次执行以下步骤记录求值时间戳_last_evaluation_time依据dependency_type分支求值UNCONDITIONAL/COMPLETION_ONLY直接为TrueSUCCESS_ONLY判定结果非NoneCONDITIONAL调用用户求值器无求值器则回退成功判定将结果写入_last_evaluation_result与_is_satisfied异常兜底整个求值体包裹在try/except中求值器抛出任何异常都会吞掉并返回False同时记录错误但不上抛。这一设计保证了星座执行引擎不会被用户求值器中的意外异常击穿。手动满足控制# Manually mark dependency as satisfied (override) dep.mark_satisfied() # Reset satisfaction status dep.reset_satisfaction() # Check satisfaction if dep.is_satisfied(): print(Dependency is satisfied)mark_satisfied()会将_is_satisfied、_last_evaluation_result置为True并刷新时间戳reset_satisfaction()则清空全部求值状态。二者常用于调试、强制放行或回滚场景。状态查询is_satisfied()具有双模式语义理解这一点至关重要# Method 1: Check using completed tasks list (for IDependency interface) # Returns True if from_task_id is in the completed_tasks list completed_tasks [task_a, task_b, task_c] if dep.is_satisfied(completed_tasks): print(Prerequisite task is completed) # Method 2: Check internal satisfaction state (without parameter) # Returns the internal _is_satisfied flag set by evaluate_condition if dep.is_satisfied(): print(Dependency condition is satisfied) # Get last evaluation details print(fLast evaluated: {dep.last_evaluation_time}) print(fResult: {dep.last_evaluation_result}) # Access metadata print(fMetadata: {dep.metadata})带参数调用传入completed_tasks列表走IDependency接口兼容路径直接检查from_task_id是否在已完成列表中task_star_line.py适合星座级“就绪任务发现”场景无参调用返回内部_is_satisfied标志反映最近一次条件求值结果。修改依赖# Change dependency type dep.dependency_type DependencyType.SUCCESS_ONLY # Update condition description dep.condition_description Updated: Deploy only after successful validation # Set new condition evaluator def new_evaluator(result): return result.get(validation_score, 0) 0.95 dep.set_condition_evaluator(new_evaluator) # Update metadata dep.update_metadata({ updated_by: admin, reason: Stricter validation threshold })警告执行期间修改需谨慎修改dependency_type或condition_evaluator会重置满足状态——源码中两个 setter 均会清空_is_satisfied与_last_evaluation_resulttask_star_line.py、L169-L180。若在星座执行过程中修改依赖可能导致已放行的任务重新被阻塞务必谨慎。metadata属性返回的是内部字典的副本self._metadata.copy()防止外部直接篡改内部状态如需修改必须走update_metadata()并触发updated_at刷新。序列化JSON、字典与 Pydantic SchemaTaskStarLine 支持三种互为补充的序列化形态便于持久化、传输与配置化加载。JSON 导出 / 导入# Export to JSON json_string dep.to_json() print(json_string) # Save to file dep.to_json(save_pathdependency_backup.json) # Load from JSON string restored_dep TaskStarLine.from_json(json_datajson_string) # Load from file loaded_dep TaskStarLine.from_json(file_pathdependency_backup.json)to_json底层会先调用_ensure_json_serializabletask_star_line.py清洗不可序列化字段复杂对象转为vars()/字符串、集合转列表、可调用对象如条件求值器标记为callable: 名称占位符。因此导出的 JSON不含可执行的求值函数体——从 JSON 还原后CONDITIONAL依赖会因缺少求值器而回退为成功判定语义跨进程迁移时需注意这一点。from_json要求json_data与file_path二选一且必须提供其一二者同时或都不提供会抛出ValueErrortask_star_line.py。字典转换# Convert to dictionary dep_dict dep.to_dict() # Create from dictionary new_dep TaskStarLine.from_dict(dep_dict) # Dictionary structure print(dep_dict) # { # line_id: uuid-string, # from_task_id: task_a, # to_task_id: task_b, # dependency_type: success_only, # condition_description: ..., # metadata: {...}, # is_satisfied: false, # last_evaluation_result: null, # created_at: 2025-11-06T..., # updated_at: 2025-11-06T... # }from_dict内部通过_parse_dependency_typetask_star_line.py兼容字符串与枚举两种输入字符串会做大小写不敏感映射未知值回退为UNCONDITIONAL。同时它还会恢复is_satisfied、last_evaluation_result及三个时间戳字段实现状态级还原。Pydantic Schema 转换# Convert to Pydantic BaseModel schema dep.to_basemodel() # Create from Pydantic schema dep_from_schema TaskStarLine.from_basemodel(schema)对应的 TaskStarLineSchema 定义了字段默认值dependency_type默认UNCONDITIONAL、condition_description默认空串并通过字段校验器将枚举值统一转为大写字符串如DependencyType.SUCCESS_ONLY→SUCCESS_ONLY通过模型校验器在缺省时自动生成line_id。from_basemodel会校验实例类型非TaskStarLineSchema实例直接抛ValueError。与 TaskConstellation 集成添加依赖from galaxy.constellation import TaskConstellation constellation TaskConstellation(namemy_workflow) # Add tasks first constellation.add_task(task_a) constellation.add_task(task_b) # Add dependency try: constellation.add_dependency(dep) print(✅ Dependency added successfully) except ValueError as e: print(f❌ Failed to add dependency: {e})依赖校验# TaskConstellation validates dependencies automatically try: # This would fail if it creates a cycle constellation.add_dependency(cyclic_dep) except ValueError as e: print(fValidation error: {e}) # Output: Adding dependency would create a cycle # Check DAG validity is_valid, errors constellation.validate_dag() if not is_valid: for error in errors: print(f❌ {error})TaskConstellation.add_dependencytask_constellation.py执行三层校验后才真正挂载边任务存在性from_task_id与to_task_id必须都已注册否则抛出ValueError如Source task xxx not found无环校验调用_would_create_cycletask_constellation.py——该函数以 DFS 检查to_task_id到from_task_id之间是否已存在可达路径若存在则说明新增边会形成环抛出Adding dependency X - Y would create a cycle引用同步挂载边后同步更新两端 TaskStar 的依赖/被依赖引用from_task.add_dependent(...)/to_task.add_dependency(...)并刷新星座状态。validate_dagtask_constellation.py则在整图层面检查通过拓扑排序get_topological_order探测环存在环时返回(DAG contains cycles, ...)错误列表。高级模式条件错误处理将SUCCESS_ONLY与CONDITIONAL组合即可实现成功走 A 分支、失败走 B 分支的经典 try/except 式工作流# Main task main_task TaskStar( task_idmain_process, descriptionProcess data ) # Success path success_task TaskStar( task_idsuccess_notification, descriptionSend success notification ) # Error path error_task TaskStar( task_iderror_recovery, descriptionAttempt recovery ) # Success-only dependency success_dep TaskStarLine.create_success_only( from_task_idmain_process, to_task_idsuccess_notification ) # Failure-only dependency (using conditional) def on_failure(result): return result is None # Task failed if result is None failure_dep TaskStarLine.create_conditional( from_task_idmain_process, to_task_iderror_recovery, condition_descriptionRun recovery if main task fails, condition_evaluatoron_failure )这里巧妙地利用了“成功 结果非None”的约定on_failure恰好取反让失败也能成为可路由的信号。基于性能的路由根据结果数据量在 GPU 与 CPU 两条处理路径间分流# Route to different processing paths based on data size def route_large_dataset(result): data_size result.get(row_count, 0) return data_size 1_000_000 # Route to GPU if 1M rows # Route to GPU for large datasets gpu_dep TaskStarLine.create_conditional( from_task_idanalyze_dataset, to_task_idprocess_on_gpu, condition_descriptionUse GPU for datasets 1M rows, condition_evaluatorroute_large_dataset ) # Route to CPU for small datasets def route_small_dataset(result): data_size result.get(row_count, 0) return data_size 1_000_000 cpu_dep TaskStarLine.create_conditional( from_task_idanalyze_dataset, to_task_idprocess_on_cpu, condition_descriptionUse CPU for datasets 1M rows, condition_evaluatorroute_small_dataset )注意两个求值器互斥且互补正好覆盖全量结果空间形成稳定的分流决策。错误处理校验失败# TaskStarLine validates on creation try: invalid_dep TaskStarLine( from_task_idtask_a, to_task_idtask_a, # Self-loop! dependency_typeDependencyType.UNCONDITIONAL ) constellation.add_dependency(invalid_dep) except ValueError as e: print(fValidation error: {e}) # TaskConstellation will detect cycle自环self-loop是_would_create_cycle的第一个捕获对象DFS 检查to_task_id from_task_id时立即返回Truetask_constellation.py因此在add_dependency阶段即被拦截。求值器异常def risky_evaluator(result): # This might raise an exception return result[complex_calculation] / result[divisor] dep TaskStarLine.create_conditional( from_task_idtask_a, to_task_idtask_b, condition_descriptionConditional with potential error, condition_evaluatorrisky_evaluator ) # evaluate_condition catches exceptions and returns False result {complex_calculation: 100} # Missing divisor is_satisfied dep.evaluate_condition(result) print(is_satisfied) # False (evaluator raised KeyError, caught internally) print(dep.last_evaluation_result) # False异常被evaluate_condition内部的except Exception捕获返回False并保持_last_evaluation_resultFalse。这确保了单条依赖的求值失败只会阻塞依赖方而不会让整个星座崩溃。示例工作流构建流水线线性依赖# checkout → build → test → deploy checkout TaskStar(task_idcheckout, descriptionCheckout code) build TaskStar(task_idbuild, descriptionBuild project) test TaskStar(task_idtest, descriptionRun tests) deploy TaskStar(task_iddeploy, descriptionDeploy to production) # Sequential success-only dependencies dep1 TaskStarLine.create_success_only(checkout, build) dep2 TaskStarLine.create_success_only(build, test) dep3 TaskStarLine.create_success_only(test, deploy)任意一环失败返回None下游即被阻断天然形成“质量门禁”。扇出模式Fan-Out# analyze → [process_gpu, process_cpu, process_edge] analyze TaskStar(task_idanalyze, descriptionAnalyze data) process_gpu TaskStar(task_idgpu, descriptionProcess on GPU) process_cpu TaskStar(task_idcpu, descriptionProcess on CPU) process_edge TaskStar(task_idedge, descriptionProcess on edge device) # All three can start after analyze completes dep1 TaskStarLine.create_unconditional(analyze, gpu) dep2 TaskStarLine.create_unconditional(analyze, cpu) dep3 TaskStarLine.create_unconditional(analyze, edge)一个前置任务同时驱动多个并行下游是 DAG 并行度的主要来源。扇入模式Fan-In# [task_a, task_b, task_c] → aggregate task_a TaskStar(task_idtask_a, descriptionProcess batch A) task_b TaskStar(task_idtask_b, descriptionProcess batch B) task_c TaskStar(task_idtask_c, descriptionProcess batch C) aggregate TaskStar(task_idaggregate, descriptionAggregate results) # Aggregate waits for all three to complete dep1 TaskStarLine.create_success_only(task_a, aggregate) dep2 TaskStarLine.create_success_only(task_b, aggregate) dep3 TaskStarLine.create_success_only(task_c, aggregate)聚合任务会等待所有入边满足后才被调度是 Map-Reduce 式工作流的天然载体。上述模式的正确性依赖TaskConstellation的get_ready_tasks调度逻辑见 IDependencyResolver 与 task_constellation.py该逻辑按“入边是否全部满足”决定任务是否可执行。最佳实践依赖设计准则选对类型根据工作流逻辑选择最贴切的依赖类型——顺序排队用UNCONDITIONAL质量门禁用SUCCESS_ONLY收尾兜底用COMPLETION_ONLY结果分流用CONDITIONAL保持求值器简单条件求值器应快速且确定性deterministic避免引入随机性或时间依赖处理好求值器异常虽然evaluate_condition内部会捕获异常并返回False但异常会污染last_evaluation_result语义失败与条件不满足难以区分务必让求值器自身做防御性处理写清条件描述使用明确的condition_description它是调试星座执行轨迹时的第一手线索避免环TaskConstellation会做无环校验但设计阶段就应规划好层级减少无效尝试。好的求值器 vs 坏的求值器✅好简单、快速、防御式def check_success(result): return result is not None and result.get(status) success❌坏复杂、缓慢、易错def check_success(result): # Slow database query db_status query_database(result[task_id]) # Complex logic with potential errors return eval(result[complex_expression]) and db_status常见陷阱循环依赖执行前务必校验 DAG 无环validate_dag缺失任务确保from_task_id与to_task_id都已加入星座否则add_dependency会抛ValueError有状态求值器避免依赖外部可变状态的求值器——同一结果两次求值可能得出不同结论慢求值器求值在调度路径上同步执行避免在求值器中做 I/O 或重计算。相关组件TaskStar—— 原子执行单元TaskStarLine 连接的对象TaskConstellation—— DAG 管理器负责校验并执行依赖ConstellationEditor—— 支持撤销/重做的安全依赖编辑Overview—— 任务星座框架总览。API 参考构造函数TaskStarLine( from_task_id: str, to_task_id: str, dependency_type: DependencyType DependencyType.UNCONDITIONAL, condition_description: Optional[str] None, condition_evaluator: Optional[Callable[[Any], bool]] None, line_id: Optional[str] None, metadata: Optional[Dict[str, Any]] None )工厂方法方法说明create_unconditional(from_id, to_id, desc)创建无条件依赖类方法create_success_only(from_id, to_id, desc)创建仅成功依赖类方法create_conditional(from_id, to_id, desc, evaluator)创建条件依赖类方法关键方法方法说明evaluate_condition(result)求值条件是否满足返回boolmark_satisfied()手动标记为已满足reset_satisfaction()重置满足状态is_satisfied(completed_tasksNone)检查依赖是否满足带参数走IDependency接口检查源任务是否完成不带参数返回内部状态set_condition_evaluator(evaluator)设置新的条件求值器update_metadata(metadata)合并更新元数据to_dict()转为字典to_json(save_path)导出 JSON可落盘from_dict(data)从字典创建类方法from_json(json_data, file_path)从 JSON 字符串/文件创建类方法to_basemodel()转为 PydanticTaskStarLineSchemafrom_basemodel(schema)从 Pydantic schema 创建类方法代码定位核心实现galaxy/constellation/task_star_line.py依赖类型枚举galaxy/constellation/enums.py接口契约IDependencygalaxy/core/interfaces.pyPydantic Schemagalaxy/agents/schema.py星座级校验与调度galaxy/constellation/task_constellation.py相关测试tests/visualization/test_dependency_property_changes.py、tests/unit/schema/test_basemodel_integration.py、tests/examples/auto_id_example.py等覆盖了依赖属性变更、Schema 往返与自动 ID 分配等场景。TaskStarLine —— 用智能依赖逻辑连接任务让星座编排既有顺序的严谨又有条件的灵活。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考