人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载本文围绕 openJiuwen agent-core 的openjiuwen.core.session.checkpointer模块系统讲解检查点Checkpointer抽象设计、内存与持久化两套实现、工厂注册扩展机制以及键命名规范并结合仓库源码说明其在 Runner 与 Session 中的实际集成方式。读完本文你将掌握如何为 Agent、Agent 团队与工作流配置断点续跑能力理解中断恢复、异常保存与强制清理等关键分支的底层原理并能够基于CheckpointerProvider自定义新的存储后端。检查点模块位于 docs/zh/2.开发指南/API文档/openjiuwen.core/session/checkpointer.md对应的源码实现分散在 openjiuwen/core/session/checkpointer/ 目录下本文以 API 文档为骨架用源码与测试佐证其实现细节。一、Checkpointer 的定位Agent 与工作流的“状态快照层”openJiuwen agent-core 的会话Session承载 Agent、Agent 团队与工作流运行时的全部状态。Checkpointer 的职责正是在执行的关键节点对这份状态做“快照”——保存、恢复、清理从而支撑三类典型场景中断恢复工作流或 Agent 在执行中等待用户交互interrupt时保存状态交互返回后从断点继续异常恢复工作流执行抛出异常时保留现场便于后续排障或重新续跑会话复用同一个session_id:workflow_id线程thread再次执行时恢复上一次的状态。从源码看Checkpointer 与 Storage 是一对抽象基类定义在 openjiuwen/core/session/checkpointer/base.pyCheckpointer面向执行生命周期的高层接口包含工作流/Agent 执行前后的钩子方法Storage面向单个实体状态的低层存取接口封装save/recover/clear/exists四类原子操作。两个内置实现InMemoryCheckpointer与PersistenceCheckpointer分别以进程内存与BaseKVStore键值存储为介质前者适合开发测试后者适合生产部署。此外仓库还提供基于 Redis 的扩展实现见 openjiuwen/extensions/checkpointer/redis/checkpointer.py说明这套抽象具备良好的后端可替换性。二、Checkpointer 抽象基类六个核心钩子与两个辅助方法Checkpointer是一个抽象基类ABC源码位置 openjiuwen/core/session/checkpointer/base.py#L14-L57。API 文档共列出 9 个成员其中 6 个是抽象执行钩子另有session_exists、release、graph_store三个通用能力仓库源码中还额外提供了 Agent 团队team维度的两个钩子。2.1 get_thread_id线程标识staticmethod get_thread_id(session: BaseSession) - str静态方法返回格式为session_id:workflow_id的线程 ID。源码实现即return :.join([session.session_id(), session.workflow_id()])该线程 ID 用于区分同一 Session 下的不同工作流运行实例是后续所有检查点键key的逻辑前缀。2.2 工作流执行钩子pre_workflow_execute / post_workflow_executeabstractmethod async pre_workflow_execute(session: BaseSession, inputs: InteractiveInput) abstractmethod async post_workflow_execute(session: BaseSession, result, exception)pre_workflow_execute在工作流执行前调用其行为由inputs的类型决定这是整个模块中最值得关注的分支逻辑若inputs是InteractiveInput类型说明这是一次带交互输入的续跑直接恢复工作流状态并将新输入合并进会话状态否则非交互输入若工作流状态不存在直接返回走全新执行若状态存在且未启用强制删除则抛出CHECKPOINTER_PRE_WORKFLOW_EXECUTION_ERROR异常提示workflow state exists but non-interactive input and cleanup is disabled——这是为了防止无意中覆盖上次未完成的现场若状态存在且启用了强制删除环境开关_force_del_workflow_state见 openjiuwen/core/session/constants.py#L22 中的FORCE_DEL_WORKFLOW_STATE_KEY则删除图状态与工作流状态后全新执行。post_workflow_execute在工作流执行后调用同样三分支exception不为None保存一次工作流检查点落盘异常现场然后重新抛出异常result中没有中断标记TASK_STATUS_INTERRUPT来自 openjiuwen/core/graph/pregel.py工作流正常结束清理本次工作流的状态与图状态result中存在中断标记工作流因等待交互而暂停保存工作流检查点以便后续恢复。2.3 Agent 执行钩子pre_agent_execute / interrupt_agent_execute / post_agent_executeabstractmethod async pre_agent_execute(session: BaseSession, inputs) abstractmethod async interrupt_agent_execute(session: BaseSession) abstractmethod async post_agent_execute(session: BaseSession)pre_agent_executeAgent 执行前恢复其状态若提供了inputs会把输入写入会话状态中内存实现写入键INTERACTIVE_INPUTinterrupt_agent_executeAgent 需要中断等待用户交互时保存状态这是人工介入-继续执行场景的关键落点post_agent_executeAgent 执行完成后保存最终状态。2.4 session_exists / release / graph_storeabstractmethod async session_exists(session_id: str) - bool abstractmethod async release(session_id: str, agent_id: str None) abstractmethod def graph_store() - Storesession_exists判断指定会话 ID 是否已有检查点数据release释放会话资源。提供agent_id时仅清理该 Agent 的检查点不提供时清理整个会话含工作流、Agent、Agent 团队的全部状态。内存实现还会把以该 session_id 为前缀的 Agent 存储一并移除见 inmemory.py#L365-L367graph_store返回图状态GraphState存储对象。工作流的图状态与工作流自身状态分开存放二者通过不同的命名空间隔离。2.5 源码中的补充Agent 团队钩子API 文档未列出的pre_agent_team_execute/post_agent_team_execute在抽象基类中同样存在base.py#L31-L45用于 Agent 团队team维度的状态恢复与保存其状态来源是会话的全局状态global_state。这说明检查点体系覆盖了 Agent、Agent 团队、工作流三个层级。三、Storage 抽象save / recover / clear / exists 四元组Storage是更底层的状态存取接口base.py#L60-L75abstractmethod async save(session: BaseSession) abstractmethod async recover(session: BaseSession, inputs: InteractiveInput None) abstractmethod async clear(session_id: str) abstractmethod async exists(session: BaseSession) - boolsave将当前会话状态序列化后写入存储recover从存储中反序列化并恢复会话状态可携带可选交互输入clear按会话 ID 清除状态exists判断指定会话的状态是否存在。两套 Checkpointer 实现各自配套了针对 Agent、Agent 团队、工作流的 Storage 实现。序列化统一走create_serializer(pickle)序列化结果为(dump_type, blob)二元组即序列化类型 字节内容便于反序列化时还原类型信息。四、InMemoryCheckpointer开发测试场景的默认选择4.1 内部结构与适用场景class openjiuwen.core.session.checkpointer.InMemoryCheckpointer()基于内存的检查点实现所有状态保存在进程内字典中进程重启后状态丢失适用于开发和测试场景。其内部维护了四类存储见 inmemory.py#L23-L30_agent_stores按 session_id 组织的 Agent 状态存储_agent_team_stores按 session_id 组织的 Agent 团队状态存储_workflow_stores按 session_id 组织的工作流状态存储_graph_storeInMemoryStore()图状态存储_session_to_workflow_ids记录 session 下已保存检查点的工作流 ID 集合用于release时精确清理。使用样例来自 API 文档可直接运行 from openjiuwen.core.session.checkpointer import InMemoryCheckpointer checkpointer InMemoryCheckpointer() # 使用 checkpointer 进行状态管理在 openJiuwen 中InMemoryCheckpointer同时也是默认检查点工厂模块底部实例化了全局单例default_inmemory_checkpointercheckpointer.py#L134任何未显式配置的场景都会回落到它。4.2 执行钩子的实现细节内存实现的钩子方法行为与抽象定义完全对齐并带完整的日志埋点事件类型如CHECKPOINT_SAVE、CHECKPOINT_RESTORE、CHECKPOINT_CLEAR、CHECKPOINTER_STORE_ADD、CHECKPOINTER_STORE_REMOVE便于通过日志追踪检查点生命周期pre_workflow_execute首次创建存储时打CHECKPOINTER_STORE_ADD日志InteractiveInput分支执行workflow_store.recover(session, inputs)恢复非交互分支按 2.2 节的三分支逻辑执行post_workflow_execute异常时调用_inner_save_workflow_checkpoint保存现场正常完成时调用_inner_clear_workflow_session清理中断时保存检查点。若工作流的父会话不是AgentSession完成清理后还会移除该 session 的工作流存储inmemory.py#L96-L99interrupt_agent_execute/post_agent_execute调用对应 AgentStorage 的save落状态release按agent_id粒度或整个会话粒度清理且会同步删除图状态。五、PersistenceCheckpointer基于 BaseKVStore 的持久化实现5.1 构造与适用后端class openjiuwen.core.session.checkpointer.PersistenceCheckpointer(kv_store: BaseKVStore)基于持久化存储的检查点实现通过BaseKVStore接口对接任意键值存储后端。BaseKVStore抽象定义于 openjiuwen/core/foundation/store/base_kv_store.py#L16提供pipeline()、batch_delete()、delete_by_prefix()等批量操作原语仓库已内置ShelveStoreshelve_store.py与DbBasedKVStoredb_based_kv_store.py两类实现。API 文档给出的经典样例 from openjiuwen.core.session.checkpointer import PersistenceCheckpointer from openjiuwen.core.foundation.store.kv import ShelveStore kv_store ShelveStore(checkpoint.db) checkpointer PersistenceCheckpointer(kv_store) # 使用 checkpointer 进行状态管理5.2 键结构设计三层命名空间隔离持久化实现把状态拆成多个 KV 条目键统一由session_id、命名空间、实体 ID 与后缀构成具体见 persistence.py实体命名空间键示例Agent 状态agentsession1:agent:agent1:agent_state_blobsAgent 团队状态agent-teamsession1:agent-team:team1:agent_team_state_blobs工作流自身状态workflowsession1:workflow:wf1:workflow_state_blobs工作流图状态workflow-graphsession1:workflow-graph:wf1:checkpoint_data_value工作流的自身状态与图状态分离存放前者存会话状态与状态更新updates后者存图执行节点状态GraphState这正是WORKFLOW_NAMESPACE_GRAPH workflow-graph常量的意义。图状态存储类GraphStore实现了 openjiuwen/core/graph/store/base.py#L41-L51 中Store接口的get/save/delete其中delete(session_id, nsNone)在ns为空时会按前缀session_id:workflow-graph级联删除该会话下所有图状态。5.3 读写性能细节pipeline 批量化持久化 Storage 的所有读写都通过pipeline()批量执行Agent / Agent 团队状态存取涉及 2 个键dump_typeblobexists要求两个键同时存在才判定状态存在工作流状态存取涉及 4 个键状态与 updates 各一对exists只要求状态键存在updates 是可选附加数据release未指定agent_id时直接delete_by_prefix(session_id :)一次清空该会话全部键。5.4 序列化与容错状态序列化格式为(dump_type, bytes)写入前通过_serde.dumps_typed(state)得到二元组读出时若blob为字符串则按 base64 解码回字节再反序列化兼容不同 KV 后端对字节的存储方式。反序列化失败会被记录CHECKPOINT_ERROR日志并返回None不会直接抛错中断流程具备一定的容错性。六、CheckpointerFactory 工厂注册、创建与默认实例管理6.1 工厂职责与注册机制classmethod register(name: str) classmethod async create(checkpointer_conf: CheckpointerConfig) - Checkpointer classmethod set_default_checkpointer(checkpointer: Checkpointer) classmethod set_checkpointer(store_type: str, checkpointer: Checkpointer) classmethod get_checkpointer(store_type: Optional[str] None) - CheckpointerCheckpointerFactorycheckpointer.py#L60-L125以注册表模式管理各种类型的检查点。register(name)返回一个装饰器用于把CheckpointerProvider子类按名称注册进_registry字典。API 文档示例 from openjiuwen.core.session.checkpointer import CheckpointerFactory, CheckpointerProvider CheckpointerFactory.register(custom) class CustomCheckpointerProvider(CheckpointerProvider): ... async def create(self, conf: dict) - Checkpointer: ... # 创建自定义检查点实例 ... return CustomCheckpointer()create先按type找到已注册的 Provider再调用其create(conf)异步构造实例未注册的类型会直接抛出异常。create内部会先导入 persistence 模块import openjiuwen.core.session.checkpointer.persistence as _确保persistence类型始终完成注册。内置注册的两个 Providerin_memoryInMemoryCheckpointerProvider无论conf内容如何都返回全局默认的内存检查点单例persistencePersistenceCheckpointerProvider根据配置构造持久化实例。仓库扩展目录 openjiuwen/extensions/checkpointer/redis/checkpointer.py#L229-L230 还通过同样方式注册了redis类型支持单机与集群两种模式。6.2 get_checkpointer 的解析优先级get_checkpointer(store_typeNone)的查找顺序与 API 文档描述一致源码见 checkpointer.py#L98-L125传入store_type时先查set_checkpointer手动设置的实例映射_type_checkpointersstore_type in_memory且无手动设置返回默认内存检查点单例否则返回set_default_checkpointer设置的默认实例若从未设置最终回落到默认内存检查点。该工厂被 Session 层直接消费AgentSession在构造时默认调用CheckpointerFactory.get_checkpointer()获取检查点openjiuwen/core/session/internal/agent.py#L45NodeSession、WorkflowSession则把调用委托给父会话openjiuwen/core/session/internal/workflow.py#L72-L77、#L169-L171。6.3 CheckpointerConfig配置对象与敏感信息脱敏class CheckpointerConfig(type: str in_memory, conf: dict {})CheckpointerConfig是 pydantic 模型type默认in_memoryconf为 Provider 透传的配置字典。值得注意的安全细节该类的__repr__与__str__会递归扫描conf中的字符串对形如scheme://的 URL 调用redact_url_password将密码替换为***checkpointer.py#L27-L51防止 Redis、数据库等连接串中的凭据泄漏到日志。对应的单元测试见 tests/unit_tests/core/session/checkpointer/test_checkpointer_config_repr.py覆盖 URL 密码脱敏、无密码 URL 原样保留、嵌套字典/列表递归脱敏等场景 from openjiuwen.core.session.checkpointer import CheckpointerFactory, CheckpointerConfig config CheckpointerConfig(typein_memory, conf{}) checkpointer await CheckpointerFactory.create(config)七、键构建工具与命名空间常量7.1 build_key 与 build_key_with_namespacefunc build_key(*parts: str) - str func build_key_with_namespace(session_id: str, namespace: str, entity_id: str, *suffixes: str) - strbuild_key用冒号连接任意多个字符串片段build_key_with_namespace在其基础上按session:namespace:entity_id:suffixes的结构组装带命名空间的键。API 文档示例 from openjiuwen.core.session.checkpointer import build_key key build_key(session1, agent, agent1) print(key) session1:agent:agent1 from openjiuwen.core.session.checkpointer import build_key_with_namespace key build_key_with_namespace(session1, agent, agent1, state) print(key) session1:agent:agent1:state这种冒号分层的键结构让同一会话下的数据天然具备前缀可扫描性是持久化实现中delete_by_prefix、get_by_prefix得以高效工作的基础。7.2 命名空间常量常量值含义SESSION_NAMESPACE_AGENTagentAgent 状态在会话下的命名空间SESSION_NAMESPACE_AGENT_TEAMagent-teamAgent 团队状态在会话下的命名空间源码补充未在 API 文档列出SESSION_NAMESPACE_WORKFLOWworkflow工作流自身状态在会话下的命名空间WORKFLOW_NAMESPACE_GRAPHworkflow-graph图状态在工作流下的命名空间与工作流自身状态分离这些常量定义在 base.py#L78-L86并通过 checkpointer/init.py 统一导出。八、生产接入在 Runner 中配置持久化检查点8.1 RunnerConfig 中的检查点配置RunnerConfigopenjiuwen/core/runner/runner_config.py#L64-L85包含checkpointer_config: Optional[CheckpointerConfig]字段。Runner 启动时openjiuwen/core/runner/runner.py#L277-L304会执行以下流程读取get_runner_config().checkpointer_config若类型为redis先尝试导入 Redis Provider 模块缺失时会提示安装 redis 依赖调用await CheckpointerFactory.create(checkpointer_config)构造实例通过CheckpointerFactory.set_default_checkpointer(checkpointer)将其设为全局默认检查点供后续创建的 Session 使用。8.2 persistence 类型的配置参数PersistenceCheckpointerProvider.createpersistence.py#L976-L1035支持的conf参数如下参数默认值说明db_typesqlite存储后端类型支持sqlite与shelve其他类型抛CHECKPOINTER_CONFIG_ERRORdb_pathcheckpointer数据库/存储文件路径sqlite 下自动补.db后缀db_client无预配置的AsyncEngine实例提供后直接包装为DbBasedKVStoredb_timeout30SQLite 被锁时等待的秒数秒降低OperationalError概率db_enable_walTrue是否启用 SQLite WAL 模式见下对 SQLite 后端实现会先确保父目录存在SQLite 无法自动创建目录再以sqliteaiosqlite:///{db_path}创建异步引擎db_enable_walTrue时通过监听引擎 connect 事件执行PRAGMA journal_modeWALpersistence.py#L962-L973WAL 允许一个写者与多个并发读者并存缓解高并发下的database is locked问题。shelve后端则把路径交给ShelveStore管理。一个完整的持久化配置示例from openjiuwen.core.runner.runner_config import RunnerConfig from openjiuwen.core.session.checkpointer import CheckpointerConfig config RunnerConfig( distributed_modeFalse, checkpointer_configCheckpointerConfig( typepersistence, conf{ db_type: sqlite, db_path: ./runtime/checkpointer.db, db_timeout: 30, db_enable_wal: True, }, ), )8.3 redis 类型的配置参数Redis Provideropenjiuwen/extensions/checkpointer/redis/checkpointer.py#L238-L284的conf结构为{ connection: { redis_client: Redis(...), # 可选预配置客户端 url: redis://..., # 未提供 redis_client 时必填 cluster_mode: True, # 可选集群模式缺省时从 URL 自动探测 connection_args: {...} # 可选额外连接参数 }, ttl: { # 可选 default_ttl: 5, # 可选TTL分钟 refresh_on_read: True # 可选读取时刷新 TTL } }connection中必须提供redis_client或url二者之一否则构造失败。需要说明的是Redis 检查点属于扩展模块使用前需确认已安装对应的 redis 依赖。九、运行机制小结一次完整的中断恢复旅程将上述知识点串联起来一次典型的工作流中断-恢复流程如下工作流执行到需人工介入的节点产生中断标记TASK_STATUS_INTERRUPTpost_workflow_execute检测到中断标记调用工作流 Storage 的save将会话状态与 updates 序列化写入后端同时记录到_session_to_workflow_ids用户提供新的交互输入InteractiveInput后再次触发工作流执行pre_workflow_execute识别到InteractiveInput类型调用recover恢复状态与 updates并把用户输入合并进对应节点的INTERACTIVE_INPUT列表工作流从断点继续最终正常完成时post_workflow_execute清理检查点与图状态一次生命周期闭环。若中途进程崩溃、下次以非交互方式执行同一工作流则会因状态存在但未启用强制删除抛出异常提示开发者显式处理历史现场此时可通过开启环境开关_force_del_workflow_state强制清理后重新执行避免脏状态污染新任务。十、扩展指引与参考路径要接入自定义存储后端只需三步继承CheckpointerProvider实现async create(conf) - Checkpointer用CheckpointerFactory.register(your_type)注册在CheckpointerConfig(typeyour_type, conf{...})中启用或在RunnerConfig.checkpointer_config中配置。相关源码与测试路径速查抽象基类与键工具openjiuwen/core/session/checkpointer/base.py工厂、配置与注册机制openjiuwen/core/session/checkpointer/checkpointer.py内存实现openjiuwen/core/session/checkpointer/inmemory.py持久化实现与 Provideropenjiuwen/core/session/checkpointer/persistence.py包导出openjiuwen/core/session/checkpointer/init.pyRunner 集成openjiuwen/core/runner/runner.py、openjiuwen/core/runner/runner_config.pyRedis 扩展openjiuwen/extensions/checkpointer/redis/checkpointer.py配置脱敏测试tests/unit_tests/core/session/checkpointer/test_checkpointer_config_repr.py总体而言openJiuwen agent-core 的 checkpointer 模块用抽象接口 工厂注册 多种后端的组合为 Agent、Agent 团队与工作流提供了统一而可插拔的状态持久化能力。理解其钩子时序、命名空间键结构与 Provider 扩展点是构建可靠的多轮交互 Agent 与可续跑工作流的关键一步。赞分享人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载相关推荐openJiuwen agent-core 的 Redis 检查点Checkpointer扩展Agent/工作流状态持久化与恢复实战指南openJiuwen agent core 的 Redis 检查点Checkpointer扩展Agent/工作流状态持久化与恢复实战指南 导读 本指南聚焦人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen agent-core 会话检查点Checkpointer机制详解状态持久化、中断恢复与工作流续跑实战指南openJiuwen agent core 会话检查点Checkpointer机制详解状态持久化、中断恢复与工作流续跑实战指南 openJiuwen ag人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen Checkpointer 检查点机制详解Agent 与 Workflow 的状态持久化与中断恢复openJiuwen Checkpointer 检查点机制详解Agent 与 Workflow 的状态持久化与中断恢复 openJiuwen agent co人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习上一篇React Toolbox Input 组件完全指南Material Design 文本字段的实现、配置与源码解析下一篇MKS Monster8完全攻略8轴主板多固件部署×3D打印爱好者的控制核心解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载本文围绕 openJiuwen agent-core 的openjiuwen.core.session.checkpointer模块系统讲解检查点Checkpointer抽象设计、内存与持久化两套实现、工厂注册扩展机制以及键命名规范并结合仓库源码说明其在 Runner 与 Session 中的实际集成方式。读完本文你将掌握如何为 Agent、Agent 团队与工作流配置断点续跑能力理解中断恢复、异常保存与强制清理等关键分支的底层原理并能够基于CheckpointerProvider自定义新的存储后端。检查点模块位于 docs/zh/2.开发指南/API文档/openjiuwen.core/session/checkpointer.md对应的源码实现分散在 openjiuwen/core/session/checkpointer/ 目录下本文以 API 文档为骨架用源码与测试佐证其实现细节。一、Checkpointer 的定位Agent 与工作流的“状态快照层”openJiuwen agent-core 的会话Session承载 Agent、Agent 团队与工作流运行时的全部状态。Checkpointer 的职责正是在执行的关键节点对这份状态做“快照”——保存、恢复、清理从而支撑三类典型场景中断恢复工作流或 Agent 在执行中等待用户交互interrupt时保存状态交互返回后从断点继续异常恢复工作流执行抛出异常时保留现场便于后续排障或重新续跑会话复用同一个session_id:workflow_id线程thread再次执行时恢复上一次的状态。从源码看Checkpointer 与 Storage 是一对抽象基类定义在 openjiuwen/core/session/checkpointer/base.pyCheckpointer面向执行生命周期的高层接口包含工作流/Agent 执行前后的钩子方法Storage面向单个实体状态的低层存取接口封装save/recover/clear/exists四类原子操作。两个内置实现InMemoryCheckpointer与PersistenceCheckpointer分别以进程内存与BaseKVStore键值存储为介质前者适合开发测试后者适合生产部署。此外仓库还提供基于 Redis 的扩展实现见 openjiuwen/extensions/checkpointer/redis/checkpointer.py说明这套抽象具备良好的后端可替换性。二、Checkpointer 抽象基类六个核心钩子与两个辅助方法Checkpointer是一个抽象基类ABC源码位置 openjiuwen/core/session/checkpointer/base.py#L14-L57。API 文档共列出 9 个成员其中 6 个是抽象执行钩子另有session_exists、release、graph_store三个通用能力仓库源码中还额外提供了 Agent 团队team维度的两个钩子。2.1 get_thread_id线程标识staticmethod get_thread_id(session: BaseSession) - str静态方法返回格式为session_id:workflow_id的线程 ID。源码实现即return :.join([session.session_id(), session.workflow_id()])该线程 ID 用于区分同一 Session 下的不同工作流运行实例是后续所有检查点键key的逻辑前缀。2.2 工作流执行钩子pre_workflow_execute / post_workflow_executeabstractmethod async pre_workflow_execute(session: BaseSession, inputs: InteractiveInput) abstractmethod async post_workflow_execute(session: BaseSession, result, exception)pre_workflow_execute在工作流执行前调用其行为由inputs的类型决定这是整个模块中最值得关注的分支逻辑若inputs是InteractiveInput类型说明这是一次带交互输入的续跑直接恢复工作流状态并将新输入合并进会话状态否则非交互输入若工作流状态不存在直接返回走全新执行若状态存在且未启用强制删除则抛出CHECKPOINTER_PRE_WORKFLOW_EXECUTION_ERROR异常提示workflow state exists but non-interactive input and cleanup is disabled——这是为了防止无意中覆盖上次未完成的现场若状态存在且启用了强制删除环境开关_force_del_workflow_state见 openjiuwen/core/session/constants.py#L22 中的FORCE_DEL_WORKFLOW_STATE_KEY则删除图状态与工作流状态后全新执行。post_workflow_execute在工作流执行后调用同样三分支exception不为None保存一次工作流检查点落盘异常现场然后重新抛出异常result中没有中断标记TASK_STATUS_INTERRUPT来自 openjiuwen/core/graph/pregel.py工作流正常结束清理本次工作流的状态与图状态result中存在中断标记工作流因等待交互而暂停保存工作流检查点以便后续恢复。2.3 Agent 执行钩子pre_agent_execute / interrupt_agent_execute / post_agent_executeabstractmethod async pre_agent_execute(session: BaseSession, inputs) abstractmethod async interrupt_agent_execute(session: BaseSession) abstractmethod async post_agent_execute(session: BaseSession)pre_agent_executeAgent 执行前恢复其状态若提供了inputs会把输入写入会话状态中内存实现写入键INTERACTIVE_INPUTinterrupt_agent_executeAgent 需要中断等待用户交互时保存状态这是人工介入-继续执行场景的关键落点post_agent_executeAgent 执行完成后保存最终状态。2.4 session_exists / release / graph_storeabstractmethod async session_exists(session_id: str) - bool abstractmethod async release(session_id: str, agent_id: str None) abstractmethod def graph_store() - Storesession_exists判断指定会话 ID 是否已有检查点数据release释放会话资源。提供agent_id时仅清理该 Agent 的检查点不提供时清理整个会话含工作流、Agent、Agent 团队的全部状态。内存实现还会把以该 session_id 为前缀的 Agent 存储一并移除见 inmemory.py#L365-L367graph_store返回图状态GraphState存储对象。工作流的图状态与工作流自身状态分开存放二者通过不同的命名空间隔离。2.5 源码中的补充Agent 团队钩子API 文档未列出的pre_agent_team_execute/post_agent_team_execute在抽象基类中同样存在base.py#L31-L45用于 Agent 团队team维度的状态恢复与保存其状态来源是会话的全局状态global_state。这说明检查点体系覆盖了 Agent、Agent 团队、工作流三个层级。三、Storage 抽象save / recover / clear / exists 四元组Storage是更底层的状态存取接口base.py#L60-L75abstractmethod async save(session: BaseSession) abstractmethod async recover(session: BaseSession, inputs: InteractiveInput None) abstractmethod async clear(session_id: str) abstractmethod async exists(session: BaseSession) - boolsave将当前会话状态序列化后写入存储recover从存储中反序列化并恢复会话状态可携带可选交互输入clear按会话 ID 清除状态exists判断指定会话的状态是否存在。两套 Checkpointer 实现各自配套了针对 Agent、Agent 团队、工作流的 Storage 实现。序列化统一走create_serializer(pickle)序列化结果为(dump_type, blob)二元组即序列化类型 字节内容便于反序列化时还原类型信息。四、InMemoryCheckpointer开发测试场景的默认选择4.1 内部结构与适用场景class openjiuwen.core.session.checkpointer.InMemoryCheckpointer()基于内存的检查点实现所有状态保存在进程内字典中进程重启后状态丢失适用于开发和测试场景。其内部维护了四类存储见 inmemory.py#L23-L30_agent_stores按 session_id 组织的 Agent 状态存储_agent_team_stores按 session_id 组织的 Agent 团队状态存储_workflow_stores按 session_id 组织的工作流状态存储_graph_storeInMemoryStore()图状态存储_session_to_workflow_ids记录 session 下已保存检查点的工作流 ID 集合用于release时精确清理。使用样例来自 API 文档可直接运行 from openjiuwen.core.session.checkpointer import InMemoryCheckpointer checkpointer InMemoryCheckpointer() # 使用 checkpointer 进行状态管理在 openJiuwen 中InMemoryCheckpointer同时也是默认检查点工厂模块底部实例化了全局单例default_inmemory_checkpointercheckpointer.py#L134任何未显式配置的场景都会回落到它。4.2 执行钩子的实现细节内存实现的钩子方法行为与抽象定义完全对齐并带完整的日志埋点事件类型如CHECKPOINT_SAVE、CHECKPOINT_RESTORE、CHECKPOINT_CLEAR、CHECKPOINTER_STORE_ADD、CHECKPOINTER_STORE_REMOVE便于通过日志追踪检查点生命周期pre_workflow_execute首次创建存储时打CHECKPOINTER_STORE_ADD日志InteractiveInput分支执行workflow_store.recover(session, inputs)恢复非交互分支按 2.2 节的三分支逻辑执行post_workflow_execute异常时调用_inner_save_workflow_checkpoint保存现场正常完成时调用_inner_clear_workflow_session清理中断时保存检查点。若工作流的父会话不是AgentSession完成清理后还会移除该 session 的工作流存储inmemory.py#L96-L99interrupt_agent_execute/post_agent_execute调用对应 AgentStorage 的save落状态release按agent_id粒度或整个会话粒度清理且会同步删除图状态。五、PersistenceCheckpointer基于 BaseKVStore 的持久化实现5.1 构造与适用后端class openjiuwen.core.session.checkpointer.PersistenceCheckpointer(kv_store: BaseKVStore)基于持久化存储的检查点实现通过BaseKVStore接口对接任意键值存储后端。BaseKVStore抽象定义于 openjiuwen/core/foundation/store/base_kv_store.py#L16提供pipeline()、batch_delete()、delete_by_prefix()等批量操作原语仓库已内置ShelveStoreshelve_store.py与DbBasedKVStoredb_based_kv_store.py两类实现。API 文档给出的经典样例 from openjiuwen.core.session.checkpointer import PersistenceCheckpointer from openjiuwen.core.foundation.store.kv import ShelveStore kv_store ShelveStore(checkpoint.db) checkpointer PersistenceCheckpointer(kv_store) # 使用 checkpointer 进行状态管理5.2 键结构设计三层命名空间隔离持久化实现把状态拆成多个 KV 条目键统一由session_id、命名空间、实体 ID 与后缀构成具体见 persistence.py实体命名空间键示例Agent 状态agentsession1:agent:agent1:agent_state_blobsAgent 团队状态agent-teamsession1:agent-team:team1:agent_team_state_blobs工作流自身状态workflowsession1:workflow:wf1:workflow_state_blobs工作流图状态workflow-graphsession1:workflow-graph:wf1:checkpoint_data_value工作流的自身状态与图状态分离存放前者存会话状态与状态更新updates后者存图执行节点状态GraphState这正是WORKFLOW_NAMESPACE_GRAPH workflow-graph常量的意义。图状态存储类GraphStore实现了 openjiuwen/core/graph/store/base.py#L41-L51 中Store接口的get/save/delete其中delete(session_id, nsNone)在ns为空时会按前缀session_id:workflow-graph级联删除该会话下所有图状态。5.3 读写性能细节pipeline 批量化持久化 Storage 的所有读写都通过pipeline()批量执行Agent / Agent 团队状态存取涉及 2 个键dump_typeblobexists要求两个键同时存在才判定状态存在工作流状态存取涉及 4 个键状态与 updates 各一对exists只要求状态键存在updates 是可选附加数据release未指定agent_id时直接delete_by_prefix(session_id :)一次清空该会话全部键。5.4 序列化与容错状态序列化格式为(dump_type, bytes)写入前通过_serde.dumps_typed(state)得到二元组读出时若blob为字符串则按 base64 解码回字节再反序列化兼容不同 KV 后端对字节的存储方式。反序列化失败会被记录CHECKPOINT_ERROR日志并返回None不会直接抛错中断流程具备一定的容错性。六、CheckpointerFactory 工厂注册、创建与默认实例管理6.1 工厂职责与注册机制classmethod register(name: str) classmethod async create(checkpointer_conf: CheckpointerConfig) - Checkpointer classmethod set_default_checkpointer(checkpointer: Checkpointer) classmethod set_checkpointer(store_type: str, checkpointer: Checkpointer) classmethod get_checkpointer(store_type: Optional[str] None) - CheckpointerCheckpointerFactorycheckpointer.py#L60-L125以注册表模式管理各种类型的检查点。register(name)返回一个装饰器用于把CheckpointerProvider子类按名称注册进_registry字典。API 文档示例 from openjiuwen.core.session.checkpointer import CheckpointerFactory, CheckpointerProvider CheckpointerFactory.register(custom) class CustomCheckpointerProvider(CheckpointerProvider): ... async def create(self, conf: dict) - Checkpointer: ... # 创建自定义检查点实例 ... return CustomCheckpointer()create先按type找到已注册的 Provider再调用其create(conf)异步构造实例未注册的类型会直接抛出异常。create内部会先导入 persistence 模块import openjiuwen.core.session.checkpointer.persistence as _确保persistence类型始终完成注册。内置注册的两个 Providerin_memoryInMemoryCheckpointerProvider无论conf内容如何都返回全局默认的内存检查点单例persistencePersistenceCheckpointerProvider根据配置构造持久化实例。仓库扩展目录 openjiuwen/extensions/checkpointer/redis/checkpointer.py#L229-L230 还通过同样方式注册了redis类型支持单机与集群两种模式。6.2 get_checkpointer 的解析优先级get_checkpointer(store_typeNone)的查找顺序与 API 文档描述一致源码见 checkpointer.py#L98-L125传入store_type时先查set_checkpointer手动设置的实例映射_type_checkpointersstore_type in_memory且无手动设置返回默认内存检查点单例否则返回set_default_checkpointer设置的默认实例若从未设置最终回落到默认内存检查点。该工厂被 Session 层直接消费AgentSession在构造时默认调用CheckpointerFactory.get_checkpointer()获取检查点openjiuwen/core/session/internal/agent.py#L45NodeSession、WorkflowSession则把调用委托给父会话openjiuwen/core/session/internal/workflow.py#L72-L77、#L169-L171。6.3 CheckpointerConfig配置对象与敏感信息脱敏class CheckpointerConfig(type: str in_memory, conf: dict {})CheckpointerConfig是 pydantic 模型type默认in_memoryconf为 Provider 透传的配置字典。值得注意的安全细节该类的__repr__与__str__会递归扫描conf中的字符串对形如scheme://的 URL 调用redact_url_password将密码替换为***checkpointer.py#L27-L51防止 Redis、数据库等连接串中的凭据泄漏到日志。对应的单元测试见 tests/unit_tests/core/session/checkpointer/test_checkpointer_config_repr.py覆盖 URL 密码脱敏、无密码 URL 原样保留、嵌套字典/列表递归脱敏等场景 from openjiuwen.core.session.checkpointer import CheckpointerFactory, CheckpointerConfig config CheckpointerConfig(typein_memory, conf{}) checkpointer await CheckpointerFactory.create(config)七、键构建工具与命名空间常量7.1 build_key 与 build_key_with_namespacefunc build_key(*parts: str) - str func build_key_with_namespace(session_id: str, namespace: str, entity_id: str, *suffixes: str) - strbuild_key用冒号连接任意多个字符串片段build_key_with_namespace在其基础上按session:namespace:entity_id:suffixes的结构组装带命名空间的键。API 文档示例 from openjiuwen.core.session.checkpointer import build_key key build_key(session1, agent, agent1) print(key) session1:agent:agent1 from openjiuwen.core.session.checkpointer import build_key_with_namespace key build_key_with_namespace(session1, agent, agent1, state) print(key) session1:agent:agent1:state这种冒号分层的键结构让同一会话下的数据天然具备前缀可扫描性是持久化实现中delete_by_prefix、get_by_prefix得以高效工作的基础。7.2 命名空间常量常量值含义SESSION_NAMESPACE_AGENTagentAgent 状态在会话下的命名空间SESSION_NAMESPACE_AGENT_TEAMagent-teamAgent 团队状态在会话下的命名空间源码补充未在 API 文档列出SESSION_NAMESPACE_WORKFLOWworkflow工作流自身状态在会话下的命名空间WORKFLOW_NAMESPACE_GRAPHworkflow-graph图状态在工作流下的命名空间与工作流自身状态分离这些常量定义在 base.py#L78-L86并通过 checkpointer/init.py 统一导出。八、生产接入在 Runner 中配置持久化检查点8.1 RunnerConfig 中的检查点配置RunnerConfigopenjiuwen/core/runner/runner_config.py#L64-L85包含checkpointer_config: Optional[CheckpointerConfig]字段。Runner 启动时openjiuwen/core/runner/runner.py#L277-L304会执行以下流程读取get_runner_config().checkpointer_config若类型为redis先尝试导入 Redis Provider 模块缺失时会提示安装 redis 依赖调用await CheckpointerFactory.create(checkpointer_config)构造实例通过CheckpointerFactory.set_default_checkpointer(checkpointer)将其设为全局默认检查点供后续创建的 Session 使用。8.2 persistence 类型的配置参数PersistenceCheckpointerProvider.createpersistence.py#L976-L1035支持的conf参数如下参数默认值说明db_typesqlite存储后端类型支持sqlite与shelve其他类型抛CHECKPOINTER_CONFIG_ERRORdb_pathcheckpointer数据库/存储文件路径sqlite 下自动补.db后缀db_client无预配置的AsyncEngine实例提供后直接包装为DbBasedKVStoredb_timeout30SQLite 被锁时等待的秒数秒降低OperationalError概率db_enable_walTrue是否启用 SQLite WAL 模式见下对 SQLite 后端实现会先确保父目录存在SQLite 无法自动创建目录再以sqliteaiosqlite:///{db_path}创建异步引擎db_enable_walTrue时通过监听引擎 connect 事件执行PRAGMA journal_modeWALpersistence.py#L962-L973WAL 允许一个写者与多个并发读者并存缓解高并发下的database is locked问题。shelve后端则把路径交给ShelveStore管理。一个完整的持久化配置示例from openjiuwen.core.runner.runner_config import RunnerConfig from openjiuwen.core.session.checkpointer import CheckpointerConfig config RunnerConfig( distributed_modeFalse, checkpointer_configCheckpointerConfig( typepersistence, conf{ db_type: sqlite, db_path: ./runtime/checkpointer.db, db_timeout: 30, db_enable_wal: True, }, ), )8.3 redis 类型的配置参数Redis Provideropenjiuwen/extensions/checkpointer/redis/checkpointer.py#L238-L284的conf结构为{ connection: { redis_client: Redis(...), # 可选预配置客户端 url: redis://..., # 未提供 redis_client 时必填 cluster_mode: True, # 可选集群模式缺省时从 URL 自动探测 connection_args: {...} # 可选额外连接参数 }, ttl: { # 可选 default_ttl: 5, # 可选TTL分钟 refresh_on_read: True # 可选读取时刷新 TTL } }connection中必须提供redis_client或url二者之一否则构造失败。需要说明的是Redis 检查点属于扩展模块使用前需确认已安装对应的 redis 依赖。九、运行机制小结一次完整的中断恢复旅程将上述知识点串联起来一次典型的工作流中断-恢复流程如下工作流执行到需人工介入的节点产生中断标记TASK_STATUS_INTERRUPTpost_workflow_execute检测到中断标记调用工作流 Storage 的save将会话状态与 updates 序列化写入后端同时记录到_session_to_workflow_ids用户提供新的交互输入InteractiveInput后再次触发工作流执行pre_workflow_execute识别到InteractiveInput类型调用recover恢复状态与 updates并把用户输入合并进对应节点的INTERACTIVE_INPUT列表工作流从断点继续最终正常完成时post_workflow_execute清理检查点与图状态一次生命周期闭环。若中途进程崩溃、下次以非交互方式执行同一工作流则会因状态存在但未启用强制删除抛出异常提示开发者显式处理历史现场此时可通过开启环境开关_force_del_workflow_state强制清理后重新执行避免脏状态污染新任务。十、扩展指引与参考路径要接入自定义存储后端只需三步继承CheckpointerProvider实现async create(conf) - Checkpointer用CheckpointerFactory.register(your_type)注册在CheckpointerConfig(typeyour_type, conf{...})中启用或在RunnerConfig.checkpointer_config中配置。相关源码与测试路径速查抽象基类与键工具openjiuwen/core/session/checkpointer/base.py工厂、配置与注册机制openjiuwen/core/session/checkpointer/checkpointer.py内存实现openjiuwen/core/session/checkpointer/inmemory.py持久化实现与 Provideropenjiuwen/core/session/checkpointer/persistence.py包导出openjiuwen/core/session/checkpointer/init.pyRunner 集成openjiuwen/core/runner/runner.py、openjiuwen/core/runner/runner_config.pyRedis 扩展openjiuwen/extensions/checkpointer/redis/checkpointer.py配置脱敏测试tests/unit_tests/core/session/checkpointer/test_checkpointer_config_repr.py总体而言openJiuwen agent-core 的 checkpointer 模块用抽象接口 工厂注册 多种后端的组合为 Agent、Agent 团队与工作流提供了统一而可插拔的状态持久化能力。理解其钩子时序、命名空间键结构与 Provider 扩展点是构建可靠的多轮交互 Agent 与可续跑工作流的关键一步。赞分享人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载相关推荐openJiuwen agent-core 的 Redis 检查点Checkpointer扩展Agent/工作流状态持久化与恢复实战指南openJiuwen agent core 的 Redis 检查点Checkpointer扩展Agent/工作流状态持久化与恢复实战指南 导读 本指南聚焦人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen agent-core 会话检查点Checkpointer机制详解状态持久化、中断恢复与工作流续跑实战指南openJiuwen agent core 会话检查点Checkpointer机制详解状态持久化、中断恢复与工作流续跑实战指南 openJiuwen ag人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen Checkpointer 检查点机制详解Agent 与 Workflow 的状态持久化与中断恢复openJiuwen Checkpointer 检查点机制详解Agent 与 Workflow 的状态持久化与中断恢复 openJiuwen agent co人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习上一篇React Toolbox Input 组件完全指南Material Design 文本字段的实现、配置与源码解析下一篇MKS Monster8完全攻略8轴主板多固件部署×3D打印爱好者的控制核心解决方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
