跳到主要内容

数据截至 (上游 commit 96983c73ed09)

第 5 章 · 跨设备编排:任务星座 DAG

这章讲什么: UFO³ Galaxy 怎么把「从两台服务器收日志、汇总成表、发邮件」这样的请求变成一张可并行执行的 DAG,并且在执行过程中根据中间结果改这张图。

5.1 要解决的小问题​

单机 UFO² 是顺序的:一步做完做下一步。这在跨设备时立刻不够用:

  • 两台服务器收日志本可以同时做,顺序做白等一倍时间。
  • 第一步的产出是第二步的输入,得有依赖概念。
  • 收完日志才发现有台机器全是错误日志,可能需要临时加一个诊断任务。

所以需要:一张有依赖的图 + 一个并行调度器 + 一条改图的通路。

5.2 三个数据结构​

名字是什么文件
TaskStar(任务星)DAG 的节点:一段自然语言描述的活 + 目标设备 + 状态galaxy/constellation/task_star.py:28
TaskStarLine(星线)DAG 的边:from → to,带依赖类型和条件galaxy/constellation/task_star_line.py
TaskConstellation(星座)整张图:节点表、边表、拓扑序、环检测、统计galaxy/constellation/task_constellation.py:31

一个 TaskStar 装什么​

关键字段(galaxy/constellation/task_star.py:42-110):

字段作用
description这个节点要干什么(自然语言,会原样发给设备)
tips给执行方的提示列表
target_device_id派到哪台设备
device_type设备类型要求
priority就绪队列的排序依据
timeout / retry_count超时与重试
_dependencies / _dependents上下游节点 id 集合

注意 description 是自然语言而不是结构化脚本。 一个 TaskStar 被派到一台 Windows 设备上,就是把这段话原样交给那台机器上的 UFO²,让它自己拆步骤。分层就这样接上了。

边有四种语义​

DependencyType(galaxy/constellation/enums.py:27-37):

类型何时放行下游
UNCONDITIONAL上游完成即可
SUCCESS_ONLY上游必须成功
COMPLETION_ONLY上游完成即可,不管成败
CONDITIONAL要满足指定条件

5.3 就绪判定:两道关​

一个节点能跑的条件被拆成两处,两处都要过:

# 关卡一:节点自己觉得自己就绪吗
# galaxy/constellation/task_star.py:411-414
@property
def is_ready_to_execute(self) -> bool:
    return self._status == TaskStatus.PENDING and len(self._dependencies) == 0
# 关卡二:星座再复核一遍依赖
# galaxy/constellation/task_constellation.py:279-295
for task in self._tasks.values():
    if task.is_ready_to_execute and self._are_dependencies_satisfied(task.task_id):
        ready_tasks.append(task)
ready_tasks.sort(key=lambda t: t.priority.value, reverse=True)   # 高优先级先跑

为什么要两道? 因为 _dependencies 是节点上的一个集合,靠 mark_task_completed 在上游完成时主动删掉(galaxy/constellation/task_constellation.py:436-481)。这套「删边」机制在并发改图时可能不同步,所以 _are_dependencies_satisfied 又从边表重新查了一遍真实状态(:1133-1156)。源码注释直接写着 Double-check dependencies are satisfied。

5.4 编排循环​

TaskConstellationOrchestrator._run_execution_loop(galaxy/constellation/orchestrator/orchestrator.py:394-431):

# 示意,非源码 —— 演示编排循环的骨架
while not constellation.is_complete():
    if cancelled():
        constellation.state = ConstellationState.CANCELLED
        break
    constellation = await self._sync_constellation_modifications(constellation)  # 先合并改图
    self._validate_existing_device_assignments(constellation)
    ready = constellation.get_ready_tasks()
    await self._schedule_ready_tasks(ready, constellation)   # 全部 create_task,不等
    await self._wait_for_task_completion()                   # 等第一个完成就返回
await self._wait_for_all_tasks()

三个要点:

  • 调度是「全发不等」:每个就绪节点 asyncio.create_task 一个协程,记进 _execution_tasks 字典(:468-482)。
  • 等待用 FIRST_COMPLETED:任何一个节点完成就回到循环顶部重新算就绪集(:484-497)。这样后继节点能第一时间被放出来。
  • 改图合并在循环最前面,后面会讲为什么必须在这个位置。

怎么读这张图:左边是编排器的循环,右边是 agent 的循环;两者靠一个队列和一个同步器耦合。

  编排器循环                          ConstellationAgent 循环
  ─────────────                       ──────────────────────
  ① 合并 agent 的改图  ◀───────────┐
  ② 取就绪节点                     │
  ③ 全部并发发出去                 │
  ④ 等任一个完成 ──▶ 发完成事件 ──▶ ⑤ 从队列取事件
         ▲                          ⑥ 让 LLM 决定要不要改图
         └──────────────────────────⑦ 执行改图工具,标记完成

