openJiuwen agent-core-java 检索批处理回调基类 BaseCallback：统计设计、内置子类与自定义实践

人工智能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 检索模块的 BaseCallback 展开它是在批量 Embedding 与文档索引流程中用于统计「批次调用次数」的回调基类。通过阅读本文你将掌握 BaseCallback 的构造方式、call/getCallCounter核心方法、线程安全实现原理、LoggingCallback与TqdmCallback两个内置子类的行为差异以及如何自定义回调并将其接入EmbedChunks、ChromaIndexer、MilvusIndexer等批处理流程。BaseCallback 的定位只统计不输出在 openJiuwen 的检索retrieval能力中当批量文档需要被切分、向量化并写入索引库时批处理是高频操作。BaseCallback 正是为这类场景设计的批处理回调基类其职责边界非常清晰只负责统计记录批次调用次数与总任务量不负责输出不写日志、不渲染进度条——这类副作用由子类实现。这一点在类注释中也有印证BaseCallback 的 Java 实现镜像自 Python 侧openjiuwen/core/retrieval/common/callbacks.py的BaseCallback源码路径为 BaseCallback.java。类签名与构造方法BaseCallback 是一个普通的公共类声明如下public class BaseCallback当前仓库源码提供如下构造入口签名说明public BaseCallback(Iterable? seq)接收任意可迭代集合如List用于初始化总任务量相关的上下文。public BaseCallback(Iterable? seq) { }从源码看构造函数本身不保存 seq 的引用也不做复制传入集合的主要作用是语义化声明“这个回调将服务于多少个文档的批处理”为子类如 TqdmCallback在构造时感知任务长度提供统一入口。官方文档同时描述了无参用法此时任务量为 0实践中也可直接new BaseCallback()配合后续批次回调使用。核心字段与方法线程安全的批次计数字段设计protected final AtomicInteger callCounter new AtomicInteger(); protected final ReentrantLock threadLock new ReentrantLock();callCounter使用AtomicInteger承载批次计数器保证多线程并发累加时的原子性threadLockReentrantLock可重入锁用于在回调执行期间保护计数更新临界区。这一设计并非冗余openJiuwen 的 API Embedding 采用异步并发批处理多个批次通过线程池并行执行见 APIEmbedding.java回调对象会被多个批次线程共享因此计数必须线程安全。核心方法public void call(int startIdx, int endIdx, ListString batch) { threadLock.lock(); try { callCounter.incrementAndGet(); } finally { threadLock.unlock(); } } public int getCallCounter() { return callCounter.get(); }方法说明call(int startIdx, int endIdx, ListString batch)处理一个批次并累计调用次数。startIdx/endIdx标识本批次在全文中的起始/结束下标前闭后开batch为本次实际处理的文本列表基类默认实现只做1计数不产生任何输出。getCallCounter()返回已回调的批次数即callCounter.get()。需要说明的是仓库文档BaseCallback.md中记录了onBatch与getTotal两个方法名当前 Java 源码中对应实现为call与getCallCounter计数即总量语义。若按文档编写代码请以本文所给源码签名为准。内置子类日志回调与进度条回调BaseCallback 自身不产生输出但仓库内置了两个开箱即用的子类均位于retrieval/src/main/java/com/openjiuwen/retrieval/common/LoggingCallbackSLF4J 日志进度public class LoggingCallback extends BaseCallback { private static final Logger LOG LoggerFactory.getLogger(LoggingCallback.class); public LoggingCallback(int total, String desc) { super(List.of()); this.total Math.max(total, 0); this.desc desc null || desc.isBlank() ? Indexing : desc; } Override public void call(int startIdx, int endIdx, ListString batch) { super.call(startIdx, endIdx, batch); LOG.info({} progress: {}/{}, desc, Math.min(endIdx, total), total); } }要点对应 LoggingCallback.java构造参数total为总任务量会被Math.max(total, 0)归一化desc为进度描述默认Indexing覆写call时先调用super.call(...)保证计数递增再输出INFO级日志desc progress: endIdx/total适合在无终端 UI、希望以日志形式跟踪索引进度的服务端场景。TqdmCallback模拟 Python tqdm 进度语义public class TqdmCallback extends BaseCallback { private final int length; private boolean closed; public TqdmCallback(List? seq, boolean useRich, String desc) { super(seq); this.length seq null ? 0 : seq.size(); this.closed length 0; } Override public void call(int startIdx, int endIdx, ListString batch) { threadLock.lock(); try { callCounter.incrementAndGet(); if (callCounter.get() length) { closed true; } } finally { threadLock.unlock(); } } }要点对应 TqdmCallback.java构造时从传入集合大小得到length即总批次数目标useRich、desc保留 Python 侧参数形态以保持镜像一致性每次call递增计数当callCounter length时置closed true表示全部批次已处理完毕提供length()与isClosed()两个辅助方法便于外层判断进度是否完成。回调如何进入批处理管线1. Embedding 批处理入口APIEmbeddingAPIEmbedding.java 是文本向量化并发批处理的核心。在embedDocuments/embedDocumentsSync中它会按批次大小切分文本并为每个批次启动异步任务BaseCallback callback resolveCallback(kwargs, indices); ListCompletableFutureBatchResult futures new ArrayList(indices.size()); for (Integer startIdx : indices) { futures.add(processBatchAsync(nonEmpty, startIdx, effectiveBatchSize, kwargs, callback, false)); }其中resolveCallback的解析规则源码位置若kwargs为空 → 直接new BaseCallback(sequence)若kwargs[callback]已是BaseCallback实例 → 直接复用该实例若kwargs[callback_cls]是类对象 → 反射实例化构造器按优先级尝试Iterable单参构造 →(List, boolean, String)三参构造 → 无参构造实例化失败则抛出RETRIEVAL_EMBEDDING_CALLBACK_INVALID错误。每个批次处理完成后会调用callback.call(startIdx, endIdx, batchTexts)同步模式下传向量化结果最终getCallCounter()即等于已完成的批次数量。同时cleanPayloadOptions会把callback/callback_cls从请求 payload 中剔除确保它们不会作为参数发送给 Embedding API。2. 文档索引管线EmbedChunksEmbedChunks.java 负责把切分后的TextChunk列表批量向量化其公开方法public static CompletableFutureVoid computeChunkEmbeddings( ListTextChunk chunks, Embedding embedModel, Class? extends BaseCallback docIndexCallback, boolean useCaptionForImages)docIndexCallback会被放入kwargs的callback_cls键传给 Embedding 模型MapString, Object kwargs docIndexCallback null ? Map.of() : Map.of(callback_cls, docIndexCallback); return embedModel.embedDocuments(texts, null, kwargs)...也就是说只要在索引构建时传入一个BaseCallback子类的 Class 对象批量向量化过程中的每一次批次调用都会被自动计数。3. 向量索引构建ChromaIndexer 与 MilvusIndexerChromaIndexer.java 的normalizeCallbackdocIndexCallback为空时默认使用TqdmCallback.class并强制校验必须继承自BaseCallback否则抛出RETRIEVAL_EMBEDDING_CALLBACK_INVALIDMilvusIndexer.java 同样以Class? extends BaseCallback docIndexCallback字段贯穿多个构造函数并提供getDocIndexCallback()读取。4. 懒加载符号表BaseCallback 还登记在 LazyLoad.java 的符号表中BaseCallback - com.openjiuwen.core.retrieval.common.BaseCallback说明它是检索模块对外暴露的标准公共 API。自定义回调与测试佐证继承 BaseCallback 即可自定义统计逻辑无需关心批次切分细节。仓库测试给出了两种典型写法写法一空实现子类仅计数APIEmbeddingTest.java 中的CountingCallbackprivate static final class CountingCallback extends BaseCallback { CountingCallback(List? seq) { super(seq); } }该测试用CountingCallback校验cleanPayloadOptions会剥离回调键、以及回调可作为callback实例传入 Embedding 流程。写法二带参构造 反射实例化ChromaIndexerTest.java、MilvusIndexerTest.java 中的RecordingCallback则作为callback_cls传入索引构建验证resolveCallback的反射构造路径。行为验证RetrievalCommonModelsTest.java 直接验证了计数与关闭语义BaseCallback baseCallback new BaseCallback(List.of(a, b)); baseCallback.call(0, 1, List.of(a)); assertThat(baseCallback.getCallCounter()).isEqualTo(1); TqdmCallback tqdmCallback new TqdmCallback(List.of(a, b), false, Indexing); tqdmCallback.call(0, 1, List.of(a)); tqdmCallback.call(1, 2, List.of(b)); assertThat(tqdmCallback.getCallCounter()).isEqualTo(2); assertThat(tqdmCallback.isClosed()).isTrue();此外 EmbedChunksTest.java 还验证了将BaseCallback.class直接作为回调类传入computeChunkEmbeddings时callback_cls会被正确写入 Embedding 调用的参数中。实践建议与注意事项选择默认实现多数索引场景直接用TqdmCallback.classChromaIndexer 的默认值即可获得完整进度语义需要落日志时改用LoggingCallback。自定义回调三要素继承BaseCallback按需提供Iterable单参、(List, boolean, String)三参或无参构造resolveCallback会依次尝试覆写call时先super.call(...)保证计数正确。线程安全基类已内置AtomicInteger ReentrantLock自定义子类若自行修改计数状态务必同样使用锁或原子类避免并发批次下计数丢失。命名差异仓库文档中的onBatch/getTotal对应源码实现为call/getCallCounter请以 BaseCallback.java 的实际签名为准。BaseCallback 以极小的 API 面一个回调方法 一个计数器撑起了 openJiuwen 检索模块从 Embedding 批处理到 Chroma/Milvus 索引构建的进度统计能力理解它即可轻松扩展出贴合自身业务的自定义批处理回调。赞分享人工智能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 知识检索批处理进度日志LoggingCallback 类原理与实战openJiuwen agent core java 知识检索批处理进度日志LoggingCallback 类原理与实战 本文聚焦 openJiuwen ag人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen agent-core-java 工作流异常处理WorkflowError 类的设计、用法与语义openJiuwen agent core java 工作流异常处理WorkflowError 类的设计、用法与语义 本文围绕 openJiuwen agen人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen agent-core 索引处理器体系Processor 抽象基类设计与文档索引 Pipeline 实战openJiuwen agent core 索引处理器体系Processor 抽象基类设计与文档索引 Pipeline 实战 本文聚焦 openJiuwen人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考