Agent 领域的讨论最近明显从“哪个模型更强”转向了另一个更具工程价值的问题一个 AI 助手能不能不只在你主动唤醒它时工作而是自己“挂”在那里连续几天处理任务、等待事件、恢复上下文、持续推进OpenAI Astra 被披露能够连续运行数日表面看是“多模态助手又进步了”但更值得开发者关注的是背后正在发生的 Agent 运行范式变化从“提问—回答”的单次交互走向“常驻后台、跨时间持续执行”的持续智能体Persistent Agent。这种变化会直接影响我们怎么设计 Agent 的架构、怎么管理上下文、怎么做状态恢复和容错。这篇文章不打算只复述新闻。我想从工程角度拆解三层问题持续智能体和传统 Agent 到底差在哪里如果自己动手做一个“能连续运行”的最小 Agent最少需要哪些组件当 Agent 真正长期运行时哪些坑会从“边缘问题”变成“核心问题”。无论你是在做 RAG 应用、自动化脚本还是已经在用 Codex CLI、Agents SDK 这类工具这篇文章都值得读完。因为“持续运行”一旦成为默认需求很多原来可以靠重启解决的麻烦都会变成必须先设计好的工程约束。1. 从“单次问答”到“持续运行”Agent 范式正在变化先看过去几年 Agent 的主流形态。无论是基于 ReAct 的推理循环还是 Function Calling 驱动的任务编排核心流程基本都是用户输入一个请求Agent 规划步骤调用工具得到结果返回最终答案。任务结束上下文清零进程退出。这种模式本质上是“请求—响应”的延伸Agent 的存在时间是秒级或分钟级。早期这样设计没问题因为大部分 Agent 处理的都是明确边界内的短任务查个天气、写段代码、整理一份纪要。但一旦任务变成“每天盯着某个数据源发现异常就自动处理”“在项目仓库里持续跟进 issue 并给出修改建议”“作为私人助理长时间记住用户的偏好和工作节奏”单次问答模式就撑不住了。这里涉及一个经常被混淆的概念多轮对话不等于持续运行。多轮对话只是把多个“请求—响应”串在一起模型并不真正拥有跨会话的目标和状态而持续智能体意味着 Agent 有一个长期目标它能在没有人类干预的情况下自行推进能感知外部事件能主动决定下一步做什么还能在中断后恢复。用一个类比来说明单次 Agent 像叫一个临时工干完活就走下次换人什么都不记得持续 Agent 像一个常驻员工有自己的任务清单会自己推进工作被叫走时还能记住进度第二天回来接着干。很多开发者第一次接触“持续运行”这个概念时会觉得这不就是写一个while True循环吗表面看确实如此但真正难的是循环里那套状态管理、记忆分层、异常恢复和成本控制。这也是为什么“连续运行数日”能被当成一个行业信号——它意味着 Agent 的可靠性已经进步到可以脱离人工陪跑了。对开发者来说这个范式变化意味着三件事设计 Agent 时不能只写“输入—处理—输出”要为长期运行设计状态和记忆评估 Agent 时不能只看单次任务准确率要看它在长时间运行后的保持能力部署 Agent 时不能当普通脚本处理要考虑监控、恢复、权限边界和资源预算。2. 持续智能体的核心概念理解了范式变化我们来拆解持续智能体背后的几个核心概念。这些概念不是新造出来的而是把分布式系统、消息队列、数据库等领域的老问题搬到了 Agent 场景里重新组合。2.1 会话上下文与长期记忆会话上下文Context是模型在单次请求中能看到的信息窗口。普通 Agent 的上下文随着请求结束就没了持续 Agent 必须把有长期价值的信息抽出来存到外部记忆里。长期记忆通常分两层工作记忆当前任务相关的临时信息比如正在处理的文件列表、当前执行到第几步长期记忆跨会话有用的信息比如用户偏好、历史决策、领域知识。持续 Agent 的设计难点在于什么时候把工作记忆沉淀为长期记忆什么时候从长期记忆中检索相关内容放回上下文。如果只靠“把所有历史都塞进 Prompt”上下文窗口迟早会被撑爆成本也会失控。2.2 状态持久化状态State是 Agent 在某一时刻的内部快照包括当前目标、已完成步骤、待办事项、中间结果。在单次 Agent 里状态存在内存中就够了进程退出就清空。但持续 Agent 必须把状态持久化到磁盘或数据库里否则一次重启、一次宕机Agent 就彻底“失忆”。状态持久化要回答几个问题状态多久落盘一次状态里可以序列化的是什么不可以序列化的是什么多个状态副本冲突时以哪个为准恢复到哪个时间点最合理。这些问题在后面的代码示例中会具体体现。2.3 事件驱动与轮询持续 Agent 需要感知外部变化。实现方式有两种轮询Polling和事件驱动Event-driven。轮询就是每隔一段时间检查一次“有没有新任务”。优点是实现简单缺点是有延迟、浪费资源。事件驱动则是让外部系统主动通知 Agent比如消息队列、Webhook 或数据库触发器。生产环境里事件驱动更优雅、更实时但依赖也更多。对于刚起步的持续 Agent轮询是完全可以接受的实现方式先把主流程跑通再逐步迁移到事件驱动。2.4 暂停、恢复与幂等持续 Agent 运行时间越长越可能遇到中断API 报错、网络抖动、进程被杀、人工介入。因此它必须具备暂停和恢复能力而且恢复后不能重复执行已完成的副作用操作。幂等Idempotency是一个关键设计原则。比如“给指定文件写摘要”这个操作如果崩溃前已经写好了摘要重启后应该跳过而不是再写一遍。实现方法通常是为每个任务记录唯一标识处理前先查重处理后再标记完成。2.5 安全边界与最小权限一个长期运行的 Agent权限风险比单次 Agent 高得多。单次 Agent 如果被输入误导最多执行一次错误操作持续 Agent 如果权限过大可能在无人值守的情况下连续造成多个错误操作。因此持续 Agent 必须采用最小权限原则只能访问它真正需要的资源危险操作必须走审批流。这里用一张表快速对比传统 Agent 和持续 Agent对比维度传统 Agent持续 Agent生命周期秒级到分钟级小时级到天数级上下文单次请求内有效跨会话持久化状态内存中随进程消失持久化存储可恢复任务来源用户显式发起用户指令 外部事件自动触发失败处理报错后用户重试自动重试、跳过、恢复资源控制单次调用成本可控必须设置预算和上限安全模型单次权限校验持续最小权限 审批流3. 为什么 Astra“连续运行”是一个分水岭从公开披露看Astra 是 OpenAI 在多模态大模型方向上的重要布局而“连续运行数日”这个信息比单纯的多模态能力提升更值得玩味。在之前的助手产品里AI 通常处于“被动响应”状态用户不调用它就不工作。Astra 连续运行数日意味着 AI 开始具备“环境感知 持续意图”的能力它能长时间观察环境变化能围绕一个长期目标推进能在不同时间点之间保持一致性。这种能力一旦成熟很多产品逻辑都会跟着改变。不过这里要守住事实边界。Astra 连续运行数日的具体内部架构并没有完全公开我们不能凭空断言它用了什么记忆框架、什么调度系统。从材料做合理推断一个能连续运行数日的多模态 Agent至少在以下几个方面做了扎实的工作会话状态管理能够跨长时间窗口保持上下文一致性记忆分层短期感知和长期记忆的配合故障恢复长时间运行中不可避免的 API 错误、网络波动、进程崩溃都需要有自愈能力资源控制多模态任务消耗的 token 和计算资源更大必须有预算和配额机制否则一个“不会自己停下来的 Agent”会变成成本黑洞。所以我更倾向于把 Astra 连续运行数日看作一个工程信号OpenAI 已经在为“Agent 不是一次性工具而是长期存在的工作单元”这个方向铺路。对普通开发者来说不必等到 Astra 正式开放才行动现在就可以用已有的 API 和开源工具先构建一个最小持续 Agent。为什么现在值得做因为持续 Agent 的核心能力本质上不是模型能力而是工程能力。模型再强如果状态管理一塌糊涂、恢复逻辑缺失、资源成本不可控Agent 依然无法投入实际使用。而这些工程能力完全可以用现有的 OpenAI API、Agents SDK 和一套可靠的状态存储方案开始积累。4. 构建持续 Agent 的环境准备接下来进入实操部分。我们用 Python 构建一个最小的持续 Agent它的任务很简单持续监控一个inbox目录发现新文档就生成摘要并保存到outbox目录同时把已处理文件记录持久化到本地状态文件。这个示例虽然简单但覆盖了持续 Agent 的四个核心机制状态持久化、轮询事件、幂等去重、暂停恢复。4.1 环境要求建议准备以下环境Python 3.10 及以上版本一个 OpenAI API Key通过环境变量注入能访问 OpenAI API 的网络环境。如果没有 API Key或者暂时不想产生调用费用也没关系。我在示例里加入了MOCK_MODE开关开启后不会真实调用模型而是用本地模拟摘要。这样你依然可以完整验证“持续运行”的机制。4.2 安装依赖本文只需要一个核心依赖openaiPython 库。创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install openai安装完成后建议先确认版本pip show openai如果希望在后续开发中用到 Agents SDK可以另行安装但本文的最小示例不需要。4.3 配置环境变量在项目根目录下创建.env文件或者在当前 shell 里导出环境变量export OPENAI_API_KEYsk-你的密钥如果你使用的是国内可正常访问的 API 转发服务也可以额外配置OPENAI_BASE_URLexport OPENAI_BASE_URLhttps://你的转发服务地址/v1注意API Key 属于敏感信息禁止提交到 Git 仓库。实际开发中推荐用环境变量或密钥管理服务。4.4 项目目录结构创建如下目录结构persistent-agent-demo/ ├── continuous_agent.py # 主程序 ├── inbox/ # 监控目录放置待处理文档 ├── outbox/ # 输出目录保存摘要结果 ├── agent_state.json # 状态文件运行后自动生成 └── pause.flag # 暂停标记文件需手动创建5. 最小持续 Agent 示例代码实现现在写主程序。代码看起来不短但每个部分都对应一个持续 Agent 的关键机制。5.1 完整代码continuous_agent.py# 文件路径continuous_agent.py import json import os import time import hashlib from datetime import datetime from pathlib import Path from openai import OpenAI # 读取环境变量 API_KEY os.getenv(OPENAI_API_KEY, ) BASE_URL os.getenv(OPENAI_BASE_URL, None) MOCK_MODE os.getenv(MOCK_MODE, 0) 1 # 模型名请替换为你有权限访问的实际模型 MODEL_NAME os.getenv(AGENT_MODEL, gpt-4o-mini) # 路径配置 STATE_FILE Path(agent_state.json) WATCH_DIR Path(./inbox) RESULT_DIR Path(./outbox) PAUSE_FLAG Path(pause.flag) # 轮询间隔单位秒 POLL_INTERVAL int(os.getenv(POLL_INTERVAL, 30)) # 初始化 OpenAI 客户端 client OpenAI(api_keyAPI_KEY, base_urlBASE_URL) def load_state() - dict: 从状态文件加载 Agent 状态。 if STATE_FILE.exists(): raw STATE_FILE.read_text(encodingutf-8) state json.loads(raw) else: state { processed: [], current_goal: 监控 inbox 目录中的新文档并生成摘要, resume_count: 0, } return state def save_state(state: dict) - None: 将 Agent 状态持久化到磁盘。 STATE_FILE.write_text( json.dumps(state, ensure_asciiFalse, indent2), encodingutf-8 ) def file_hash(path: Path) - str: 计算文件内容的 MD5用于判断文件是否已被处理。 return hashlib.md5(path.read_bytes()).hexdigest() def summarize(text: str) - str: 调用模型生成摘要。MOCK_MODE 开启时不访问外部 API。 if MOCK_MODE: return f[mock] 文档摘要{text[:80]} resp client.chat.completions.create( modelMODEL_NAME, messages[ { role: system, content: 你是文档摘要助手请输出简洁、准确的中文摘要。 }, { role: user, content: f请为以下文档生成摘要\n\n{text[:3000]} } ], temperature0.3 ) return resp.choices[0].message.content def process_file(path: Path, state: dict) - None: 处理单个文件读取内容、生成摘要、保存结果、更新状态。 text path.read_text(encodingutf-8, errorsignore) summary summarize(text) RESULT_DIR.mkdir(exist_okTrue) result_file RESULT_DIR / f{path.stem}_summary.md content ( f# 摘要\n\n f{summary}\n\n f来源{path.name}\n f生成时间{datetime.now().isoformat()}\n ) result_file.write_text(content, encodingutf-8) state[processed].append({ file: path.name, hash: file_hash(path), summary_file: result_file.name, processed_at: datetime.now().isoformat(), }) print(f[{datetime.now().isoformat()}] 已处理 {path.name}摘要写入 {result_file.name}) def should_pause() - bool: 检测暂停标记。存在 pause.flag 文件时Agent 在完成当前轮次后暂停。 return PAUSE_FLAG.exists() def run_once(state: dict) - None: 执行一轮监控任务扫描目录、处理新文件、保存状态。 WATCH_DIR.mkdir(exist_okTrue) RESULT_DIR.mkdir(exist_okTrue) for path in sorted(WATCH_DIR.iterdir()): if not path.is_file(): continue h file_hash(path) # 幂等检查文件已处理且内容未变化则跳过 if any( item[file] path.name and item[hash] h for item in state[processed] ): continue try: process_file(path, state) save_state(state) except Exception as exc: print(f[{datetime.now().isoformat()}] 处理 {path.name} 失败{exc}) def main() - None: state load_state() state[resume_count] 1 save_state(state) print(f[{datetime.now().isoformat()}] 持续 Agent 启动第 {state[resume_count]} 次运行 f监控目录{WATCH_DIR}轮询间隔{POLL_INTERVAL} 秒) while True: if should_pause(): print(f[{datetime.now().isoformat()}] 检测到暂停标记完成当前轮次后暂停。) run_once(state) break run_once(state) time.sleep(POLL_INTERVAL) if __name__ __main__: main()5.2 代码关键逻辑说明第一状态持久化。load_state和save_state负责把 Agent 状态读写到agent_state.json。每次处理完一个文件就立刻save_state避免 Agent 在处理中途崩溃时丢失已完成的工作。第二幂等去重。file_hash计算文件内容的 MD5run_once中通过“文件名 内容哈希”判断文件是否处理过。这样即使文件名重复只要内容不同Agent 也会重新处理如果内容没变即使 Agent 重启多次也不会重复生成摘要。第三暂停恢复。should_pause检测pause.flag文件。创建这个文件后Agent 会在完成当前一轮后退出而不是立即被强杀。因为退出前状态已经保存下次启动时会从上次的位置继续不会丢任务。第四异常隔离。process_file被try...except包裹单个文件处理失败不会拖垮整个主循环。这在持续运行中非常重要一个坏文件不应该让 Agent 整体停止。5.3 启动脚本run_agent.sh为了方便在后台运行和查看日志可以创建一个启动脚本# 文件路径run_agent.sh #!/usr/bin/env bash export OPENAI_API_KEY${OPENAI_API_KEY:?请先设置 OPENAI_API_KEY} # 如果暂时没有 API Key可以开启 mock 模式 # export MOCK_MODE1 mkdir -p inbox outbox nohup python continuous_agent.py agent.log 21 echo $! agent.pid echo Agent 已启动PID: $(cat agent.pid)给脚本添加执行权限并启动chmod x run_agent.sh ./run_agent.sh5.4 暂停与恢复命令创建暂停标记touch pause.flagAgent 会在当前轮次结束后退出。查看进程是否退出cat agent.pid ps -p $(cat agent.pid) || echo Agent 已停止恢复运行rm pause.flag ./run_agent.sh6. 运行结果与效果验证6.1 启动 Agent假设你已经启动了 Agent。启动日志类似[2025-01-15T10:00:0108:00] 持续 Agent 启动第 1 次运行监控目录inbox轮询间隔30 秒6.2 验证持续监控和处理在inbox目录下创建一个测试文档cat inbox/hello.txt EOF 这是一个关于持续智能体的介绍文档。它讨论了 Agent 如何从单次问答走向长时间运行。 EOF等待最多一个轮询周期30 秒然后查看outbox目录ls -la outbox/预期会生成一个新文件hello_summary.md内容类似# 摘要 这是一个关于持续智能体的介绍文档。它讨论了 Agent 如何从单次问答走向长时间运行。 来源hello.txt 生成时间2025-01-15T10:01:2308:00再查看状态文件cat agent_state.json会看到类似于下面的内容{ processed: [ { file: hello.txt, hash: c4ca4238a0b923820dcc509a6f75849b, summary_file: hello_summary.md, processed_at: 2025-01-15T10:01:2308:00 } ], current_goal: 监控 inbox 目录中的新文档并生成摘要, resume_count: 1 }这说明 Agent 已经成功完成了一次“感知新文件、调用模型、保存结果、更新状态”的完整闭环。6.3 验证幂等去重再次手动运行一次处理逻辑或者直接重启 Agentkill $(cat agent.pid) ./run_agent.sh这时候可以发现即使inbox里还有hello.txtAgent 也不会重复生成摘要。因为状态文件中已经记录了该文件的哈希值run_once会跳过它。6.4 验证暂停恢复创建暂停标记并观察日志touch pause.flag tail -f agent.logAgent 会在当前轮次结束后打印暂停日志并退出。此时再向inbox放入一个新文件然后删除pause.flag并重启 Agent新文件会被正常处理旧文件依然不会重复处理。6.5 验证 Mock 模式如果没有 API Key可以设置MOCK_MODE1export MOCK_MODE1 ./run_agent.sh此时摘要内容会是[mock] 文档摘要...但持续运行、状态持久化、暂停恢复这些机制完全正常。这非常适合先在本机把工程框架打通再接入真实模型。7. 持续 Agent 常见问题与排查思路长时间运行的 Agent 和普通脚本不同它的故障往往是“慢性的”而不是“一次性的”。下面是几类最容易遇到的典型问题。问题现象可能原因排查方式解决方案启动后立即退出API Key 无效或未设置查看 agent.log 和启动脚本报错确认环境变量已导出检查 Key 是否有效单文件处理失败导致主循环退出异常未隔离直接抛出到顶层查看错误堆栈是否发生在 process_file 中用 try...except 包裹单文件处理逻辑同一文件被重复处理只判断文件名没判断内容哈希查看 agent_state.json 中的 hash 字段使用文件名 内容 MD5 双重判断状态文件无限膨胀processed 列表只增不减查看 agent_state.json 的行数定期归档历史记录只保留近期状态模型调用报 429 限流请求频率超限或余额不足查看日志中是否有 rate_limit 字样增加退避重试降低轮询频率检查账户额度进程被系统杀掉依赖 nohup 仍不稳定或系统重启查看 agent.pid 和系统日志使用 supervisor/systemd 托管进程pause.flag 不生效Agent 正在 sleep 或 API 调用中等待当前轮次结束缩短 POLL_INTERVAL或实现信号量中断API 调用费用异常增长每个文件都调用高价模型且无预算控制统计每日 token 消耗模型分级简单任务用便宜模型设置预算上限首当其冲的排查步骤永远是一样的打开日志。我见过很多“持续 Agent 莫名其妙停了”的情况最后几乎都是小问题——API Key 过期、网络抖动、磁盘写满。所以从第一天起就要把日志和状态文件当作核心资产而不是可有可无的调试工具。8. 生产环境最佳实践与工程建议如果只是学习和验证上面的示例已经够了。但如果要把持续 Agent 放到生产环境还有几个关键工程问题需要认真对待。8.1 状态持久化要用数据库而不是 JSON 文件示例中的 JSON 文件适合单机、单进程、低并发场景。生产环境里状态应该放在 PostgreSQL、Redis 或对象存储中。原因很简单多个 Agent 实例同时运行时需要一个共享状态源而且数据库的原子更新、事务、备份恢复能力都是 JSON 文件不具备的。状态表至少应该包含任务 ID、状态、输入摘要、输出结果、时间戳、重试次数。用任务 ID 做唯一约束天然支持幂等。8.2 记忆分层不要让上下文无限膨胀持续 Agent 运行的每一秒都在产生新信息如果全部塞进上下文上下文窗口和成本都会失控。推荐的做法是分层处理当前任务上下文只保留本轮任务相关的对话和工具调用工作记忆保留短期需要引用的中间结果用 Redis 缓存长期记忆定期把重要信息抽取成摘要或结构化记录存入向量数据库按需检索。简单任务用便宜的小模型复杂推理和决策用大模型。这个“模型分级”策略在长期运行中能节省大量成本。8.3 用事件驱动替代轮询示例中用了 30 秒轮询简单可靠但生产环境不够优雅。更合理的方案是用消息队列如 RabbitMQ、Kafka推送任务事件用 Webhook 接收外部系统通知用数据库触发器和 CDC 捕获数据变更用定时任务调度框架如 APScheduler、Celery Beat处理定时类任务。轮询并没有错只是当 Agent 管理的任务量上来之后事件驱动的实时性和资源利用率会明显更好。8.4 可观测性是持续 Agent 的生命线一个只运行 10 秒的脚本可以靠 print 调试一个运行 7×24 小时的 Agent 必须拥有完整的可观测性体系结构化日志JSON 格式带上 trace_id、task_id、时间戳指标监控任务处理数、失败率、平均耗时、token 消耗分布式追踪任务跨多个服务和模型调用时能定位瓶颈和错误源头。每条关键日志都要有任务 ID这样排查问题时才能把“状态文件”“模型调用”“外部工具调用”串成一条完整链路。8.5 安全边界必须比单次 Agent 更严格持续 Agent 权限过大的后果是灾难性的。无人值守时一个错误决定可能触发一连串错误操作。生产环境至少要做好这几件事限制文件系统、数据库、网络资源的访问范围所有删除、写生产库、发送对外消息等危险操作必须进入人工审批流API Key 使用最小权限定期轮换为每个 Agent 实例设置独立的身份和审计日志。最稳妥的原则是默认拒绝按需放行。8.6 成本控制给 Agent 上一道“预算锁”持续 Agent 的成本模型是“运行时间 × 单位时间 token 消耗”如果不加控制一个失控的 Agent 可以在几小时内烧掉大量费用。建议做三层控制单次请求限制限制每次模型调用的最大 token 数周期预算限制按小时、按天设定 token 预算超限自动暂停任务数量限制每个轮次最多处理 N 个任务防止突发流量把 Agent 压垮。在示例代码基础上可以很自然地加入这些限制这比上线后再补救容易得多。8.7 故障恢复要做到幂等和可重放崩溃不可怕可怕的是恢复后重复执行或状态错乱。所有会产生副作用的操作都应该具备幂等性写文件前检查目录和文件名调用外部 API 前生成请求 ID服务端去重状态落盘和副作用执行尽量保持同一事务或者先记“将要执行”状态再更新为“已执行”。当你能做到“任意时刻崩溃重启后从最近一个稳定状态继续”时持续 Agent 才算是真正生产可用的。9. 下一步可以做什么到这里我们已经把一个最小的持续 Agent 从零跑通了。它虽然简单但包含了持续智能体最核心的几条骨架状态持久化、幂等去重、轮询感知、暂停恢复。接下来你可以沿着三条线继续深入。第一条线是“替换模型和场景”。把监控文件夹换成监控数据库新增记录把摘要任务换成自动分类或自动回复看看持续 Agent 在你的业务里能跑通什么流程。第二条线是“引入工程基础设施”。把 JSON 状态换成 PostgreSQL把轮询换成消息队列把 print 日志换成结构化日志和监控面板。这一步能让你的持续 Agent 从“能跑”变成“能稳定跑”。第三条线是“研究 OpenAI 生态里的持续 Agent 工具”。比如 Codex CLI 已经把 Agent 带入了命令行开发环境Agents SDK 提供了构建多 Agent 协作的基础设施。OpenAI 在 Agent 领域的动作越来越密集Astra 连续运行数日只是其中一个信号。尽早掌握状态管理、事件驱动、可观测性这些底层能力你会在下一轮 Agent 应用爆发时更有主动权。持续智能体不会取代单次 Agent但它会接管那些过去需要人盯着的重复性工作。与其等框架成熟再学不如现在就动手把最小闭环跑通。建议把这篇文章收藏备用等你有 API Key 或者遇到 Agent 长任务场景时直接照示例改造即可。