Hermes Agent 作为 Nous Research 推出的自进化 AI 代理，其「Scheduled automations」（定时自动化）模块是核心差异化特性之一——它并非简单封装系统 Cron，而是将传统定时调度与 AI 代理的记忆、技能、跨平台能力深度融合，实现了「自然语言定义任务、AI 编排执行、跨平台投递、自优化迭代」的全链路自动化。本文将从架构设计、核心算法、代码实现、业务流程四个维度，完整拆解这一模块的技术实现。

一、背景：为什么 Hermes Agent 的定时任务与众不同？

传统 Cron 调度仅能执行预设脚本/命令，存在三大痛点：

1. 任务定义门槛高（需手动编写 Cron 表达式）；
2. 执行逻辑固定，无动态调整能力；
3. 结果投递孤立，无 AI 记忆/上下文联动。

Hermes Agent 针对这些痛点重构了定时任务系统：

• 自然语言转 Cron 规则，降低使用门槛；
• AI 驱动的任务编排与异常自愈；
• 跨平台结果投递 + 记忆联动，形成「执行-反馈-优化」闭环；
• 子代理并行执行，支撑复杂任务链。

二、核心架构：分层设计与模块交互

Hermes Agent 定时任务系统采用四层分层架构，各模块解耦且可扩展：

层级	核心模块	职责
交互层	cli.py/网关模块	接收用户自然语言指令（CLI/电报/Discord 等），返回任务执行状态
解析层	cron/parser.py	LLM 解析自然语言 → 标准 Cron 表达式 + 任务元数据（执行逻辑/投递平台）
调度层	cron/scheduler.py	基于时间轮算法管理定时任务，触发执行并做异常兜底
执行层	agent//gateway/	子代理并行执行任务链，跨平台投递结果，同步执行状态到记忆模块

模块交互流程：

三、关键算法与核心代码实现

以下基于 Hermes Agent 技术栈（Python、APScheduler、LLM 调用、FTS5 记忆），拆解核心算法的实现逻辑，并提供符合仓库风格的核心代码片段（注：代码基于仓库文档/生态推导，贴合实际实现逻辑）。

3.1 算法1：自然语言转 Cron 规则（LLM 解析）

算法原理

核心是通过 Prompt 引导 LLM 提取「时间特征」「执行动作」「投递平台」三大维度信息，映射为标准 Cron 表达式 + 任务元数据，并通过语法校验保证合法性。

核心代码实现

# cron/parser.py
import re
from hermes.llm.client import get_llm_client  # Hermes 内置LLM客户端
from croniter import croniter  # Cron语法校验

class NL2CronParser:
    def __init__(self):
        self.llm_client = get_llm_client()
        self.cron_pattern = re.compile(r'^(\*|\d+|\d+-\d+|\d+/\d+|\?|\L+|\#)+(\s+(\*|\d+|\d+-\d+|\d+/\d+|\?|\L+|\#)){4}$')

    def build_prompt(self, user_query: str) -> str:
        """构建Prompt，引导LLM输出结构化结果"""
        prompt = f"""
        你需要将用户的自然语言定时任务指令解析为JSON格式，包含以下字段：
        1. cron_expression: 标准5位Cron表达式（分 时 日 月 周），无法解析时返回null；
        2. task_action: 具体执行的动作（如"备份聊天记录"、"生成周报"）；
        3. delivery_platform: 结果投递平台（如telegram/discord/cli）；
        4. task_priority: 优先级（1-5，5最高）。

        用户指令：{user_query}
        输出要求：仅返回JSON，无多余文字。
        示例：
        用户指令：每天晚上8点生成上周工作周报并发送到Telegram
        输出：{{"cron_expression":"0 20 * * *","task_action":"生成上周工作周报","delivery_platform":"telegram","task_priority":3}}
        """
        return prompt

    def parse(self, user_query: str) -> dict:
        """核心解析逻辑：LLM生成 + Cron校验"""
        # 1. 调用LLM获取结构化结果
        response = self.llm_client.generate(
            prompt=self.build_prompt(user_query),
            temperature=0.1,  # 低温度保证确定性
            max_tokens=500
        )
        try:
            result = eval(response)  # 解析JSON（Hermes实际用json.loads，此处简化）
        except:
            raise ValueError("LLM输出格式错误，无法解析")
        
        # 2. Cron表达式合法性校验
        cron_expr = result.get("cron_expression")
        if not cron_expr or not croniter.is_valid(cron_expr):
            # 兜底：让LLM重新生成
            retry_prompt = f"你生成的Cron表达式 {cron_expr} 不合法，请重新生成符合5位标准的表达式：{user_query}"
            retry_response = self.llm_client.generate(prompt=retry_prompt)
            result["cron_expression"] = eval(retry_response)["cron_expression"]
        
        return result

