✅AgentScope Java进阶：Human-in-the-Loop



Human-in-the-Loop我们之前也讲过，这是一种允许在 Agent 执行过程中暂停、人工审查、交互确认后继续执行的机制。这一机制使得 AI Agent 可以在涉及敏感操作、信息缺失或需要人类决策的场景下安全、可控地运行。



前面在介绍ReAct Agent的实现原理的时候提到过，ReActAgent每一轮迭代只干2件事，推理和行动。



整个过程对应用层是自动运转的，agent.call(userMsg).block() 一次调用要么成功返回最终答案，要么抛异常。HITL 要解决的核心问题就一句话：怎么在这个全自动循环里安全地插入一个”暂停—等用户—继续”的缝隙。



注意”安全”二字 —— 暂停后 memory 里可能留着没填 result 的 ToolUseBlock，下一次调用必须能恢复，不能让模型看到悬空状态。



敏感动作前确认



HIL的一个最常见的场景：模型想调 delete_file，我希望先弹出 “确认/取消”。



在ReActAgent.reasoning() 里，看 510-543 行：

1
2
3
4
5
6
7
.flat
Map(this::notifyPostReasoning).flat
Map(event -> {
    Msg msg = event.getReasoningMessage();
    if (msg != null) memory.addMessage(msg);     // ① 先把推理结果写进 memory
    if (event.isStopRequested()) {
                // ② 检查暂停标志        return Mono.just(msg.withGenerateReason(                GenerateReason.REASONING_STOP_REQUESTED));   // ③ 直接返回，不进 acting    }    ...    return acting(iter);});



这里面有一个暂停标记检查的动作，也就是说，如果这时候event.isStopRequested() 如果为true的话，那么就会直接进入HIL了，



那么这个event是啥呢？通过查看代码我们能知道他是PostReasoningEvent，而如果我们有一个Hook可以监听这个事件，并且把他的isStopRequested设置为true，就能触发HIL了。



于是我们可以这么做：



1
2
3
4
5
6
7
8
9
10
11
12
13
14
Hook confirmationHook = new Hook() {
    private static final List<String> SENSITIVE_TOOLS = List.of("delete_file", "send_email");

    @Override    public <T extends HookEvent> Mono<T> onEvent(T event) {
        if (event instanceof PostReasoningEvent e) {
            Msg reasoningMsg = e.getReasoningMessage();
            List<ToolUseBlock> toolCalls = reasoningMsg.getContentBlocks(ToolUseBlock.class);
            // 如果包含敏感工具，暂停等待确认            boolean hasSensitive = toolCalls.stream()                .anyMatch(t -> SENSITIVE_TOOLS.contains(t.getName()));
            if (hasSensitive) {
                e.stopAgent();            }        }
        return Mono.just(event);    }
};

ReActAgent agent = ReActAgent.builder()    .name("Assistant")    .model(model)    .toolkit(toolkit)    .hook(confirmationHook)    .build();



通过PostReasoningEvent的stopAgent方法，来把isStopRequested设置为true：



1
2
3
public void stopAgent() {
    this.stopRequested = true;
}



Agent在执行时，如果触发了HIL用户拿到的 Msg 里一定带着 ToolUseBlock（还没执行），所以判断条件就是它。如果用户同意，则继续执行，否则提示失败。



1
2
3
4
5
Msg resp = agent.call(userMsg).block();
while (resp.hasContentBlocks(ToolUseBlock.class)) {
    showPendingToUser(resp);
    if (userClickYes()) {
        resp = agent.call().block();           // 无参 → 继续跑 acting    } else {        resp = agent.call(buildCancelMsg(resp)).block();  // 注入伪 result    }}



如果想要继续执行，会执行到acting方法中，而这个方法的第一句就是获取pending的ToolUseBlock，然后把他执行完。

1
2
3
private Mono<Msg> acting(int iter) {
        // Extract only pending tool calls (those without results in memory)        List<ToolUseBlock> pendingToolCalls = extractPendingToolCalls();
}



关键动作后确认



前面是在执行前人工审查，但有的需求是工具已经跑完，结果也要给人审一遍（比如生成了一封邮件草稿，让人看完再决定要不要发下一步动作）。



ReActAgent.acting() 605-619 行：



1
2
return Flux.fromIterable(successPairs)        .concat
Map(this::notifyPostActingHook)   // 串行通知每个工具的 PostActing        .last()                                  // 取最后一个事件        .flatMap(event -> {            if (event.isStopRequested()) {                return Mono.just(event.getToolResultMsg()                        .withGenerateReason(GenerateReason.ACTING_STOP_REQUESTED));            }            ...        });



同样判断event.isStopRequested() ，如果为true，触发HIL。只不过这里的Event是PostActingEvent。



LLM主动找用户确认



前面两种都是”ASJ框架替用户拦截”，但还有一类需求：LLM 自己意识到信息不够，想反过来问用户。例如健身教练 agent 缺少”用户身高体重”，想弹出一个表单让用户填。



如果还走 Hook 路径，相当于要在外面写状态机猜什么时候问什么——很别扭。设计者的解法是给 LLM 一把钥匙：让它调一个工具，工具的实现就是抛ToolSuspendException异常。



1
// 取自 examples/advanced/hitl/UserInteractionTool.java@Tool(name = "ask_user", description = "Ask the user for clarification ...")public String askUser(@ToolParam("question") String question,                      @ToolParam("ui_type")  String uiType,                      @ToolParam("options")  List<String> options,                      @ToolParam("fields")   List<Map<String,Object>> fields, ...) {    throw new ToolSuspendException(question);}



1
// tool/ToolSuspendException.javapublic class ToolSuspendException extends RuntimeException {    private final String reason;    public ToolSuspendException(String reason) { ... }}



LLM 在系统提示里被告知 “需要更多信息时调 ask_user，参数自己填”，于是它生成的 ToolUseBlock.input 里就带着 question/ui_type/options——这些参数本身就是 UI 描述。



在ASJ中会单独处理这种ToolSuspendException：



acting() 588-630 行：



1
2
3
4
5
List<...> successPairs = results.stream().filter(e -> !e.getValue().isSuspended()).to
List();
List<...> pendingPairs = results.stream().filter(e ->  e.getValue().isSuspended()).to
List();...if (!pendingPairs.isEmpty()) {
    return Mono.just(buildSuspendedMsg(pendingPairs));   // 直接返回，TOOL_SUSPENDED}

buildSuspendedMsg 把 ToolUseBlock + 占位的 pending ToolResultBlock 一起塞进消息：



1
2
3
4
5
private Msg buildSuspendedMsg(...) {
    List<ContentBlock> content = new Array
List<>();
    for (var pair : pendingPairs) {
        content.add(pair.getKey());     // ToolUseBlock：含 question/ui_type        content.add(pair.getValue());   // 占位 ToolResultBlock：含 toolId    }    return Msg.builder().generateReason(GenerateReason.TOOL_SUSPENDED) ... .build();}



前端读 ToolUseBlock.input 渲染 UI，用户填完，按 toolId 构造 ToolResult 注回：



1
ToolResultBlock answer = ToolResultBlock.of(        toolId, "ask_user",        TextBlock.builder().text("User responded: " + userInput).build());Msg responseMsg = Msg.builder().role(MsgRole.TOOL).content(answer).build();agent.call(responseMsg).block();



兜底策略



看完HIL的实现之后，不知道你会不会和我一样有个疑问：如果应用层忘了恢复怎么办？



比如暂停后用户直接刷新页面，新一轮发了一条普通问题进来。这时 memory 里还有”悬空的 ToolUseBlock”，下一轮 reasoning 一进 LLM 就会报错。



为此ASJ默认搞了一个 PendingToolRecoveryHook（priority=10，最先跑）：



1
2
3
4
5
6
7
private Mono<PreCallEvent> handlePreCall(PreCallEvent event) {
    Set<String> pendingIds = findPendingToolUseIds(memory);
    if (pendingIds.isEmpty()) return Mono.just(event);
    List<Msg> input = event.getInputMessages();
    // ① 输入为空 = 用户在用 agent.call() 恢复 → 不要管    if (input == null || input.isEmpty()) return Mono.just(event);
    // ② 输入里已经有 ToolResult = 应用层自己处理过 → 不要管    if (input.stream().anyMatch(m -> m.hasContentBlocks(ToolResultBlock.class)))        return Mono.just(event);
    // ③ 否则注入合成 error result，避免崩溃    patchPendingToolCalls(reactAgent, memory, pendingIds);    return Mono.just(event);}



它只在”用户没按规则恢复”时兜底。如果你做了完整的状态持久化（比如把 session 序列化到 DB），可以 Builder.enablePendingToolRecovery(false) 关掉，自己掌控。



Demo



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import io.agentscope.core.ReActAgent;
import io.agentscope.core.formatter.dashscope.DashScopeChatFormatter;
import io.agentscope.core.hook.*;
import io.agentscope.core.memory.InMemoryMemory;
import io.agentscope.core.message.*;
import io.agentscope.core.model.DashScopeChatModel;
import io.agentscope.core.tool.*;
import java.util.List;
import java.util.Scanner;
import reactor.core.publisher.Mono;
public class HitlDemo {
    public static class Tools {
        @Tool(name = "delete_file", description = "Delete a file")        public String del(
@ToolParam(name = "filename") String f) {
            return "deleted " + f;        }
        @Tool(name = "search_web", description = "Search the web")        public String search(
@ToolParam(name = "q") String q) {
            return "results for " + q;        }    }
    static class ConfirmHook implements Hook {
        static final List<String> SENSITIVE = List.of("delete_file");
        @Override public <T extends HookEvent> Mono<T> onEvent(T e) {
            if (e instanceof PostReasoningEvent p) {
                boolean dangerous = p.getReasoningMessage()                        .getContentBlocks(ToolUseBlock.class).stream()                        .anyMatch(t -> SENSITIVE.contains(t.getName()));
                if (dangerous) p.stopAgent();            }
            return Mono.just(e);        }    }
    public static void main(String[] args) {
        Toolkit toolkit = new Toolkit();        toolkit.registerTool(new Tools());
        ReActAgent agent = ReActAgent.builder()                .name("SafeAgent")                .model(DashScopeChatModel.builder()                        .apiKey(System.getenv("DASHSCOPE_API_KEY"))                        .modelName("qwen-plus").stream(true)                        .formatter(new DashScopeChatFormatter()).build())                .toolkit(toolkit)                .memory(new InMemoryMemory())                .hook(new ConfirmHook())                .build();
        Scanner sc = new Scanner(System.in);
        while (true) {
            System.out.pr
int("You: ");            Msg userMsg = Msg.builder().role(MsgRole.USER)                    .content(TextBlock.builder().text(sc.nextLine()).build()).build();            Msg resp = agent.call(userMsg).block();
            while (resp != null && resp.hasContentBlocks(ToolUseBlock.class)) {
                System.out.println("⚠ Pending tools: " + resp.getContentBlocks(ToolUseBlock.class));                System.out.pr
int("Confirm? (y/n): ");
                if (sc.nextLine().equalsIgnoreCase("y")) {
                    resp = agent.call().block();   // 无参恢复                } else {                    List<ToolResultBlock> cancel = resp.getContentBlocks(ToolUseBlock.class).stream()                            .map(t -> ToolResultBlock.of(t.getId(), t.getName(),                                    TextBlock.builder().text("Cancelled").build()))                            .toList();                    Msg cancelMsg = Msg.builder().role(MsgRole.TOOL)                            .content(cancel.toArray(new ToolResultBlock[0])).build();                    resp = agent.call(cancelMsg).block();                }            }            System.out.println("Agent: " + resp.getTextContent());        }    }}