5.5 ConstellationAgent 的两种模式​

它是唯一的 Galaxy 大脑,但干两件不同的事,靠一个 WEAVING_MODE 上下文变量区分:

模式何时进入输出什么
CREATION第一次,还没有图一整张 DAG(JSON)
EDITING每次有节点完成一串工具调用(改图动作)

对应两个入口:process_creation(galaxy/agents/constellation_agent.py:307-337)与 process_editing(:340-415)。

两个模式的输出格式不一样,这是重点​

创建模式要求模型直接吐一整个 constellation 对象(galaxy/prompts/constellation/share/constellation_creation.yaml):

输出一个 JSON,匹配 ConstellationAgentResponse,其中 constellation 含 tasks 和 dependencies。

编辑模式要求模型吐一串动作(galaxy/prompts/constellation/share/constellation_editing.yaml):

提出一个有序的工具调用序列来修改星座;不需要改就返回空的 action 列表。

为什么不都用「重发整张图」?因为执行中的图有节点正在跑、有节点已完成——整体替换会把运行态覆盖掉。改成增量动作,就能只碰能碰的部分。

5.6 改图 = MCP 工具调用​

这是整个 Galaxy 设计里最漂亮的一处复用。

改图的七个动作,是七个 MCP 工具(ufo/client/mcp/local_servers/constellation_mcp_server.py:29-330):

工具名干什么
add_task加一个节点
remove_task删一个节点
update_task改节点字段
add_dependency加一条边
remove_dependency删一条边
update_dependency改边
build_constellation整体构建

它们注册在命名空间 ConstellationEditor 下,而 config/ufo/mcp.yaml:114-120 把这个命名空间挂给了 ConstellationAgent。

于是:ConstellationAgent 改 DAG,跟 AppAgent 点鼠标,走的是完全同一条链路——同样的 ActionCommandInfo → Command → dispatcher → Computer → MCP(对比 galaxy/agents/processors/strategies/base_constellation_strategy.py:418-459 与第 4 章的 _action_to_command,两处代码几乎逐行对应)。

收益是实打实的: 改图自动获得了工具 schema 注入 prompt、错误结果回喂模型、执行日志、记忆记录这一整套设施,不用重写。

5.7 安全改图:三层防护​

边跑边改一张正在执行的图,是最容易出事的地方。UFO 叠了三层。

第一层:数据结构层面拒绝改运行中的节点​

TaskStar 的几个 setter 直接抛异常(galaxy/constellation/task_star.py:124-172):

@name.setter
def name(self, value: str) -> None:
    if self._status == TaskStatus.RUNNING:
        raise ValueError(f"Cannot modify name of running task {self._task_id}")

description、tips 同理。

第二层:告诉模型哪些能改​

拼 prompt 时,每个节点后面直接标上 ✏️ [MODIFIABLE] 或 🔒 [READ-ONLY],可改的还额外附一行提示(_format_constellation,galaxy/agents/prompters/base_constellation_prompter.py:87-186)。

可改的判据是状态属于 {PENDING, WAITING_DEPENDENCY}(get_modifiable_tasks,galaxy/constellation/task_constellation.py:332-342)。边的可改性看它的下游节点还没开始(get_modifiable_dependencies,:343-359)。

这是把运行态直接编码进 prompt。 比起「让模型乱改再拒绝」,先告诉它规则要有效得多。

第三层:每个改图命令都可回滚​

改图命令是可撤销命令对象(IUndoableCommand,galaxy/constellation/editor/commands.py:21 BaseConstellationCommand)。以 UpdateTaskCommand 为例(:296-388):

# 示意,非源码 —— 演示改图命令的自保
self._create_backup()                      # 先备份整张图
for field, value in updates.items():
    self._original_values[field] = getattr(task, field)
    setattr(task, field, value)

ok, errors = self._constellation.validate_dag()   # 改完立刻验图
if not ok:
    self._restore_backup()                  # 图坏了就整体回滚
    raise CommandExecutionError(...)

加边时还有一道单独的环检测:add_dependency 先跑 _would_create_cycle(DFS 找反向通路),会成环就直接拒绝(galaxy/constellation/task_constellation.py:210-242、:1158-1180)。

5.8 竞态:两份图怎么合并​

问题​

编排器和 agent 各持有一份 constellation 引用:

  • 编排器知道最新的运行状态(谁跑完了、结果是什么)。
  • agent知道最新的结构(刚加了哪个节点、删了哪条边)。

谁覆盖谁都会丢东西。

解法:按字段来源分工合并​

ConstellationModificationSynchronizer.merge_and_sync_constellation_states(galaxy/session/observers/constellation_sync_observer.py:384-450):

