OpenAI Astra 连续运行数日这个消息在智能体开发者圈子里讨论度很高。它传递的最重要信号不是某个模型又刷新了分数而是持续智能体这个概念正在从演示走向工程落地。一个能连续数天保持任务的智能体意味着它必须处理上下文膨胀、状态持久化、工具调用恢复、资源回收和异常重试等一系列问题这些恰恰是普通 Chatbot 很少关心的内容。这里不讨论发布会本身而是围绕“连续运行数日”这个结果拆解持续智能体背后的工程难点并给出一个可以在本地跑起来的最小原型让读者理解为什么持续运行听起来容易做起来却需要一套完整工程支撑。1. 先看懂“连续运行数日”为什么是一个工程事件1.1 Astra 在智能体赛道里的定位从公开信息来看Astra 通常被理解为 OpenAI 在实时多模态智能体方向上的产品形态。它做的事情可以概括成一句话让模型不再局限于一次问答而是通过摄像头、麦克风、屏幕画面等多模态输入持续理解当前环境并主动推进目标任务。连续运行数日这个演示结果意味着这类智能体已经具备类似后台服务的形态每天持续接收输入、做判断、调用工具、更新状态。不过要注意公开演示里展示的效果和开发者自己能在项目里复现的效果之间还有明显距离。演示环境通常有专门优化而工程落地面对的是网络波动、接口限流、进程重启、磁盘写满等真实问题。理解这一点才能把注意力放在正确的方向上。1.2 从“一轮问答”到“持续任务”的差异普通对话系统的流程是调用一次模型接口生成一段回复任务结束。持续智能体完全不同它的任务时间跨度从秒级变成小时、天甚至周输入不是一次性 prompt而是不断追加的消息流模型上下文窗口有限不能让消息无限堆积进程可能崩溃、重启、被部署系统杀掉状态不能只存在内存里模型输出也不再只是文本还可能是工具调用、外部系统操作、等待人工确认等动作。维度普通对话持续智能体单次任务时长秒到分钟小时到天上下文单轮或短会话长会话加记忆管理状态可以不保存必须持久化工具调用可有可无核心能力需可恢复故障恢复失败后重新开始断点续跑监控不需要必需1.3 持续智能体改变了哪些工程假设如果之前只做过 prompt 调优现在需要面对的是状态机设计、数据库选型、任务队列、幂等控制、日志追踪、资源回收。这些是后端工程师很熟悉的词汇但它们第一次被集中引入到“和模型交互”这一层。换句话说持续智能体的开发本质上是把一次 API 调用包装成一个长期运行的业务系统。这个转变有一个实际影响团队配置会出现变化。只懂 prompt 的人负责不了长时间运行的任务系统至少还要有人熟悉存储、调度、监控和故障恢复。Astra 连续运行数日看起来是一个自然结果但在工程上这背后通常是一整套可观测、可恢复、可治理的基础设施。2. 持续智能体的四个核心难点上下文、状态、资源、恢复2.1 上下文窗口有限记忆必须分层管理LLM 的上下文窗口再大也有限。假设一个任务每 5 分钟产生一轮对话一天就是 288 轮一周超过 2000 轮。按平均每轮 500 token 估算一周就有上百万 token没有任何模型能无限接收。所以持续智能体必须做记忆管理短期记忆保留最近 N 轮原始消息让模型对当前现场有完整感知。工作记忆保存当前任务的中间结论、已完成步骤、待办事项用结构化字段而不是自由文本。长期记忆保存任务目标、用户偏好、历史结果摘要以压缩后的文本或向量形式存储。判断标准很简单模型每次调用时只拿到“当前需要的上下文”而不是“全部历史”。盲目裁剪会丢失关键信息全文塞入又会超出窗口因此需要压缩、摘要、检索三种手段配合。2.2 任务状态必须持久化内存状态在崩溃后等于零在普通脚本里变量即状态。但持续运行的进程一旦被重启所有变量都会消失。必须把任务状态落到数据库或对象存储里至少包括任务当前处于哪个阶段。已经执行了哪些步骤。每一步的工具调用参数和结果。下一次应该从哪一步继续。最容易犯的错是只在代码里维护一个 Python dict 保存状态看起来方便进程一挂就全丢。更隐蔽的问题是消息和状态不同步比如先更新了任务状态还没写最后一条消息进程就崩了恢复后消息不完整。2.3 长时间运行要求资源可回收连续运行几天消息列表会越来越长数据库连接不能一直占用日志不能无限增长HTTP 连接要有超时并发要有上限。典型资源问题包括内存持续增长每轮循环都把全部历史消息加载进内存。连接泄漏每轮调用都新建数据库连接却不关闭。日志膨胀每步都把完整模型响应打到日志文件。API 配额耗尽没有限速和重试退避短时间触发大量请求。这些问题的共同特点是初期看不出来运行一天后才会暴露。所以在设计阶段就要给资源的数量级设上限而不是等出问题再排查。2.4 故障恢复是持续运行的底线持续智能体运行中会遇到三类故障进程崩溃或发布重启。外部 API 超时、限流、返回异常。工具执行失败比如调用的接口返回 500。恢复策略首先要保证消息不丢。模型调用产生的消息、工具调用的参数和结果都应该在产生后立刻写入持久化存储。其次要保证任务能续跑启动时扫描未完成任务把状态从 running 重置为 pending重新进入调度。最后要保证工具不会重复执行否则一次崩溃恢复可能导致同一笔订单被创建两次。3. 构建一个可连续运行的智能体最小原型3.1 技术选型用框架还是自己搭智能体平台和框架很多Dify、Coze、LangChain、AutoGen 等都提供了不同程度的抽象。它们解决的层级不同方案适合场景局限Dify工作流编排、知识库、可视化调试自定义长生命周期逻辑受限Coze平台化 bot、快速上线持久化控制较弱LangChain灵活组合各类工具抽象层级多调试成本高自研OpenAI SDK SQLite深度控制状态与恢复机制需要自己写调度与持久化如果目标是快速做一个对话机器人直接用平台最省事。如果目标是理解“连续运行”背后的机制自研一个最小循环更合适。本文选自研方案核心依赖只有一个 openai SDK 和标准库 sqlite3。3.2 项目结构与依赖使用 Python 3.10 以上版本目录结构如下persistent_agent/ ├── main.py ├── db.py ├── agent.py ├── tools.py ├── memory.py └── requirements.txtrequirements.txtopenai1.30.03.3 定义任务与消息的状态模型先建三张表tasks 保存任务主状态messages 保存完整的对话消息对象tool_results 保存工具执行结果用于幂等。import sqlite3 from datetime import datetime, timezone DB_PATH agent_state.db def now(): return datetime.now(timezone.utc).isoformat() def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, objective TEXT NOT NULL, status TEXT NOT NULL DEFAULT pending, step INTEGER NOT NULL DEFAULT 0, created_at TEXT, updated_at TEXT ) ) conn.execute( CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT ) ) conn.execute( CREATE TABLE IF NOT EXISTS tool_results ( task_id TEXT NOT NULL, tool_call_id TEXT PRIMARY KEY, name TEXT NOT NULL, args TEXT NOT NULL, result TEXT NOT NULL, created_at TEXT ) ) conn.commit() return conn关键点status 控制任务生命周期这里用 pending、running、completed、failed 四种状态。step 记录进度恢复时可以判断任务已经走了多远。messages 的 payload 存完整消息对象 JSON而不是只存文本这样才能完整还原 assistant 的 tool_calls 和 tool 消息。tool_results 用 tool_call_id 做主键重复调度时能查到已有结果避免工具重复执行。3.4 持久化会话消息保存消息时不要只在内存里 append要立即写入数据库。import json def save_message(conn, task_id, message): conn.execute( INSERT INTO messages (task_id, payload, created_at) VALUES (?, ?, ?), (task_id, json.dumps(message, ensure_asciiFalse), now()), ) conn.commit() def load_recent_messages(conn, task_id, limit20): rows conn.execute( SELECT payload FROM messages WHERE task_id? ORDER BY id DESC LIMIT ?, (task_id, limit), ).fetchall() return [json.loads(r[0]) for r in reversed(rows)]从后往前读取最近 limit 条再反转回正序。限制条数的目的就是控制每次调用模型的 token 量这是“短期记忆窗口”最朴素的落地方案。3.5 上下文压缩防止历史无限膨胀只做截断会丢失早期关键信息正确做法是定期把历史压缩成摘要。压缩后的摘要作为系统提示的一部分注入原始消息只保留最近 N 轮。def compress_history(client, messages): resp client.chat.completions.create( modelMODEL_NAME, messages[ {role: system, content: 压缩以下对话为任务摘要保留关键决定、已完成步骤和未完成事项。}, *messages, ], temperature0, ) return resp.choices[0].message.content实际触发时机可以按估算 token 数判断当最近消息超过 MAX_CONTEXT_TOKENS 的六成时先压缩再截断。3.6 单步执行调用模型并