第 22 章：自治任务

发表于 更新于 分类于 阅读次数：
Agent 不再被动等待——四种自治机制的完整实现：Cron 表达式的定时调度、心跳任务的周期检查、Mission 自主多步骤工作流、以及主动消息的推送触发策略。
个人公众号
1
2
卷四：纵深
[20] 配置的秘密 -> [21] 记忆的宫殿 -> [22] 自治任务 <- you are here

Agent 不只是被动等待用户输入。它可以定时执行任务、定期检查状态、甚至自主完成复杂的多步骤工作。这一章看四种自治机制：Cron 调度、心跳、Mission 系统、主动消息。

四种自治机制

1
2
3
4
1. Cron 调度   用户定义的定时任务（如每天早上发送天气）
2. 心跳        定期读取 HEARTBEAT.md 并执行
3. Mission     自主完成复杂的多步骤编程任务
4. 主动消息    用户空闲一段时间后主动联系

什么时候用哪个？

场景推荐机制原因
“每天早上9点发天气预报”Cron（text）固定时间发固定内容，不需要 AI 推理
“每2小时检查邮件并总结”Cron（agent）需要 AI 理解邮件内容，但触发时间是固定的
“持续监控某个事情，条件触发时行动”心跳（HEARTBEAT.md）Markdown 指令灵活，可包含复杂条件和多步逻辑
“帮我实现一个完整功能”Mission需要 AI 自主分步、写代码、验证
“你之前问的问题，我查到了更多资料”主动消息基于历史对话的主动联系

简单来说：Cron 是定时触发器，心跳是定期自主思考，Mission 是有目标的自主任务，主动消息是机会主义联系。

Cron 调度——APScheduler 驱动

CronManager 基于 apscheduler 的 AsyncIOScheduler：

1
2
3
4
5
6
7
8
9
10
11
class CronManager:
    async def start(self):
        # 加载用户定义的任务
        jobs = await self.repo.list_jobs()
        for job in jobs:
            if job.enabled:
                self._scheduler.add_job(...)
        # 注册心跳和 Dream 任务
        self._register_heartbeat()
        self._register_dream()
        self._scheduler.start()

每个 Cron 任务有两种类型：

  • text：定时发送固定文本到指定渠道
  • agent：定时运行 Agent 处理请求，结果发送到渠道

任务定义（CronJobSpec）：

1
2
3
4
5
6
7
8
9
class CronJobSpec(BaseModel):
    id: str
    name: str
    enabled: bool
    schedule: ScheduleSpec        # 5 字段 cron 表达式 + 时区
    task_type: Literal["text", "agent"]
    text: str                     # text 类型的内容
    request: CronJobRequest       # agent 类型的请求
    dispatch: DispatchSpec        # 发送到哪个渠道/用户

执行流程：

1
2
3
Cron 触发 -> CronExecutor.execute(job)
  if text:  channel_manager.send_text(text)
  if agent: runner.stream_query(request) -> channel_manager.send_event(event)

心跳——定期自主行动

心跳是一种特殊的 Cron 任务——定期读取 HEARTBEAT.md 并用其内容作为 Agent 的输入：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async def run_heartbeat_once(runner, channel_manager, ...):
    # 1. 检查是否在活跃时段（如 08:00-22:00）
    if not _in_active_hours():
        return

    # 2. 读取 HEARTBEAT.md
    heartbeat = (workspace_dir / "HEARTBEAT.md").read_text()

    # 3. 构建 Agent 请求
    request = build_request(content=heartbeat)

    # 4. 执行并发送结果
    if target == "last" and last_dispatch:
        # 发送到上次对话的渠道
        await runner.stream_query(request, dispatch=last_dispatch)
    else:
        # 仅执行，不发送
        await runner.stream_query(request)

配置：

1
2
3
4
5
class HeartbeatConfig(BaseModel):
    enabled: bool = False
    every: str = "6h"              # 间隔或 cron 表达式
    target: str = "main"           # "main" 或 "last"
    active_hours: ActiveHoursConfig  # 如 08:00-22:00

用户在 HEARTBEAP.md 中写指令，如”检查我的邮箱并总结重要邮件”——Agent 会定期自动执行。

Mission 系统——自主编程

Mission 是 QwenPaw 最复杂的自治机制——用户给一个任务描述，Agent 自主完成从分析到实现的全过程。

两阶段执行

Phase 1：需求分析与 PRD 生成

1
2
3
4
5
6
7
8
9
10
用户：/mission 实现一个 CSV 文件解析工具
  |
  v
Agent 自主探索代码库，理解现有结构
  |
  v
生成 prd.json（结构化任务列表 + 用户故事）
  |
  v
等待用户确认

Phase 2：自主实现

1
2
3
4
5
6
7
8
用户确认 PRD
  |
  v
迭代循环（最多 max_iterations 轮）：
  1. Agent 派遣 worker 执行子任务
  2. 检查 prd.json 中的验收标准
  3. 全部通过 -> 完成
  4. 未通过 -> 注入继续消息

关键设计：Phase 2 中主 Agent 是”控制器”——实现工具（edit_file、browser_use）被移到 "mission_impl" 组并禁用。主 Agent 通过 submit_to_agent 派遣 worker 执行实际编码。

状态文件布局：

1
2
3
4
5
<workspace>/missions/<loop_id>/
  task.md            # 原始任务描述
  prd.json           # 结构化任务列表
  progress.txt       # 迭代日志
  loop_config.json   # 环境元数据

主动消息——空闲时主动联系

proactive_trigger_loop 每 30 秒检查一次，如果用户空闲超过配置的 idle_minutes：

1
2
3
4
5
6
7
8
9
async def proactive_trigger_loop(session_id):
    while True:
        await asyncio.sleep(30)
        if is_agent_busy():
            continue
        idle = time_since_last_message()
        if idle >= idle_minutes:
            await generate_proactive_response(workspace)
            break  # 触发一次后退出

Agent 会根据历史对话内容主动发起相关话题——比如”你之前问的那个问题，我查到了更多资料”。

工程权衡

为什么心跳用 HEARTBEAT.md 而非配置字段？

Markdown 文件让用户可以用自然语言写指令，支持复杂的条件和逻辑。配置字段只能存简单参数。用户可以随时编辑 HEARTBEAP.md 改变 Agent 的行为，不需要重启服务。

为什么 Mission 要分两阶段？

Phase 1 的 PRD 让用户在 Agent 开始编码前确认方向。没有这一步，Agent 可能花大量时间走错方向。PRD 是”合同”——用户和 Agent 都知道要做什么、怎么验收。

自检

  • 知道 CronManager 用 APScheduler 驱动，支持 text 和 agent 两种任务
  • 知道心跳读取 HEARTBEAT.md 作为 Agent 输入
  • 知道 Mission 分两阶段：PRD 生成 + 自主实现
  • 知道主动消息在用户空闲时触发

下一章我们看多智能体协作——MultiAgentManager 怎么并行加载多个 Agent？Agent 之间怎么通过 HTTP API 通信？ACP 协议怎么连接 Claude Code 等外部 Agent？