# 示意,非源码 —— 演示合并规则
merged = agent_constellation                     # 以 agent 的结构为底
for tid, orch_task in orchestrator_constellation.tasks.items():
    if tid in merged.tasks:
        agent_task = merged.tasks[tid]
        if _is_state_more_advanced(orch_task.status, agent_task.status):
            agent_task._status = orch_task.status        # 运行态以编排器为准
            agent_task._result = orch_task.result
            agent_task._error  = orch_task.error
            agent_task._execution_start_time = orch_task.execution_start_time
            agent_task._execution_end_time   = orch_task.execution_end_time
merged.update_state()

「更靠前」怎么定义? 用一张状态等级表(_is_state_more_advanced,:452-478):

状态等级
PENDING0
WAITING_DEPENDENCY1
RUNNING2
COMPLETED / FAILED / CANCELLED3

只有编排器的状态等级更高才覆盖。这保证了「已完成」不会被 agent 那份陈旧的「运行中」盖回去。

还有一道等待闸​

光合并不够。可能出现:节点 A 完成 → 编排器立刻算就绪集 → 把 B 发出去 → 但 agent 正在改 B 的描述。

所以编排器每轮开头先 wait_for_pending_modifications()(:250-310)。这个方法有超时保护:超时后打警告并清空 pending,继续跑——注释写明是 to prevent permanent deadlock。

这是很诚实的工程取舍: 宁可偶尔在改图未完成时放行,也不接受整个编排卡死。

5.9 设备分配​

三种策略(ConstellationManager.assign_devices_automatically,galaxy/constellation/orchestrator/constellation_manager.py:132-186):

策略名怎么分
round_robin(默认)轮流
capability_match按能力标签匹配
load_balance看负载

但实际路径通常不走这里——DAG 生成时模型就已经把 target_device_id 填好了。创建 prompt 里会把 device_info_list(每台设备的 device_id / os / capabilities / metadata)一并给模型,让它自己决定哪一步派给谁(galaxy/prompts/constellation/share/constellation_creation.yaml)。

设备清单来自 config/galaxy/devices.yaml,每台设备写明 server_url、os、capabilities 和一堆自由格式的 metadata(日志路径、开发目录、日志关键字等)。这些 metadata 会原样进 prompt,相当于给模型的设备说明书。

5.10 图上的分析能力​

TaskConstellation 自带一组图算法,主要用于可视化与统计:

方法算什么位置
get_topological_orderKahn 算法拓扑排序,发现环就抛异常:508-547
has_cycle靠拓扑排序是否成功来判定:1181-1187
get_longest_path关键路径(最长路径):549-609
get_max_width图的最大宽度(并行度上限):610-650
get_parallelism_metrics并行度指标:737-798
get_critical_path_length_with_time带时间权重的关键路径:651-723

可视化在 galaxy/visualization/dag_visualizer.py,终端里直接画图(display_dag,:1202)。

5.11 代码地图​

主题文件路径符号名
节点定义与运行态保护galaxy/constellation/task_star.pyTaskStar、TaskStar.execute、is_ready_to_execute
边定义galaxy/constellation/task_star_line.pyTaskStarLine
图本体与就绪计算galaxy/constellation/task_constellation.pyTaskConstellation、get_ready_tasks、mark_task_completed、_are_dependencies_satisfied
环检测与拓扑序galaxy/constellation/task_constellation.py_would_create_cycle、get_topological_order、has_cycle
可改性判定galaxy/constellation/task_constellation.pyget_modifiable_tasks、get_modifiable_dependencies
状态枚举galaxy/constellation/enums.pyTaskStatus、DependencyType、ConstellationState
编排循环galaxy/constellation/orchestrator/orchestrator.pyTaskConstellationOrchestrator._run_execution_loop、_schedule_ready_tasks
设备分配策略galaxy/constellation/orchestrator/constellation_manager.pyConstellationManager.assign_devices_automatically
Galaxy 大脑galaxy/agents/constellation_agent.pyConstellationAgent.process_creation、process_editing
Galaxy 状态机galaxy/agents/constellation_agent_states.pyStartConstellationAgentState、ContinueConstellationAgentState
改图动作管线galaxy/agents/processors/strategies/base_constellation_strategy.pyBaseConstellationActionExecutionStrategy
改图 prompt 组装galaxy/agents/prompters/base_constellation_prompter.py_format_constellation
可撤销改图命令galaxy/constellation/editor/commands.pyBaseConstellationCommand、UpdateTaskCommand、AddDependencyCommand
改图 MCP 工具ufo/client/mcp/local_servers/constellation_mcp_server.pycreate_constellation_mcp_server
竞态合并galaxy/session/observers/constellation_sync_observer.pyConstellationModificationSynchronizer、merge_and_sync_constellation_states、_is_state_more_advanced
Galaxy 会话galaxy/session/galaxy_session.pyGalaxySession、GalaxyRound
创建/编辑 promptgalaxy/prompts/constellation/share/constellation_creation.yaml、constellation_editing.yaml
设备与运行参数config/galaxy/devices.yaml、constellation.yaml