wenjin272 commented on code in PR #1363:
URL: https://github.com/apache/flink-agents/pull/1363#discussion_r4522188169
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ChatModelAction.java:
##########
@@ -357,8 +368,24 @@ private static void chat(
int retryWaitIntervalSec =
ChatModelInvoker.configuredRetryWaitSec(ctx);
List<String> triedModels = new ArrayList<>();
+ // Candidates whose answer a quality signal rejected this round:
fallback skips them, since
+ // they already answered the request.
+ Set<String> rejected = new HashSet<>();
+ List<Map<String, Object>> hops = new ArrayList<>();
+ long escalatedTokens = 0L;
+ int escalations = selection.getEscalationsSoFar();
+ Deque<String> pending = new ArrayDeque<>(selection.attemptOrder());
Review Comment:
Could we keep the routing-specific orchestration out of `ChatModelAction`?
This change adds candidate queues, rejected-model tracking, escalation limits,
fallback transitions, and routing metadata handling to an already shared chat
execution path.
My main concern is the growing complexity of `ChatModelAction`. It should
remain focused on the common chat and tool-call lifecycle. The routing layer
should encapsulate model selection, fallback, escalation, and the routing state
carried across tool rounds, returning an accepted response for the action to
process.
Some integration changes in `ChatModelAction` are expected, but adding
routing capabilities should not require it to understand and maintain the
individual routing policies.
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/ModelRouter.java:
##########
@@ -81,18 +84,31 @@ public ModelRouter(ResourceDescriptor descriptor,
ResourceContext resourceContex
// with the candidate message instead of failing the cast here.
Object defaultModel = descriptor.getArgument(DEFAULT_MODEL_KEY);
RoutingCandidateValidator.validate("ModelRouter", names, defaultModel);
+ this.escalation =
+ Escalation.fromArguments("ModelRouter",
descriptor.getInitialArguments())
+ .orElse(null);
+ if (this.escalation != null) {
+ this.escalation.requireTiersAreCandidates("ModelRouter", names);
Review Comment:
Could we separate the user-facing router declaration from the framework
resource implementation?
`ModelRouter` currently serves both as the fluent builder entry point and as
the `Resource` instantiated from a `ResourceDescriptor`. I suggest keeping
`ModelRouter.of(...).build()` and the declaration types in `api`, and moving
the resource implementation into `plan`. Both classes can retain the name
`ModelRouter` in their respective packages, preserving the existing builder
API. The plan-side constructor would continue to accept `ResourceDescriptor`
and `ResourceContext`, following the existing resource convention, with
descriptor-to-resource resolution updated accordingly.
Along with this split, could we consolidate descriptor validation in
`AgentPlan`? Currently, both `AgentPlan` and the `ModelRouter` constructor call
`Escalation.fromArguments(...)` and `requireTiersAreCandidates(...)`. Once the
descriptor has been validated during plan construction, the resource
constructor should only decode the configuration and initialize its fields.
This would require separating parsing from validation in the current helpers
and ensuring every supported construction path passes through plan validation.
It should not require introducing a separate configuration object in place of
`ResourceDescriptor`.
##########
plan/src/main/java/org/apache/flink/agents/plan/routing/EscalationGate.java:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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.plan.routing;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.networknt.schema.Schema;
+import com.networknt.schema.SchemaRegistry;
+import com.networknt.schema.SpecificationVersion;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.ChatResult;
+import org.apache.flink.agents.api.chat.messages.ContentBlock;
+import org.apache.flink.agents.api.chat.messages.DataContentBlock;
+import org.apache.flink.agents.api.chat.messages.MediaBlock;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.chat.messages.TextBlock;
+import org.apache.flink.agents.api.chat.messages.ToolCallBlock;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.chat.model.routing.Escalation;
+import org.apache.flink.agents.api.chat.model.routing.QualityContext;
+import org.apache.flink.agents.api.chat.model.routing.QualitySignal;
+import org.apache.flink.agents.api.chat.model.routing.QualityVerdict;
+import org.apache.flink.agents.api.context.DurableCallable;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.decision.BaseDecisionModelSetup;
+import org.apache.flink.agents.api.decision.DecisionRequest;
+import org.apache.flink.agents.api.decision.DecisionResult;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.tools.Tool;
+import org.apache.flink.agents.plan.actions.ChatModelAction;
+import org.apache.flink.agents.plan.actions.ChatModelInvoker;
+import org.apache.flink.agents.plan.actions.DecisionModelInvoker;
+import org.apache.flink.agents.plan.resource.python.PythonChatModelSetup;
+import org.apache.flink.agents.plan.utils.CancellationUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Runs an {@link Escalation} policy after one model round: the structural
checks first, then the
+ * declared signals in order. The first escalating verdict names the next
tier; the caller discards
+ * the round's answer before any of its tool calls is dispatched and re-sends
the conversation to
+ * that tier.
+ *
+ * <p>Structural checks are pure functions of the response and recomputed on
replay. A
+ * decision-model signal runs through {@link DecisionModelInvoker} under {@code
+ * escalate:<router>:<candidate>}; a custom signal's outcome (verdict or
failure) is persisted under
+ * {@code escalate:<router>:<candidate>:<n>}. Both are reused on replay, so a
recovered request
+ * reaches the same tier without re-judging.
+ */
+public final class EscalationGate {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(EscalationGate.class);
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+ /** Serializes tool arguments with object keys sorted, so equal arguments
render alike. */
+ private static final ObjectMapper CANONICAL =
+ new
ObjectMapper().enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
+
+ private static final SchemaRegistry SCHEMAS =
+
SchemaRegistry.withDefaultDialect(SpecificationVersion.DRAFT_2020_12);
+
+ /** Signal name recorded for the built-in checks. */
+ public static final String STRUCTURAL = "structural";
+
+ /** Question key used for the decision-model signal. */
+ static final String QUESTION_KEY = "quality";
+
+ static final String DEFAULT_INSTRUCTIONS =
+ "Does the ANSWER fully and correctly satisfy the REQUEST, with no
missing steps,"
+ + " refusals, or placeholders? Judge the answer, not the
question.";
+
+ private EscalationGate() {}
+
+ /** The decision to escalate: where to, which signal said so, and why. */
+ public static final class Outcome {
+ public final String fromTier;
+ public final String toTier;
+ public final String signal;
+ public final String reason;
+ @Nullable public final Double score;
+
+ Outcome(
+ String fromTier,
+ String toTier,
+ String signal,
+ String reason,
+ @Nullable Double score) {
+ this.fromTier = fromTier;
+ this.toTier = toTier;
+ this.signal = signal;
+ this.reason = reason;
+ this.score = score;
+ }
+ }
+
+ /**
+ * Decide whether the round produced by {@code candidate} should escalate.
+ *
+ * @return the escalation to perform, or empty to accept the round
+ * @throws Exception a failing signal (resource lookup, provider error
after retries, a custom
+ * signal throwing) fails the request; cancellation propagates
+ */
+ public static Optional<Outcome> check(
+ Escalation policy,
+ UUID requestId,
+ String router,
+ String candidate,
+ int escalationsSoFar,
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ ChatModelInvoker.ChatAttemptResult result,
+ RunnerContext ctx)
+ throws Exception {
+ int tierIndex = policy.tierIndex(candidate);
+ if (tierIndex < 0) {
+ // A fallback candidate outside the tiers, or a model the policy
does not know.
+ return Optional.empty();
+ }
+ Optional<String> next = policy.nextTier(candidate);
+ if (next.isEmpty() || escalationsSoFar >= policy.getMaxEscalations()) {
+ // The last tier always accepts on quality; so does a request out
of hops.
+ return Optional.empty();
+ }
+ ChatResult response = Objects.requireNonNull(result.response,
"response");
+ // A Python-declared candidate binds its tools on the Python side; the
Java setup never
+ // opens them, so the tool checks cannot be applied to it.
+ boolean toolsKnown =
+ result.chatModel != null && !(result.chatModel instanceof
PythonChatModelSetup);
+ List<Tool> tools = toolsKnown ? safeTools(result.chatModel) :
List.of();
+ QualityContext context =
+ new QualityContext(
+ requestId,
+ router,
+ candidate,
+ tierIndex,
+ policy.getTiers(),
+ effectiveMessages(result.chatModel, messages,
promptArgs),
+ promptArgs,
+ response,
+ tools);
+
+ QualityVerdict structural = structural(context, toolsKnown);
+ if (structural.isEscalate()) {
+ return Optional.of(outcome(candidate, next.get(), STRUCTURAL,
structural));
+ }
+ int index = 0;
+ for (Escalation.Signal signal : policy.getSignals()) {
+ QualityVerdict verdict;
+ String name;
+ if
(Escalation.Signal.TYPE_DECISION_MODEL.equals(signal.getType())) {
+ name = Escalation.Signal.TYPE_DECISION_MODEL + ":" +
signal.decisionModel();
+ verdict = decisionSignal(signal, context, router, candidate,
ctx);
Review Comment:
Could we skip the decision-model quality signal for tool-only responses by
default, as discussed in #1328?
Currently, a valid tool call passes the structural checks and then reaches
`decisionSignal(...)` before the tool is executed. The default verifier
question asks whether the answer fully satisfies the request, but a tool-only
response is an intermediate step. A low score therefore discards a legitimate
tool call and escalates the request before the model has had a chance to use
its result.
The structural checks should still run, and custom signals can define their
own behavior. The default decision-model signal should evaluate the answer
produced after the tools return.
Please also add a regression test through `ChatModelAction`: a valid
tool-only response should dispatch the tool without invoking the verifier,
while the subsequent answer should invoke it.
--
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]