# 调用示例
if __name__ == "__main__":
    parser = NL2CronParser()
    user_query = "每周一早上9点备份我的聊天记录并发送到Discord"
    parsed_result = parser.parse(user_query)
    print(parsed_result)
    # 输出：{"cron_expression":"0 9 * * 1","task_action":"备份聊天记录","delivery_platform":"discord","task_priority":4}

3.2 算法2：时间轮调度算法（Time Wheel）

算法原理

传统 Cron 调度采用「轮询检查」方式，大量任务时性能损耗大；时间轮算法将时间划分为「轮盘刻度」，任务按触发时间挂载到对应刻度，调度器仅需遍历当前刻度的任务，时间复杂度从 O(n) 降至 O(1)，是 Hermes 调度层的核心优化。

核心代码实现

# cron/scheduler.py
import time
import threading
from dataclasses import dataclass
from typing import Dict, List, Callable

@dataclass
class ScheduledTask:
    """任务元数据类"""
    task_id: str
    cron_expr: str
    task_action: str
    priority: int
    execute_func: Callable  # 任务执行函数
    delivery_platform: str

class TimeWheelScheduler:
    def __init__(self, tick_interval: int = 1):
        self.tick_interval = tick_interval  # 时间轮刻度间隔（秒）
        self.wheel_size = 60  # 时间轮大小（60个刻度，适配分钟级调度）
        self.wheel: Dict[int, List[ScheduledTask]] = {i: [] for i in range(self.wheel_size)}
        self.current_tick = 0
        self.running = False
        self.lock = threading.Lock()

    def add_task(self, task: ScheduledTask):
        """将任务添加到时间轮对应刻度"""
        with self.lock:
            # 计算任务触发时间对应的刻度（简化版：取Cron表达式的分钟位）
            cron_minute = task.cron_expr.split()[0]
            target_tick = int(cron_minute) if cron_minute != "*" else self.current_tick
            self.wheel[target_tick].append(task)
            # 按优先级排序（高优先级先执行）
            self.wheel[target_tick].sort(key=lambda x: x.priority, reverse=True)

    def _run_tick(self):
        """执行当前刻度的所有任务"""
        with self.lock:
            current_tasks = self.wheel[self.current_tick]
            if current_tasks:
                # 启动子线程执行任务（避免阻塞调度器）
                for task in current_tasks:
                    threading.Thread(
                        target=self._execute_task,
                        args=(task,),
                        daemon=True  # 守护线程，不阻塞主进程
                    ).start()
                # 清空当前刻度任务（重复任务会重新添加）
                self.wheel[self.current_tick] = []

    def _execute_task(self, task: ScheduledTask):
        """执行单个任务，包含异常兜底"""
        try:
            # 调用子代理执行任务动作
            result = task.execute_func(task.task_action)
            # 跨平台投递结果（调用gateway模块）
            from gateway.delivery import DeliveryRouter
            DeliveryRouter().deliver(result, task.delivery_platform)
            # 同步执行状态到记忆模块
            from hermes.memory import MemoryManager
            MemoryManager().write(
                content=f"定时任务 {task.task_id} 执行成功：{result}",
                tags=["scheduled_task", task.delivery_platform]
            )
        except Exception as e:
            # 异常自愈：调用LLM分析错误并重试
            error_msg = f"任务 {task.task_id} 执行失败：{str(e)}"
            self._retry_task(task, error_msg)

    def _retry_task(self, task: ScheduledTask, error_msg: str):
        """AI驱动的任务重试"""
        from hermes.llm.client import get_llm_client
        llm_client = get_llm_client()
        retry_prompt = f"""
        定时任务执行失败，错误信息：{error_msg}
        任务动作：{task.task_action}
        请分析错误原因，并返回修正后的执行逻辑（仅返回Python可执行代码片段）。
        """
        fix_code = llm_client.generate(prompt=retry_prompt)
        try:
            # 执行修正后的逻辑
            exec(fix_code)
            # 同步重试结果到记忆
            MemoryManager().write(content=f"任务 {task.task_id} 重试成功", tags=["scheduled_task"])
        except:
            # 最终兜底：记录失败状态
            MemoryManager().write(content=f"任务 {task.task_id} 重试失败：{error_msg}", tags=["scheduled_task"])

    def start(self):
        """启动时间轮调度器"""
        self.running = True
        while self.running:
            self._run_tick()
            time.sleep(self.tick_interval)
            self.current_tick = (self.current_tick + 1) % self.wheel_size

    def stop(self):
        """停止调度器"""
        self.running = False

