Skip to content

fix(core): fail model empty-completion before retry #2968

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Buktal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(core): fail model empty-completion before retry #2968

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 101 additions & 3 deletions agentscope-core/src/main/java/io/agentscope/core/model/ModelUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,14 @@
package io.agentscope.core.model;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

/**
Expand All @@ -43,6 +47,17 @@ private ModelUtils() {
* configuration in GenerateOptions. Both timeout and retry are optional and only applied if
* configured.
*
* <p><b>Empty-completion detection:</b> A stream that completes without any content-bearing
* {@code ChatResponse} (no chunk carries a non-null, non-empty content list) is converted to
* a {@code ModelException} after the timeout check and before retry. Contentless chunks are
* withheld until the first content-bearing chunk arrives (then released in order), so an
* empty completion surfaces the error as the first downstream signal and switchOnFirst-based
* fallback models engage. The check is deliberately type-agnostic: any content block counts,
* so multimodal completions carrying only image/video/audio blocks are not misclassified as
* empty. This detection is unconditional (applies regardless of retry configuration), since
* an empty completion is an upstream transport anomaly that should always surface as an
* error.
*
* <p><b>Timeout Behavior:</b>
* <ul>
* <li>If requestTimeout is configured, the entire request will fail if it exceeds the
Expand Down Expand Up @@ -77,9 +92,11 @@ public static Flux<ChatResponse> applyTimeoutAndRetry(
GenerateOptions effectiveOptions = GenerateOptions.mergeOptions(options, defaultOptions);

// Extract execution config
ExecutionConfig execConfig = effectiveOptions.getExecutionConfig();
ExecutionConfig execConfig =
effectiveOptions != null ? effectiveOptions.getExecutionConfig() : null;

// Apply timeout if configured
if (execConfig != null) {
// Apply timeout if configured
Duration timeout = execConfig.getTimeout();
if (timeout != null) {
responseFlux =
Expand All @@ -92,8 +109,12 @@ public static Flux<ChatResponse> applyTimeoutAndRetry(
provider)));
LOG.debug("Applied timeout: {} for model: {}", timeout, modelName);
}
}

// Apply retry if configured (maxAttempts > 1 means retry is enabled)
responseFlux = ensureNonEmptyCompletion(responseFlux, modelName, provider);

// Apply retry if configured
if (execConfig != null) {
Integer maxAttempts = execConfig.getMaxAttempts();
if (maxAttempts != null && maxAttempts > 1) {
Duration initialBackoff = execConfig.getInitialBackoff();
Expand Down Expand Up @@ -138,6 +159,83 @@ public static Flux<ChatResponse> applyTimeoutAndRetry(
return responseFlux;
}

/**
* Ensures the completion contains at least one content-bearing response.
*
* <p>Chunks that carry no content block are withheld until the first content-bearing chunk
* arrives (then released in order); a stream that completes without one — zero chunks, or
* only chunks whose {@code getContent()} is null/empty — is converted to a {@code
* ModelException}. Withholding is what keeps fallback effective: switchOnFirst-based
* fallback commits on the first downstream signal, so an empty completion must reach it as
* an error, not preceded by contentless chunks that would already commit the primary model.
* Any content block counts as content-bearing (multimodal blocks included). This detection
* sits after timeout (which would have already fired) and before retryWhen, so an empty
* completion triggers retry just like any other model error.
*
* @param responseFlux the response stream to guard
* @param modelName the model name for error messages
* @param provider the provider name for error messages
* @return guarded flux that errors on empty completions
*/
private static Flux<ChatResponse> ensureNonEmptyCompletion(
Flux<ChatResponse> responseFlux, String modelName, String provider) {
// Per-subscription state: retryWhen resubscribes this chain on each attempt, so the
// withheld-chunk buffer and the released flag must be recreated per subscription —
// chunks withheld by an attempt that failed mid-stream must not leak into a later,
// empty attempt. Signals are delivered serially per subscriber, so unsynchronized state
// is safe.
return Flux.defer(
() -> {
List<ChatResponse> withheld = new ArrayList<>();
AtomicBoolean released = new AtomicBoolean(false);
return responseFlux
.concatMap(
chunk -> {
if (released.get()) {
return Flux.just(chunk);
}
withheld.add(chunk);
if (!isContentBearing(chunk)) {
return Flux.empty();
}
released.set(true);
return Flux.fromIterable(withheld);
})
.concatWith(
Mono.defer(
() -> {
if (released.get()) {
return Mono.empty();
}
LOG.warn(
"Model {} ({}) returned an empty completion"
+ " (zero content-bearing chunks)",
modelName,
provider);
return Mono.error(
new ModelException(
"Model returned empty completion"
+ " (zero content-bearing"
+ " chunks)",
modelName,
provider));
}));
});
}

