From ba91a09d5a6f91b8e52a478c2a314bf9e57e273f Mon Sep 17 00:00:00 2001 From: wangminghui <2768495589@qq.com> Date: Tue, 1 Sep 2026 12:35:07 +0800 Subject: [PATCH] feat(middleware): add final answer filter for ReAct streams --- .../FinalAnswerFilterMiddleware.java | 140 ++++++++++++++++ .../ReActAgentMiddlewareIntegrationTest.java | 105 ++++++++++++ .../FinalAnswerFilterMiddlewareTest.java | 151 ++++++++++++++++++ docs/v2/en/docs/building-blocks/middleware.md | 18 +++ docs/v2/zh/docs/building-blocks/middleware.md | 18 +++ 5 files changed, 432 insertions(+) create mode 100644 agentscope-core/src/main/java/io/agentscope/core/middleware/FinalAnswerFilterMiddleware.java create mode 100644 agentscope-core/src/test/java/io/agentscope/core/middleware/FinalAnswerFilterMiddlewareTest.java diff --git a/agentscope-core/src/main/java/io/agentscope/core/middleware/FinalAnswerFilterMiddleware.java b/agentscope-core/src/main/java/io/agentscope/core/middleware/FinalAnswerFilterMiddleware.java new file mode 100644 index 0000000000..422bb3df8a --- /dev/null +++ b/agentscope-core/src/main/java/io/agentscope/core/middleware/FinalAnswerFilterMiddleware.java @@ -0,0 +1,140 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.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.ModelCallStartEvent; +import io.agentscope.core.event.TextBlockDeltaEvent; +import io.agentscope.core.event.TextBlockEndEvent; +import io.agentscope.core.event.TextBlockStartEvent; +import io.agentscope.core.event.ToolCallStartEvent; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.function.Function; +import reactor.core.publisher.Flux; + +/** + * Exposes only the text from the final reasoning round of a ReAct stream. + * + *

Text events are buffered until the current model call completes. If the model produces a + * tool call during that round, the buffered and subsequent text events are suppressed. Otherwise, + * the buffered text events are emitted before the corresponding {@link ModelCallEndEvent}. + * + *

This middleware is opt-in and does not change the default behavior of {@code ReActAgent}. + * Because a round can only be identified as intermediate after a tool call is observed, text from + * the current round is not emitted until the model call completes or a tool call is detected. + */ +public class FinalAnswerFilterMiddleware implements MiddlewareBase { + + @Override + public Flux onReasoning( + Agent agent, + RuntimeContext ctx, + ReasoningInput input, + Function> next) { + return Flux.defer( + () -> { + RoundState state = new RoundState(); + return next.apply(input) + .concatMap(state::handle) + .doFinally(signal -> state.clear()); + }); + } + + private static boolean isTextBlockEvent(AgentEvent event) { + return event instanceof TextBlockStartEvent + || event instanceof TextBlockDeltaEvent + || event instanceof TextBlockEndEvent; + } + + private static final class RoundState { + private final List bufferedTextEvents = new ArrayList<>(); + private String replyId; + private boolean toolCallSeen; + + private Flux handle(AgentEvent event) { + if (event instanceof ModelCallStartEvent start) { + replyId = start.getReplyId(); + toolCallSeen = false; + bufferedTextEvents.clear(); + return Flux.just(event); + } + + if (isTextBlockEvent(event)) { + if (isCurrentReply(event) && !toolCallSeen) { + bufferedTextEvents.add(event); + return Flux.empty(); + } + if (isCurrentReply(event)) { + return Flux.empty(); + } + return Flux.just(event); + } + + if (event instanceof ToolCallStartEvent toolCall + && Objects.equals(replyId, toolCall.getReplyId())) { + toolCallSeen = true; + bufferedTextEvents.clear(); + return Flux.just(event); + } + + if (event instanceof ModelCallEndEvent end && isCurrentReply(end)) { + if (toolCallSeen) { + clear(); + return Flux.just(event); + } + + List finalEvents = new ArrayList<>(bufferedTextEvents); + finalEvents.add(event); + clear(); + return Flux.fromIterable(finalEvents); + } + + return Flux.just(event); + } + + private boolean isCurrentReply(AgentEvent event) { + String eventReplyId = getReplyId(event); + return replyId != null && Objects.equals(replyId, eventReplyId); + } + + private static String getReplyId(AgentEvent event) { + if (event instanceof TextBlockStartEvent textStart) { + return textStart.getReplyId(); + } + if (event instanceof TextBlockDeltaEvent textDelta) { + return textDelta.getReplyId(); + } + if (event instanceof TextBlockEndEvent textEnd) { + return textEnd.getReplyId(); + } + if (event instanceof ModelCallEndEvent modelEnd) { + return modelEnd.getReplyId(); + } + return null; + } + + private void clear() { + bufferedTextEvents.clear(); + replyId = null; + toolCallSeen = false; + } + } +} diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentMiddlewareIntegrationTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentMiddlewareIntegrationTest.java index ac4e82e451..0b1b4844bc 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentMiddlewareIntegrationTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentMiddlewareIntegrationTest.java @@ -26,13 +26,18 @@ import io.agentscope.core.event.AgentEvent; import io.agentscope.core.event.AgentStartEvent; import io.agentscope.core.event.HintBlockEvent; +import io.agentscope.core.event.ModelCallEndEvent; import io.agentscope.core.event.ModelCallStartEvent; import io.agentscope.core.event.TextBlockDeltaEvent; +import io.agentscope.core.event.ToolCallStartEvent; import io.agentscope.core.message.ContentBlock; import io.agentscope.core.message.Msg; import io.agentscope.core.message.TextBlock; +import io.agentscope.core.message.ToolResultBlock; +import io.agentscope.core.message.ToolUseBlock; import io.agentscope.core.middleware.ActingInput; import io.agentscope.core.middleware.AgentInput; +import io.agentscope.core.middleware.FinalAnswerFilterMiddleware; import io.agentscope.core.middleware.MiddlewareBase; import io.agentscope.core.middleware.ModelCallInput; import io.agentscope.core.middleware.ReasoningInput; @@ -40,12 +45,18 @@ import io.agentscope.core.model.ChatResponse; import io.agentscope.core.model.GenerateOptions; import io.agentscope.core.model.ToolSchema; +import io.agentscope.core.permission.PermissionContextState; +import io.agentscope.core.permission.PermissionDecision; import io.agentscope.core.state.AgentState; +import io.agentscope.core.tool.ToolBase; +import io.agentscope.core.tool.ToolCallParam; import io.agentscope.core.tool.Toolkit; import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; @@ -80,6 +91,71 @@ protected Flux doStream( } } + private static final class ToolThenFinalModel extends ChatModelBase { + private final AtomicInteger calls = new AtomicInteger(); + + @Override + public String getModelName() { + return "tool-then-final"; + } + + @Override + protected Flux doStream( + List messages, List tools, GenerateOptions options) { + if (calls.getAndIncrement() == 0) { + return Flux.just( + ChatResponse.builder() + .content( + List.of( + TextBlock.builder() + .text("intermediate text") + .build(), + ToolUseBlock.builder() + .id("tool-call-1") + .name("lookup") + .input(Map.of("query", "AgentScope")) + .build())) + .build()); + } + return Flux.just( + ChatResponse.builder() + .content( + List.of( + TextBlock.builder().text("final answer").build())) + .build()); + } + } + + private static final class LookupTool extends ToolBase { + LookupTool() { + super( + "lookup", + "Looks up a query", + Map.of( + "type", + "object", + "properties", + Map.of("query", Map.of("type", "string"))), + true, + true, + false, + null, + false, + false); + } + + @Override + public Mono checkPermissions( + Map input, PermissionContextState ctx) { + return Mono.just(PermissionDecision.allow("allowed")); + } + + @Override + public Mono callAsync(ToolCallParam param) { + return Mono.just(ToolResultBlock.text("lookup result")); + } + } + private static final class InterruptedAfterTextModel extends ChatModelBase { @Override public String getModelName() { @@ -167,6 +243,35 @@ private static ReActAgent buildAgent(ChatModelBase model, List m .build(); } + @Test + void finalAnswerFilterSuppressesIntermediateReactRoundText() { + ToolThenFinalModel model = new ToolThenFinalModel(); + Toolkit toolkit = new Toolkit(); + toolkit.registerAgentTool(new LookupTool()); + ReActAgent agent = + ReActAgent.builder() + .name("asst") + .sysPrompt("hello-system") + .model(model) + .toolkit(toolkit) + .middleware(new FinalAnswerFilterMiddleware()) + .build(); + + List events = agent.streamEvents(List.of()).collectList().block(); + + assertNotNull(events); + assertEquals(2, model.calls.get()); + assertEquals( + List.of("final answer"), + events.stream() + .filter(TextBlockDeltaEvent.class::isInstance) + .map(TextBlockDeltaEvent.class::cast) + .map(TextBlockDeltaEvent::getDelta) + .toList()); + assertEquals(2L, events.stream().filter(ModelCallEndEvent.class::isInstance).count()); + assertTrue(events.stream().anyMatch(ToolCallStartEvent.class::isInstance)); + } + @Test void singleMiddlewareSeesReplyAndReasoningAndModelCall() { List trace = new ArrayList<>(); diff --git a/agentscope-core/src/test/java/io/agentscope/core/middleware/FinalAnswerFilterMiddlewareTest.java b/agentscope-core/src/test/java/io/agentscope/core/middleware/FinalAnswerFilterMiddlewareTest.java new file mode 100644 index 0000000000..c8a8cdf02b --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/middleware/FinalAnswerFilterMiddlewareTest.java @@ -0,0 +1,151 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.middleware; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.agentscope.core.event.AgentEvent; +import io.agentscope.core.event.ModelCallEndEvent; +import io.agentscope.core.event.ModelCallStartEvent; +import io.agentscope.core.event.TextBlockDeltaEvent; +import io.agentscope.core.event.TextBlockEndEvent; +import io.agentscope.core.event.TextBlockStartEvent; +import io.agentscope.core.event.ThinkingBlockDeltaEvent; +import io.agentscope.core.event.ToolCallStartEvent; +import io.agentscope.core.model.ChatUsage; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; + +class FinalAnswerFilterMiddlewareTest { + + private static final String REPLY_ID = "reply-1"; + + private final FinalAnswerFilterMiddleware middleware = new FinalAnswerFilterMiddleware(); + + @Test + void finalRoundEmitsBufferedTextBeforeModelCallEnd() { + List events = + apply( + Flux.just( + new ModelCallStartEvent(REPLY_ID), + new TextBlockStartEvent(REPLY_ID, "text"), + new TextBlockDeltaEvent(REPLY_ID, "text", "final answer"), + new TextBlockEndEvent(REPLY_ID, "text"), + new ModelCallEndEvent(REPLY_ID, (ChatUsage) null))); + + assertEquals(5, events.size()); + assertTrue(events.get(0) instanceof ModelCallStartEvent); + assertTrue(events.get(1) instanceof TextBlockStartEvent); + assertTrue(events.get(2) instanceof TextBlockDeltaEvent); + assertTrue(events.get(3) instanceof TextBlockEndEvent); + assertTrue(events.get(4) instanceof ModelCallEndEvent); + } + + @Test + void intermediateRoundSuppressesTextWhenToolCallIsObserved() { + List events = + apply( + Flux.just( + new ModelCallStartEvent(REPLY_ID), + new TextBlockStartEvent(REPLY_ID, "text"), + new TextBlockDeltaEvent(REPLY_ID, "text", "intermediate"), + new TextBlockEndEvent(REPLY_ID, "text"), + new ToolCallStartEvent(REPLY_ID, "tool-1", "search"), + new ModelCallEndEvent(REPLY_ID, (ChatUsage) null))); + + assertEquals(3, events.size()); + assertTrue(events.get(0) instanceof ModelCallStartEvent); + assertTrue(events.get(1) instanceof ToolCallStartEvent); + assertTrue(events.get(2) instanceof ModelCallEndEvent); + assertFalse(events.stream().anyMatch(TextBlockStartEvent.class::isInstance)); + assertFalse(events.stream().anyMatch(TextBlockDeltaEvent.class::isInstance)); + assertFalse(events.stream().anyMatch(TextBlockEndEvent.class::isInstance)); + } + + @Test + void nonTextEventsAreForwarded() { + ThinkingBlockDeltaEvent thinking = + new ThinkingBlockDeltaEvent(REPLY_ID, "thinking", "reasoning"); + ToolCallStartEvent toolCall = new ToolCallStartEvent(REPLY_ID, "tool-1", "search"); + + List events = + apply( + Flux.just( + new ModelCallStartEvent(REPLY_ID), + thinking, + new TextBlockDeltaEvent(REPLY_ID, "text", "intermediate"), + toolCall, + new ModelCallEndEvent(REPLY_ID, (ChatUsage) null))); + + assertTrue(events.contains(thinking)); + assertTrue(events.contains(toolCall)); + } + + @Test + void stateIsolatedAcrossSubscriptions() { + AtomicInteger subscriptionCount = new AtomicInteger(); + Flux events = + middleware.onReasoning( + null, + null, + new ReasoningInput(List.of(), List.of(), null), + ignored -> + Flux.defer( + () -> { + if (subscriptionCount.getAndIncrement() == 0) { + return Flux.just( + new ModelCallStartEvent(REPLY_ID), + new TextBlockDeltaEvent( + REPLY_ID, "text", "intermediate"), + new ToolCallStartEvent( + REPLY_ID, "tool-1", "search"), + new ModelCallEndEvent( + REPLY_ID, (ChatUsage) null)); + } + return Flux.just( + new ModelCallStartEvent(REPLY_ID), + new TextBlockDeltaEvent( + REPLY_ID, "text", "final answer"), + new ModelCallEndEvent( + REPLY_ID, (ChatUsage) null)); + })); + + List first = events.collectList().block(); + List second = events.collectList().block(); + + assertFalse(first.stream().anyMatch(TextBlockDeltaEvent.class::isInstance)); + assertTrue( + second.stream() + .filter(TextBlockDeltaEvent.class::isInstance) + .map(TextBlockDeltaEvent.class::cast) + .anyMatch(event -> "final answer".equals(event.getDelta()))); + } + + private List apply(Flux source) { + return middleware + .onReasoning( + null, + null, + new ReasoningInput(List.of(), List.of(), null), + ignored -> source) + .collectList() + .block(); + } +} diff --git a/docs/v2/en/docs/building-blocks/middleware.md b/docs/v2/en/docs/building-blocks/middleware.md index 129b22b878..3fa4ea25b4 100644 --- a/docs/v2/en/docs/building-blocks/middleware.md +++ b/docs/v2/en/docs/building-blocks/middleware.md @@ -171,6 +171,24 @@ ReActAgent agent = .build(); ``` +### FinalAnswerFilterMiddleware + +`FinalAnswerFilterMiddleware` exposes only the text from the final ReAct reasoning round. Text from rounds that produce tool calls is suppressed, while tool and other non-text events continue to stream normally. + +```java +import io.agentscope.core.middleware.FinalAnswerFilterMiddleware; + +ReActAgent agent = + ReActAgent.builder() + .name("assistant") + .model(model) + .toolkit(toolkit) + .middleware(new FinalAnswerFilterMiddleware()) + .build(); +``` + +The middleware buffers each round's text until the model call ends, because it cannot know whether the round is final until no tool call is observed. + ## Custom middleware Implement `MiddlewareBase` (`io.agentscope.core.middleware`) and override only the hooks you need. diff --git a/docs/v2/zh/docs/building-blocks/middleware.md b/docs/v2/zh/docs/building-blocks/middleware.md index 7d962bfa0d..3a1efde56c 100644 --- a/docs/v2/zh/docs/building-blocks/middleware.md +++ b/docs/v2/zh/docs/building-blocks/middleware.md @@ -171,6 +171,24 @@ ReActAgent agent = .build(); ``` +### FinalAnswerFilterMiddleware + +`FinalAnswerFilterMiddleware` 仅输出 ReAct 最终推理轮次的文本。产生工具调用的中间轮次文本会被过滤，工具事件及其他非文本事件仍会正常流式输出。 + +```java +import io.agentscope.core.middleware.FinalAnswerFilterMiddleware; + +ReActAgent agent = + ReActAgent.builder() + .name("assistant") + .model(model) + .toolkit(toolkit) + .middleware(new FinalAnswerFilterMiddleware()) + .build(); +``` + +由于只有在未观察到工具调用时才能确定当前轮次是最终轮次，该 middleware 会将每轮文本缓冲到模型调用结束。 + ## 自定义 Middleware 实现 `MiddlewareBase` 接口（位于 `io.agentscope.core.middleware`），只重写需要的 hook 即可，其它的不用管。