# 调度器使用示例
if __name__ == "__main__":
    scheduler = TimeWheelScheduler()
    # 定义任务执行函数
    def backup_chat():
        return "聊天记录备份完成，路径：/data/backup/chat_20240501.tar.gz"
    # 创建任务
    task = ScheduledTask(
        task_id="backup_chat_001",
        cron_expr="0 20 * * *",
        task_action="备份聊天记录",
        priority=3,
        execute_func=backup_chat,
        delivery_platform="telegram"
    )
    # 添加任务并启动调度器
    scheduler.add_task(task)
    threading.Thread(target=scheduler.start, daemon=True).start()

3.3 算法3：子代理并行执行与任务链编排

算法原理

复杂定时任务（如“拉取数据→统计分析→生成报告→投递”）需要多步骤编排，Hermes Agent 采用「子代理隔离 + 异步回调 + 任务链编排」策略：

1. 每个子任务启动独立子代理进程（避免单任务崩溃影响全局）；
2. 基于 RPC 调用子代理，支持串行/并行任务链；
3. 超时重试 + 回调通知，保证任务链完整性。

核心代码实现

# agent/subagent/executor.py
import multiprocessing
import time
from typing import List, Callable
from hermes.agent.base import BaseAgent

class TaskChainExecutor:
    """任务链执行器：支持串行/并行子代理任务"""
    def __init__(self):
        self.subagents: Dict[str, BaseAgent] = {}  # 子代理池

    def create_subagent(self, agent_id: str, skills: List[str]) -> BaseAgent:
        """创建隔离的子代理（进程级隔离）"""
        subagent = BaseAgent(agent_id=agent_id, skills=skills)
        # 启动子代理进程
        subagent_process = multiprocessing.Process(
            target=subagent.run,
            daemon=True
        )
        subagent_process.start()
        self.subagents[agent_id] = subagent
        return subagent

    def execute_parallel(self, tasks: List[dict]) -> List[dict]:
        """并行执行子任务"""
        results = []
        # 使用进程池执行并行任务
        with multiprocessing.Pool(processes=len(tasks)) as pool:
            for task in tasks:
                subagent = self.create_subagent(
                    agent_id=f"subagent_{task['id']}",
                    skills=task["skills"]
                )
                # 异步提交任务
                result = pool.apply_async(
                    subagent.execute,
                    args=(task["action"],),
                    callback=lambda res: results.append({"task_id": task["id"], "result": res})
                )
            # 等待所有任务完成
            pool.close()
            pool.join()
        return results

    def execute_sequential(self, tasks: List[dict]) -> List[dict]:
        """串行执行子任务（上一步结果作为下一步输入）"""
        results = []
        prev_result = None
        for task in tasks:
            subagent = self.create_subagent(
                agent_id=f"subagent_{task['id']}",
                skills=task["skills"]
            )
            # 串行执行，传递上一步结果
            current_result = subagent.execute(task["action"], input_data=prev_result)
            results.append({"task_id": task["id"], "result": current_result})
            prev_result = current_result
            # 销毁子代理，释放资源
            self.subagents[task["id"]].stop()
            del self.subagents[task["id"]]
        return results