/**
* Returns whether the chunk carries any content block.
*
* <p>Type-agnostic by design: multimodal completions may carry only image/video/audio
* blocks, which are valid completions, not empty ones.
*
* @param chunk the response chunk to inspect
* @return true if the chunk carries at least one content block
*/
private static boolean isContentBearing(ChatResponse chunk) {
return chunk.getContent() != null && !chunk.getContent().isEmpty();
}

/**
* Ensures GenerateOptions has MODEL_DEFAULTS for executionConfig applied.
*
Expand Down
49 changes: 49 additions & 0 deletions agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,10 @@
import io.agentscope.core.message.ToolUseBlock;
import io.agentscope.core.model.ChatResponse;
import io.agentscope.core.model.ChatUsage;
import io.agentscope.core.model.GenerateOptions;
import io.agentscope.core.model.Model;
import io.agentscope.core.model.ModelUtils;
import io.agentscope.core.model.ToolSchema;
import io.agentscope.core.tool.Toolkit;
import io.agentscope.core.util.JsonUtils;
import java.time.Duration;
Expand All @@ -52,6 +56,7 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
Expand Down Expand Up @@ -578,6 +583,50 @@ void testFallbackModel() {
assertEquals(1, fallbackModel.getCallCount(), "Fallback model should be called once");
}

@Test
@DisplayName("Should switch to fallback model when the primary returns an empty completion")
void testFallbackModelOnEmptyCompletion() {
// Primary emits only contentless chunks (role-only gateway response) and completes —
// the empty-completion guard must surface the error as the first signal so the
// switchOnFirst fallback engages instead of the run ending as silent success
Model primaryModel =
new Model() {
@Override
public Flux<ChatResponse> stream(
List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
Flux<ChatResponse> responseFlux =
Flux.just(
new ChatResponse("chunk-1", List.of(), null, null, null),
new ChatResponse("chunk-2", null, null, null, null));
return ModelUtils.applyTimeoutAndRetry(
responseFlux, options, null, "empty-primary", "test");
}

@Override
public String getModelName() {
return "empty-primary";
}
};
MockModel fallbackModel = new MockModel("Fallback response");

agent =
ReActAgent.builder()
.name(TestConstants.TEST_REACT_AGENT_NAME)
.sysPrompt(TestConstants.DEFAULT_SYS_PROMPT)
.model(primaryModel)
.fallbackModel(fallbackModel)
.toolkit(mockToolkit)
.build();

Msg userMsg = TestUtils.createUserMessage("User", TestConstants.TEST_USER_INPUT);
Msg response =
agent.call(userMsg).block(Duration.ofMillis(TestConstants.DEFAULT_TEST_TIMEOUT_MS));

assertNotNull(response, "Response should not be null");
assertEquals("Fallback response", TestUtils.extractTextContent(response));
assertEquals(1, fallbackModel.getCallCount(), "Fallback model should be called once");
}

@Test
@DisplayName("Should support streaming responses")
void testStreaming() {
Expand Down
Loading
Loading