agno Agent Working State 实战基于 FileSystem 检查点文件实现跨会话任务续跑与监控基线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南围绕 agno当前仓库GitHub_Trending/ag/agnocookbook 中 03_working_state 示例目录 及其 TEST_LOG.md 的真实测试记录展开。你会掌握一个核心能力让 Agent 把做到哪一步写进持久化文件在下一次会话、下一次进程甚至下一个调度周期中从断点无缝续跑而完全不依赖会话历史。文中给出的两套模式——任务检查点checkpoint与上次见过值监控last-seen baseline都是经过 gpt-5.5 agno 2.8.1 实际验证、可复制运行的完整方案。一、为什么需要 Working State会话会死文件不会在 agno 中会话状态随会话消亡一次对话结束后会话内的上下文、记忆都不再可用而定时调度的 Agent 每次运行拿到的是一个全新的会话fresh session per run。如果任务比一次运行更长例如数据迁移、审计、回填或者两次运行之间观测数据发生了变化例如延迟监控Agent 就需要一种比会话更长寿、比进程更持久的状态载体。Working State 模式的答案非常朴素把进度放进文件。Agno 的 FileSystem 为每个 Agent 提供一套私有、持久的文件系统底层由可插拔的BaseFS后端实现默认是数据库。从 Agent 视角看它就是一组普通文件工具从存储视角看文件以(namespace, path)为粒度落库天然跨越会话与进程存活。关键区别原文 README 的总结会话状态随会话结束而消失而检查点文件两者都扛得住。03_working_state的两个示例都用每次运行新建的 SQLite 文件以便演示从零开始真实部署时应该固定一个共享数据库让状态比进程活得更久——这正是 01_getting_started/basic.py 演示的模式。适用场景依据 README以下场景适合 Working State多轮运行任务迁移、审计、回填——凡是你会用任务队列做 checkpoint 的事情都可以交给 Agent 用文件方式完成可重启的部署同一模式配合一个固定共享数据库进程重启后状态仍在监控与看护类 Agent对变化发出告警此时上次看到的值属于 Agent 的工作状态working state而非用户记忆精确去重哪些记录我已经处理过则不属于本目录职责应改用 02_durable_records/ 的check_linesFileSystem 的基础挂载方式见 01_getting_started/。二、测试验证基准在什么环境、以什么标准确认这两个示例可用在深入代码之前先交代 TEST_LOG.md 记录的验证基准测试日期2026-07-24模型gpt-5.5通过OpenAIResponses调用agno 版本2.8.1源码树分支feat/agent-fs提交7df2fad3a结论该目录下每个文件均为 PASS测试方式测试日志引用真实的工具调用与打印的状态内容模型措辞每次运行会变化日志中已改写而非逐字引用。这意味着下面看到的工具调用序列list_files→write_file、read_file→write_file不是设计文档的想象而是 gpt-5.5 在 agno 2.8.1 上的真实行为。模型输出虽每次略有差异但行为契约是稳定的先读检查点、执行增量、覆写检查点。三、模式一任务检查点basic.py——四步迁移分两轮跑3.1 问题设定basic.py 模拟一个数据迁移任务共四步1. Export the users table 2. Export the orders table 3. Verify row counts match 4. Write the summary report约束是每个会话只允许做两步。第一次会话做完步骤 1、2 后必须把进度写入state/checkpoint.md第二次会话是全新会话session_id 不同、与第一次没有任何共享历史它必须通过读取该文件得知已做到第 2 步从而从第 3 步继续而不是从头重跑。3.2 代码拆解1创建 FileSystem。关键点是数据库文件使用uuid4().hex随机命名保证每次执行都从空存储开始、演示可复现DB_FILE ftmp/agent_fs_checkpoint_{uuid4().hex}.db fs FileSystem(SqliteDb(db_fileDB_FILE))源码注释明确提示真实可续跑任务应固定一个共享数据库让检查点比进程活得更久而不只是比会话活得更久。2创建 Agent。挂载fs.tools()并把fs.instructions()与自己的指令组合进instructions这正是 fs.py 推荐的挂载方式工具不带自带指令system prompt 的措辞由开发者掌控agent Agent( modelOpenAIResponses(idgpt-5.5), tools[fs.tools()], instructions[ You run a data migration with these steps:\n \n.join(STEPS) \n Each session you have time for exactly TWO steps. Read state/checkpoint.md first (it may not exist on the first run) to see what is already done. Perform the next two pending steps (performing describing the work as done), then overwrite state/checkpoint.md with the full list of completed steps using write_file. Reply with which steps you completed this session., fs.instructions(), ], )这里的指令把协议说透了先读文件可能不存在、做两步、用write_file覆写完整清单。fs.instructions()提供的默认指引见 fs.py其中包含路径规范相对路径、notes/decisions.md风格、文件维护约定先读后改、用replace_lines原地修正而非追加矛盾内容、搜索优先先search_content再read_file以及绝不存密钥/密码/API Key的安全红线。3两次会话。使用两个互不相关的session_id杜绝任何会话级上下文泄漏agent.print_response(Continue the migration., session_idmigration-1) print(fs.read(state/checkpoint.md)) # 编程式读取用于打印验证 agent.print_response(Continue the migration., session_idmigration-2) print(fs.read(state/checkpoint.md))注意fs.read()是 FileSystem 的编程式 APIfs.py文件不存在时返回None它在这里只用于人的视角打印验证Agent 自己的读写在工具调用层完成。3.3 测试日志观测到的真实工具调用根据 TEST_LOG.mdgpt-5.5 的实际行为是Session 1migration-1先调用list_files(directorystate, patterncheckpoint.md, ...)——不是直接read_file而是先探测检查点文件是否存在文件尚不存在直接读会得到 file not found随后调用write_file(pathstate/checkpoint.md, content1. Export the users table\n2. Export the orders table, overwriteTrue)打印出的检查点内容精确等于这两步。Session 2migration-2全新会话第一步就是read_file(pathstate/checkpoint.md, ...)从文件中得知进度完成后把四步全部写回打印出的检查点列出步骤 1~4——Session 2 从第 3 步续跑而不是重头开始。这个序列给出了一个可复用的工程细节探测文件是否存在用list_files globpatterncheckpoint.md而不是直接read_file因为read_file对缺失文件返回错误字符串。Toolkit 层确实如此实现toolkit.py 中read_file在内容为None时返回Error: file not found: ...。四、模式二Last-Seen 监控last_seen_monitor.py——只报告发生了什么变化4.1 问题设定last_seen_monitor.py 是一个按计划运行的延迟监控器它不关心当前 p95 延迟的绝对值而是对比本次读数与上次运行记录在state/last-run.md的基线只标记相对上次变化超过 20% 的服务然后把本次读数覆写为新的基线。两次运行的输入数据READINGS_MONDAY {checkout-api: 210ms, billing-api: 180ms, search-api: 95ms} READINGS_TUESDAY {checkout-api: 540ms, billing-api: 185ms, search-api: 96ms}肉眼可见checkout-api从 210ms 涨到 540ms157.1%其余两个服务变化在 1ms 以内。正确的行为是只告警 checkout-api。4.2 代码拆解Agent 的指令明确写清了监控协议agent Agent( modelOpenAIResponses(idgpt-5.5), tools[fs.tools()], instructions[ You are a latency monitor that runs on a schedule. Each run you receive current p95 readings. Read state/last-run.md (it may not exist on the first run), then report which services changed by more than 20 percent since last run, or say that this is the baseline run. Finally overwrite state/last-run.md with the current readings using write_file, one service: value per line., fs.instructions(), ], )执行器把两次运行封装成一个函数每次使用独立session_id——完全模拟定时调度的行为每次运行全新会话比较只能来自存储的基线def run_monitor(readings, session_id): lines [name : value for name, value in readings.items()] agent.print_response(Current readings:\n \n.join(lines), session_idsession_id)同样地数据库文件使用随机名tmp/agent_fs_monitor_{uuid4().hex}.db让演示每次从空存储开始。源码注释点明了真实部署的坑若每次进程都新建存储监控器会忘记基线永远只输出 baseline永远不告警。定时监控必须固定同一个数据库。4.3 测试日志观测到的真实行为根据 TEST_LOG.mdRun 1monitor-monday没有历史读数报告这是 baseline run并把三个服务全部保存。Run 2monitor-tuesday调用read_file(pathstate/last-run.md, ...)恰好标记一个服务checkout-api: 210ms → 540ms (157.1%)明确说明没有其他服务变化超过 20%billing-api180→185ms和search-api95→96ms完全未被提及——模型没有把无关信息塞进报告这正是报告变化而非报告当前值的监控语义随后调用write_file(pathstate/last-run.md, ...)打印出的新基线为checkout-api: 540ms / billing-api: 185ms / search-api: 96ms。一个值得注意的细节测试日志中的变化百分比157.1%是模型在回复里计算的而阈值判断20%同样由模型依据指令完成。这意味着该模式没有硬编码的数值比较逻辑告警语义完全由instructions承载——好处是零配置、纯声明式代价是依赖模型的指令遵循能力因此fs.instructions()与业务指令的组合措辞需要精心编写这也解释了为什么 TEST_LOG 强调每次运行模型措辞略有差异。五、FileSystem 底层原理这套文件究竟存在哪里、怎么工作理解 Working State 能跨会话续跑需要知道 FileSystem 的几个设计点1存储后端是可插拔的。FileSystem接受一个BaseFS后端或任何它能识别的存储句柄例如SqliteDb/PostgresDb会被自动包装成 DbFileSystem。由于数据库表使用独立的fsschemaPostgres 下Agent 的文件与会话、记忆、评估数据在同一数据库内物理隔离便于单独备份或清理。FileSystem(SqliteDb(db_file...))这种写法正是示例采用的零后端导入便捷路径。2namespace 是隔离单元。同一个后端 同一个 namespace 同一批文件不同 namespace 完全隔离。namespace 还支持{user_id}、{agent_id}、{team_id}模板占位符运行时由框架注入上下文解析缺值时 fail-closed 抛错匿名运行绝不会悄悄塌缩进共享命名空间。对多租户系统这是把各用户的 working state 隔开的关键机制。3配额与安全默认值。构造参数max_file_bytes1_000_000单文件 1MB、max_namespace_bytes20_000_000单命名空间 20MB。写入超限会触发QuotaExceededErrorToolkit 层会把错误转成Error: ...字符串返回给模型工具错误永不抛出见 toolkit.py并引导模型拆小文件或归档而不是覆写仍可能需要的旧文件。4工具面与指令面分离。FileSystem.tools() 默认注册笔记七件套read_file、write_file、append_file、replace_lines、list_files、search_content、move_filedelete_file是破坏性操作必须allow_deleteTrue才注册check_lines批量精确行匹配去重原语需通过include_tools显式点名——这正是 02_durable_records/ 场景的工具。read_onlyTrue时只注册三个读工具配合instructions(read_onlyTrue)构成消费方 Agent 只读查阅他人 namespace的表面。而FileSystem.instructions()返回的是与 namespace 无关的使用指引用于与开发者自己的指令组合两个示例都用了这种组合方式。5覆写语义。write默认overwriteTruelast-writer-wins传expected_version可获得乐观并发控制overwriteFalse时文件已存在会抛FileExistsError。两个示例的覆写检查点文件正是默认语义的直接使用。六、运行与验证如何复现测试结果前置条件需要OPENAI_API_KEY两个示例都走OpenAIResponses默认模型gpt-5.5。python cookbook/13_filesystem/03_working_state/basic.py python cookbook/13_filesystem/03_working_state/last_seen_monitor.py运行后按测试日志的验证清单核对basic.pysession 1 打印的检查点恰好是步骤 1、2session 2migration-2打印的检查点是步骤 1~4且 Agent 的回复应表明它从第 3 步续跑last_seen_monitor.pyrun 1 输出 baselinerun 2 只标记checkout-api最终打印的基线为三个服务的最新值。若想验证真实部署形态状态跨进程存活把DB_FILE换成固定路径如tmp/agent_fs_checkpoint.db然后连续执行两次同一脚本——第二次运行应直接从检查点续跑这正是 01_getting_started/basic.py 演示的run twice, same DB模式。七、从示例到生产的三个注意事项结合 README 与源码注释落地时需把握演示用随机库、生产用固定库。两个示例的uuid4()数据库命名是为了每次演示从零开始真实定时任务/多进程部署必须固定一个共享数据库SQLite 或 Postgres否则状态随进程消亡checkpoint 与 baseline 全部失效。先探测再读取。检查点文件在首次运行时不存在测试日志显示模型用list_files(..., patterncheckpoint.md)探测而非直接read_file在自定义指令中明确文件可能不存在两个示例都写进了 instructions能显著降低首轮运行出错概率。监控语义写在指令里而非代码里。变化超过 20% 才告警是自然语言约定若需要严格的数值阈值与可审计的百分比计算应把比较逻辑放进工具函数agno 支持 callable tools让模型只负责读取与汇报这是从演示可用走向生产可靠的分水岭。八、小结Working State 是 agno FileSystem 在跨会话、跨进程、跨调度周期场景下的核心用法。通过 basic.py 与 last_seen_monitor.py 两个经过 gpt-5.5 实测 PASS 的示例本文还原了检查点续跑与 last-seen 监控的完整协议先读状态文件允许不存在→ 执行增量工作 → 用 write_file 覆写状态文件。这套模式把任务做到哪一步上次观测是什么从易逝的会话状态中剥离出来放进由 fs.py 与 toolkit.py 实现的持久化文件存储让 Agent 成为真正可续跑、可重启、可调度的工作单元。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考