Agent 控制流与动态编排

Agent 控制流与动态编排

学习目标

读完本章后，你将能够：

• 理解控制流与数据流的分离设计
• 设计 Commander/Controller 模式的动态编排系统
• 掌握运行时 Agent 启用、禁用、配置的操作模式
• 实现基于条件的动态路由与自适应工作流

---

核心概念

控制流 vs 数据流

在 Agent 自动化系统中，存在两种独立的关系：

维度	数据流	控制流
连接实体	Link（Agent → Agent）	ControlLink（Commander → Target）
传递内容	Event（数据 payload）	Control Action（run/enable/disable/configure）
触发时机	事件创建后自动传播	Commander 被调度或收到事件时
方向性	单向（source → receiver）	单向（controller → control_target）

四种控制操作

操作	效果	适用场景
run	触发目标 Agent 的 check 方法	定时任务编排、按需执行
enable	将 disabled 设为 false	条件激活、季节性工作流
disable	将 disabled 设为 true	条件暂停、节能模式
configure	options.deep_merge(new_config)	动态参数调整、位置感知

---

实现架构

ControlLink 数据模型

ControlLink
├── controller_id      → 控制者 Agent（如 CommanderAgent）
├── control_target_id  → 被控制 Agent（任意类型）
└── (无事件相关字段)

与 Link 的关键区别：

• Link 记录 event_id_at_creation（事件水位线），ControlLink 不需要
• Link 是 Event 传播路径，ControlLink 是控制指令路径
• 一个 Agent 可以同时拥有 Link 和 ControlLink 连接

AgentControllerConcern 模块

提供控制能力的 Concern 模块，可混入任何 Agent 类：

module AgentControllerConcern
  included do
    can_control_other_agents!  # 声明控制能力
    validate :validate_control_action
  end

  def control!
    control_targets.each do |target|
      interpolate_with('target' => target) do
        case control_action
        when 'run'
          Agent.async_check(target.id)
        when 'enable'
          target.update!(disabled: false)
        when 'disable'
          target.update!(disabled: true)
        when 'configure'
          target.update! options: target.options.deep_merge(configure_options)
        end
      end
    end
  end
end

关键设计：

• can_control_other_agents! 类方法声明控制能力
• control_targets 通过 ControlLink 自动获取
• interpolate_with('target' => target) 允许在动作中使用 Liquid 模板引用目标 Agent 的属性

---

动态编排模式

模式一：条件工作流

根据外部条件动态启用/禁用 Agent 分支：

Liquid 动态动作：

{
  "action": "{% if conditions contains 'Rain' %}enable{% else %}disable{% endif %}"
}

模式二：动态参数调整

根据输入事件动态调整目标 Agent 的配置：

配置合并：

{
  "action": "configure",
  "configure_options": {
    "location": "{{_location_.latlng}}"
  }
}

模式三：级联调度

一个 Commander 按顺序触发多个 Agent：

模式四：自适应调度

SchedulerAgent 支持运行时动态 cron 调度：

与 CommanderAgent 的区别：

• CommanderAgent 是一次性触发（收到事件时执行动作）
• SchedulerAgent 是持续性调度（按 cron 周期反复执行）
• SchedulerAgent 使用 Rufus::Scheduler 的 schedule_cron 方法在运行时注册/注销定时任务

---

陷阱与对策

陷阱一：控制循环

Commander A 控制 Commander B，B 又控制 A，形成控制循环。

对策：设计上避免——控制流应是有向无环图（DAG）。系统层面不做自动检测，依赖用户理解。

陷阱二：竞态条件

Commander 在 Agent 正在执行 receive 时修改其 options。

对策：ActiveRecord 的行级锁保证 update! 的原子性。但 options 的修改不会影响正在进行的 receive 调用——那是修改前的快照。

陷阱三：禁用 Agent 的积压事件

禁用 Agent 后，上游 Agent 仍在向其发送事件，导致大量积压。

对策：

• drop_pending_events 选项：启用 Agent 时自动跳过积压事件
• 启用时调用 set_last_checked_event_id，从最新水位线开始消费

---

设计取舍

集中式编排 vs 分散式控制

维度	集中式（单一编排器）	分散式（每个 Agent 自控制）
可视性	高（一个入口看全局）	低（控制逻辑分散）
灵活性	低（编排器是瓶颈）	高（每个 Agent 独立决策）
调试难度	低	高

Huginn 的选择：分散式——每个 CommanderAgent 只控制自己的 targets，编排逻辑分布在多个 Commander 中。

配置合并 vs 配置替换

configure 操作使用 deep_merge 而非完全替换：

target.update! options: target.options.deep_merge(configure_options)

维度	deep_merge	完全替换
安全性	高（未修改的字段保留）	低（可能丢失配置）
灵活性	中（只能增量修改）	高（可以完全重写）
可预测性	高	中

选择 deep_merge 的原因：安全性优先——动态配置通常只修改个别参数，不应影响其他配置。

---

参考来源

• IFTTT Applets — 条件触发式工作流的典型产品
• Apache Airflow — 对比参考：静态 DAG vs 动态控制流