# 任务链使用示例
if __name__ == "__main__":
    executor = TaskChainExecutor()
    # 定义周报生成任务链（并行拉取数据 + 串行生成报告）
    parallel_tasks = [
        {"id": "pull_data_1", "skills": ["data_extract"], "action": "拉取本周工作记录"},
        {"id": "pull_data_2", "skills": ["data_extract"], "action": "拉取本周项目进度"}
    ]
    sequential_tasks = [
        {"id": "stat_analysis", "skills": ["data_analysis"], "action": "统计本周工作时长"},
        {"id": "generate_report", "skills": ["report_writing"], "action": "生成周报"}
    ]

    # 1. 并行拉取数据
    parallel_results = executor.execute_parallel(parallel_tasks)
    # 2. 串行生成报告（传入并行结果）
    sequential_tasks[0]["action"] += f"，输入数据：{parallel_results}"
    sequential_results = executor.execute_sequential(sequential_tasks)
    print("周报生成完成：", sequential_results[-1]["result"])

3.4 算法4：跨平台投递的路由与格式适配

算法原理

不同平台（Telegram/Discord/CLI）的消息格式、长度限制不同，Hermes Agent 采用「路由算法 + 格式适配算法」保证结果正确投递：

1. 路由算法：基于任务标签/用户默认配置，将结果路由到对应网关；
2. 格式适配：自动裁剪长文本、转换文件格式、适配富文本规则。

核心代码实现

# gateway/delivery.py
import os
from typing import Dict
from hermes.gateway.telegram import TelegramGateway
from hermes.gateway.discord import DiscordGateway
from hermes.gateway.cli import CLIGateway

class DeliveryRouter:
    """跨平台投递路由器"""
    def __init__(self):
        # 初始化各平台网关（从配置读取密钥）
        self.gateways: Dict[str, BaseGateway] = {
            "telegram": TelegramGateway(api_key=os.getenv("TELEGRAM_API_KEY")),
            "discord": DiscordGateway(bot_token=os.getenv("DISCORD_BOT_TOKEN")),
            "cli": CLIGateway()
        }
        # 平台格式限制配置
        self.platform_limits = {
            "telegram": {"max_msg_len": 4096, "support_file": True},
            "discord": {"max_msg_len": 2000, "support_rich_text": True},
            "cli": {"max_msg_len": None, "support_file": False}
        }

    def adapt_content(self, content: str, platform: str) -> dict:
        """格式适配算法：根据平台限制调整内容"""
        limits = self.platform_limits[platform]
        adapted = {"type": "text", "content": content}
        
        # 1. 文本长度适配
        if limits["max_msg_len"] and len(content) > limits["max_msg_len"]:
            if limits["support_file"]:
                # 长文本转为文件
                file_path = "/tmp/hermes_task_result.txt"
                with open(file_path, "w") as f:
                    f.write(content)
                adapted = {"type": "file", "path": file_path}
            else:
                # 裁剪文本并添加摘要
                summary = self._generate_summary(content)
                adapted["content"] = f"{content[:limits['max_msg_len']-100]}...\n\n摘要：{summary}"
        
        # 2. 富文本适配（仅Discord）
        if platform == "discord" and limits["support_rich_text"]:
            adapted["content"] = self._convert_to_rich_text(adapted["content"])
        
        return adapted

    def _generate_summary(self, content: str) -> str:
        """调用LLM生成长文本摘要"""
        from hermes.llm.client import get_llm_client
        return get_llm_client().generate(prompt=f"生成以下文本的简短摘要（50字内）：{content}", max_tokens=50)

    def _convert_to_rich_text(self, content: str) -> str:
        """转换为Discord富文本格式"""
        # 示例：将markdown转为Discord嵌入格式
        content = content.replace("**", "**")  # Discord支持markdown粗体
        return f"```markdown\n{content}\n```"

    def deliver(self, content: str, platform: str):
        """核心投递逻辑：路由 + 适配 + 发送"""
        if platform not in self.gateways:
            raise ValueError(f"不支持的投递平台：{platform}")
        
        # 1. 格式适配
        adapted_content = self.adapt_content(content, platform)
        # 2. 路由到对应网关发送
        gateway = self.gateways[platform]
        if adapted_content["type"] == "text":
            gateway.send_message(adapted_content["content"])
        elif adapted_content["type"] == "file":
            gateway.send_file(adapted_content["path"])

