mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-11-02 14:11:49 +08:00
fix(agent): retry finalization after reasoning exhausts output budget (#672)
This commit is contained in:
matevip
parent
ccf33f8357
commit
f023b6685a
@ -1709,12 +1709,16 @@ public class NodeStreamingChatHelper {
|
||||
: truncatedByContentRepeat ? "content_repetition"
|
||||
: thinkingTokenLimit ? "thinking_token_limit"
|
||||
: null;
|
||||
// Retain the provider's end condition even when it emitted some content.
|
||||
// A verifier JSON cut off by length is not a malformed completed verdict.
|
||||
Map<String, Object> resultMetadata = new LinkedHashMap<>(providerMetadata.get());
|
||||
if (finishReason.get() != null) resultMetadata.put("generationFinishReason", finishReason.get());
|
||||
return assembleResult(contentAccum, thinkingAccum, toolCallAccumulators,
|
||||
promptTokens.get(), completionTokens.get(),
|
||||
cacheReadTokens.get(), cacheWriteTokens.get(),
|
||||
reasoningTokens.get(), phase,
|
||||
truncated,
|
||||
truncationReason, providerMetadata.get());
|
||||
truncationReason, resultMetadata);
|
||||
}
|
||||
|
||||
/** 组装 stopped partial 结果（用户主动停止，有已累积内容） */
|
||||
@ -2388,6 +2392,14 @@ public class NodeStreamingChatHelper {
|
||||
return errorType == ErrorType.PROMPT_TOO_LONG;
|
||||
}
|
||||
|
||||
/** Explicit provider output-limit stop, distinct from transport errors and loop guards. */
|
||||
public boolean isOutputTokenLimit() {
|
||||
if (stopped || hasToolCalls || errorType != ErrorType.NONE) return false;
|
||||
if ("thinking_token_limit".equals(errorMessage)) return true;
|
||||
return errorMessage == null && assistantMessage != null
|
||||
&& "length".equalsIgnoreCase(String.valueOf(assistantMessage.getMetadata().get("generationFinishReason")));
|
||||
}
|
||||
|
||||
/** 是否有任何可保存的内容（text/thinking/toolCalls） */
|
||||
public boolean hasAnyContent() {
|
||||
return (text != null && !text.isBlank())
|
||||
|
||||
@ -18,6 +18,7 @@ import java.util.concurrent.CancellationException;
|
||||
/** Call-scoped publication review. No conversation or attempt state is retained here. */
|
||||
public final class TaskFinalizationCoordinator {
|
||||
private static final int OUTPUT_TOKENS = 1024;
|
||||
private static final int MAX_RETRY_OUTPUT_TOKENS = 8192;
|
||||
private static final String RECOVERY_PROSE_RULES = """
|
||||
Recovery prose must match the specific failed original request. Unqualified phrases such as
|
||||
"随后补齐 language="python" 重试", "补齐参数后成功", "fixed the parameters and retried", or
|
||||
@ -95,7 +96,7 @@ public final class TaskFinalizationCoordinator {
|
||||
ACCEPTED, REVISION_REQUIRED, RECOVERY_RELATION_CORRECTION, CONTINUATION_REQUIRED, INVALID_VERDICT, FAILED_CALL_COUNT_MISMATCH,
|
||||
UNSUPPORTED_RECOVERY, CONTRADICTORY_OUTCOME, TASK_UNFINISHED, INVALID_CONTINUATION,
|
||||
CORRECTION_LIMIT, CONTINUATION_NOT_AVAILABLE, VERIFICATION_REJECTED,
|
||||
CHECKER_UNAVAILABLE, CHECKER_PARTIAL, CHECKER_ERROR, CHECKER_TOOL_CALL, CHECKER_EXCEPTION, CANCELLED
|
||||
CHECKER_UNAVAILABLE, CHECKER_PARTIAL, CHECKER_OUTPUT_LIMIT, CHECKER_ERROR, CHECKER_TOOL_CALL, CHECKER_EXCEPTION, CANCELLED
|
||||
}
|
||||
public record Review(TaskFinalizationGate.Result result, NodeStreamingChatHelper.StreamResult usage,
|
||||
int calls, boolean stopped) {}
|
||||
@ -183,21 +184,46 @@ public final class TaskFinalizationCoordinator {
|
||||
"progressHintsTokens",includedProgressTokens,"progressHintsOmitted",progressOmitted,
|
||||
"canContinue",canContinue,"correctiveAttempts",attempts));
|
||||
NodeStreamingChatHelper.StreamResult response=null;
|
||||
NodeStreamingChatHelper.StreamResult usage=null;
|
||||
int calls=0;
|
||||
String outcome="INCOMPLETE";
|
||||
Diagnostic diagnostic=Diagnostic.CHECKER_UNAVAILABLE;
|
||||
TaskFinalizationGate.ParseFailureCode parseFailureCode=TaskFinalizationGate.ParseFailureCode.NOT_EVALUATED;
|
||||
try {
|
||||
var checkState = new vip.mate.agent.runtime.state.RunState(runState.data());
|
||||
checkState.replace(vip.mate.agent.graph.state.MateClawStateKeys.REQUESTED_THINKING_LEVEL,"off");
|
||||
Prompt checkPrompt = new Prompt(messages,options);
|
||||
response=vip.mate.agent.runtime.state.RuntimeExecutionScope.call(checkState,
|
||||
() -> helper.streamCallSilent(model,checkPrompt,conversationId,"finalization_check"));
|
||||
if(response==null) return incomplete("Checker unavailable",null,1,false);
|
||||
if(stopped(conversationId) || response.stopped()) { diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",response,1,true); }
|
||||
for (int attempt=0; ; attempt++) {
|
||||
Prompt checkPrompt = new Prompt(messages,options);
|
||||
calls++;
|
||||
response=vip.mate.agent.runtime.state.RuntimeExecutionScope.call(checkState,
|
||||
() -> helper.streamCallSilent(model,checkPrompt,conversationId,"finalization_check"));
|
||||
usage=usage==null?response:addUsage(usage,response);
|
||||
if(response==null) return incomplete("Checker unavailable",usage,calls,false);
|
||||
if(stopped(conversationId) || response.stopped()) { diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",usage,calls,true); }
|
||||
// Some compatible providers cannot disable reasoning. Their output
|
||||
// limit includes thinking, so the small JSON allowance can expire
|
||||
// before a complete verdict is emitted. Retry only that explicit condition,
|
||||
// once, without replaying tools or dropping evidence from the prompt.
|
||||
if(attempt==0 && response.isOutputTokenLimit() && callAvailable(maxIterations,usedCalls+calls)) {
|
||||
int configured=vip.mate.agent.context.BudgetedCompactionModel.outputReserveTokens(
|
||||
ChatModelDefaults.optionsFor(model),4096);
|
||||
int available=window-TokenEstimator.estimateTokens(messages)-128;
|
||||
int retryTokens=Math.min(Math.min(configured,MAX_RETRY_OUTPUT_TOKENS),available);
|
||||
if(retryTokens>OUTPUT_TOKENS) {
|
||||
options=toolFreeOptions(model,retryTokens);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
if(response.isOutputTokenLimit()) {
|
||||
diagnostic=Diagnostic.CHECKER_OUTPUT_LIMIT;
|
||||
return incomplete(diagnostic.name(),usage,calls,false);
|
||||
}
|
||||
if(response.partial() || response.errorMessage()!=null || response.hasToolCalls()) {
|
||||
diagnostic=response.errorMessage()!=null?Diagnostic.CHECKER_ERROR
|
||||
:response.partial()?Diagnostic.CHECKER_PARTIAL:Diagnostic.CHECKER_TOOL_CALL;
|
||||
return incomplete(diagnostic.name(),response,1,false);
|
||||
return incomplete(diagnostic.name(),usage,calls,false);
|
||||
}
|
||||
var verdict=gate.parse(response.text()).orElse(null);
|
||||
parseFailureCode=verdict==null?gate.parseFailureCode(response.text()):TaskFinalizationGate.ParseFailureCode.NONE;
|
||||
@ -205,16 +231,16 @@ public final class TaskFinalizationCoordinator {
|
||||
outcome=evaluated.outcome().name();
|
||||
diagnostic=diagnostic(verdict,evaluated,ledger,attempts);
|
||||
if (evaluated.outcome() == TaskFinalizationGate.Outcome.INCOMPLETE)
|
||||
return incomplete(diagnostic.name(), response, 1, false);
|
||||
return new Review(evaluated,response,1,false);
|
||||
} catch(CancellationException cancelled) {diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",response,1,true);}
|
||||
catch(Exception unavailable) {diagnostic=Diagnostic.CHECKER_EXCEPTION;return incomplete("CHECKER_EXCEPTION",response,1,false);}
|
||||
return incomplete(diagnostic.name(), usage, calls, false);
|
||||
return new Review(evaluated,usage,calls,false);
|
||||
} catch(CancellationException cancelled) {diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",usage,calls,true);}
|
||||
catch(Exception unavailable) {diagnostic=Diagnostic.CHECKER_EXCEPTION;return incomplete("CHECKER_EXCEPTION",usage,calls,false);}
|
||||
finally {
|
||||
var completed=new java.util.LinkedHashMap<String,Object>(Map.of("logicalCalls",1,"promptTokens",response==null?0:response.promptTokens(),
|
||||
"completionTokens",response==null?0:response.completionTokens(),"usageReported",response!=null,
|
||||
"cacheReadTokens",response==null?0:response.cacheReadTokens(),
|
||||
"cacheWriteTokens",response==null?0:response.cacheWriteTokens(),
|
||||
"reasoningTokens",response==null?0:response.reasoningTokens(),"outcome",outcome));
|
||||
var completed=new java.util.LinkedHashMap<String,Object>(Map.of("logicalCalls",calls,"promptTokens",usage==null?0:usage.promptTokens(),
|
||||
"completionTokens",usage==null?0:usage.completionTokens(),"usageReported",usage!=null,
|
||||
"cacheReadTokens",usage==null?0:usage.cacheReadTokens(),
|
||||
"cacheWriteTokens",usage==null?0:usage.cacheWriteTokens(),
|
||||
"reasoningTokens",usage==null?0:usage.reasoningTokens(),"outcome",outcome));
|
||||
completed.put("diagnostic",diagnostic.name());
|
||||
completed.put("parseFailureCode",parseFailureCode.name());
|
||||
completed.put("responseTextChars",response==null || response.text()==null?0:response.text().length());
|
||||
@ -256,19 +282,22 @@ public final class TaskFinalizationCoordinator {
|
||||
return options;
|
||||
}
|
||||
public static ChatOptions toolFreeOptions(ChatModel model) {
|
||||
return toolFreeOptions(model,OUTPUT_TOKENS);
|
||||
}
|
||||
private static ChatOptions toolFreeOptions(ChatModel model,int outputTokens) {
|
||||
ChatOptions defaults=ChatModelDefaults.optionsFor(model);
|
||||
ChatOptions options=defaults==null?ChatOptions.builder().maxTokens(OUTPUT_TOKENS).build():defaults.mutate().maxTokens(OUTPUT_TOKENS).build();
|
||||
ChatOptions options=defaults==null?ChatOptions.builder().maxTokens(outputTokens).build():defaults.mutate().maxTokens(outputTokens).build();
|
||||
if(options instanceof ToolCallingChatOptions tools) options=tools.mutate().toolCallbacks(List.of()).build();
|
||||
// OpenAI reasoning providers use maxCompletionTokens instead of maxTokens.
|
||||
if(options instanceof org.springframework.ai.openai.OpenAiChatOptions openAi) {
|
||||
options = openAi.getMaxCompletionTokens()!=null
|
||||
? openAi.mutate().maxTokens(null).maxCompletionTokens(OUTPUT_TOKENS).toolChoice("none").build()
|
||||
: openAi.mutate().maxTokens(OUTPUT_TOKENS).toolChoice("none").build();
|
||||
? openAi.mutate().maxTokens(null).maxCompletionTokens(outputTokens).toolChoice("none").build()
|
||||
: openAi.mutate().maxTokens(outputTokens).toolChoice("none").build();
|
||||
}
|
||||
if(options instanceof org.springframework.ai.anthropic.AnthropicChatOptions anthropic)
|
||||
options=anthropic.mutate().thinkingDisabled().maxTokens(OUTPUT_TOKENS).toolCallbacks(List.of()).build();
|
||||
options=anthropic.mutate().thinkingDisabled().maxTokens(outputTokens).toolCallbacks(List.of()).build();
|
||||
if(options instanceof vip.mate.llm.dashscope.NativeDashScopeChatOptions dashscope)
|
||||
options=dashscope.mutate().enableThinking(false).maxTokens(OUTPUT_TOKENS).toolCallbacks(List.of()).build();
|
||||
options=dashscope.mutate().enableThinking(false).maxTokens(outputTokens).toolCallbacks(List.of()).build();
|
||||
return options;
|
||||
}
|
||||
private void event(String id,String status,Map<String,Object> details) {
|
||||
@ -295,6 +324,7 @@ public final class TaskFinalizationCoordinator {
|
||||
case "Invalid verification facts", "INVALID_VERDICT", "FAILED_CALL_COUNT_MISMATCH",
|
||||
"UNSUPPORTED_RECOVERY", "CONTRADICTORY_OUTCOME", "INVALID_CONTINUATION", "VERIFICATION_REJECTED" -> "最终核验返回格式无效或与实际工具记录不一致，未发布未经核实的回答";
|
||||
case "CHECKER_PARTIAL" -> "最终核验响应被截断，无法确认核验结果";
|
||||
case "CHECKER_OUTPUT_LIMIT" -> "最终核验在生成核验结果前耗尽输出 token 预算，请适当增加模型最大输出 token 数或拆分任务后重试";
|
||||
case "CHECKER_TOOL_CALL" -> "最终核验意外返回工具调用，未获得有效核验结果";
|
||||
case "Checker unavailable", "CHECKER_ERROR", "CHECKER_EXCEPTION", "Verification unavailable",
|
||||
"Summary verification unavailable", "Summary unavailable" -> "最终核验或汇总调用失败，任务是否完成尚未确认";
|
||||
|
||||
@ -47,6 +47,21 @@ class NodeStreamingChatHelperThinkingCapTest {
|
||||
return m;
|
||||
}
|
||||
|
||||
@Test
|
||||
void lengthFinishIsRetainedEvenWhenPartialContentWasEmitted() {
|
||||
var msg = AssistantMessage.builder().content("{\"decision\":\"ACCEPT\",")
|
||||
.properties(Map.of("reasoningContent", "Reviewing fifteen forecast rows")).build();
|
||||
var metadata = ChatGenerationMetadata.builder().finishReason("length").build();
|
||||
var model = mock(ChatModel.class);
|
||||
when(model.stream(any(Prompt.class))).thenReturn(Flux.just(
|
||||
new ChatResponse(List.of(new Generation(msg, metadata)))));
|
||||
var result = new NodeStreamingChatHelper(streamTracker)
|
||||
.streamCallSilent(model, smallPrompt(), "c", "finalization_check");
|
||||
assertEquals("length", result.assistantMessage().getMetadata().get("generationFinishReason"));
|
||||
assertEquals(msg.getText(), result.text());
|
||||
assertFalse(result.partial(), "Do not change live chunk/continuation semantics when retaining finish metadata");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("thinking >= cap + tool_call in same chunk: cap must NOT trigger; tool_call survives")
|
||||
void thinkingAndToolCallSameChunk_doesNotTripSoftCap() {
|
||||
|
||||
@ -26,6 +26,135 @@ class TaskFinalizationCoordinatorTest {
|
||||
static String json(String decision,String claims){return "{\"decision\":\""+decision+"\",\"taskComplete\":true,\"remainingWork\":false,\"failedCallCount\":1,\"recoveryClaims\":"+claims+",\"reason\":\"checked\"}";}
|
||||
NodeStreamingChatHelper.StreamResult result(String text){return new NodeStreamingChatHelper.StreamResult(text,"checker thinking",null,List.of(),false,7,3,false,null,NodeStreamingChatHelper.ErrorType.NONE,false,2,1,1);}
|
||||
TaskFinalizationCoordinator.Review review(int window,int attempts,int used){return coordinator.review(model,"c","Calculate all stages","candidate",ledger,"untrusted progress",window,attempts,10,used,true,new RunState());}
|
||||
@Test void reasoningTokenExhaustionRetriesOnceWithConfiguredBudget() {
|
||||
var defaults = OpenAiChatOptions.builder().model("sensenova-6.8-flash-lite").maxTokens(8192).build();
|
||||
when(model.getDefaultOptions()).thenReturn(defaults);
|
||||
var calls = new java.util.concurrent.atomic.AtomicInteger();
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> {
|
||||
Prompt prompt = call.getArgument(1);
|
||||
assertEquals("off", ThinkingLevelHolder.get());
|
||||
assertTrue(((ToolCallingChatOptions) prompt.getOptions()).getToolCallbacks().isEmpty());
|
||||
if (calls.getAndIncrement() == 0) {
|
||||
assertEquals(1024, prompt.getOptions().getMaxTokens());
|
||||
return new NodeStreamingChatHelper.StreamResult("", "Reviewing all fifteen forecast days", null,
|
||||
List.of(), false, 10, 1024, true, "thinking_token_limit",
|
||||
NodeStreamingChatHelper.ErrorType.NONE, false, 2, 0, 1024);
|
||||
}
|
||||
assertEquals(8192, prompt.getOptions().getMaxTokens());
|
||||
return result(json("ACCEPT", "[]"));
|
||||
});
|
||||
var check = review(16000, 0, 0);
|
||||
assertEquals(TaskFinalizationGate.Outcome.ACCEPT, check.result().outcome());
|
||||
assertEquals(2, check.calls());
|
||||
assertEquals(17, check.usage().promptTokens());
|
||||
assertEquals(1027, check.usage().completionTokens());
|
||||
assertEquals(4, check.usage().cacheReadTokens());
|
||||
assertEquals(1025, check.usage().reasoningTokens());
|
||||
assertEquals(8192, defaults.getMaxTokens());
|
||||
}
|
||||
|
||||
@Test void tokenRetryStaysWithinWindowAndKeepsObjectiveAndEvidence() {
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(32000).build());
|
||||
var calls = new java.util.concurrent.atomic.AtomicInteger();
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> {
|
||||
Prompt prompt = call.getArgument(1);
|
||||
assertTrue(prompt.getContents().contains("Calculate all stages"));
|
||||
assertTrue(prompt.getContents().contains("missing language"));
|
||||
assertTrue(vip.mate.agent.context.TokenEstimator.estimateTokens(prompt.getInstructions())
|
||||
+ prompt.getOptions().getMaxTokens() <= 8000);
|
||||
if (calls.getAndIncrement() == 0) {
|
||||
return new NodeStreamingChatHelper.StreamResult("", "Review in progress", null,
|
||||
List.of(), false, 10, 1024, true, "thinking_token_limit",
|
||||
NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024);
|
||||
}
|
||||
assertTrue(prompt.getOptions().getMaxTokens() > 1024);
|
||||
return result(json("ACCEPT", "[]"));
|
||||
});
|
||||
assertEquals(TaskFinalizationGate.Outcome.ACCEPT, review(8000, 0, 0).result().outcome());
|
||||
assertEquals(2, calls.get());
|
||||
}
|
||||
|
||||
@Test void exhaustedCallBudgetReportsTokenLimitWithoutRetrying() {
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(8192).build());
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn(
|
||||
new NodeStreamingChatHelper.StreamResult("", "Review in progress", null,
|
||||
List.of(), false, 10, 1024, true, "thinking_token_limit",
|
||||
NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024));
|
||||
var check = review(16000, 0, 49);
|
||||
assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome());
|
||||
assertEquals("CHECKER_OUTPUT_LIMIT", check.result().reason());
|
||||
assertEquals(1, check.calls());
|
||||
assertTrue(TaskFinalizationCoordinator.incompleteAnswer(check.result().reason()).contains("token"));
|
||||
verify(helper, times(1)).streamCallSilent(any(), any(), anyString(), anyString());
|
||||
}
|
||||
|
||||
@Test void repeatedExhaustionIsBoundedAndReportsAggregateUsage() {
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(32000).build());
|
||||
var tracker = mock(vip.mate.channel.web.ChatStreamTracker.class);
|
||||
var observed = new TaskFinalizationCoordinator(helper, tracker);
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn(
|
||||
new NodeStreamingChatHelper.StreamResult("", "Still reviewing", null,
|
||||
List.of(), false, 10, 1024, true, "thinking_token_limit",
|
||||
NodeStreamingChatHelper.ErrorType.NONE, false, 2, 1, 1024));
|
||||
var check = observed.review(model, "c", "objective", "candidate", ledger, "", 16000, 0, 10, 0, true, new RunState());
|
||||
assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome());
|
||||
assertEquals("CHECKER_OUTPUT_LIMIT", check.result().reason());
|
||||
assertEquals(2, check.calls());
|
||||
assertEquals(20, check.usage().promptTokens());
|
||||
assertEquals(2048, check.usage().completionTokens());
|
||||
var prompts = org.mockito.ArgumentCaptor.forClass(Prompt.class);
|
||||
verify(helper, times(2)).streamCallSilent(any(), prompts.capture(), anyString(), anyString());
|
||||
assertEquals(8192, prompts.getAllValues().get(1).getOptions().getMaxTokens());
|
||||
var events = org.mockito.ArgumentCaptor.forClass(Map.class);
|
||||
verify(tracker, times(2)).broadcastObject(eq("c"), eq("finalization_check"), events.capture());
|
||||
var completed = events.getAllValues().get(1);
|
||||
assertEquals(2, completed.get("logicalCalls"));
|
||||
assertEquals(20, completed.get("promptTokens"));
|
||||
assertEquals(2048, completed.get("completionTokens"));
|
||||
assertEquals("CHECKER_OUTPUT_LIMIT", completed.get("diagnostic"));
|
||||
}
|
||||
|
||||
@Test void retryCannotTurnInvalidVerdictIntoAcceptance() {
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(8192).build());
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn(
|
||||
new NodeStreamingChatHelper.StreamResult("", "Review", null, List.of(), false,
|
||||
10, 1024, true, "thinking_token_limit", NodeStreamingChatHelper.ErrorType.NONE,
|
||||
false, 0, 0, 1024), result("not a verdict"));
|
||||
var check = review(16000, 0, 0);
|
||||
assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome());
|
||||
assertEquals("INVALID_VERDICT", check.result().reason());
|
||||
assertEquals(2, check.calls());
|
||||
}
|
||||
|
||||
@Test void noBudgetGrowthAndSafetyCapsAreNotRetried() {
|
||||
for (String reason : List.of("thinking_token_limit", "thinking_only_no_content", "content_repetition")) {
|
||||
reset(helper);
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(1024).build());
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn(
|
||||
new NodeStreamingChatHelper.StreamResult("", "Review", null, List.of(), false,
|
||||
10, 1024, true, reason, NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024));
|
||||
assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, review(16000, 0, 0).result().outcome());
|
||||
verify(helper, times(1)).streamCallSilent(any(), any(), anyString(), anyString());
|
||||
}
|
||||
}
|
||||
|
||||
@Test void completionTokenRetryPreservesOpenAiParameterFamily() {
|
||||
when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().model("gpt-5")
|
||||
.maxCompletionTokens(8192).build());
|
||||
var calls = new java.util.concurrent.atomic.AtomicInteger();
|
||||
when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> {
|
||||
var options = (OpenAiChatOptions) ((Prompt) call.getArgument(1)).getOptions();
|
||||
assertNull(options.getMaxTokens());
|
||||
assertEquals(calls.get() == 0 ? 1024 : 8192, options.getMaxCompletionTokens());
|
||||
if (calls.getAndIncrement() == 0) return new NodeStreamingChatHelper.StreamResult("", "Review", null,
|
||||
List.of(), false, 10, 1024, true, "thinking_token_limit", NodeStreamingChatHelper.ErrorType.NONE,
|
||||
false, 0, 0, 1024);
|
||||
return result(json("ACCEPT", "[]"));
|
||||
});
|
||||
assertEquals(TaskFinalizationGate.Outcome.ACCEPT, review(16000, 0, 0).result().outcome());
|
||||
assertEquals(2, calls.get());
|
||||
}
|
||||
|
||||
@Test void incompleteExplanationDistinguishesFixedCausesWithoutLeakingReviewerText() {
|
||||
var context=TaskFinalizationCoordinator.incompleteAnswer("Required execution facts or objective/candidate cannot fit context");
|
||||
var correction=TaskFinalizationCoordinator.incompleteAnswer("Correction allowance exhausted");
|
||||
|
||||
@ -0,0 +1,113 @@
|
||||
package vip.mate.llm.chatmodel;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import io.micrometer.observation.ObservationRegistry;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.support.StaticListableBeanFactory;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.web.client.RestClient;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
import vip.mate.agent.GraphEventPublisher;
|
||||
import vip.mate.agent.graph.NodeStreamingChatHelper;
|
||||
import vip.mate.agent.graph.guard.TaskFinalizationCoordinator;
|
||||
import vip.mate.agent.graph.guard.TaskFinalizationGate;
|
||||
import vip.mate.agent.graph.state.ActionExecutionLedger;
|
||||
import vip.mate.agent.runtime.state.RunState;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.llm.model.ModelConfigEntity;
|
||||
import vip.mate.llm.model.ModelProviderEntity;
|
||||
import vip.mate.llm.service.ModelProviderService;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/** Exercises the real SDK and SSE helper against a reasoning-provider protocol fixture. */
|
||||
class ReasoningFinalizationTransportTest {
|
||||
private static final ObjectMapper JSON = new ObjectMapper();
|
||||
|
||||
@Test
|
||||
void reasoningOnlyLengthResponseRecoversBeforePublishingVerdict() throws Exception {
|
||||
assertRecovers("");
|
||||
}
|
||||
|
||||
@Test
|
||||
void lengthResponseWithTruncatedJsonRecoversBeforePublishingVerdict() throws Exception {
|
||||
assertRecovers("{\"decision\":\"ACCEPT\",\"taskComplete\":true,");
|
||||
}
|
||||
|
||||
private void assertRecovers(String exhaustedContent) throws Exception {
|
||||
var requests = new CopyOnWriteArrayList<JsonNode>();
|
||||
var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
server.createContext("/v1/chat/completions", exchange -> {
|
||||
JsonNode request = JSON.readTree(exchange.getRequestBody());
|
||||
requests.add(request);
|
||||
boolean exhausted = request.path("max_tokens").asInt() <= 1024;
|
||||
String verdict = "{\"decision\":\"ACCEPT\",\"taskComplete\":true,\"remainingWork\":false,"
|
||||
+ "\"failedCallCount\":0,\"recoveryClaims\":[],\"reason\":\"Forecast matches tool evidence\"}";
|
||||
String content = exhausted ? exhaustedContent : verdict;
|
||||
// SenseNova documents reasoning (rather than reasoning_content) and
|
||||
// an empty finish_reason on intermediate chunks. max_tokens includes reasoning.
|
||||
String events = "data: " + chunk(Map.of("reasoning", "Checking the fifteen daily forecasts"), "")
|
||||
+ "\n\ndata: " + chunk(Map.of("content", content), exhausted ? "length" : "stop")
|
||||
+ "\n\ndata: {\"id\":\"fixture\",\"object\":\"chat.completion.chunk\",\"created\":1,"
|
||||
+ "\"model\":\"sensenova-6.8-flash-lite\",\"choices\":[],\"usage\":{\"prompt_tokens\":200,"
|
||||
+ "\"completion_tokens\":" + (exhausted ? 1024 : 1400)
|
||||
+ ",\"total_tokens\":" + (exhausted ? 1224 : 1600) + "}}\n\ndata: [DONE]\n\n";
|
||||
byte[] body = events.getBytes(StandardCharsets.UTF_8);
|
||||
exchange.getResponseHeaders().set("Content-Type", "text/event-stream; charset=utf-8");
|
||||
exchange.sendResponseHeaders(200, body.length);
|
||||
try (var out = exchange.getResponseBody()) { out.write(body); }
|
||||
});
|
||||
server.start();
|
||||
try {
|
||||
var service = mock(ModelProviderService.class);
|
||||
var provider = new ModelProviderEntity();
|
||||
provider.setProviderId("sensenova-fixture");
|
||||
provider.setRequireApiKey(false);
|
||||
provider.setBaseUrl("http://127.0.0.1:" + server.getAddress().getPort());
|
||||
when(service.isProviderConfigured(provider.getProviderId())).thenReturn(true);
|
||||
when(service.readProviderGenerateKwargs(provider)).thenReturn(Map.of());
|
||||
var beans = new StaticListableBeanFactory();
|
||||
beans.addBean("observations", ObservationRegistry.NOOP);
|
||||
var builder = new OpenAiCompatibleChatModelBuilder(service,
|
||||
beans.getBeanProvider(RestClient.Builder.class), beans.getBeanProvider(WebClient.Builder.class),
|
||||
beans.getBeanProvider(ObservationRegistry.class));
|
||||
var config = new ModelConfigEntity();
|
||||
config.setModelName("sensenova-6.8-flash-lite");
|
||||
config.setMaxTokens(8192);
|
||||
var model = builder.build(config, provider, RetryTemplate.builder().maxAttempts(1).build());
|
||||
var tracker = mock(ChatStreamTracker.class);
|
||||
var helper = new NodeStreamingChatHelper(tracker);
|
||||
var coordinator = new TaskFinalizationCoordinator(helper, tracker);
|
||||
var ledger = ActionExecutionLedger.fromEvents(List.of(
|
||||
GraphEventPublisher.toolComplete("forecast", "execute_code", "15 days of Shanghai forecast data", true)));
|
||||
var result = coordinator.review(model, "test-672", "上海未来15天天气", "Forecast summary", ledger,
|
||||
"", 16000, 0, 8, 3, true, new RunState());
|
||||
assertEquals(TaskFinalizationGate.Outcome.ACCEPT, result.result().outcome());
|
||||
assertEquals(2, result.calls());
|
||||
assertEquals(2, requests.size());
|
||||
assertEquals(1024, requests.get(0).path("max_tokens").asInt());
|
||||
assertEquals(8192, requests.get(1).path("max_tokens").asInt());
|
||||
assertEquals(requests.get(0).path("messages"), requests.get(1).path("messages"));
|
||||
assertTrue(requests.stream().allMatch(request -> request.path("tools").isEmpty()));
|
||||
assertNull(ThinkingLevelHolder.get());
|
||||
} finally {
|
||||
ThinkingLevelHolder.clear();
|
||||
server.stop(0);
|
||||
}
|
||||
}
|
||||
|
||||
private static String chunk(Map<String, String> delta, String finish) throws java.io.IOException {
|
||||
return JSON.writeValueAsString(Map.of("id", "fixture", "object", "chat.completion.chunk", "created", 1,
|
||||
"model", "sensenova-6.8-flash-lite", "choices", List.of(Map.of("index", 0, "delta", delta,
|
||||
"finish_reason", finish))));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user