尧图网站建设 
UFO³ Galaxy 中的 Constellation Agent 状态机四态 FSM 驱动的动态任务编排生命周期详解【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOUFO³ Galaxy 的 Constellation Agent 是贯穿分布式设备编排的织网中枢它依靠一个4 态有限状态机FSM来管理自身生命周期从初始化建图START、稳态监控与动态改图CONTINUE到成功收尾FINISH或异常终止FAIL。本文以 documents/docs/galaxy/constellation_agent/state.md 为核心骨架结合 galaxy/agents/constellation_agent_states.py、galaxy/agents/constellation_agent.py 等源码实现深入讲解每个状态的处理逻辑、转移矩阵、事件批处理与状态合并机制并给出状态查询、错误恢复与最佳实践帮助你掌握这一确定性生命周期控制的完整实现。为什么需要状态机把 LLM 推理与确定性控制解耦在 Constellation Agent 总览 中Constellation Agent 被描述为集中式的星座织造者Centralized Constellation Weaver它既要理解用户自然语言意图、生成可执行的 Task Constellation又要在任务执行过程中根据反馈动态增删改任务与依赖。如果这些行为完全交给大模型自由发挥就会出现状态不可预期、难以调试的问题。4 态 FSM 的核心价值在于把 LLM 的语义推理与确定性的控制逻辑分离LLM 负责想在创建/编辑两种模式下产出星座结构与修改动作见 WeavingMode 中的CREATION/EDITING。FSM 负责控确保 Agent 在任何时刻只处于一个明确状态所有转移都有清晰的触发条件提升安全性与可调试性。从源码看状态机的实现位于 galaxy/agents/constellation_agent_states.py而驱动它的运行循环位于 galaxy/session/galaxy_session.py其核心逻辑为# Initialize agent in START state self._agent.set_state(StartConstellationAgentState()) # Run agent state machine until completion while not self.is_finished(): # Execute current state await self._agent.handle(self._context) # Transition to next state self.state self._agent.state.next_state(self._agent) # Update agent state self._agent.set_state(self.state) await asyncio.sleep(0.01) # prevent busy waiting也就是说执行当前状态 → 依据 agent.status 决定下一个状态 → 切换状态构成了一个确定性的闭环循环持续直到 Agent 进入终态。图Constellation Agent 生命周期状态转换示意START → CONTINUE → FINISH/FAIL两个终态无出边。对应的完整状态转移图与 Mermaid 状态图定义见原文档与下文状态转换矩阵章节。状态空间与状态枚举四个状态的语义划分State类型描述进入条件START初始态初始化并创建星座Agent 实例化、完成后重启CONTINUE稳态监控事件并处理反馈星座创建成功FINISH终态成功终止所有任务完成、无需再编辑FAIL终态错误终止不可恢复错误、校验失败状态枚举定义状态枚举ConstellationAgentStatus定义在 galaxy/agents/constellation_agent_states.pyclass ConstellationAgentStatus(Enum): Galaxy Agent states START START CONTINUE CONTINUE FINISH FINISH FAIL FAIL注意这里的枚举值是字符串如START、CONTINUE它们与agent.status字段直接对应。在 ConstellationAgent 构造函数 中_status初始为START并调用self.set_state(StartConstellationAgentState())让 Agent 一开始就处于 START 状态。状态转移图MermaidSTART 状态初始化与建图阶段职责START 是初始化与创建阶段Agent 在该状态中完成四件事根据用户请求生成初始 Task Constellation校验 DAG 结构的正确性环路检测启动后台编排任务转移到监控模式。状态处理实现StartConstellationAgentState的实现galaxy/agents/constellation_agent_states.py核心逻辑如下ConstellationAgentStateManager.register class StartConstellationAgentState(ConstellationAgentState): async def handle(self, agent, context) - None: # 已处于终态则直接返回No-op if agent.status in [ ConstellationAgentStatus.FINISH.value, ConstellationAgentStatus.FAIL.value, ]: return timing_info {} # 若还没有星座则进入创建模式 if not agent.current_constellation: context.set(ContextNames.WEAVING_MODE, WeavingMode.CREATION) agent._current_constellation, timing_info ( await agent.process_creation(context) ) # 启动后台编排非阻塞 if agent.current_constellation: asyncio.create_task( agent.orchestrator.orchestrate_constellation( agent.current_constellation, metadatatiming_info ) ) agent.status ConstellationAgentStatus.CONTINUE.value elif agent.status ConstellationAgentStatus.CONTINUE.value: agent.status ConstellationAgentStatus.FAIL.value关键点分析幂等保护如果 Agent 已经处于 FINISH/FAIL 终态handle直接返回避免重复创建。创建模式标记通过context.set(ContextNames.WEAVING_MODE, WeavingMode.CREATION)把上下文切到创建模式随后process_creation会经过 ConstellationAgent.process_creation初始化 prompter → 加载 MCP 上下文 → 交给ConstellationAgentProcessor处理 → 通过_validate_and_update_constellation执行constellation.validate_dag()校验。非阻塞编排asyncio.create_task()把orchestrate_constellation作为后台任务启动Agent 随即把状态置为CONTINUE立即进入监控。注意timing_info初始化在前源码注释明确Initialize timing_info to avoid UnboundLocalError否则星座已存在时该变量未定义会抛错。行为与错误处理场景动作下一状态首次执行通过 LLM 生成星座CONTINUE成功/FAIL出错重启触发复用已有星座CONTINUE创建失败记录错误无星座产生FAIL校验失败DAG 含环或结构非法FAIL已处于终态空操作立即返回保持原状态错误处理采用多层try/exceptAttributeError如上下文字段缺失、KeyError如字典缺键、兜底Exception三者均把状态置为FAIL并输出完整 traceback源码实现。CONTINUE 状态稳态监控与动态改图职责CONTINUE 是稳态监控与编辑阶段Agent 在此状态中等待编排器发来的任务完成/失败事件从队列批量收集事件把编排器星座与自身最新修改进行状态合并处理事件并应用编辑循环直到全部任务完成或发生致命错误。事件批处理为什么一次只调用一次 LLMContinueConstellationAgentState.handlegalaxy/agents/constellation_agent_states.py的第一步是先阻塞等待至少一个事件再非阻塞地收走队列中所有积压事件# Wait for at least one event (blocking) first_event await agent.task_completion_queue.get() completed_task_events.append(first_event) # Collect other pending events (non-blocking) while not agent.task_completion_queue.empty(): try: event agent.task_completion_queue.get_nowait() completed_task_events.append(event) except asyncio.QueueEmpty: break为什么要批量处理当多个任务并行完成时例如 3 个任务几乎同时结束不批处理3 次 LLM 调用、3 次编辑会话批处理1 次 LLM 调用、1 次编辑会话处理全部 3 个事件。带来的收益是单次 LLM 调用即可反映多个完成事件、修改具备原子性、降低延迟与 API 成本。源码随后把task_ids一次性传给process_editing并在日志中输出收集到的任务数。事件源在 galaxy/core/events.pyTaskEvent的event_type只允许TASK_COMPLETED或TASK_FAILED进入任务完成队列由 add_task_completion_event 做类型与事件类型双重校验这保证了队列内容的合法性。状态合并避免编辑看到旧状态在批处理事件之后Agent 不会直接用编排器快照而是通过modification synchronizer做实时合并async def _get_merged_constellation(self, agent, orchestrator_constellation): synchronizer agent.orchestrator._modification_synchronizer if not synchronizer: return orchestrator_constellation merged_constellation synchronizer.merge_and_sync_constellation_states( orchestrator_constellationorchestrator_constellation ) agent.logger.info( f Real-time merged constellation for editing. fTasks before: {len(orchestrator_constellation.tasks)}, fTasks after merge: {len(merged_constellation.tasks)} ) return merged_constellation为什么合并至关重要考虑如下竞态场景任务 A 完成 → Agent 编辑星座新增任务 C任务 B 在编辑进行中完成不合并任务 B 的编辑基于旧状态看不到任务 C可能产生冲突修改合并任务 B 的编辑基于合并后的状态包含任务 C保证全局一致。从源码结构看合并由orchestrator._modification_synchronizer提供相关实现分布在 galaxy/constellation/orchestrator/orchestrator.py 与 galaxy/session/observers/constellation_sync_observer.py 中这正是文档强调的状态同步是关键的实现依据。编辑处理与转移判定合并后Agent 调用process_editing(context, task_ids, before_constellationmerged_constellation)。编辑完成后Agent 根据分析结果设置状态if constellation.is_complete() and no_more_edits_needed: agent.status ConstellationAgentStatus.FINISH.value elif critical_error_occurred: agent.status ConstellationAgentStatus.FAIL.value elif new_constellation_needed: agent.status ConstellationAgentStatus.START.value else: agent.status ConstellationAgentStatus.CONTINUE.value # Keep monitoring在process_editing内部[galaxy/agents/constellation_agent.py#L340-L415]还包含一条重要的重启链路_handle_constellation_completion会在旧星座完成但新星座未完成时把状态重置为START从而触发新一轮创建/复用星座 → 后台编排的循环。CONTINUE 行为表场景动作下一状态任务完成处理事件、应用编辑CONTINUE多个任务完成批量处理、单次编辑会话CONTINUE全部任务完成Agent 判定结束FINISH致命错误处理期间抛异常FAIL需要重启需要新星座STARTFINISH 状态成功终止职责与实现FINISH 表示成功终止前提是星座内所有任务成功完成、无需进一步编辑、用户目标已达成。ConstellationAgentStateManager.register class FinishConstellationAgentState(ConstellationAgentState): async def handle(self, agent, contextNone) - None: agent.logger.info(Galaxy task completed successfully) agent._status ConstellationAgentStatus.FINISH.value def next_state(self, agent) - AgentState: return self # Terminal state - no transitions def is_round_end(self) - bool: return True def is_subtask_end(self) - bool: return True关键特性next_state返回自身且is_round_end()与is_subtask_end()均为True——这意味着 FINISH 是终态执行轮次与子任务全部收尾不会有任何出边。进入条件示例LLM 决策LLM 基于星座状态决定结束例如{ thought: All tasks completed successfully. No further actions needed., status: FINISH, result: { summary: Dataset downloaded, model trained, deployed to production, total_tasks: 5, completed: 5, failed: 0 } }优雅关闭FINISH 状态保证资源全部释放、最终结果聚合、记忆日志持久化、成功指标记录。FAIL 状态错误终止职责与实现FAIL 表示错误终止适用于创建/编辑阶段出现不可恢复错误、DAG 校验失败、系统级致命故障。ConstellationAgentStateManager.register class FailConstellationAgentState(ConstellationAgentState): async def handle(self, agent, contextNone) - None: agent.logger.error(Galaxy task failed) agent._status ConstellationAgentStatus.FAIL.value def next_state(self, agent) - AgentState: return self # Terminal state - no transitions def is_round_end(self) - bool: return True def is_subtask_end(self) - bool: return True与 FINISH 相同FAIL 也是终态next_state返回自身避免 Agent 在失败后意外复活。失败场景与恢复策略场景触发原因恢复方式创建失败LLM 无法分解请求用户重新表述请求校验失败生成的 DAG 含环Agent 重试或人工修复致命异常意外系统错误查日志、重启 Agent超时处理超出限制增大超时或简化任务在星座层面任务还有独立的失败处理机制TaskStatus枚举galaxy/constellation/enums.py定义了PENDING、RUNNING、COMPLETED、FAILED、CANCELLED、WAITING_DEPENDENCY六种状态而整个星座的ConstellationState则包括COMPLETED、FAILED、PARTIALLY_FAILED等供编辑模式判断是否需要新增诊断任务。状态转移机制转移矩阵From ↓ / To →STARTCONTINUEFINISHFAILSTART❌✅ (success)❌✅ (error)CONTINUE✅ (restart)✅ (loop)✅ (done)✅ (error)FINISH❌❌✅ (stay)❌FAIL❌❌❌✅ (stay)转移规则由状态而非动作驱动与常见动作驱动的 FSM 不同这里的转移是状态驱动的next_state只读取agent.status再交给状态管理器解析出对应的状态对象galaxy/agents/constellation_agent_states.pyclass ConstellationAgentState(AgentState): def next_state(self, agent) - AgentState: status agent.status state ConstellationAgentStateManager().get_state(status) return state状态管理器与 register 装饰器class ConstellationAgentStateManager(AgentStateManager): _state_mapping: Dict[str, Type[AgentState]] {} property def none_state(self) - AgentState: return StartConstellationAgentState()状态类通过register装饰器模式自动注册进_state_mapping键为状态名如START、CONTINUE。该机制继承自 ufo/agents/states/basic.py 中的AgentStateManagerregister把state_class.name()映射到类本身get_state采用懒加载——首次访问才实例化并缓存。none_state默认指向StartConstellationAgentState因此未知状态会回退到 START。ConstellationAgentStateManager.register class StartConstellationAgentState(ConstellationAgentState): classmethod def name(cls) - str: return ConstellationAgentStatus.START.value状态接口参考AgentState 基类所有状态都实现AgentState抽象基类ufo/agents/states/basic.pyclass AgentState(ABC): abstractmethod async def handle(self, agent, context) - None: 执行状态专属逻辑 def next_state(self, agent) - AgentState: 基于 agent.status 决定下一状态 def next_agent(self, agent): 多 Agent 场景下的下一 Agent return agent abstractmethod def is_round_end(self) - bool: 该状态是否标记轮次结束 abstractmethod def is_subtask_end(self) - bool: 该状态是否标记子任务结束 classmethod abstractmethod def name(cls) - str: 状态标识其中next_agent默认返回当前 Agent说明该状态机是单 Agent 自循环模型星座内部的多设备执行交给编排器与各设备 Agent而非状态机跳转。is_round_end/is_subtask_end则被上层会话用于判断整个执行是否终结。状态度量与典型耗时执行时间线Gantt 示意典型耗时状态典型耗时影响因素START2-5 秒LLM 响应时间、校验复杂度CONTINUE可变10 秒 - 10 分钟任务执行时长、并行度FINISH 1 秒日志与清理FAIL 1 秒错误日志从源码推断START 的耗时主体是process_creation中的 LLM 推理与 DAG 校验validate_dag为 O(ne) 的环检测CONTINUE 的耗时则取决于任务执行与事件批处理节奏。状态查询与可观测性运行时状态查询# 查看当前状态对象 current_state agent.current_state print(fState: {current_state.name()}) # 判断是否轮次结束终态 if current_state.is_round_end(): print(Agent execution completed) # 直接读取状态字符串 status agent.status print(fStatus: {status}) # START, CONTINUE, FINISH, or FAIL状态历史记录Agent 在记忆日志中维护状态转移历史每条记录包含步骤、状态、时间戳与星座 ID{ step: 1, state: START, timestamp: 2024-01-01T10:00:00, constellation_id: constellation_abc123 }配合 galaxy/session/galaxy_session.py 中每次转移的日志Transitioning from X to Y与 galaxy/core/events.py 的CONSTELLATION_MODIFIED事件携带编辑前后快照、修改类型、涉及的 task_ids 与 timing 信息可实现对状态与修改的全链路追溯。错误处理与恢复策略异常层级状态处理统一采用捕获 → 记录 traceback → 置 FAIL的模式START 状态尤为典型try: constellation, timing await agent.process_creation(context) except AttributeError as e: agent.logger.error(fAttribute error: {e}) agent.status ConstellationAgentStatus.FAIL.value except KeyError as e: agent.logger.error(fMissing key: {e}) agent.status ConstellationAgentStatus.FAIL.value except Exception as e: agent.logger.error(fUnexpected error: {e}) agent.status ConstellationAgentStatus.FAIL.valueCONTINUE 状态同样把异常统一导向 FAIL源码确保任何未预期异常都不会让 Agent 停留在半死不活的中间态。分类型恢复策略错误类型所处状态恢复动作临时网络故障CONTINUE带退避的重试LLM 响应非法CONTINUE带示例重新提示DAG 检测到环START快速失败需人工干预任务执行超时CONTINUE标记任务失败星座继续致命系统错误任意立即转 FAIL最佳实践与常见陷阱状态机设计建议保持状态聚焦每个状态只承担单一清晰职责最小化转移转移越少调试越简单记录所有转移带上下文记录状态变更显式处理错误不要依赖隐式错误传播使用终态确保执行不会意外恢复。常见陷阱规避CONTINUE 中的死循环务必检查终止条件is_complete()等缺失错误处理未捕获异常会导致状态不可预期阻塞操作使用 async/await 防止死锁如task_completion_queue.get()必须 await状态污染不要在状态处理器之外随意修改 Agent 状态。转移日志示例agent.logger.info( fState transition: {old_state.name()} → {new_state.name()} )测试验证状态机行为的单元测试覆盖状态机的行为在仓库中有完整的测试支撑见 tests/unit/galaxy/agents/test_galaxy_agent_states.py覆盖了START 成功路径星座创建后置为执行态编排任务被启动一次START 失败路径创建返回None或抛出异常时置为 FAILCONTINUE 事件处理任务完成事件驱动编辑、异常时置 FAILCONTINUE 转移判定Agent 决定继续 → 回 START重启决定结束 → 转 FINISH终态语义FINISH / FAIL 的is_round_end()与is_subtask_end()均为True状态管理器none_state回退到StartConstellationAgentState、register注册映射正确超时配置不同优先级任务分配不同超时如GALAXY_TASK_TIMEOUT1800.0与GALAXY_CRITICAL_TASK_TIMEOUT3600.0。此外tests/unit/galaxy/session 下的会话测试覆盖了START → CONTINUE → FINISH的完整状态周期与带延续的重启周期与本文所述的循环驱动模型互相印证。总结Constellation Agent 的四态 FSM 是 UFO³ Galaxy 分布式编排中确定性控制的基石START 建图、CONTINUE 稳态改图、FINISH/FAIL 双终态收口配合事件批处理、状态合并modification synchronizer与register装饰器注册机制既保证了 LLM 拥有充分的动态适应能力又确保了生命周期的可预期、可审计与可恢复。深入理解这套状态机是掌握 Galaxy 多设备任务编排、排查运行问题、乃至扩展自定义 Agent 行为的必经之路。相关文档Constellation Agent 总览 — 双模式创建/编辑控制循环与整体架构Prompter 实现细节 — Prompter 架构命令参考 — MCP 工具规格Task Constellation 概览 — DAG 数据模型Constellation Orchestrator 概览 — 任务执行引擎【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 UFO³ Galaxy 中的 Constellation Agent 状态机：四态 FSM 驱动的动态任务编排生命周期详解
UFO³ Galaxy 中的 Constellation Agent 状态机四态 FSM 驱动的动态任务编排生命周期详解【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOUFO³ Galaxy 的 Constellation Agent 是贯穿分布式设备编排的织网中枢它依靠一个4 态有限状态机FSM来管理自身生命周期从初始化建图START、稳态监控与动态改图CONTINUE到成功收尾FINISH或异常终止FAIL。本文以 documents/docs/galaxy/constellation_agent/state.md 为核心骨架结合 galaxy/agents/constellation_agent_states.py、galaxy/agents/constellation_agent.py 等源码实现深入讲解每个状态的处理逻辑、转移矩阵、事件批处理与状态合并机制并给出状态查询、错误恢复与最佳实践帮助你掌握这一确定性生命周期控制的完整实现。为什么需要状态机把 LLM 推理与确定性控制解耦在 Constellation Agent 总览 中Constellation Agent 被描述为集中式的星座织造者Centralized Constellation Weaver它既要理解用户自然语言意图、生成可执行的 Task Constellation又要在任务执行过程中根据反馈动态增删改任务与依赖。如果这些行为完全交给大模型自由发挥就会出现状态不可预期、难以调试的问题。4 态 FSM 的核心价值在于把 LLM 的语义推理与确定性的控制逻辑分离LLM 负责想在创建/编辑两种模式下产出星座结构与修改动作见 WeavingMode 中的CREATION/EDITING。FSM 负责控确保 Agent 在任何时刻只处于一个明确状态所有转移都有清晰的触发条件提升安全性与可调试性。从源码看状态机的实现位于 galaxy/agents/constellation_agent_states.py而驱动它的运行循环位于 galaxy/session/galaxy_session.py其核心逻辑为# Initialize agent in START state self._agent.set_state(StartConstellationAgentState()) # Run agent state machine until completion while not self.is_finished(): # Execute current state await self._agent.handle(self._context) # Transition to next state self.state self._agent.state.next_state(self._agent) # Update agent state self._agent.set_state(self.state) await asyncio.sleep(0.01) # prevent busy waiting也就是说执行当前状态 → 依据 agent.status 决定下一个状态 → 切换状态构成了一个确定性的闭环循环持续直到 Agent 进入终态。图Constellation Agent 生命周期状态转换示意START → CONTINUE → FINISH/FAIL两个终态无出边。对应的完整状态转移图与 Mermaid 状态图定义见原文档与下文状态转换矩阵章节。状态空间与状态枚举四个状态的语义划分State类型描述进入条件START初始态初始化并创建星座Agent 实例化、完成后重启CONTINUE稳态监控事件并处理反馈星座创建成功FINISH终态成功终止所有任务完成、无需再编辑FAIL终态错误终止不可恢复错误、校验失败状态枚举定义状态枚举ConstellationAgentStatus定义在 galaxy/agents/constellation_agent_states.pyclass ConstellationAgentStatus(Enum): Galaxy Agent states START START CONTINUE CONTINUE FINISH FINISH FAIL FAIL注意这里的枚举值是字符串如START、CONTINUE它们与agent.status字段直接对应。在 ConstellationAgent 构造函数 中_status初始为START并调用self.set_state(StartConstellationAgentState())让 Agent 一开始就处于 START 状态。状态转移图MermaidSTART 状态初始化与建图阶段职责START 是初始化与创建阶段Agent 在该状态中完成四件事根据用户请求生成初始 Task Constellation校验 DAG 结构的正确性环路检测启动后台编排任务转移到监控模式。状态处理实现StartConstellationAgentState的实现galaxy/agents/constellation_agent_states.py核心逻辑如下ConstellationAgentStateManager.register class StartConstellationAgentState(ConstellationAgentState): async def handle(self, agent, context) - None: # 已处于终态则直接返回No-op if agent.status in [ ConstellationAgentStatus.FINISH.value, ConstellationAgentStatus.FAIL.value, ]: return timing_info {} # 若还没有星座则进入创建模式 if not agent.current_constellation: context.set(ContextNames.WEAVING_MODE, WeavingMode.CREATION) agent._current_constellation, timing_info ( await agent.process_creation(context) ) # 启动后台编排非阻塞 if agent.current_constellation: asyncio.create_task( agent.orchestrator.orchestrate_constellation( agent.current_constellation, metadatatiming_info ) ) agent.status ConstellationAgentStatus.CONTINUE.value elif agent.status ConstellationAgentStatus.CONTINUE.value: agent.status ConstellationAgentStatus.FAIL.value关键点分析幂等保护如果 Agent 已经处于 FINISH/FAIL 终态handle直接返回避免重复创建。创建模式标记通过context.set(ContextNames.WEAVING_MODE, WeavingMode.CREATION)把上下文切到创建模式随后process_creation会经过 ConstellationAgent.process_creation初始化 prompter → 加载 MCP 上下文 → 交给ConstellationAgentProcessor处理 → 通过_validate_and_update_constellation执行constellation.validate_dag()校验。非阻塞编排asyncio.create_task()把orchestrate_constellation作为后台任务启动Agent 随即把状态置为CONTINUE立即进入监控。注意timing_info初始化在前源码注释明确Initialize timing_info to avoid UnboundLocalError否则星座已存在时该变量未定义会抛错。行为与错误处理场景动作下一状态首次执行通过 LLM 生成星座CONTINUE成功/FAIL出错重启触发复用已有星座CONTINUE创建失败记录错误无星座产生FAIL校验失败DAG 含环或结构非法FAIL已处于终态空操作立即返回保持原状态错误处理采用多层try/exceptAttributeError如上下文字段缺失、KeyError如字典缺键、兜底Exception三者均把状态置为FAIL并输出完整 traceback源码实现。CONTINUE 状态稳态监控与动态改图职责CONTINUE 是稳态监控与编辑阶段Agent 在此状态中等待编排器发来的任务完成/失败事件从队列批量收集事件把编排器星座与自身最新修改进行状态合并处理事件并应用编辑循环直到全部任务完成或发生致命错误。事件批处理为什么一次只调用一次 LLMContinueConstellationAgentState.handlegalaxy/agents/constellation_agent_states.py的第一步是先阻塞等待至少一个事件再非阻塞地收走队列中所有积压事件# Wait for at least one event (blocking) first_event await agent.task_completion_queue.get() completed_task_events.append(first_event) # Collect other pending events (non-blocking) while not agent.task_completion_queue.empty(): try: event agent.task_completion_queue.get_nowait() completed_task_events.append(event) except asyncio.QueueEmpty: break为什么要批量处理当多个任务并行完成时例如 3 个任务几乎同时结束不批处理3 次 LLM 调用、3 次编辑会话批处理1 次 LLM 调用、1 次编辑会话处理全部 3 个事件。带来的收益是单次 LLM 调用即可反映多个完成事件、修改具备原子性、降低延迟与 API 成本。源码随后把task_ids一次性传给process_editing并在日志中输出收集到的任务数。事件源在 galaxy/core/events.pyTaskEvent的event_type只允许TASK_COMPLETED或TASK_FAILED进入任务完成队列由 add_task_completion_event 做类型与事件类型双重校验这保证了队列内容的合法性。状态合并避免编辑看到旧状态在批处理事件之后Agent 不会直接用编排器快照而是通过modification synchronizer做实时合并async def _get_merged_constellation(self, agent, orchestrator_constellation): synchronizer agent.orchestrator._modification_synchronizer if not synchronizer: return orchestrator_constellation merged_constellation synchronizer.merge_and_sync_constellation_states( orchestrator_constellationorchestrator_constellation ) agent.logger.info( f Real-time merged constellation for editing. fTasks before: {len(orchestrator_constellation.tasks)}, fTasks after merge: {len(merged_constellation.tasks)} ) return merged_constellation为什么合并至关重要考虑如下竞态场景任务 A 完成 → Agent 编辑星座新增任务 C任务 B 在编辑进行中完成不合并任务 B 的编辑基于旧状态看不到任务 C可能产生冲突修改合并任务 B 的编辑基于合并后的状态包含任务 C保证全局一致。从源码结构看合并由orchestrator._modification_synchronizer提供相关实现分布在 galaxy/constellation/orchestrator/orchestrator.py 与 galaxy/session/observers/constellation_sync_observer.py 中这正是文档强调的状态同步是关键的实现依据。编辑处理与转移判定合并后Agent 调用process_editing(context, task_ids, before_constellationmerged_constellation)。编辑完成后Agent 根据分析结果设置状态if constellation.is_complete() and no_more_edits_needed: agent.status ConstellationAgentStatus.FINISH.value elif critical_error_occurred: agent.status ConstellationAgentStatus.FAIL.value elif new_constellation_needed: agent.status ConstellationAgentStatus.START.value else: agent.status ConstellationAgentStatus.CONTINUE.value # Keep monitoring在process_editing内部[galaxy/agents/constellation_agent.py#L340-L415]还包含一条重要的重启链路_handle_constellation_completion会在旧星座完成但新星座未完成时把状态重置为START从而触发新一轮创建/复用星座 → 后台编排的循环。CONTINUE 行为表场景动作下一状态任务完成处理事件、应用编辑CONTINUE多个任务完成批量处理、单次编辑会话CONTINUE全部任务完成Agent 判定结束FINISH致命错误处理期间抛异常FAIL需要重启需要新星座STARTFINISH 状态成功终止职责与实现FINISH 表示成功终止前提是星座内所有任务成功完成、无需进一步编辑、用户目标已达成。ConstellationAgentStateManager.register class FinishConstellationAgentState(ConstellationAgentState): async def handle(self, agent, contextNone) - None: agent.logger.info(Galaxy task completed successfully) agent._status ConstellationAgentStatus.FINISH.value def next_state(self, agent) - AgentState: return self # Terminal state - no transitions def is_round_end(self) - bool: return True def is_subtask_end(self) - bool: return True关键特性next_state返回自身且is_round_end()与is_subtask_end()均为True——这意味着 FINISH 是终态执行轮次与子任务全部收尾不会有任何出边。进入条件示例LLM 决策LLM 基于星座状态决定结束例如{ thought: All tasks completed successfully. No further actions needed., status: FINISH, result: { summary: Dataset downloaded, model trained, deployed to production, total_tasks: 5, completed: 5, failed: 0 } }优雅关闭FINISH 状态保证资源全部释放、最终结果聚合、记忆日志持久化、成功指标记录。FAIL 状态错误终止职责与实现FAIL 表示错误终止适用于创建/编辑阶段出现不可恢复错误、DAG 校验失败、系统级致命故障。ConstellationAgentStateManager.register class FailConstellationAgentState(ConstellationAgentState): async def handle(self, agent, contextNone) - None: agent.logger.error(Galaxy task failed) agent._status ConstellationAgentStatus.FAIL.value def next_state(self, agent) - AgentState: return self # Terminal state - no transitions def is_round_end(self) - bool: return True def is_subtask_end(self) - bool: return True与 FINISH 相同FAIL 也是终态next_state返回自身避免 Agent 在失败后意外复活。失败场景与恢复策略场景触发原因恢复方式创建失败LLM 无法分解请求用户重新表述请求校验失败生成的 DAG 含环Agent 重试或人工修复致命异常意外系统错误查日志、重启 Agent超时处理超出限制增大超时或简化任务在星座层面任务还有独立的失败处理机制TaskStatus枚举galaxy/constellation/enums.py定义了PENDING、RUNNING、COMPLETED、FAILED、CANCELLED、WAITING_DEPENDENCY六种状态而整个星座的ConstellationState则包括COMPLETED、FAILED、PARTIALLY_FAILED等供编辑模式判断是否需要新增诊断任务。状态转移机制转移矩阵From ↓ / To →STARTCONTINUEFINISHFAILSTART❌✅ (success)❌✅ (error)CONTINUE✅ (restart)✅ (loop)✅ (done)✅ (error)FINISH❌❌✅ (stay)❌FAIL❌❌❌✅ (stay)转移规则由状态而非动作驱动与常见动作驱动的 FSM 不同这里的转移是状态驱动的next_state只读取agent.status再交给状态管理器解析出对应的状态对象galaxy/agents/constellation_agent_states.pyclass ConstellationAgentState(AgentState): def next_state(self, agent) - AgentState: status agent.status state ConstellationAgentStateManager().get_state(status) return state状态管理器与 register 装饰器class ConstellationAgentStateManager(AgentStateManager): _state_mapping: Dict[str, Type[AgentState]] {} property def none_state(self) - AgentState: return StartConstellationAgentState()状态类通过register装饰器模式自动注册进_state_mapping键为状态名如START、CONTINUE。该机制继承自 ufo/agents/states/basic.py 中的AgentStateManagerregister把state_class.name()映射到类本身get_state采用懒加载——首次访问才实例化并缓存。none_state默认指向StartConstellationAgentState因此未知状态会回退到 START。ConstellationAgentStateManager.register class StartConstellationAgentState(ConstellationAgentState): classmethod def name(cls) - str: return ConstellationAgentStatus.START.value状态接口参考AgentState 基类所有状态都实现AgentState抽象基类ufo/agents/states/basic.pyclass AgentState(ABC): abstractmethod async def handle(self, agent, context) - None: 执行状态专属逻辑 def next_state(self, agent) - AgentState: 基于 agent.status 决定下一状态 def next_agent(self, agent): 多 Agent 场景下的下一 Agent return agent abstractmethod def is_round_end(self) - bool: 该状态是否标记轮次结束 abstractmethod def is_subtask_end(self) - bool: 该状态是否标记子任务结束 classmethod abstractmethod def name(cls) - str: 状态标识其中next_agent默认返回当前 Agent说明该状态机是单 Agent 自循环模型星座内部的多设备执行交给编排器与各设备 Agent而非状态机跳转。is_round_end/is_subtask_end则被上层会话用于判断整个执行是否终结。状态度量与典型耗时执行时间线Gantt 示意典型耗时状态典型耗时影响因素START2-5 秒LLM 响应时间、校验复杂度CONTINUE可变10 秒 - 10 分钟任务执行时长、并行度FINISH 1 秒日志与清理FAIL 1 秒错误日志从源码推断START 的耗时主体是process_creation中的 LLM 推理与 DAG 校验validate_dag为 O(ne) 的环检测CONTINUE 的耗时则取决于任务执行与事件批处理节奏。状态查询与可观测性运行时状态查询# 查看当前状态对象 current_state agent.current_state print(fState: {current_state.name()}) # 判断是否轮次结束终态 if current_state.is_round_end(): print(Agent execution completed) # 直接读取状态字符串 status agent.status print(fStatus: {status}) # START, CONTINUE, FINISH, or FAIL状态历史记录Agent 在记忆日志中维护状态转移历史每条记录包含步骤、状态、时间戳与星座 ID{ step: 1, state: START, timestamp: 2024-01-01T10:00:00, constellation_id: constellation_abc123 }配合 galaxy/session/galaxy_session.py 中每次转移的日志Transitioning from X to Y与 galaxy/core/events.py 的CONSTELLATION_MODIFIED事件携带编辑前后快照、修改类型、涉及的 task_ids 与 timing 信息可实现对状态与修改的全链路追溯。错误处理与恢复策略异常层级状态处理统一采用捕获 → 记录 traceback → 置 FAIL的模式START 状态尤为典型try: constellation, timing await agent.process_creation(context) except AttributeError as e: agent.logger.error(fAttribute error: {e}) agent.status ConstellationAgentStatus.FAIL.value except KeyError as e: agent.logger.error(fMissing key: {e}) agent.status ConstellationAgentStatus.FAIL.value except Exception as e: agent.logger.error(fUnexpected error: {e}) agent.status ConstellationAgentStatus.FAIL.valueCONTINUE 状态同样把异常统一导向 FAIL源码确保任何未预期异常都不会让 Agent 停留在半死不活的中间态。分类型恢复策略错误类型所处状态恢复动作临时网络故障CONTINUE带退避的重试LLM 响应非法CONTINUE带示例重新提示DAG 检测到环START快速失败需人工干预任务执行超时CONTINUE标记任务失败星座继续致命系统错误任意立即转 FAIL最佳实践与常见陷阱状态机设计建议保持状态聚焦每个状态只承担单一清晰职责最小化转移转移越少调试越简单记录所有转移带上下文记录状态变更显式处理错误不要依赖隐式错误传播使用终态确保执行不会意外恢复。常见陷阱规避CONTINUE 中的死循环务必检查终止条件is_complete()等缺失错误处理未捕获异常会导致状态不可预期阻塞操作使用 async/await 防止死锁如task_completion_queue.get()必须 await状态污染不要在状态处理器之外随意修改 Agent 状态。转移日志示例agent.logger.info( fState transition: {old_state.name()} → {new_state.name()} )测试验证状态机行为的单元测试覆盖状态机的行为在仓库中有完整的测试支撑见 tests/unit/galaxy/agents/test_galaxy_agent_states.py覆盖了START 成功路径星座创建后置为执行态编排任务被启动一次START 失败路径创建返回None或抛出异常时置为 FAILCONTINUE 事件处理任务完成事件驱动编辑、异常时置 FAILCONTINUE 转移判定Agent 决定继续 → 回 START重启决定结束 → 转 FINISH终态语义FINISH / FAIL 的is_round_end()与is_subtask_end()均为True状态管理器none_state回退到StartConstellationAgentState、register注册映射正确超时配置不同优先级任务分配不同超时如GALAXY_TASK_TIMEOUT1800.0与GALAXY_CRITICAL_TASK_TIMEOUT3600.0。此外tests/unit/galaxy/session 下的会话测试覆盖了START → CONTINUE → FINISH的完整状态周期与带延续的重启周期与本文所述的循环驱动模型互相印证。总结Constellation Agent 的四态 FSM 是 UFO³ Galaxy 分布式编排中确定性控制的基石START 建图、CONTINUE 稳态改图、FINISH/FAIL 双终态收口配合事件批处理、状态合并modification synchronizer与register装饰器注册机制既保证了 LLM 拥有充分的动态适应能力又确保了生命周期的可预期、可审计与可恢复。深入理解这套状态机是掌握 Galaxy 多设备任务编排、排查运行问题、乃至扩展自定义 Agent 行为的必经之路。相关文档Constellation Agent 总览 — 双模式创建/编辑控制循环与整体架构Prompter 实现细节 — Prompter 架构命令参考 — MCP 工具规格Task Constellation 概览 — DAG 数据模型Constellation Orchestrator 概览 — 任务执行引擎【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考