# 投递示例
if __name__ == "__main__":
    router = DeliveryRouter()
    # 长文本结果投递到Telegram
    long_report = """（此处省略10000字的工作周报）"""
    router.deliver(long_report, "telegram")  # 自动转为文件发送
    # 短消息投递到CLI
    short_msg = "聊天记录备份完成！"
    router.deliver(short_msg, "cli")  # 直接输出到终端

3.5 算法5：记忆联动与技能迭代

算法原理

Hermes Agent 将定时任务的执行状态（成功/失败/结果）写入 FTS5 索引的持久化记忆，通过「周期性记忆梳理」和「技能迭代算法」，自动优化后续任务：

1. FTS5 索引：快速检索任务执行历史；
2. 记忆梳理：定期总结任务执行规律（如“每周三备份失败，原因是磁盘不足”）；
3. 技能迭代：从重复任务中提取通用逻辑，生成可复用技能。

核心代码实现

# hermes/memory/scheduled_task_memory.py
import sqlite3
from datetime import datetime, timedelta

class ScheduledTaskMemory:
    def __init__(self, db_path: str = "hermes_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_table()

    def _init_table(self):
        """初始化FTS5索引表"""
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS scheduled_tasks USING fts5(
                task_id,
                execute_time,
                status,  # success/failed/retry
                result,
                error_msg,
                platform
            );
        """)
        self.conn.commit()

    def write(self, task_id: str, status: str, result: str = "", error_msg: str = "", platform: str = ""):
        """写入任务执行状态"""
        execute_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        self.conn.execute("""
            INSERT INTO scheduled_tasks (task_id, execute_time, status, result, error_msg, platform)
            VALUES (?, ?, ?, ?, ?, ?);
        """, (task_id, execute_time, status, result, error_msg, platform))
        self.conn.commit()

    def search(self, query: str) -> list:
        """FTS5检索任务历史"""
        cursor = self.conn.execute("""
            SELECT * FROM scheduled_tasks WHERE scheduled_tasks MATCH ?;
        """, (query,))
        return cursor.fetchall()

    def summarize_task_patterns(self, days: int = 7):
        """周期性记忆梳理：总结近N天任务执行规律"""
        start_time = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
        # 1. 查询近N天失败任务
        failed_tasks = self.conn.execute("""
            SELECT task_id, error_msg FROM scheduled_tasks
            WHERE status = 'failed' AND execute_time >= ?;
        """, (start_time,)).fetchall()
        
        # 2. 调用LLM总结规律
        from hermes.llm.client import get_llm_client
        llm_client = get_llm_client()
        summary_prompt = f"""
        以下是近{days}天定时任务的失败记录：{failed_tasks}
        请总结失败规律（如重复失败的任务、共性原因），并给出优化建议。
        """
        summary = llm_client.generate(prompt=summary_prompt)
        
        # 3. 写入记忆，供后续技能迭代使用
        self.write(
            task_id="memory_summary",
            status="success",
            result=summary,
            platform="internal"
        )
        return summary

    def generate_skill_from_tasks(self, task_id_prefix: str):
        """技能迭代：从重复任务中提取通用技能"""
        # 1. 检索相同前缀的任务执行记录
        task_records = self.search(f"task_id:{task_id_prefix}")
        # 2. 调用LLM生成技能代码
        skill_prompt = f"""
        以下是重复执行的定时任务记录：{task_records}
        请提取通用执行逻辑，生成Hermes Agent技能代码（符合agentskills.io标准）。
        技能名称：{task_id_prefix}_skill
        技能描述：自动执行{task_id_prefix}相关定时任务
        """
        skill_code = llm_client.generate(prompt=skill_prompt)
        # 3. 注册技能到Agent
        from hermes.skills.registry import SkillRegistry
        SkillRegistry().register_skill(
            skill_name=f"{task_id_prefix}_skill",
            skill_code=skill_code
        )
        return f"技能 {task_id_prefix}_skill 生成并注册成功"

# 记忆使用示例
if __name__ == "__main__":
    memory = ScheduledTaskMemory()
    # 1. 写入任务执行状态
    memory.write(
        task_id="backup_chat_001",
        status="failed",
        error_msg="磁盘空间不足，备份失败",
        platform="telegram"
    )
    # 2. 总结近7天失败规律
    summary = memory.summarize_task_patterns(days=7)
    print("失败规律总结：", summary)
    # 3. 从重复备份任务生成技能
    memory.generate_skill_from_tasks(task_id_prefix="backup_chat")

四、完整业务流程：从用户指令到任务自优化

以「每周一9点生成工作周报并投递到Telegram」为例，拆解全链路业务流程：

步骤1：用户输入自然语言指令

用户在 Telegram 发送：每周一早上9点生成上周工作周报并发送到我的Telegram

步骤2：解析层处理

1. CLI/网关模块接收指令，转发到 cron/parser.py；
2. LLM 解析指令，生成：
   • Cron 表达式：0 9 * * 1；
   • 任务动作：生成上周工作周报；
   • 投递平台：telegram；
   • 优先级：3；
3. 校验 Cron 表达式合法性，返回解析结果。

步骤3：调度层注册任务

1. cron/scheduler.py 将任务添加到时间轮对应刻度（9点0分）；
2. 调度器以守护线程运行，监听时间轮刻度变化。

步骤4：任务触发与执行

1. 每周一9点0分，时间轮触发当前刻度任务；
2. 调度器启动子线程，调用 TaskChainExecutor 执行任务链：
   • 并行子任务：子代理1拉取本周工作记录、子代理2拉取项目进度；
   • 串行子任务：子代理3统计工作时长、子代理4生成周报；
3. 执行过程中，异常则触发 LLM 错误分析 + 重试。

步骤5：跨平台投递

1. DeliveryRouter 接收周报结果，检测到内容长度超过 Telegram 限制（4096字）；
2. 自动将周报转为 hermes_weekly_report.txt 文件；
3. 路由到 Telegram 网关，发送文件到用户聊天窗口。

步骤6：记忆联动与技能迭代

1. ScheduledTaskMemory 写入任务执行状态（成功/失败）；
2. 每周日，Agent 执行「记忆梳理技能」，总结近7周周报生成规律；
3. 重复执行4次后，Agent 提取通用逻辑，生成「周报生成技能」，后续任务直接调用该技能（无需重复LLM推理）。

Hermes Agent 定时任务系统的核心价值，是将「传统 Cron 调度」升级为「AI 驱动的全链路自动化」——它不仅解决了“什么时候执行”的问题，更通过自然语言解析、子代理编排、记忆联动，解决了“执行什么”“怎么优化执行”的问题。

这一设计思路也为 AI 代理的定时任务场景提供了参考：调度器不再是简单的时间触发器，而是与 AI 记忆、技能、跨平台能力深度融合的核心组件，最终实现“一次指令，持续优化，全场景落地”的自进化能力。

未来，随着 LLM 解析精度和技能系统的迭代，Hermes Agent 有望进一步降低定时任务的使用门槛，让非技术用户也能通过自然语言定义复杂的跨平台自动化流程。