mirror of https://gitee.com/dromara/liteFlow
feat(agent): Hook→Middleware（Logging/ChatUsage/SkillTracking）+ 恢复 ctx.getChatUsage()/usedSkills() + 旧 Hook 经 builder.hooks() 桥接
Task 5.1：用 v2 RC3 的 MiddlewareBase 重建 Task 0 删除的三个 1.0 Hook 能力。 新增 middleware 包（com.yomahub.liteflow.agent.middleware）： - LoggingMiddleware：onReasoning/onActing 打印 reason/act/error 日志（替代 ReActLoggingHook，格式对齐 1.0），由 cmp.enableReActLogging() 控制。 - ChatUsageMiddleware：onModelCall 订阅 ModelCallEndEvent.getUsage() 累加（替代 ChatUsageTrackingHook）。per-invocation ThreadLocal 累加器，process() 入口 bind、 出口 unbind；snapshot() 静态读累计值。 - SkillTrackingMiddleware：onActing 跟踪 load_skill_through_path 工具调用（替代 SkillTrackingHook）。构造期从注入的 AgentSkillRepository 建 skillId→name 映射， per-invocation ThreadLocal 集合，usedSkills() 静态读。 探针结论（findings R-skill-track，已追加）："技能被使用"的最干净可观测信号是 onActing 的 ActingInput.toolCalls()——load_skill_through_path 工具名固定、 input.skillId 即 AgentSkill.getSkillId()。未降级，usedSkills() 真实落地。 接入与恢复： - ReActAgentComponent：process() 入口 ChatUsageMiddleware.bind() + SkillTrackingMiddleware.bind()，出口 finally unbind；新增 protected middlewares()（默认空）；hooks() 标 @Deprecated(since=2.16.0, forRemoval=true)， 指引改用 middlewares()；新增 protected final usedSkills()。 - ReActAgentContext：getChatUsage() 恢复为 ChatUsageMiddleware.snapshot()； 删除 stub setChatUsageTrackingHook；class javadoc 从"迁移期"改为"v2（RC3）"。 - ReactAgentFactory.build：组装 [LoggingMiddleware(if enabled), ChatUsageMiddleware, SkillTrackingMiddleware(repo), ...cmp.middlewares()] 逐个 builder.middleware()， 再 builder.hooks(cmp.hooks()) 直接桥接旧 Hook（RC3 Hook 软弃用但 builder.hooks 仍存在，无需 adapter）。 - SkillRepositoryResolver.configure 签名 void→AgentSkillRepository（返回注册的 repo 供 SkillTrackingMiddleware 建映射）。 un-stub：SkillsAgentCmp.handleReply 的 USED_SKILLS_SNAPSHOT 从 List.of() stub 改回 usedSkills()；ChatUsageAgentCmp/HookAgentCmp 无需改源（随核心恢复自动生效）。 校验：core compile + test-compile SUCCESS； -Dtest=ChatUsageMiddlewareTest,ProcessIntegrationTest,ShellPermissionBehaviorTest, SkillLoadingTest → 14 用例全绿；ReactAgentFactoryTest 2 用例无回归。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
everywhere.z
parent
96159b9499
commit
a37df41628
|
|
@ -508,3 +508,54 @@ Task 4.1 的 `strict` 语义在 <b>resolver 层</b>实现，分两段：
|
|||
- 接入点：`ReactAgentFactory.build`，在 `.permissionContext(...)` 之后、`.build()` 之前调 `SkillRepositoryResolver.configure(builder, this, cfg)`。
|
||||
- **不**碰 `usedSkills()` 跟踪（依赖 Task 5.1 SkillTrackingMiddleware，brief 明确推迟）。
|
||||
|
||||
|
||||
---
|
||||
|
||||
## R-skill-track：技能"被使用"的可观测信号（Task 5.1 探针确认）
|
||||
|
||||
**来源：** 对 `agentscope-2.0.0-RC3-sources.jar` 中 `io/agentscope/core/skill/SkillToolFactory.java`、`DynamicSkillMiddleware.java`、`io/agentscope/core/middleware/ActingInput.java`、`io/agentscope/core/event/ToolCallEndEvent.java`、`io/agentscope/core/skill/AgentSkill.java`、`ReActAgent.java`（中间件链 build 段）通读（JDK 21）。结论已用于 Task 5.1 `SkillTrackingMiddleware`。
|
||||
|
||||
**结论：RC3 暴露"技能被使用"的最干净可观测信号是 `MiddlewareBase.onActing` 收到的 `ActingInput.toolCalls()`。**
|
||||
|
||||
### (a) 技能加载工具的固定名 + 入参
|
||||
|
||||
技能加载由 `DynamicSkillMiddleware`（在 `ReActAgent.build()` 段 4268–4276 自动安装，见 R-skill (e)）注册的内置工具承担：
|
||||
- 工具名固定为 **`load_skill_through_path`**（`SkillToolFactory.createSkillAccessToolAgentTool()` 的 `getName()`，RC3 全 jar 唯一来源）。
|
||||
- 入参 schema（`getParameters()`）含两个 key：
|
||||
- **`skillId`**（string）：技能唯一 id，= `AgentSkill.getSkillId()` = `name + "_" + source`（如 `research_filesystem-tmpXXX_skills`）。注意<b>不是</b>裸 name。
|
||||
- **`path`**（string）：技能内资源路径，`SKILL.md` 表示加载技能说明。
|
||||
|
||||
### (b) 信号来源对比（为何选 onActing）
|
||||
|
||||
| 信号源 | 可行性 | 评估 |
|
||||
|---|---|---|
|
||||
| **`MiddlewareBase.onActing` 的 `ActingInput.toolCalls()`** | ✅ 最佳 | `ActingInput` 是 record `{List<ToolUseBlock> toolCalls}`。`ToolUseBlock.getName()` 返回工具名、`getInput():Map<String,Object>` 含 `skillId`。与 1.0 `SkillTrackingHook` 监听 `PostActingEvent` 取 `toolUse.getName()/getInput()` 逻辑<b>等价</b>——v2 把这层从 hook 挪到了 middleware。在工具真正被调用前即可记录（onActing 在 acting 阶段入口）。 |
|
||||
| `onModelCall` 下游 Flux 里的 `ToolCallEndEvent` | 可行但绕 | `ToolCallEndEvent` 带 `toolCallName` 但<b>不带 input</b>（只有 `replyId/toolCallId/toolCallName`），拿不到 `skillId`。需额外用 `toolCallId` 回查，不直接。 |
|
||||
| `ToolResultEndEvent` | 不合适 | 工具执行<b>之后</b>才发，且只带 `toolCallName`，无 input；失败的工具调用也可能不发。 |
|
||||
| `DynamicSkillMiddleware` 自发事件 | 无 | RC3 源码里 `DynamicSkillMiddleware` 只在 `onSystemPrompt` 钩子里加载技能清单/注册工具，<b>不发任何"技能被加载"事件</b>。 |
|
||||
| `AgentEvent.metadata` | 无约定 | 无技能相关约定 key。 |
|
||||
|
||||
### (c) skillId → name 映射的建立
|
||||
|
||||
用户在 `cmp.skills()` 写的是<b>裸 name</b>（如 `"research"`），而 `load_skill_through_path` 的 input `skillId` 是 `name + "_" + source`。故 `SkillTrackingMiddleware` 构造期需从注入的 `AgentSkillRepository`（`ReactAgentFactory` 调 `SkillRepositoryResolver.configure` 拿到，Task 5.1 让该 configure 返回注册的 repo）遍历一次 `repo.getAllSkills()`，建 `skillId → name` 不可变表；命中则记裸 name，未命中兜底记 skillId 原样。repo 为 null（技能未启用）时映射为空，middleware 仍可注册（onActing 见非 load_skill 工具自然跳过）。
|
||||
|
||||
### (d) 对 Task 5.1（SkillTrackingMiddleware）的指导（已采用）
|
||||
|
||||
- 覆写 `onAgent/onReasoning/onActing/onModelCall` 中只覆写 **`onActing`**：
|
||||
```java
|
||||
Flux<AgentEvent> onActing(Agent, RuntimeContext, ActingInput input, Function next) {
|
||||
Set<String> used = BOUND.get(); // per-invocation ThreadLocal
|
||||
if (used != null && input.toolCalls() != null) {
|
||||
for (ToolUseBlock t : input.toolCalls()) {
|
||||
if (t != null && "load_skill_through_path".equals(t.getName())) {
|
||||
Object id = t.getInput() == null ? null : t.getInput().get("skillId");
|
||||
if (id != null) used.add(skillIdToName.getOrDefault(String.valueOf(id), String.valueOf(id)));
|
||||
}
|
||||
}
|
||||
}
|
||||
return next.apply(input);
|
||||
}
|
||||
```
|
||||
- per-invocation 绑定同 `ChatUsageMiddleware`：`bind()`/`unbind()` 操作 ThreadLocal `Set<String>`，`process()` 入口 bind、出口 finally unbind。`usedSkills()` 静态读当前线程集合（未 bind 返回空 List）。
|
||||
- 工具名常量 `LOAD_SKILL_TOOL_NAME = "load_skill_through_path"`、入参 key `skillId`（与 1.0 `SkillTrackingHook` 一致）。
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
package com.yomahub.liteflow.agent.component;
|
||||
|
||||
import com.yomahub.liteflow.agent.exception.AgentConfigException;
|
||||
import com.yomahub.liteflow.agent.middleware.ChatUsageMiddleware;
|
||||
import com.yomahub.liteflow.agent.middleware.SkillTrackingMiddleware;
|
||||
import com.yomahub.liteflow.agent.model.ModelSpec;
|
||||
import com.yomahub.liteflow.core.NodeComponent;
|
||||
import com.yomahub.liteflow.property.LiteflowConfigGetter;
|
||||
|
|
@ -12,6 +14,7 @@ import io.agentscope.core.agent.RuntimeContext;
|
|||
import io.agentscope.core.hook.Hook;
|
||||
import io.agentscope.core.message.Msg;
|
||||
import io.agentscope.core.message.UserMessage;
|
||||
import io.agentscope.core.middleware.MiddlewareBase;
|
||||
import io.agentscope.core.model.Model;
|
||||
|
||||
import java.nio.file.Path;
|
||||
|
|
@ -198,8 +201,47 @@ public abstract class ReActAgentComponent extends NodeComponent {
|
|||
protected int maxIterations() { return -1; }
|
||||
protected boolean enableShellTool() { return true; }
|
||||
protected boolean enableWorkspaceFileTools() { return true; }
|
||||
|
||||
/**
|
||||
* 业务侧遗留 hooks 接入点。RC3 的 {@code io.agentscope.core.hook.Hook} 软弃用但仍可编译，
|
||||
* {@code ReActAgent.builder().hooks(List<Hook>)} 仍存在——{@link ReactAgentFactory#build}
|
||||
* 把这里返回的 hooks 经 {@code builder.hooks(...)} 直接桥接，无需 adapter。
|
||||
*
|
||||
* <p><b>推荐改用 {@link #middlewares()}</b>（v2 洋葱模型，{@link MiddlewareBase}）。
|
||||
* 本方法保留以兼容 1.0 业务子类，自 2.16.0 标记弃用、计划移除。
|
||||
*
|
||||
* @deprecated 改用 {@link #middlewares()}；Hook 体系在 v2 软弃用，未来版本移除。
|
||||
*/
|
||||
@Deprecated(since = "2.16.0", forRemoval = true)
|
||||
protected List<Hook> hooks() { return List.of(); }
|
||||
|
||||
/**
|
||||
* 业务侧自定义 middleware 接入点（v2 RC3 推荐）。{@link ReactAgentFactory#build} 会在
|
||||
* 框架内置 middleware（{@code ChatUsageMiddleware}/{@code SkillTrackingMiddleware}/
|
||||
* 可选 {@code LoggingMiddleware}）之后追加这里返回的 middleware。
|
||||
*
|
||||
* <p>默认返回空列表。返回的 middleware 实例会随 agent 单例缓存——<b>不要</b>在
|
||||
* middleware 里持有 per-invocation 状态（如累加器、已用技能集），那些应通过
|
||||
* {@link ChatUsageMiddleware#bind()} / {@link SkillTrackingMiddleware#bind()} 等
|
||||
* per-invocation 绑定机制管理。
|
||||
*
|
||||
* @since 2.16.0
|
||||
*/
|
||||
protected List<MiddlewareBase> middlewares() { return List.of(); }
|
||||
|
||||
/**
|
||||
* 返回本次 {@code process()} 截至当前已用到的技能名列表（裸 name，保持插入顺序）。
|
||||
*
|
||||
* <p>由 {@link SkillTrackingMiddleware} 在 {@code onActing} 钩子里跟踪
|
||||
* {@code load_skill_through_path} 工具调用得到（findings R-skill-track）。
|
||||
* 必须在 {@code process()} 生命周期内调用；未观察到任何技能加载时返回空列表。
|
||||
*
|
||||
* @since 2.16.0
|
||||
*/
|
||||
protected final List<String> usedSkills() {
|
||||
return SkillTrackingMiddleware.usedSkills();
|
||||
}
|
||||
|
||||
protected boolean enableReActLogging() {
|
||||
return agentConfig().getLogging().isReactEnabled();
|
||||
}
|
||||
|
|
@ -250,10 +292,16 @@ public abstract class ReActAgentComponent extends NodeComponent {
|
|||
ReActAgentContext ctx = new ReActAgentContext(slot, cid, akey, workspaceRoot(cfg, cid));
|
||||
ctx.setRuntimeContext(rc);
|
||||
slot.setAttachment(ctxKey(), ctx);
|
||||
// per-invocation 绑定 ChatUsage / Skill 累加器——agent 是单例、跨 process() 复用，
|
||||
// 不能跨 invocation 累加。在入口 bind、出口 finally unbind（见 findings R5）。
|
||||
ChatUsageMiddleware.bind();
|
||||
SkillTrackingMiddleware.bind();
|
||||
try {
|
||||
Msg reply = agent.call(List.of(new UserMessage(userPrompt())), rc).block();
|
||||
handleReply(reply);
|
||||
} finally {
|
||||
ChatUsageMiddleware.unbind();
|
||||
SkillTrackingMiddleware.unbind();
|
||||
slot.removeAttachment(ctxKey());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package com.yomahub.liteflow.agent.component;
|
||||
|
||||
import com.yomahub.liteflow.agent.middleware.ChatUsageMiddleware;
|
||||
import com.yomahub.liteflow.slot.Slot;
|
||||
import io.agentscope.core.agent.RuntimeContext;
|
||||
import io.agentscope.core.model.ChatUsage;
|
||||
|
|
@ -17,9 +18,9 @@ import java.util.Objects;
|
|||
* <li>{@link #getWorkspaceDir()}：按 conversationId 创建，同一段对话中的多个 agent 共享。</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p><b>状态：v2 迁移期。</b>对 1.0 {@code ChatUsageTrackingHook}（已删除）的依赖临时移除，
|
||||
* {@link #getChatUsage()} 返回 {@code null}；Task 5.1 会基于 v2 middleware 恢复。
|
||||
* {@link #setChatUsageTrackingHook} 暂为 no-op，签名保留以兼容历史调用点。
|
||||
* <p><b>状态：v2（RC3）。</b>{@link #getChatUsage()} 由 {@link ChatUsageMiddleware}
|
||||
* （v2 middleware，替代 1.0 {@code ChatUsageTrackingHook}）在 per-invocation 累加器上提供，
|
||||
* 累加器由 {@link ReActAgentComponent#process()} 在入口 bind、出口 unbind。
|
||||
*
|
||||
* <p><b>勿在跨 invocation 缓存的对象中持有 {@code ReActAgentContext} 引用</b>
|
||||
* （例如自定义工具实例、Hook、Model 实现）。这些对象会被缓存的 agent 跨次复用，
|
||||
|
|
@ -64,25 +65,20 @@ public class ReActAgentContext {
|
|||
this.runtimeContext = runtimeContext;
|
||||
}
|
||||
|
||||
/**
|
||||
* 由框架注入：本次 {@code process()} 调用使用的 token 累加 hook。
|
||||
*
|
||||
* <p><b>v2 迁移期 no-op。</b>1.0 的 {@code ChatUsageTrackingHook} 已删除，
|
||||
* 这里仅保留方法签名以兼容调用点；Task 5.1 重建后改回真实注入。
|
||||
*/
|
||||
public void setChatUsageTrackingHook(Object hook) {
|
||||
// no-op: 1.0 ChatUsageTrackingHook 已删除；Task 5.1 恢复
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回本次 {@code process()} 截至当前已累计的 token 用量。
|
||||
*
|
||||
* <p><b>v2 迁移期返回 {@code null}。</b>1.0 的 {@code ChatUsageTrackingHook}
|
||||
* 已删除，待 Task 5.1 基于 v2 middleware 重建后再恢复真实累计值。
|
||||
* <p>由 {@link ChatUsageMiddleware}（v2 RC3 middleware，替代 1.0
|
||||
* {@code ChatUsageTrackingHook}）在 {@code onModelCall} 钩子里累加每次模型调用的
|
||||
* {@code ModelCallEndEvent.getUsage()}（findings R5）。累加器是 per-invocation 的
|
||||
* ThreadLocal，由 {@link ReActAgentComponent#process()} 在入口 {@code bind}、
|
||||
* 出口 {@code unbind}。
|
||||
*
|
||||
* @return 当前固定返回 {@code null}（Task 5.1 恢复后给出累计 ChatUsage）
|
||||
* <p>必须在 {@code process()} 生命周期内调用（典型时机：{@code handleReply}，
|
||||
* 此时所有 reasoning step 已完成）。在生命周期外、或本次调用未观察到任何 usage
|
||||
* （模型/网关未上报）时返回 {@code null}。
|
||||
*/
|
||||
public ChatUsage getChatUsage() {
|
||||
return null;
|
||||
return ChatUsageMiddleware.snapshot();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
package com.yomahub.liteflow.agent.component;
|
||||
|
||||
import com.yomahub.liteflow.agent.exception.AgentConfigException;
|
||||
import com.yomahub.liteflow.agent.middleware.ChatUsageMiddleware;
|
||||
import com.yomahub.liteflow.agent.middleware.LoggingMiddleware;
|
||||
import com.yomahub.liteflow.agent.middleware.SkillTrackingMiddleware;
|
||||
import com.yomahub.liteflow.agent.permission.PermissionConfigMapper;
|
||||
import com.yomahub.liteflow.agent.skill.SkillRepositoryResolver;
|
||||
import com.yomahub.liteflow.agent.state.AgentStateStoreResolver;
|
||||
|
|
@ -9,13 +12,16 @@ import com.yomahub.liteflow.agent.tool.WorkspaceFileTools;
|
|||
import com.yomahub.liteflow.property.agent.AgentConfig;
|
||||
import com.yomahub.liteflow.property.agent.ShellMode;
|
||||
import io.agentscope.core.ReActAgent;
|
||||
import io.agentscope.core.middleware.MiddlewareBase;
|
||||
import io.agentscope.core.model.Model;
|
||||
import io.agentscope.core.permission.PermissionContextState;
|
||||
import io.agentscope.core.skill.repository.AgentSkillRepository;
|
||||
import io.agentscope.core.state.AgentStateStore;
|
||||
import io.agentscope.core.tool.Toolkit;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
|
|
@ -164,7 +170,36 @@ public final class ReactAgentFactory {
|
|||
// SkillsConfig + cmp.skills() allow-list → v2 skillRepository/skillFilter/dynamicSkillsEnabled
|
||||
// （findings R-skill）。本类同 component 包，可读 cmp 的 protected enableSkills()/skills()，
|
||||
// 把这两个值透传给 resolver（resolver 在 skill 包，无法直接访问 protected 钩子）。
|
||||
SkillRepositoryResolver.configure(builder, cmp.enableSkills(), cmp.skills(), name, cfg);
|
||||
// resolver 返回注册的 repo（未启用/构造失败为 null），供 SkillTrackingMiddleware
|
||||
// 建立 skillId→name 映射（findings R-skill-track）。
|
||||
AgentSkillRepository skillRepo = SkillRepositoryResolver.configure(
|
||||
builder, cmp.enableSkills(), cmp.skills(), name, cfg);
|
||||
|
||||
// Task 5.1：v2 middleware（替代 1.0 三个 Hook）。
|
||||
// 顺序：Logging（最外层，包裹整个 reasoning/acting）→ ChatUsage（onModelCall 累加）→
|
||||
// SkillTracking（onActing 跟踪 load_skill）。之后追加业务侧 cmp.middlewares()。
|
||||
List<MiddlewareBase> middlewares = new ArrayList<>();
|
||||
if (cmp.enableReActLogging()) {
|
||||
middlewares.add(new LoggingMiddleware());
|
||||
}
|
||||
middlewares.add(new ChatUsageMiddleware());
|
||||
middlewares.add(new SkillTrackingMiddleware(skillRepo));
|
||||
List<MiddlewareBase> extra = cmp.middlewares();
|
||||
if (extra != null) {
|
||||
for (MiddlewareBase mw : extra) {
|
||||
if (mw != null) {
|
||||
middlewares.add(mw);
|
||||
}
|
||||
}
|
||||
}
|
||||
for (MiddlewareBase mw : middlewares) {
|
||||
builder.middleware(mw);
|
||||
}
|
||||
|
||||
// 旧 Hook 桥接：RC3 的 io.agentscope.core.hook.Hook 软弃用但 ReActAgent.builder()
|
||||
// 仍提供 .hooks(List<Hook>)（findings R5）。业务侧 cmp.hooks() 经此直接接入，无需 adapter。
|
||||
builder.hooks(cmp.hooks());
|
||||
|
||||
return builder.build();
|
||||
} catch (Exception e) {
|
||||
throw new AgentConfigException(
|
||||
|
|
|
|||
|
|
@ -0,0 +1,127 @@
|
|||
package com.yomahub.liteflow.agent.middleware;
|
||||
|
||||
import io.agentscope.core.agent.Agent;
|
||||
import io.agentscope.core.agent.RuntimeContext;
|
||||
import io.agentscope.core.event.AgentEvent;
|
||||
import io.agentscope.core.event.ModelCallEndEvent;
|
||||
import io.agentscope.core.middleware.MiddlewareBase;
|
||||
import io.agentscope.core.middleware.ModelCallInput;
|
||||
import io.agentscope.core.model.ChatUsage;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* 累加单次 {@code process()} 调用内所有模型调用的 token 用量（替代 1.0
|
||||
* {@code ChatUsageTrackingHook}）。
|
||||
*
|
||||
* <p>RC3 把 token 用量挂在每次模型调用的 {@link ModelCallEndEvent#getUsage()} 上（由
|
||||
* {@code ReActAgent} 在 {@code new ModelCallEndEvent(replyId, context.getChatUsage())}
|
||||
* 处填充，见 findings R5）。本 middleware 覆写 {@link #onModelCall}，在
|
||||
* {@code next.apply(input)} 返回的 {@code Flux<AgentEvent>} 里订阅
|
||||
* {@link ModelCallEndEvent} 并把当步 usage 累加到一个 per-invocation 累加器。
|
||||
*
|
||||
* <h2>per-invocation 绑定</h2>
|
||||
* {@code ReActAgent} 被 {@code ReactAgentFactory} 按 cmp 子类缓存为单例、跨
|
||||
* {@code process()} 复用，<b>不能</b>把累加器做成实例字段直接累加——否则会把上次
|
||||
* 调用的余量带入下一次。故累加器用 {@link ThreadLocal} 持有，{@link #bind()} 在
|
||||
* {@code process()} 入口调用（push 新累加器），{@link #unbind()} 在出口 {@code finally}
|
||||
* 调用（pop 并丢弃）。
|
||||
*
|
||||
* <p>之所以用 ThreadLocal 而非 RuntimeContext：RC3-core 下 {@code process()} 用
|
||||
* {@code .block()} 同步执行整条 ReAct 循环，模型调用在同一线程上完成；middleware
|
||||
* 在该线程上被调用。子类若未来引入异步流式（Task 6.1），需相应把累加器改成随
|
||||
* reactor {@code Context} 传播——RC3-core 不在此范围内。
|
||||
*
|
||||
* <h2>读累计值</h2>
|
||||
* {@link com.yomahub.liteflow.agent.component.ReActAgentContext#getChatUsage()} 通过
|
||||
* {@link #snapshot()} 读当前线程绑定的累加器；未 {@code bind} 或尚未观察到任何
|
||||
* usage 时返回 {@code null}。
|
||||
*/
|
||||
public class ChatUsageMiddleware implements MiddlewareBase {
|
||||
|
||||
/** per-invocation 累加器栈：bind push、unbind pop（栈结构支持嵌套，虽当前 process() 不嵌套）。 */
|
||||
private static final ThreadLocal<Accumulator> BOUND = new ThreadLocal<>();
|
||||
|
||||
public ChatUsageMiddleware() {
|
||||
}
|
||||
|
||||
/**
|
||||
* 在当前线程绑定一个新的 per-invocation 累加器。必须在 {@code process()} 入口调用，
|
||||
* 出口 {@link #unbind()} 清零。
|
||||
*/
|
||||
public static void bind() {
|
||||
BOUND.set(new Accumulator());
|
||||
}
|
||||
|
||||
/**
|
||||
* 摘除当前线程绑定的累加器。必须在 {@code process()} 出口（{@code finally}）调用，
|
||||
* 避免单例 middleware 跨 invocation 累加。
|
||||
*/
|
||||
public static void unbind() {
|
||||
BOUND.remove();
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回当前线程累加器截至当前累计的 token 用量；未 bind 或未观察到任何 usage 时
|
||||
* 返回 {@code null}。
|
||||
*
|
||||
* <p>静态访问：累加器本身是 ThreadLocal，与具体 middleware 实例无关；故
|
||||
* {@link com.yomahub.liteflow.agent.component.ReActAgentContext#getChatUsage()}
|
||||
* 可直接读，无需持有 middleware 引用。
|
||||
*/
|
||||
public static ChatUsage snapshot() {
|
||||
Accumulator acc = BOUND.get();
|
||||
return acc == null ? null : acc.snapshot();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AgentEvent> onModelCall(
|
||||
Agent agent,
|
||||
RuntimeContext ctx,
|
||||
ModelCallInput input,
|
||||
Function<ModelCallInput, Flux<AgentEvent>> next) {
|
||||
Flux<AgentEvent> downstream = next.apply(input);
|
||||
Accumulator acc = BOUND.get();
|
||||
if (acc == null) {
|
||||
// 未 bind（例如被独立使用、或 process() 外触发的模型调用）——不累加，透传。
|
||||
return downstream;
|
||||
}
|
||||
// 订阅下游流：每个 ModelCallEndEvent 累加到当前线程的累加器；不影响事件本身。
|
||||
return downstream.doOnNext(event -> {
|
||||
if (event instanceof ModelCallEndEvent end) {
|
||||
ChatUsage usage = end.getUsage();
|
||||
if (usage != null) {
|
||||
acc.add(usage);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/* ----- 累加器（线程不安全，仅由持有它的 ThreadLocal 保证单线程访问）----- */
|
||||
|
||||
private static final class Accumulator {
|
||||
private int inputTokens;
|
||||
private int outputTokens;
|
||||
private double time;
|
||||
private int steps;
|
||||
|
||||
synchronized void add(ChatUsage usage) {
|
||||
this.inputTokens += usage.getInputTokens();
|
||||
this.outputTokens += usage.getOutputTokens();
|
||||
this.time += usage.getTime();
|
||||
this.steps++;
|
||||
}
|
||||
|
||||
synchronized ChatUsage snapshot() {
|
||||
if (steps == 0) {
|
||||
return null;
|
||||
}
|
||||
return ChatUsage.builder()
|
||||
.inputTokens(inputTokens)
|
||||
.outputTokens(outputTokens)
|
||||
.time(time)
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,108 @@
|
|||
package com.yomahub.liteflow.agent.middleware;
|
||||
|
||||
import io.agentscope.core.agent.Agent;
|
||||
import io.agentscope.core.agent.RuntimeContext;
|
||||
import io.agentscope.core.event.AgentEvent;
|
||||
import io.agentscope.core.event.ModelCallEndEvent;
|
||||
import io.agentscope.core.event.ToolCallEndEvent;
|
||||
import io.agentscope.core.event.ToolResultEndEvent;
|
||||
import io.agentscope.core.message.ToolUseBlock;
|
||||
import io.agentscope.core.middleware.ActingInput;
|
||||
import io.agentscope.core.middleware.MiddlewareBase;
|
||||
import io.agentscope.core.middleware.ReasoningInput;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* 把 ReActAgent 的 reasoning / acting 阶段输出到日志（替代 1.0 {@code ReActLoggingHook}）。
|
||||
*
|
||||
* <p>覆写 {@link MiddlewareBase#onReasoning} 与 {@link MiddlewareBase#onActing}，
|
||||
* 在进入时打印 {@code >>>} 概要，在下游事件流的关键事件上打印 {@code <<<} 概要。
|
||||
* 日志格式对齐 1.0 {@code ReActLoggingHook}：
|
||||
* <ul>
|
||||
* <li>reasoning 入口：{@code [agent:reason] >>> model=<name> messages=<n>}</li>
|
||||
* <li>reasoning 出口（ModelCallEndEvent）：{@code [agent:reason] <<< usage=...}</li>
|
||||
* <li>acting 入口：{@code [agent:act] >>> tool=<name> input=<...>}</li>
|
||||
* <li>acting 出口（ToolResultEndEvent）：{@code [agent:act] <<< tool=<name>}</li>
|
||||
* <li>异常：{@code [agent:error] ...}</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>{@code sessionId} 取自 {@link RuntimeContext#getSessionId()}（= agentKey），
|
||||
* 便于在日志中区分同对话内不同 agent。无 ctx 时记 {@code "-"}。
|
||||
*
|
||||
* <p>由 {@link ReActAgentComponent#enableReActLogging()} 控制——为 false 时 factory 不注册本 middleware。
|
||||
*/
|
||||
public class LoggingMiddleware implements MiddlewareBase {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(LoggingMiddleware.class);
|
||||
private static final int MAX_LEN = 500;
|
||||
|
||||
@Override
|
||||
public Flux<AgentEvent> onReasoning(
|
||||
Agent agent,
|
||||
RuntimeContext ctx,
|
||||
ReasoningInput input,
|
||||
Function<ReasoningInput, Flux<AgentEvent>> next) {
|
||||
String sid = sessionId(ctx);
|
||||
// ReasoningInput 不含 model 实例（model 在 onModelCall 的 ModelCallInput 里），
|
||||
// 故这里只记消息数；model 名字由 ModelCallEndEvent 的后续日志间接体现。
|
||||
int msgCount = input == null || input.messages() == null ? 0 : input.messages().size();
|
||||
LOG.info("[agent:reason][{}] >>> messages={}", sid, msgCount);
|
||||
|
||||
return next.apply(input)
|
||||
.doOnNext(event -> {
|
||||
if (event instanceof ModelCallEndEvent end) {
|
||||
LOG.info("[agent:reason][{}] <<< usage={}", sid, end.getUsage());
|
||||
}
|
||||
})
|
||||
.doOnError(e -> LOG.warn("[agent:error][{}] {}", sid, e.toString(), e));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AgentEvent> onActing(
|
||||
Agent agent,
|
||||
RuntimeContext ctx,
|
||||
ActingInput input,
|
||||
Function<ActingInput, Flux<AgentEvent>> next) {
|
||||
String sid = sessionId(ctx);
|
||||
List<ToolUseBlock> toolCalls = input == null ? null : input.toolCalls();
|
||||
if (toolCalls != null) {
|
||||
for (ToolUseBlock t : toolCalls) {
|
||||
if (t == null) {
|
||||
continue;
|
||||
}
|
||||
LOG.info("[agent:act][{}] >>> tool={} input={}",
|
||||
sid, t.getName(), truncate(String.valueOf(t.getInput())));
|
||||
}
|
||||
}
|
||||
return next.apply(input)
|
||||
.doOnNext(event -> {
|
||||
if (event instanceof ToolResultEndEvent end) {
|
||||
LOG.info("[agent:act][{}] <<< tool={}", sid, end.getToolCallName());
|
||||
} else if (event instanceof ToolCallEndEvent end) {
|
||||
LOG.info("[agent:act][{}] <<< toolCall={}", sid, end.getToolCallName());
|
||||
}
|
||||
})
|
||||
.doOnError(e -> LOG.warn("[agent:error][{}] {}", sid, e.toString(), e));
|
||||
}
|
||||
|
||||
private static String sessionId(RuntimeContext ctx) {
|
||||
try {
|
||||
return ctx == null || ctx.getSessionId() == null ? "-" : ctx.getSessionId();
|
||||
} catch (Throwable t) {
|
||||
return "-";
|
||||
}
|
||||
}
|
||||
|
||||
private static String truncate(String s) {
|
||||
if (s == null) {
|
||||
return "";
|
||||
}
|
||||
s = s.replaceAll("\\s+", " ").trim();
|
||||
return s.length() <= MAX_LEN ? s : s.substring(0, MAX_LEN) + "...(truncated)";
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,151 @@
|
|||
package com.yomahub.liteflow.agent.middleware;
|
||||
|
||||
import io.agentscope.core.agent.Agent;
|
||||
import io.agentscope.core.agent.RuntimeContext;
|
||||
import io.agentscope.core.event.AgentEvent;
|
||||
import io.agentscope.core.message.ToolUseBlock;
|
||||
import io.agentscope.core.middleware.ActingInput;
|
||||
import io.agentscope.core.middleware.MiddlewareBase;
|
||||
import io.agentscope.core.skill.AgentSkill;
|
||||
import io.agentscope.core.skill.repository.AgentSkillRepository;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* 跟踪本次 {@code process()} 调用内被用到的技能（替代 1.0 {@code SkillTrackingHook}，
|
||||
* 恢复 {@code ReActAgentComponent.usedSkills()}）。
|
||||
*
|
||||
* <h2>探测结论（R-skill-track）</h2>
|
||||
* RC3 暴露"技能被使用"的最干净可观测信号是 {@link MiddlewareBase#onActing} 收到的
|
||||
* {@link ActingInput#toolCalls()}：技能加载由 {@code DynamicSkillMiddleware} 注册的
|
||||
* {@code load_skill_through_path} 工具承担（findings R-skill + RC3 源码
|
||||
* {@code SkillToolFactory.createSkillAccessToolAgentTool}，工具名常量
|
||||
* {@code load_skill_through_path}），其输入 schema 含 {@code skillId}（= 技能的
|
||||
* {@code name + "_" + source}，见 {@code AgentSkill.getSkillId()}）。当 ReActAgent 选择
|
||||
* 调用该工具时，{@code onActing} 的 {@code toolCalls} 里就会出现一个
|
||||
* {@link ToolUseBlock}，其 {@code getName()=="load_skill_through_path"} 且
|
||||
* {@code getInput().get("skillId")} 为本次加载的技能 id。
|
||||
*
|
||||
* <p>（对比 1.0：旧的 {@code SkillTrackingHook} 监听 {@code PostActingEvent}，取
|
||||
* {@code toolUse.getName()/getInput()}，逻辑等价——只是 v2 把这层从 hook 挪到了 middleware。）
|
||||
*
|
||||
* <h2>per-invocation 绑定</h2>
|
||||
* {@code ReActAgent} 单例、跨 {@code process()} 复用；本 middleware 同样单例。故已用技能集
|
||||
* 用 {@link ThreadLocal} 持有，{@link #bind()} 在 {@code process()} 入口 push、
|
||||
* {@link #unbind()} 在出口 pop。{@link #usedSkills()} 读当前线程绑定的集合（未 bind 返回空）。
|
||||
*
|
||||
* <h2>skillId → name 映射</h2>
|
||||
* 构造期从注入的 {@link AgentSkillRepository}（若有）遍历一次，建 {@code skillId → name} 表。
|
||||
* 用户在 {@code cmp.skills()} 写的是<b>裸 name</b>（findings R-skill (c)），故 {@code usedSkills()}
|
||||
* 也返回裸 name。repo 为 {@code null}（技能未启用）时映射为空，{@code load_skill_through_path}
|
||||
* 即便被调用也只能记 skillId 原样（兜底）。
|
||||
*/
|
||||
public class SkillTrackingMiddleware implements MiddlewareBase {
|
||||
|
||||
/** RC3 技能加载工具的固定名（{@code SkillToolFactory.createSkillAccessToolAgentTool}）。 */
|
||||
public static final String LOAD_SKILL_TOOL_NAME = "load_skill_through_path";
|
||||
private static final String SKILL_ID_INPUT_KEY = "skillId";
|
||||
|
||||
/** skillId → 裸 name 映射（不可变；构造期一次性建立）。 */
|
||||
private final Map<String, String> skillIdToName;
|
||||
|
||||
/** per-invocation 已用技能名集合，按 bind/unbind 生命周期隔离。 */
|
||||
private static final ThreadLocal<Set<String>> BOUND = new ThreadLocal<>();
|
||||
|
||||
public SkillTrackingMiddleware(AgentSkillRepository repository) {
|
||||
this.skillIdToName = buildSkillIdToName(repository);
|
||||
}
|
||||
|
||||
/**
|
||||
* 在当前线程绑定一个新的空"已用技能"集合。{@code process()} 入口调用。
|
||||
*/
|
||||
public static void bind() {
|
||||
BOUND.set(Collections.synchronizedSet(new LinkedHashSet<>()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 摘除当前线程绑定的集合。{@code process()} 出口（{@code finally}）调用。
|
||||
*/
|
||||
public static void unbind() {
|
||||
BOUND.remove();
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回当前线程已用技能名列表（保持插入顺序）。未 bind 时返回空列表。
|
||||
*
|
||||
* <p>静态访问：已用集合本身是 ThreadLocal，与具体 middleware 实例无关；故
|
||||
* {@link com.yomahub.liteflow.agent.component.ReActAgentComponent#usedSkills()}
|
||||
* 可直接读，无需持有 middleware 引用。
|
||||
*/
|
||||
public static List<String> usedSkills() {
|
||||
Set<String> set = BOUND.get();
|
||||
if (set == null) {
|
||||
return List.of();
|
||||
}
|
||||
synchronized (set) {
|
||||
return List.copyOf(set);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AgentEvent> onActing(
|
||||
Agent agent,
|
||||
RuntimeContext ctx,
|
||||
ActingInput input,
|
||||
Function<ActingInput, Flux<AgentEvent>> next) {
|
||||
Set<String> used = BOUND.get();
|
||||
if (used != null && input != null && input.toolCalls() != null) {
|
||||
for (ToolUseBlock t : input.toolCalls()) {
|
||||
recordIfSkillLoad(t, used);
|
||||
}
|
||||
}
|
||||
return next.apply(input);
|
||||
}
|
||||
|
||||
private void recordIfSkillLoad(ToolUseBlock toolUse, Set<String> used) {
|
||||
if (toolUse == null || !LOAD_SKILL_TOOL_NAME.equals(toolUse.getName())) {
|
||||
return;
|
||||
}
|
||||
Map<String, Object> in = toolUse.getInput();
|
||||
if (in == null) {
|
||||
return;
|
||||
}
|
||||
Object skillId = in.get(SKILL_ID_INPUT_KEY);
|
||||
if (skillId == null) {
|
||||
return;
|
||||
}
|
||||
String id = String.valueOf(skillId);
|
||||
String name = skillIdToName.get(id);
|
||||
synchronized (used) {
|
||||
used.add(name != null ? name : id);
|
||||
}
|
||||
}
|
||||
|
||||
private static Map<String, String> buildSkillIdToName(AgentSkillRepository repository) {
|
||||
if (repository == null) {
|
||||
return Map.of();
|
||||
}
|
||||
Map<String, String> map = new LinkedHashMap<>();
|
||||
try {
|
||||
List<AgentSkill> all = repository.getAllSkills();
|
||||
if (all != null) {
|
||||
for (AgentSkill s : all) {
|
||||
if (s == null) {
|
||||
continue;
|
||||
}
|
||||
map.put(s.getSkillId(), s.getName());
|
||||
}
|
||||
}
|
||||
} catch (Throwable ignored) {
|
||||
// repo 读失败不影响主流程；映射留空，usedSkills() 兜底记 skillId。
|
||||
}
|
||||
return Collections.unmodifiableMap(map);
|
||||
}
|
||||
}
|
||||
|
|
@ -107,10 +107,13 @@ public final class SkillRepositoryResolver {
|
|||
* @param allowList 组件级 allow-list（来自 {@code cmp.skills()}），空/null = 全部可见
|
||||
* @param componentLabel 组件标签（用于日志/异常消息，通常是 {@code cmp.getClass().getSimpleName()}）
|
||||
* @param cfg agent 配置（读 {@link AgentConfig#getSkills()}）
|
||||
* @return 注册到 builder 的 {@link AgentSkillRepository}；未启用或构造失败时返回 {@code null}。
|
||||
* 调用方（{@code ReactAgentFactory}）可把它传给 {@code SkillTrackingMiddleware}
|
||||
* 以建立 skillId→name 映射（Task 5.1）。
|
||||
* @throws AgentConfigException enabled=true 且 strict=true 时技能路径不可用/无可解析技能
|
||||
*/
|
||||
public static void configure(ReActAgent.Builder builder, boolean enabled,
|
||||
List<String> allowList, String componentLabel, AgentConfig cfg) {
|
||||
public static AgentSkillRepository configure(ReActAgent.Builder builder, boolean enabled,
|
||||
List<String> allowList, String componentLabel, AgentConfig cfg) {
|
||||
if (builder == null) {
|
||||
throw new AgentConfigException("ReActAgent.Builder is null; cannot configure skills");
|
||||
}
|
||||
|
|
@ -118,7 +121,7 @@ public final class SkillRepositoryResolver {
|
|||
if (!enabled) {
|
||||
// 显式关闭：即便后续代码塞了 repo（目前无此路径），middleware 也不会装。
|
||||
builder.dynamicSkillsEnabled(false);
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
if (cfg == null || cfg.getSkills() == null) {
|
||||
// enabled 但无配置：保守按 strict 抛，避免静默装上空 middleware。
|
||||
|
|
@ -148,7 +151,7 @@ public final class SkillRepositoryResolver {
|
|||
LOG.warn("[SkillRepositoryResolver] strict=false, skipping invalid skill path {}: {}",
|
||||
skillsPath, e.getMessage());
|
||||
builder.dynamicSkillsEnabled(true);
|
||||
return;
|
||||
return null;
|
||||
} catch (Exception e) {
|
||||
if (strict) {
|
||||
throw new AgentConfigException(
|
||||
|
|
@ -159,7 +162,7 @@ public final class SkillRepositoryResolver {
|
|||
LOG.warn("[SkillRepositoryResolver] strict=false, skipping skill repository init for {}: {}",
|
||||
skillsPath, e.getMessage());
|
||||
builder.dynamicSkillsEnabled(true);
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
|
||||
// 构造成功但空（无 SKILL.md 子目录）。
|
||||
|
|
@ -177,7 +180,7 @@ public final class SkillRepositoryResolver {
|
|||
LOG.warn("[SkillRepositoryResolver] strict=false, getAllSkillNames failed for {}: {}",
|
||||
skillsPath, e.getMessage());
|
||||
builder.dynamicSkillsEnabled(true);
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
if (skillNames == null || skillNames.isEmpty()) {
|
||||
if (strict) {
|
||||
|
|
@ -228,6 +231,7 @@ public final class SkillRepositoryResolver {
|
|||
+ "for component {} (no allow-list, all skills visible)",
|
||||
skillsPath, skillNames == null ? 0 : skillNames.size(), label);
|
||||
}
|
||||
return repo;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -14,10 +14,10 @@ import java.util.concurrent.atomic.AtomicReference;
|
|||
* 验证 skills.enabled=true 时 {@code load_skill_through_path} 工具会被注册到 Agent。
|
||||
* 是否真的 load 取决于模型是否选择调用工具；本测试只关注工具集与组件级 skills() 过滤。
|
||||
*
|
||||
* <p><b>v2 迁移期占位（Task 2.2c）：</b>原 1.0 的 {@code usedSkills()} 已从
|
||||
* {@link com.yomahub.liteflow.agent.component.ReActAgentComponent} 移除，待 Task 4.1
|
||||
* 重建 v2 skillRepository 后恢复组件级已用 skills 的读取语义。当前 {@code handleReply}
|
||||
* 把 {@code USED_SKILLS_SNAPSHOT} 填成空 List 占位，维持 {@link SkillsFeatureTest} 编译。
|
||||
* <p>{@code usedSkills()} 由 {@link com.yomahub.liteflow.agent.middleware.SkillTrackingMiddleware}
|
||||
* 在 {@code onActing} 钩子里跟踪 {@code load_skill_through_path} 工具调用得到
|
||||
* （findings R-skill-track，Task 5.1 重建）。{@code handleReply} 里读一次快照供
|
||||
* {@link SkillsFeatureTest} 断言。
|
||||
*/
|
||||
@Component("skillsAgent")
|
||||
public class SkillsAgentCmp extends ReActAgentComponent {
|
||||
|
|
@ -86,10 +86,9 @@ public class SkillsAgentCmp extends ReActAgentComponent {
|
|||
|
||||
@Override
|
||||
protected void handleReply(Msg reply) {
|
||||
// TODO(Task 4.1): 1.0 usedSkills() 已从 ReActAgentComponent 移除，待 v2 skillRepository
|
||||
// 重建后恢复组件级已用 skills 读取：
|
||||
// USED_SKILLS_SNAPSHOT.set(usedSkills());
|
||||
USED_SKILLS_SNAPSHOT.set(List.of());
|
||||
// usedSkills() 由 SkillTrackingMiddleware 在 onActing 跟踪 load_skill_through_path
|
||||
// 得到（Task 5.1）。在 handleReply 读一次快照——此时所有 reasoning/acting step 已完成。
|
||||
USED_SKILLS_SNAPSHOT.set(usedSkills());
|
||||
super.handleReply(reply);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
package com.yomahub.liteflow.test.agent.v2;
|
||||
|
||||
import com.yomahub.liteflow.agent.middleware.ChatUsageMiddleware;
|
||||
import io.agentscope.core.event.AgentEvent;
|
||||
import io.agentscope.core.event.ModelCallEndEvent;
|
||||
import io.agentscope.core.middleware.ModelCallInput;
|
||||
import io.agentscope.core.model.ChatUsage;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* Task 5.1 单元测试：验证 {@link ChatUsageMiddleware} 在单次 invocation 内累加多步
|
||||
* {@link ModelCallEndEvent#getUsage()} 的 token 用量。
|
||||
*
|
||||
* <p><b>纯单元测试</b>：不构建 ReActAgent、不起 Spring、不调 LLM。直接构造
|
||||
* {@link ChatUsageMiddleware}，覆写 {@code onModelCall}，用一个发出两个
|
||||
* {@link ModelCallEndEvent}（各带不同 {@link ChatUsage}）的 mock {@code next}，
|
||||
* 断言 {@link ChatUsageMiddleware#snapshot()}（通过 bind 的 invocation 累加器）= 两者之和。
|
||||
*
|
||||
* <p>关键点（findings R5）：
|
||||
* <ul>
|
||||
* <li>{@code onModelCall(agent, ctx, input, next)} 的 {@code next.apply(input)} 返回
|
||||
* {@code Flux<AgentEvent>}，从其中订阅 {@link ModelCallEndEvent} 取 usage。</li>
|
||||
* <li>agent 是单例、跨 invocation 复用，故累加器必须 per-invocation 绑定
|
||||
* （{@code bind}/{@code unbind}），不能跨次调用累加。</li>
|
||||
* </ul>
|
||||
*/
|
||||
class ChatUsageMiddlewareTest {
|
||||
|
||||
private final ChatUsageMiddleware middleware = new ChatUsageMiddleware();
|
||||
|
||||
@AfterEach
|
||||
void clean() {
|
||||
// 兜底解绑，避免 ThreadLocal 跨用例泄漏（即便测试中途断言失败）。
|
||||
ChatUsageMiddleware.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
void accumulatesUsageAcrossMultipleModelCallEndEvents() {
|
||||
ChatUsage step1 = ChatUsage.builder().inputTokens(100).outputTokens(50).time(0.5).build();
|
||||
ChatUsage step2 = ChatUsage.builder().inputTokens(200).outputTokens(80).time(1.2).build();
|
||||
|
||||
// bind 一个 per-invocation 累加器（模拟 process() 入口）。
|
||||
ChatUsageMiddleware.bind();
|
||||
try {
|
||||
// 两次 onModelCall（模拟 ReAct 循环里的多步 reasoning）。
|
||||
runModelCall(middleware, step1);
|
||||
runModelCall(middleware, step2);
|
||||
|
||||
ChatUsage snapshot = ChatUsageMiddleware.snapshot();
|
||||
Assertions.assertNotNull(snapshot, "累加至少一次 usage 后 snapshot 不应为 null");
|
||||
Assertions.assertEquals(300, snapshot.getInputTokens(), "inputTokens 应为两步之和");
|
||||
Assertions.assertEquals(130, snapshot.getOutputTokens(), "outputTokens 应为两步之和");
|
||||
Assertions.assertEquals(430, snapshot.getTotalTokens(), "totalTokens 应为两步之和");
|
||||
Assertions.assertEquals(1.7, snapshot.getTime(), 1e-9, "time 应为两步之和");
|
||||
} finally {
|
||||
ChatUsageMiddleware.unbind();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void snapshotNullWhenNoUsageObserved() {
|
||||
ChatUsageMiddleware.bind();
|
||||
try {
|
||||
Assertions.assertNull(ChatUsageMiddleware.snapshot(),
|
||||
"未观察到任何 ModelCallEndEvent 时 snapshot 应为 null");
|
||||
} finally {
|
||||
ChatUsageMiddleware.unbind();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void snapshotNullWhenNotBound() {
|
||||
// 不 bind 直接读：单例跨 invocation 不应残留上次调用的累计。
|
||||
Assertions.assertNull(ChatUsageMiddleware.snapshot(),
|
||||
"未 bind 时 snapshot 应为 null（单例不能跨 invocation 累加）");
|
||||
}
|
||||
|
||||
@Test
|
||||
void unbindClearsAccumulator() {
|
||||
ChatUsage step = ChatUsage.builder().inputTokens(10).outputTokens(5).time(0.1).build();
|
||||
ChatUsageMiddleware.bind();
|
||||
runModelCall(middleware, step);
|
||||
Assertions.assertNotNull(ChatUsageMiddleware.snapshot());
|
||||
ChatUsageMiddleware.unbind();
|
||||
|
||||
// unbind 后再读 → null（per-invocation 累加器已被摘除）。
|
||||
Assertions.assertNull(ChatUsageMiddleware.snapshot(),
|
||||
"unbind 后 snapshot 应为 null");
|
||||
}
|
||||
|
||||
/* ----- helpers ----- */
|
||||
|
||||
/** 调 middleware.onModelCall，next 发出单个带 usage 的 ModelCallEndEvent，然后 block 完成订阅。 */
|
||||
@SuppressWarnings({"rawtypes", "unchecked"})
|
||||
private static void runModelCall(ChatUsageMiddleware mw, ChatUsage usage) {
|
||||
Function<ModelCallInput, Flux<AgentEvent>> next = input -> Flux.<AgentEvent>just(
|
||||
new ModelCallEndEvent("reply-" + System.nanoTime(), usage));
|
||||
Flux<AgentEvent> flux = mw.onModelCall(null, null, new ModelCallInput(
|
||||
List.of(), List.of(), null, null), next);
|
||||
flux.collectList().block(); // 同步消费完整个流，触发累加副作用
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue