weiqingy commented on code in PR #964:
URL: https://github.com/apache/flink-agents/pull/964#discussion_r3726267216
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -400,56 +669,199 @@ public ChatMessage call() throws Exception {
Thread.sleep(currentWaitSec * 1000L);
totalWaitTimeSec += currentWaitSec;
}
- } else {
- LOG.debug(
- "Chat request {} failed, the input chat messages
are {}.",
- initialRequestId,
- messages);
- throw e;
+ continue;
}
+ throw new ChatAttemptFailed(
+ model, chatModel, e, actualRetryCount,
totalWaitTimeSec);
}
}
+ throw new IllegalStateException("Unreachable chat retry state.");
+ }
- if (actualRetryCount > 0) {
- accumulateRetryStats(
- ctx.getSensoryMemory(), initialRequestId,
actualRetryCount, totalWaitTimeSec);
+ private static void recordAttemptRetryStats(
+ RunnerContext ctx,
+ UUID initialRequestId,
+ BaseChatModelSetup chatModel,
+ int retryCount,
+ int retryWaitSec)
+ throws Exception {
+ if (retryCount <= 0) {
+ return;
}
+ accumulateRetryStats(ctx.getSensoryMemory(), initialRequestId,
retryCount, retryWaitSec);
+ String metricModel = chatModel.getConnectionName();
+ recordRetryMetrics(
+ ctx,
+ metricModel == null || metricModel.isEmpty() ? "unknown" :
metricModel,
+ retryCount,
+ retryWaitSec);
+ }
- if (!Objects.requireNonNull(response).getToolCalls().isEmpty()) {
- handleToolCalls(
- response,
- initialRequestId,
- model,
- chatModel,
- messages,
- promptArgs,
- outputSchema,
- ctx);
- } else {
- Map<String, Long> retryStats =
getRetryStats(ctx.getSensoryMemory(), initialRequestId);
- int totalRetryCount = retryStats.get(TOTAL_RETRY_COUNT).intValue();
- int totalRetryWaitSec =
retryStats.get(TOTAL_RETRY_WAIT_SEC).intValue();
+ private static List<String> candidateAttemptOrder(RoutingSelection
selection) {
+ List<String> order = new ArrayList<>();
+ order.add(selection.selectedModel);
+ if (selection.isRouter && selection.fallbackEnabled) {
+ for (String candidate : selection.candidates) {
+ if (!candidate.equals(selection.selectedModel)) {
+ order.add(candidate);
+ }
+ }
+ }
+ return order;
+ }
- recordRetryMetrics(
- ctx, chatModel.getConnectionName(), totalRetryCount,
totalRetryWaitSec);
+ private static String durableChatCallId(RoutingSelection selection, String
candidate) {
+ if (!selection.isRouter) {
+ return "chat";
+ }
+ return "chat:" + selection.requestedModel + ":" + candidate;
+ }
- ctx.sendEvent(
- new ChatResponseEvent(
- initialRequestId, response, totalRetryCount,
totalRetryWaitSec));
+ private static void attachRoutingMetadata(
+ ChatMessage response,
+ RoutingSelection selection,
+ String finalModel,
+ List<String> triedModels) {
+ boolean fallbackAttempted =
!finalModel.equals(selection.selectedModel);
+ List<String> fallbackModelsTried = new ArrayList<>();
+ for (int i = 1; i < triedModels.size(); i++) {
+ fallbackModelsTried.add(triedModels.get(i));
+ }
+ Map<String, Object> routing = new LinkedHashMap<>();
+ routing.put("router", selection.requestedModel);
+ routing.put("selected_model", selection.selectedModel);
+ routing.put("initial_selected_model", selection.selectedModel);
+ routing.put("final_model", finalModel);
+ routing.put("candidates", new ArrayList<>(selection.candidates));
+ routing.put(
+ "decision_source",
+ fallbackAttempted ? ModelRoutingEvent.SOURCE_FALLBACK :
selection.decisionSource);
+ routing.put("fallback_enabled", selection.fallbackEnabled);
+ routing.put("fallback_attempted", fallbackAttempted);
+ routing.put("fallback_models_tried", fallbackModelsTried);
+ routing.put("metadata", new LinkedHashMap<>(selection.metadata));
+ if (selection.reason != null) {
+ routing.put("reason", selection.reason);
+ }
+ if (selection.score != null) {
+ routing.put("score", selection.score);
}
+ response.getExtraArgs().put("model_routing", routing);
}
private static void processChatRequest(ChatRequestEvent event,
RunnerContext ctx)
throws Exception {
+ RoutingSelection selection =
+ resolveRouter(
+ event.getId(),
+ event.getModel(),
+ event.getMessages(),
+ event.getPromptArgs(),
+ ctx);
chat(
event.getId(),
- event.getModel(),
+ selection,
event.getMessages(),
event.getPromptArgs(),
event.getOutputSchema(),
ctx);
}
+ /**
+ * If {@code model} names a {@link ModelRouter}, run its strategy (as a
durable {@code "route"}
+ * call so the decision replays deterministically on recovery), normalize
the result (abstain ->
+ * default model, non-candidate -> fail clearly), emit an
observability-only {@link
+ * ModelRoutingEvent}, and return the selected concrete model. Otherwise
returns a direct
+ * selection.
+ *
+ * <p>Routing runs once for the initial chat request; tool-call rounds
reuse the selected
+ * concrete model because it is saved in the tool-request context (see
{@link
+ * #handleToolCalls}), so this method is only reached with a router name
on the initial request.
+ */
+ private static RoutingSelection resolveRouter(
+ UUID requestId,
+ String model,
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ RunnerContext ctx)
+ throws Exception {
+ if (!ctx.hasResource(model, ResourceType.MODEL_ROUTER)) {
+ return RoutingSelection.direct(model);
+ }
+ ModelRouter router = (ModelRouter) ctx.getResource(model,
ResourceType.MODEL_ROUTER);
+ RoutingContext routingContext =
+ new RoutingContext(requestId, model, messages, promptArgs,
router.getCandidates());
+
+ DurableCallable<RoutingDecision> routeCallable =
+ new DurableCallable<>() {
+ @Override
+ public String getId() {
+ return "route:" + requestId + ":" + model;
+ }
+
+ @Override
+ public Class<RoutingDecision> getResultClass() {
+ return RoutingDecision.class;
+ }
+
+ @Override
+ public RoutingDecision call() throws Exception {
+ // Timed inside the durable call so the latency is
persisted with the
+ // decision: a replayed run reports the original
strategy wall time.
+ long start = System.nanoTime();
+ RoutingDecision decision =
router.route(routingContext);
+ return decision.withDecisionMs((System.nanoTime() -
start) / 1_000_000.0);
+ }
+ };
+
+ RoutingDecision decision = ctx.durableExecute(routeCallable);
Review Comment:
`resolveRouter` (781-863) has no try/catch and runs before `chat()`, so the
strategy call sits outside the `FAIL`/`RETRY`/`IGNORE` machinery. A job
configured `error-handling-strategy: ignore` still dies if the strategy throws.
That isn't limited to user code. `RuleBasedRoutingStrategy.java:61-66`
throws `IllegalArgumentException` when a matched rule key isn't a candidate,
and nothing validates rule keys at build time: `Strategies.rules`
(`Strategies.java:43-47`) packs the map into descriptor args, and
`ModelRouter.Builder` validates candidate names in `describe()`
(`ModelRouter.java:176-185`) but not in `strategy()`. So
`ModelRouter.of("small","big").strategy(Strategies.rules(Map.of("bg", "...")))`
builds fine and then throws on the first matching request, per record,
regardless of the configured error handling.
Two independent questions. `Builder.build()` already holds both the
candidate list and `strategy_args` (`ModelRouter.java:209`, `:218`), so is
there a reason to leave rule-key validation in the per-request path? And is a
strategy failure meant to be subject to `error-handling-strategy`, say treated
as an abstain so the default model answers, or is fail-fast the intended
contract? Both look defensible to me; right now it's neither declared nor
tested.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -344,15 +505,129 @@ public static void chat(
: 0;
}
- ChatMessage response = null;
+ List<String> triedModels = new ArrayList<>();
+ Exception lastError = null;
+ for (String candidate : candidateAttemptOrder(selection)) {
+ triedModels.add(candidate);
+ try {
+ ChatAttemptResult result =
+ chatWithRetries(
+ initialRequestId,
+ candidate,
+ durableChatCallId(selection, candidate),
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx,
+ strategy,
+ numRetries,
+ retryWaitIntervalSec);
+ recordAttemptRetryStats(
+ ctx,
+ initialRequestId,
+ result.chatModel,
+ result.retryCount,
+ result.totalRetryWaitSec);
+ if (selection.isRouter) {
+ attachRoutingMetadata(result.response, selection,
result.model, triedModels);
+ if (!result.model.equals(selection.selectedModel)) {
+ // The strategy's pick failed and another candidate
answered; record the
+ // outcome in the event log, not just on the response.
+ ctx.sendEvent(
+ new ModelRoutingEvent(
+ initialRequestId,
+ selection.requestedModel,
+ selection.candidates,
+ result.model,
+ ModelRoutingEvent.SOURCE_FALLBACK,
+ selection.fallbackEnabled,
+ String.format(
+ "fallback after selected model
'%s' failed",
+ selection.selectedModel),
+ null,
+ selection.metadata,
+ null));
+ }
+ } else if (selection.carriedRouting != null) {
+ result.response
+ .getExtraArgs()
+ .put("model_routing", new
LinkedHashMap<>(selection.carriedRouting));
+ }
+
+ if
(!Objects.requireNonNull(result.response).getToolCalls().isEmpty()) {
+ handleToolCalls(
+ result.response,
+ initialRequestId,
+ result.model,
+ result.chatModel,
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx);
+ } else {
+ Map<String, Long> retryStats =
+ getRetryStats(ctx.getSensoryMemory(),
initialRequestId);
+ int totalRetryCount =
retryStats.get(TOTAL_RETRY_COUNT).intValue();
+ int totalRetryWaitSec =
retryStats.get(TOTAL_RETRY_WAIT_SEC).intValue();
+
+ ctx.sendEvent(
+ new ChatResponseEvent(
+ initialRequestId,
+ result.response,
+ totalRetryCount,
+ totalRetryWaitSec));
+ }
+ return;
+ } catch (ChatAttemptFailed e) {
+ recordAttemptRetryStats(
+ ctx, initialRequestId, e.chatModel, e.retryCount,
e.totalRetryWaitSec);
+ lastError = e.error;
+ LOG.debug(
+ "Chat request {} failed for model {}, the input chat
messages are {}.",
+ initialRequestId,
+ e.model,
+ messages);
+ }
+ }
+
+ if (strategy == Agent.ErrorHandlingStrategy.IGNORE) {
+ LOG.warn(
+ "Chat request {} failed with error: {}, ignored.",
initialRequestId, lastError);
+ return;
+ }
+ throw Objects.requireNonNull(lastError);
+ }
+
+ private static ChatAttemptResult chatWithRetries(
+ UUID initialRequestId,
+ String model,
+ String durableCallId,
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ @Nullable Object outputSchema,
+ RunnerContext ctx,
+ Agent.ErrorHandlingStrategy strategy,
+ int numRetries,
+ int retryWaitIntervalSec)
+ throws ChatAttemptFailed, Exception {
+ BaseChatModelSetup chatModel =
+ (BaseChatModelSetup) ctx.getResource(model,
ResourceType.CHAT_MODEL);
Review Comment:
The `ctx.getResource(model, ResourceType.CHAT_MODEL)` lookup is outside the
try/catch that produces `ChatAttemptFailed` (644-676), and the call site
catches only `ChatAttemptFailed` (581). So a lookup failure for candidate B
propagates straight out of `chat()`: candidate A's real failure is discarded,
the caller sees "resource not found" instead, and the candidate loop that
exists to keep a failing model from killing the request is bypassed, including
under `IGNORE`. Nothing catches a typo upstream either. `ModelRouter`'s
constructor (57-89) checks non-empty, duplicate and default-is-a-candidate, but
never touches the resource registry.
The signature already hints at the seam: line 612 declares `throws
ChatAttemptFailed, Exception`, where the second clause subsumes the first.
Could the resource and config resolution move inside the same conversion, so a
candidate that can't be resolved is treated as that candidate failing and the
signature collapses to `throws ChatAttemptFailed`?
Is there also room to cross-validate candidate names at plan construction?
The PR makes exactly that locality argument for the clash check at
`Agent.java:135-139`, and `AgentPlan` has the full provider map in hand by the
end of `extractResourceProvidersFromAgent`.
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/RuleBasedRoutingStrategy.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.agents.api.chat.model.routing;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/**
+ * Built-in keyword/regex routing strategy. Configured with a map of {@code
candidateModel ->
+ * regex}; the first candidate whose regex matches the most recent user
message (case-insensitive
+ * find) wins, evaluated in the map's iteration order (pass a {@code
LinkedHashMap} when precedence
+ * matters). If nothing matches, the strategy abstains so the router uses its
default model.
+ *
+ * <p>Constructed reflectively from a {@link RoutingStrategyDescriptor} via
the {@code
+ * (Map<String,Object>)} constructor; use {@link Strategies#rules(Map)} to
build one.
+ */
+public class RuleBasedRoutingStrategy implements RoutingStrategy {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Map<String, Pattern> rules;
+
+ @SuppressWarnings("unchecked")
+ public RuleBasedRoutingStrategy(Map<String, Object> args) {
+ this.rules = new LinkedHashMap<>();
+ Object raw = args == null ? null : args.get("rules");
+ if (raw instanceof Map) {
+ for (Map.Entry<String, ?> entry : ((Map<String, ?>)
raw).entrySet()) {
+ String candidate = entry.getKey();
+ String regex = String.valueOf(entry.getValue());
Review Comment:
`String.valueOf(entry.getValue())` returns the three-character string
`"null"` for a null value, never `null`, so the `regex != null` guard on the
next line is dead. A rules entry with a null value compiles to
`Pattern.compile("null", CASE_INSENSITIVE)`, and any user message containing
"null" then routes to that candidate. `String.valueOf` also coerces non-String
values silently (`42` becomes the pattern `42`), which matters because `rules`
arrives as `Map<String,Object>` off the plan JSON. Rejecting a null or
non-String value would be clearer than producing a pattern from it.
##########
api/src/main/java/org/apache/flink/agents/api/EventType.java:
##########
@@ -39,6 +39,8 @@ public final class EventType {
org.apache.flink.agents.api.event.ContextRetrievalRequestEvent.EVENT_TYPE;
public static final String ContextRetrievalResponseEvent =
org.apache.flink.agents.api.event.ContextRetrievalResponseEvent.EVENT_TYPE;
+ public static final String ModelRoutingEvent =
Review Comment:
After a rebase onto current `main`, `ModelRoutingEvent` will be missing from
`allConstants()`. The branch predates that change, so this PR isn't removing
anything, but `main` now carries a hand-maintained map at
`EventType.java:45-54`, and `ConditionExpressionValidator.java:125` rejects any
`EventType.X` in a trigger condition that isn't a key of it. So `@Action("type
== EventType.ModelRoutingEvent && ...")` compiles and then fails plan
validation with "Unknown EventType constant". Plain
`@Action(EventType.ModelRoutingEvent)` is unaffected, since only CEL trigger
conditions go through the validator. One line in `ALL_CONSTANTS` covers it.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -344,15 +505,129 @@ public static void chat(
: 0;
}
- ChatMessage response = null;
+ List<String> triedModels = new ArrayList<>();
+ Exception lastError = null;
+ for (String candidate : candidateAttemptOrder(selection)) {
+ triedModels.add(candidate);
+ try {
+ ChatAttemptResult result =
+ chatWithRetries(
+ initialRequestId,
+ candidate,
+ durableChatCallId(selection, candidate),
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx,
+ strategy,
+ numRetries,
+ retryWaitIntervalSec);
+ recordAttemptRetryStats(
+ ctx,
+ initialRequestId,
+ result.chatModel,
+ result.retryCount,
+ result.totalRetryWaitSec);
+ if (selection.isRouter) {
+ attachRoutingMetadata(result.response, selection,
result.model, triedModels);
+ if (!result.model.equals(selection.selectedModel)) {
+ // The strategy's pick failed and another candidate
answered; record the
+ // outcome in the event log, not just on the response.
+ ctx.sendEvent(
+ new ModelRoutingEvent(
+ initialRequestId,
+ selection.requestedModel,
+ selection.candidates,
+ result.model,
+ ModelRoutingEvent.SOURCE_FALLBACK,
+ selection.fallbackEnabled,
+ String.format(
+ "fallback after selected model
'%s' failed",
+ selection.selectedModel),
+ null,
+ selection.metadata,
+ null));
+ }
+ } else if (selection.carriedRouting != null) {
+ result.response
+ .getExtraArgs()
+ .put("model_routing", new
LinkedHashMap<>(selection.carriedRouting));
+ }
+
+ if
(!Objects.requireNonNull(result.response).getToolCalls().isEmpty()) {
+ handleToolCalls(
+ result.response,
+ initialRequestId,
+ result.model,
+ result.chatModel,
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx);
+ } else {
+ Map<String, Long> retryStats =
+ getRetryStats(ctx.getSensoryMemory(),
initialRequestId);
+ int totalRetryCount =
retryStats.get(TOTAL_RETRY_COUNT).intValue();
+ int totalRetryWaitSec =
retryStats.get(TOTAL_RETRY_WAIT_SEC).intValue();
+
+ ctx.sendEvent(
+ new ChatResponseEvent(
+ initialRequestId,
+ result.response,
+ totalRetryCount,
+ totalRetryWaitSec));
+ }
+ return;
+ } catch (ChatAttemptFailed e) {
+ recordAttemptRetryStats(
Review Comment:
Retry metric recording moved. `origin/main:433-434` called
`recordRetryMetrics` once on the final-response branch with cumulative totals;
this records per attempt inside `recordAttemptRetryStats` (681-698), which also
runs on the failure path. The total over a completed request is unchanged, but
retries on requests that ultimately throw now increment where they previously
didn't, which is visible to anyone with a dashboard on it. `AGENTS.md` asks PRs
to describe any compatibility impact. Worth a line in the description?
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -51,14 +55,43 @@
import static org.apache.flink.agents.api.agents.Agent.STRUCTURED_OUTPUT;
import static org.apache.flink.agents.plan.actions.Utils.supportAsync;
-/** Built-in action for processing chat request and tool call result. */
+/**
+ * Built-in action for processing chat request and tool call result.
+ *
+ * <h2>Model routing overview</h2>
+ *
+ * <p>When a {@link ChatRequestEvent} names a {@code MODEL_ROUTER} instead of
a chat model, this
+ * action layers five jobs on top of the normal chat path; each is localized
to one place:
+ *
+ * <ol>
+ * <li><b>Decide</b> — {@code resolveRouter} runs the router's strategy and
normalizes the result
+ * (abstain → default model; non-candidate → fail).
+ * <li><b>Durably</b> — the strategy runs inside a durable call ({@code
"route:<requestId>:
Review Comment:
The durability guarantee is stated unconditionally here (69-71) and again on
`RoutingDecision.java:93-95`, but the journal is off in the default
configuration: `actionStateStoreBackend` defaults to null
(`AgentConfigOptions.java:65-66`), `setupDurableExecutionContext` returns
immediately when the store is null (`DurableExecutionManager.java:289-292`),
and `matchNextOrClearSubsequentCallResult` / `recordCallCompletion` then no-op
(`RunnerContextImpl.java:389-396`, `:408-415`).
The default path is still self-consistent: with no journal the decision is
re-derived and the chat call re-runs, so a stale pairing can't occur. It's the
javadoc that overpromises for a reader deploying with defaults.
Suggested wording, in case it helps: recovery replays the persisted decision
when an action state store backend is configured, and without one the strategy
re-runs on recovery.
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/RoutingContext.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.agents.api.chat.model.routing;
+
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Read-only view a {@link RoutingStrategy} sees when deciding which model to
route to.
Review Comment:
The collections are genuinely unmodifiable (55-58, 59-62), but the copy is
shallow and `ChatMessage` is mutable on main (`setContent`, `setToolCalls`).
The elements are the same instances that are then sent to the selected model:
`ChatModelAction.java:752-767` passes `event.getMessages()` into
`resolveRouter` at 758 and into `chat(...)` at 764, and
`ChatRequestEvent.getMessages()` returns the live attribute list rather than a
copy. So a `setContent(...)` inside a strategy silently rewrites the prompt
that reaches the model, with no event and no trace. `getPromptArgs()` has the
same shape, since `new HashMap<>(promptArgs)` is shallow.
The strategy's reach to other resources is already closed by the type. This
is the same class of thing one level down, and narrowing it is free before v1
ships and breaking after.
The context already exposes `firstUserMessage()` and `lastUserMessage()`
(100, 114), both returning `String`, which is what a selection strategy
actually needs. Was deep-copying the message list into the context considered,
or dropping `getMessages()` until a strategy needs more than the text?
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -344,15 +505,129 @@ public static void chat(
: 0;
}
- ChatMessage response = null;
+ List<String> triedModels = new ArrayList<>();
+ Exception lastError = null;
+ for (String candidate : candidateAttemptOrder(selection)) {
+ triedModels.add(candidate);
+ try {
+ ChatAttemptResult result =
+ chatWithRetries(
+ initialRequestId,
+ candidate,
+ durableChatCallId(selection, candidate),
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx,
+ strategy,
+ numRetries,
+ retryWaitIntervalSec);
+ recordAttemptRetryStats(
+ ctx,
+ initialRequestId,
+ result.chatModel,
+ result.retryCount,
+ result.totalRetryWaitSec);
+ if (selection.isRouter) {
+ attachRoutingMetadata(result.response, selection,
result.model, triedModels);
+ if (!result.model.equals(selection.selectedModel)) {
+ // The strategy's pick failed and another candidate
answered; record the
+ // outcome in the event log, not just on the response.
+ ctx.sendEvent(
+ new ModelRoutingEvent(
+ initialRequestId,
+ selection.requestedModel,
+ selection.candidates,
+ result.model,
+ ModelRoutingEvent.SOURCE_FALLBACK,
+ selection.fallbackEnabled,
+ String.format(
+ "fallback after selected model
'%s' failed",
+ selection.selectedModel),
+ null,
+ selection.metadata,
+ null));
+ }
+ } else if (selection.carriedRouting != null) {
+ result.response
+ .getExtraArgs()
+ .put("model_routing", new
LinkedHashMap<>(selection.carriedRouting));
+ }
+
+ if
(!Objects.requireNonNull(result.response).getToolCalls().isEmpty()) {
+ handleToolCalls(
+ result.response,
+ initialRequestId,
+ result.model,
+ result.chatModel,
+ messages,
+ promptArgs,
+ outputSchema,
+ ctx);
+ } else {
+ Map<String, Long> retryStats =
+ getRetryStats(ctx.getSensoryMemory(),
initialRequestId);
+ int totalRetryCount =
retryStats.get(TOTAL_RETRY_COUNT).intValue();
+ int totalRetryWaitSec =
retryStats.get(TOTAL_RETRY_WAIT_SEC).intValue();
+
+ ctx.sendEvent(
+ new ChatResponseEvent(
+ initialRequestId,
+ result.response,
+ totalRetryCount,
+ totalRetryWaitSec));
+ }
+ return;
+ } catch (ChatAttemptFailed e) {
+ recordAttemptRetryStats(
+ ctx, initialRequestId, e.chatModel, e.retryCount,
e.totalRetryWaitSec);
+ lastError = e.error;
Review Comment:
`lastError = e.error` is a plain overwrite and line 598 throws only that, so
with candidates `[A, B, C]` all failing the operator sees C's exception. A's
and B's are dropped entirely: the debug line at 585 logs the model and the
messages, not the error. So the routed model's own failure is the one
discarded, and when the failures differ in kind (A: 401 auth, C: connection
refused) the surviving message actively misleads.
The event log doesn't compensate: the second `ModelRoutingEvent` only fires
in the success branch (536-549), so on exhaustion nothing records which
candidates were tried or how they failed.
Something like this before the overwrite, if it helps:
```java
if (lastError != null && lastError != e.error) {
e.error.addSuppressed(lastError);
}
```
Worth knowing `fallbackExhaustedRethrows` can't catch this today: it asserts
`hasMessageContaining("down")` and both scripted errors are `"big is down"` /
`"small is down"` (`ChatModelActionRoutingTest.java:408-410`), so it passes
whichever surfaces. Distinct messages would turn it into a real assertion about
which error the caller gets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
