ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

agno AgentOS 如何配置 cron 定时任务并查看每次执行的历史记录?

agno AgentOS 如何配置 cron 定时任务并查看每次执行的历史记录? agno AgentOS 如何配置 cron 定时任务并查看每次执行的历史记录【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno如果你的 Agent 部署在 agno 的 AgentOS 上需要在固定时间自动触发一个 Agent / Team / Workflow 的 run例如每天 9 点生成日报并且要能查到每一次触发的执行结果那么 AgentOS 的 scheduler 模块就是对应的功能它可以持久化 cron 计划、按轮询间隔认领到期的任务、调用 AgentOS 端点并把每次执行尝试写入历史表。本文基于 cookbook/05_agent_os/12_scheduler 中的官方示例给出一条连续路径启动带 scheduler 的 AgentOS → 创建 cron 计划 → 通过 REST 或 Python 查看每次执行的历史记录。准备条件文档要求的环境来自 README启动 Postgres。官方示例通过 Docker 脚本启动agnohq/pgvector:18映射到本地 5532 端口./cookbook/scripts/run_pgvector.sh导出 OpenAI key示例中的 Agent 使用gpt-5.5模型export OPENAI_API_KEY...注意03_manage_with_python.py不调用模型只需要 Postgres其余示例会发起真实的gpt-5.5调用。示例统一使用.venvs/demo/bin/python作为解释器如果你没有这个虚拟环境需要替换为你自己装好 agno 的 Python。选择存储时按部署形态区分文档 Deployment notes 明确给出SQLite 只适合本地开发和单个本地 scheduler 进程多个 AgentOS worker 共享计划时必须用 Postgres它的认领操作基于FOR UPDATE SKIP LOCKED的原子更新保证不同 worker 不会执行同一条到期记录。启动带 scheduler 的 AgentOS主路径直接使用 01_run_in_agentos.py它演示了完整的配置面。关键代码是db PostgresDb( idscheduler-agent-os-db, db_urlDATABASE_URL, # 默认 postgresqlpsycopg://ai:ailocalhost:5532/ai schedules_tableagent_os_scheduler_schedules, schedule_runs_tableagent_os_scheduler_runs, ) agent_os AgentOS( idOS_ID, descriptionAgentOS with a Postgres-backed schedule poller., agents[scheduled_agent], dbdb, schedulerTrue, scheduler_poll_intervalSCHEDULER_POLL_INTERVAL_SECONDS, # 5 scheduler_base_urlBASE_URL, # 默认 http://127.0.0.1:7777 ) app agent_os.get_app()其中几个配置的用途均来自文档schedulerTrue打开内置轮询器scheduler_poll_interval是轮询间隔示例为 5 秒。scheduler_base_url是进程内执行器回调 AgentOS 时使用的地址默认http://127.0.0.1:7777AgentOS 监听在其他地址时要显式设置。开启 scheduling 后AgentOS 会在未提供internal_service_token时创建一个内部服务令牌执行器以 bearer 凭证发送它。该令牌只用于 scheduler 到 AgentOS 的内部流量不是终端用户 API key显式提供时要当作机密保管。每个被调用的 run 端点其 payload 里必须包含message字段。执行器会强制把计划内的 Agent / Team / Workflow 调用设为streamfalse、backgroundtrue然后轮询持久化的 run 直到到达终态。运行方式# 启动服务端会先种入一条 * * * * * 的分钟级计划 .venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/01_run_in_agentos.py # 另一个终端运行观察器 .venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/01_run_in_agentos.py --demoseed_schedule()通过ScheduleManager.create()创建计划参数包括name、cron示例为* * * * *、endpoint/agents/scheduled-greeter/runs、payload必须带message示例还带了session_id、timezone和timeout_seconds示例 120 秒。轮询器每 5 秒检查一次认领下一个自然到期的分钟触发 Agent 并在历史表中持久化结果。查看每次执行的历史记录历史记录的 REST 路由是GET /schedules/{schedule_id}/runs返回对象而不是裸列表顶层只有data和meta两个键分页参数是page不是offset配合limit使用。02_rest_api.py 展示了完整的 REST 生命周期在01_run_in_agentos.py服务端仍在运行时执行.venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/02_rest_api.py核心请求形态如下# 创建计划注意REST 字段名是 cron_expr不是 cron client.post(/schedules, json{ name: SCHEDULE_NAME, cron_expr: 0 0 1 1 *, endpoint: f/agents/{AGENT_ID}/runs, payload: {message: ..., session_id: rest-scheduler-session}, timezone: UTC, timeout_seconds: 120, max_retries: 1, retry_delay_seconds: 5, }) # 手动触发一次不需要等到 cron 到期 client.post(f/schedules/{schedule_id}/trigger) # 分页读取执行历史 client.get(f/schedules/{schedule_id}/runs, params{limit: 1, page: 1}) client.get(f/schedules/{schedule_id}/runs, params{limit: 1, page: 2})其余可用路由GET /schedules列表、GET /schedules/{id}详情、PATCH /schedules/{id}更新、POST /schedules/{id}/enable/disable、DELETE /schedules/{id}。验证方式脚本内置的断言 TEST_LOG 中记录的实际运行结果均为文档示例01_run_in_agentos.py --demo的观察器会轮询 runs 路由等待最多约 200 秒覆盖分钟边界、5 秒轮询间隔、120 秒 run 超时和少量余量出现一条状态为success的新历史记录若终态是failed等其他状态则抛出错误。成功时打印Schedule run、Agent run对应的 Agent run id、History total和Status。02_rest_api.py连续触发两次后断言两次记录分别出现在page1和page2meta.total_count 2且两页的 run id 不同。TEST_LOG 中记录的实际运行结果为GET /health返回ok两次触发的记录分别落到了第 1、2 页顶层键恰好是data和meta。用 Python 直接管理计划可选路径不经过 HTTP 时可以用ScheduleManager同步用PostgresDb异步用AsyncPostgresDb示例见 03_manage_with_python.py.venvs/demo/bin/python cookbook/05_agent_os/12_scheduler/03_manage_with_python.py这条路径有两个容易踩的坑文档明确说明字段名不一致create()接收cron但更新走数据库字段名所以更新 cron 必须传cron_expr...。示例先disable()再update(cron_expr...)最后enable()让next_run_at按新 cron 重新计算。验证行为非法 cron 表达式、不存在的时区、重复的计划名都会抛出ValueError示例验证了这三类错误。查询历史用get_runs(schedule_id, limit..., page...)从未执行过的计划查不到任何记录示例对此做了断言。边界与限制时区timezone使用 IANA 名称示例用UTC、America/New_York传不存在的名会校验失败。并发安全多 worker 共享计划必须用 Postgres见准备条件一节。执行语义计划触发永远是后台非流式 run执行器轮询到终态为止status的可能取值在观察器代码中体现为success/failed/paused/timeout非 success 时记录里带error字段。重试配置max_retries和retry_delay_seconds在创建时传入REST 和ScheduleManager.create()均支持示例值为 1 次重试、5 秒延迟timeout_seconds控制单次 run 的超时。文档还提供了一个 agentic 路径 04_scheduler_tools_agent.py给 Agent 挂SchedulerTools工具用自然语言创建计划并通过SchedulerTools子类收窄 create schema让调用方无法覆盖配置好的默认 endpoint 和 payload。它属于可选玩法历史查看方式与本文 REST 路径相同。完成上述路径后你的验收标准就是GET /schedules/{id}/runs返回的data里能找到对应触发记录status为success且meta.total_count随触发次数递增。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表