CMDCOrchestrator.DAG (cmdc_orchestrator v0.6.0)

Copy Markdown View Source

DAG 定义：节点 + 边 + 拓扑排序。

Summary

Types

edge_def()

边定义。

node_def()
node_type()
t()

Functions

from_snapshot(snapshot)

从 snapshot/1 的结果恢复 DAG。

get_dependencies(dag, node_id)
get_dependents(dag, node_id)
get_node(dag, id)
layered_order(dag)

返回拓扑分层结果：每一层内部所有节点彼此独立，可并行执行；层与层串行。

snapshot(dag)

把 DAG 转成 JSON 安全快照,供 RunStore 持久化恢复使用。

topological_sort(dag)

Types

edge_def()

@type edge_def() :: %{
  :from => String.t(),
  :to => String.t(),
  optional(:label) => String.t() | nil,
  optional(:branch) => String.t() | nil,
  optional(:signal) => String.t() | nil,
  optional(:when) => String.t() | nil
}

边定义。

  • :from / :to — 必填，节点 ID
  • :label — 可选展示名
  • :branch — 可选；若上游为 :router 节点，仅当 router 输出 %{route: branch} 时该边才被激活。v0.5 起也作为通用 signal 匹配字段。
  • :signal / :when — 可选；当上游节点输出 signal 匹配时激活该边。

node_def()

@type node_def() :: %{id: String.t(), type: node_type(), config: map()}

node_type()

@type node_type() ::
  :agent
  | :aggregator
  | :router
  | :gate
  | :debate
  | :hierarchy
  | :tool
  | :eval_gate
  | :condition
  | :human_task
  | :fork
  | :join
  | atom()

t()

@type t() :: %CMDCOrchestrator.DAG{
  edges: [edge_def()],
  id: String.t(),
  name: String.t(),
  nodes: [node_def()]
}

Functions

from_snapshot(snapshot)

@spec from_snapshot(map()) :: t()

从 snapshot/1 的结果恢复 DAG。

get_dependencies(dag, node_id)

@spec get_dependencies(t(), String.t()) :: [String.t()]

get_dependents(dag, node_id)

@spec get_dependents(t(), String.t()) :: [String.t()]

get_node(dag, id)

@spec get_node(t(), String.t()) :: node_def() | nil

layered_order(dag)

@spec layered_order(t()) :: {:ok, [[String.t()]]} | {:error, :cycle_detected}

返回拓扑分层结果：每一层内部所有节点彼此独立，可并行执行；层与层串行。

对应 Agentic Design Patterns 第 3 章 Parallelization 模式。

snapshot(dag)

@spec snapshot(t()) :: map()

把 DAG 转成 JSON 安全快照,供 RunStore 持久化恢复使用。

topological_sort(dag)

@spec topological_sort(t()) :: {:ok, [String.t()]} | {:error, :cycle_detected}