人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载本文围绕com.openjiuwen.core.graph.store.InMemoryStore展开。它实现了 agent-core-java 图执行引擎Pregel的图状态存储接口Store用两级ConcurrentHashMap在内存中维护“会话 → 命名空间 → 状态快照”的图执行状态是工作流中断恢复、会话检查点等能力的内存级实现底座。读完本文你能理解get/save/delete三个方法的语义与深拷贝细节并能将其接入InMemoryCheckpointer、GraphStore装饰器等真实调用链中使用。定位Store 抽象与 InMemoryStore 的职责agent-core-java 的图状态存储以 Store 接口为最小契约只定义三个异步方法public interface Store { CompletionStageOptionalGraphStoreState get(String sessionId, String ns); CompletionStageVoid save(String sessionId, String ns, GraphStoreState state); CompletionStageVoid delete(String sessionId, String ns); }该接口是异步图状态存储抽象对应 Python 版openjiuwen/core/graph/store/base.py中的Store三个操作分别是读取、保存和删除。所有实现都以CompletionStage返回异步结果便于同步与异步存储后端如 Redis、KV 存储共用同一套调用代码。InMemoryStore是该接口的内存实现类声明与注释见 InMemoryStore.javapublic class InMemoryStore implements Store { private final MapString, MapString, GraphStoreState storeCk new ConcurrentHashMap();源码注释明确其定位“In-memory implementation of the graph state store”适合测试或单进程场景。它不依赖任何外部存储进程退出后状态即丢失因此主要用于本地调试、单元测试以及短生命周期会话的检查点保存。存储的数据结构两级 Map唯一字段storeCk是一个MapString, MapString, GraphStoreState类型的多级映射字段类型默认值说明storeCkMapString, MapString, GraphStoreStatenew ConcurrentHashMap()sessionId - (ns - state)的多级内存映射两级均为ConcurrentHashMap保证并发读写安全。其中外层 key 为sessionId会话/对话标识如conv_123一个会话下可同时保存多个命名空间的图状态内层 key 为nsnamespace通常对应 graphId/工作流 idvalue 为一份 GraphStoreState 状态快照。每份快照包含恢复/续跑 Pregel 图所需的全部信息字段类型说明nsString状态所属命名空间stepint当前已持久化的 super-step 编号channelValuesMapString, Object已落盘的 channel 快照值pendingBufferListMessage尚未完全消费的消息缓冲区pendingNodeMapString, PendingNode待恢复节点信息映射节点名 状态 异常列表nodeVersionMapString, Integer节点版本号映射三个核心方法详解get返回状态副本Override public CompletionStageOptionalGraphStoreState get(String sessionId, String ns) { MapString, GraphStoreState sessionMap storeCk.get(sessionId); if (sessionMap null) { return CompletableFuture.completedFuture(Optional.empty()); } GraphStoreState state sessionMap.get(ns); return CompletableFuture.completedFuture(Optional.ofNullable(deepCopy(state))); }行为要点见 InMemoryStore.java#L26-L34sessionId不存在时直接返回已完成 future 包装的空Optional不会抛异常命中时返回的是deepCopy(state)生成的状态副本而不是存储中的原对象调用方对返回值的修改不会污染存储内容由于底层是内存 Map该实现总是返回已完成completed的CompletableFuture异步签名仅为满足Store接口的统一契约。save以深拷贝方式写入Override public CompletionStageVoid save(String sessionId, String ns, GraphStoreState state) { storeCk.computeIfAbsent(sessionId, ignored - new ConcurrentHashMap()) .put(ns, deepCopy(state)); return CompletableFuture.completedFuture(null); }computeIfAbsent会按需懒创建该 session 的内层 Map因此首次保存某会话无需任何初始化。写入同样走深拷贝存储中持有的是快照的独立副本调用方后续对传入state的修改不会影响已保存内容。这与get的副本语义配合形成“读副本 写副本”的双向隔离。deepCopy的具体实现见 InMemoryStore.java#L71-L83private static GraphStoreState deepCopy(GraphStoreState state) { if (state null) { return null; } return GraphStoreState.create( state.getNs(), state.getStep(), new HashMap(state.getChannelValues()), new java.util.ArrayList(state.getPendingBuffer()), new HashMap(state.getPendingNode()), new HashMap(state.getNodeVersion()) ); }需要注意的是这里是集合级别的浅拷贝新建了HashMap/ArrayList容器但channelValues中的值对象本身不做递归复制且GraphStoreState的 getter 本身返回的是防御性拷贝如getChannelValues()每次new LinkedHashMap(channelValues)见 GraphStoreState.java#L70-L84。对于检查点场景状态在保存前已定型、恢复后重新执行这一深度是足够的。delete整会话删除与前缀匹配删除Override public CompletionStageVoid delete(String sessionId, String ns) { MapString, GraphStoreState sessionMap storeCk.get(sessionId); if (sessionMap null) { return CompletableFuture.completedFuture(null); } if (ns null) { storeCk.remove(sessionId); return CompletableFuture.completedFuture(null); } deleteNsByPrefix(sessionMap, ns); if (sessionMap.isEmpty()) { storeCk.remove(sessionId); } return CompletableFuture.completedFuture(null); }delete的语义分两档ns nullstoreCk.remove(sessionId)删除整个 session即该会话下所有命名空间的状态一次性清空ns ! null按startsWith(prefix)前缀规则批量删除。deleteNsByPrefixInMemoryStore.java#L62-L69遍历内层 Map 的 keySet凡是iterator.next().startsWith(prefix)的命名空间条目都被移除——即删除apple会同时删掉apple和apple:orange这类子命名空间前缀删除后若内层 Map 变空外层的 session 条目也会被回收避免空壳 Map 常驻内存。对不存在的 session 调用delete是幂等安全的空操作。真实调用链从 Checkpointer 到 Pregel 恢复InMemoryStore不是孤立组件它在会话检查点体系中承担图状态层的存储角色InMemoryCheckpointer 持有实例InMemoryCheckpointer.java#L69 中private final Store graphStore new InMemoryStore();并通过graphStore()方法对外暴露。检查点器还会在会话释放release、工作流结束clearWorkflowSession以及 TTL/容量驱逐evictSession时调用graphStore.delete(...)清理对应状态——这里正是ns null删除整个 session 语义的用武之地。编译时包装为 GraphStorePregelGraph.java#L161-L168 在运行时从 session 的 checkpointer 取出底层Store包装成new GraphStore(checkpointer.graphStore())。GraphStore 是日志装饰器在每次 get/save/delete 前后打印 super-step 编号、sessionId 等诊断日志并在构造时自动用 KeyLockedStore 加上按 sessionId 分条默认 64 条 stripe的写锁保证同一会话的读/写/删操作互斥、不同会话并发执行。PregelLoop 恢复与保存PregelLoop.java#L51-L56 在循环启动时按sessionId ns读取既有快照实现续跑执行出错时则构造 error state 后saver.save(...)落盘便于后续排查与恢复。一个典型组合是InMemoryCheckpointer内存会话检查点→InMemoryStore图状态→GraphStore日志 会话级锁→PregelLoop消费快照。使用示例下面这段代码与 GraphMemoryExampleSupport 中的saveCheckpoint/loadCheckpoint及单元测试中的用法一致可直接复制InMemoryStore store new InMemoryStore(); // 1. 构造一份图状态快照channel 值、待消费消息、待恢复节点 GraphStoreState checkpoint GraphStoreState.create( default, // ns 1, // 当前 super-step Map.of(ch1, 42), // channelValues List.of(new Message(, , pending msg)), // pendingBuffer Map.of(node1, new PendingNode(n1, running)), // pendingNode null); // nodeVersion 传 null 会得到空 Map // 2. 异步保存内存实现返回已完成 futurejoin 立即返回 store.save(conv_123, checkpoint.getNs(), checkpoint) .toCompletableFuture().join(); // 3. 读取副本 OptionalGraphStoreState loaded store.get(conv_123, default) .toCompletableFuture().join(); loaded.ifPresent(state - System.out.println(step state.getStep() , pending state.getPendingBuffer().size())); // 4. 删除传 ns 按前缀删除传 null 删除整个会话 store.delete(conv_123, default).toCompletableFuture().join(); store.delete(conv_123, null).toCompletableFuture().join();注意GraphStoreState.create对null的集合字段会填充默认空集合构造器内channelValues ! null ? new LinkedHashMap(channelValues) : new LinkedHashMap()因此示例中nodeVersion传null是安全的。测试验证GraphStoreTest该类的行为由 GraphStoreTest 完整覆盖与 Python 版test_memory_checkpoint_saver_basic对齐关键用例包括inMemoryStoreSavesLoadsAndDeletesCheckpoint完整走一遍 save → get → deletens 为 null→ 验证get返回空Optional并逐项断言step、channelValues、pendingBufferpayload、PendingNode的节点名与状态inMemoryStoreDeletesNamespaceAndChildrenByPrefix同时保存apple与apple:orange两个命名空间执行delete(conv_123, apple)后两者均消失直接验证了前缀删除语义graphStoreWrapperSavesThroughInMemoryStore验证GraphStore装饰器包裹InMemoryStore后保存的数据可被原 store 正确读回graphStoreDelegatesAsyncOperations/graphStorePropagatesDelegateFailures验证异步委托与底层失败的异常传播CompletionException包装后透传。适用场景与限制结合源码可以确认InMemoryStore的定位与边界适用单元测试、本地调试、单进程短生命周期会话InMemoryCheckpointer的默认组合就是纯内存链路。它的并发安全性来自ConcurrentHashMap多线程 get/save/delete 不会破坏 Map 结构同会话的有序性则由外层KeyLockedStore的条纹锁保障。限制数据仅存在于 JVM 堆内进程重启即丢失不能用于需要跨进程/重启存活的生产断点恢复生产可替换为PersistenceCheckpointer/RedisCheckpointer等提供其他Store实现的检查点器如 PersistenceCheckpointer 中的PersistenceGraphStore接口契约不变没有内置容量上限与 TTL。内存治理由外层InMemoryCheckpointer的 Guava Cache 驱逐策略默认 TTL 7 天、最多 100 个会话驱逐时回调graphStore.delete(tid, null)清理图状态负责直接裸用InMemoryStore则需自行控制生命周期深拷贝是集合容器级别的channelValues内部复杂对象不做递归复制调用方不应在保存后继续修改这些共享对象并期望存储内容随之变化也不会副本已隔离但也不要假设存储副本与源对象完全独立于引用层面。从源码结构看InMemoryStore与 Python 版openjiuwen/core/graph/store/inmemory.py行为对齐类注释中的 “Mirrors Pythons InMemoryStore” 说明了这一设计渊源Java 侧以此为基线实现了图状态存储层的语义一致性。赞分享人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载相关推荐openJiuwen agent-core-java 图状态持久化存储graph.store全面解析从 Store 抽象到 Pregel 恢复续跑openJiuwen agent core java 图状态持久化存储graph.store全面解析从 Store 抽象到 Pregel 恢复续跑 导读人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习BareMetal OS安全特性分析exokernel架构如何提供更强的安全隔离BareMetal OS安全特性分析exokernel架构如何提供更强的安全隔离 BareMetal OS是一个基于exokernel架构的轻量级操作系统专操作系统PyTorch Lightning 保存与恢复 DataModule 状态基于 state_dict / load_state_dict 的断点续训实践PyTorch Lightning 保存与恢复 DataModule 状态基于 state_dict / load_state_dict 的断点续训实践 导读人工智能深度学习机器学习预训练分布式训练微调创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考