weiqingy commented on code in PR #1042:
URL: https://github.com/apache/flink-agents/pull/1042#discussion_r3921418687
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/ModelRouter.java:
##########
@@ -84,30 +99,117 @@ public ModelRouter(ResourceDescriptor descriptor,
ResourceContext resourceContex
}
this.fallbackEnabled =
Boolean.TRUE.equals(descriptor.getArgument("fallback",
Boolean.FALSE));
- String strategyClazz = descriptor.getArgument("strategy_clazz");
+ String typeTag = descriptor.getArgument(STRATEGY_TYPE_KEY);
+ if (typeTag == null || typeTag.isEmpty()) {
+ throw new IllegalArgumentException("ModelRouter requires a routing
strategy.");
+ }
Map<String, Object> strategyArgs =
- descriptor.getArgument("strategy_args",
Collections.emptyMap());
- this.strategy = instantiateStrategy(strategyClazz, strategyArgs);
+ descriptor.getArgument(STRATEGY_ARGS_KEY,
Collections.emptyMap());
+ String executorClass =
descriptor.getArgument(STRATEGY_EXECUTOR_CLASS_KEY);
+ // The declaration constructor owns the per-type argument rules, so a
structurally invalid
+ // configuration fails here (resource construction) with the same
message as at build().
+ this.strategy =
+ new RoutingStrategy(
+ RoutingStrategyType.fromTag(typeTag), strategyArgs,
executorClass);
+ this.compiledRules = compileRules(this.strategy);
+ this.customExecutor = instantiateCustomExecutor(this.strategy);
}
- @SuppressWarnings("unchecked")
- private static RoutingStrategy instantiateStrategy(String clazz,
Map<String, Object> args)
+ /**
+ * Instantiates the user's {@link CustomRoutingExecutor} once per router
instance. Routers are
+ * cached per subtask — at parallelism N that is N router (and executor)
instances, so executor
+ * instance state spans the requests of one subtask, not the whole
TaskManager. The construction
+ * contract is a {@code (Map<String,Object>)} constructor fed the
declaration's arguments, then
+ * a no-arg constructor, via the thread context classloader — plan-time
validation checks the
+ * same contract without instantiating.
+ */
+ private static CustomRoutingExecutor
instantiateCustomExecutor(RoutingStrategy strategy)
throws Exception {
- if (clazz == null || clazz.isEmpty()) {
- throw new IllegalArgumentException("ModelRouter requires a routing
strategy.");
+ if (strategy.getType() != RoutingStrategyType.CUSTOM) {
+ return null;
+ }
+ Class<?> clazz =
+ Class.forName(
+ strategy.getExecutorClass(),
+ true,
+ Thread.currentThread().getContextClassLoader());
+ if (!CustomRoutingExecutor.class.isAssignableFrom(clazz)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Custom routing executor '%s' does not implement
%s.",
+ strategy.getExecutorClass(),
CustomRoutingExecutor.class.getName()));
}
- Class<?> c = Class.forName(clazz, true,
Thread.currentThread().getContextClassLoader());
try {
- Constructor<?> ctor = c.getConstructor(Map.class);
- return (RoutingStrategy) ctor.newInstance(args);
+ return (CustomRoutingExecutor)
+
clazz.getConstructor(Map.class).newInstance(strategy.getArguments());
} catch (NoSuchMethodException noMapCtor) {
- return (RoutingStrategy) c.getConstructor().newInstance();
+ return (CustomRoutingExecutor)
clazz.getConstructor().newInstance();
}
}
- /** Run the strategy for the given context. */
- public RoutingDecision route(RoutingContext context) throws Exception {
- return strategy.route(context);
+ /** The user's custom executor instance ({@code null} unless the strategy
type is CUSTOM). */
+ public CustomRoutingExecutor getCustomExecutor() {
+ return customExecutor;
+ }
+
+ /**
+ * The single validation/compilation path for rule maps: null/empty keys,
non-String values and
+ * invalid regex all fail here with the same diagnostics everywhere it is
called — the builder
+ * ({@code build()}), the router constructor, and plan-time validation
({@code
+ * AgentPlan#validateRuleKeys}). Called once per router instance (routers
are cached per
+ * subtask), so rule evaluation stays regex-match-only per request.
Patterns were validated at
+ * build(); this re-validates defensively for descriptors constructed
outside the builder.
Review Comment:
nit: does the last sentence still fit? The first one says `compileRules` is
the single validation path, and `build()` does validate by calling it (`:319`).
So "Patterns were validated at build(); this re-validates" seems to argue with
it.
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ModelRoutingResolver.java:
##########
@@ -111,12 +188,312 @@ public RoutingDecision call() throws Exception {
if (!router.isCandidate(selectedModel)) {
throw new IllegalStateException(
String.format(
- "Routing strategy for router '%s' returned
non-candidate model '%s'; candidates are %s.",
+ "Routing decision for router '%s' selected
non-candidate model '%s'; candidates are %s.",
model, selectedModel,
router.getCandidateNames()));
}
- decisionSource = ModelRoutingEvent.SOURCE_STRATEGY;
+ decisionSource = concreteSource;
}
+ return finish(requestId, model, router, decision, selectedModel,
decisionSource, ctx);
+ }
+
+ /** Records the decision latency histogram sample (also for decisions the
guards reject). */
+ private static void recordDecisionLatency(RunnerContext ctx,
RoutingDecision decision) {
+ Double decisionMs = decision.getDecisionMs();
+ FlinkAgentsMetricGroup actionMetrics = ctx.getActionMetricGroup();
+ if (actionMetrics != null && decisionMs != null) {
+
actionMetrics.getHistogram("routingDecisionLatencyMs").update(Math.round(decisionMs));
+ }
+ }
+ /**
+ * LLM-as-judge path (framework-managed, per discussion #897): the engine
runs the judge chat
+ * itself through the normal durable/metered/observable invoker path —
durable id {@code
+ * "judge:<router>"} so a recovered run replays the original verdict
instead of re-calling the
+ * judge (with a durable action-state store configured; without one the
judge re-runs on replay,
+ * like any non-deterministic strategy) — then derives the decision from
the verdict as a pure
+ * function ({@link LlmJudgeRoutingExecutor}). The decision (including its
wall time, which
+ * covers the judge call) is persisted under the standard {@code
"route:<router>"} durable call,
+ * preserving the replay-fingerprint property. Verdict abstains are
persisted <i>as
+ * abstains</i>, so a replay after a candidate-set change re-resolves to
the current default
+ * exactly like the strategy path.
+ *
+ * <p>Failure policy: an unparseable or non-candidate verdict always
abstains to the router's
+ * default model. A judge call that exhausts its retries honors the
request's error-handling
+ * strategy, exactly like a throwing rule/custom strategy: {@code FAIL}
surfaces the outage
+ * loudly, {@code IGNORE} degrades to the default with the cause recorded.
Cancellation
+ * propagates and is never persisted as a routing outcome.
+ */
+ private static ResolvedModelRoute resolveViaJudge(
+ UUID requestId,
+ String model,
+ ModelRouter router,
+ RoutingStrategy strategy,
+ RoutingContext routingContext,
+ RunnerContext ctx)
+ throws Exception {
+ long start = System.nanoTime();
+ Agent.ErrorHandlingStrategy errorStrategy =
+
ctx.getConfig().get(AgentExecutionOptions.ERROR_HANDLING_STRATEGY);
+ int numRetries = ChatModelInvoker.configuredRetries(ctx,
errorStrategy);
+ int retryWaitIntervalSec =
ChatModelInvoker.configuredRetryWaitSec(ctx, errorStrategy);
+ String judgeModel = LlmJudgeRoutingExecutor.judgeModel(strategy);
+
+ Map<String, Object> judgeMetadata = new LinkedHashMap<>();
+ judgeMetadata.put("judge_model", judgeModel);
+ String verdictModel = null;
+ String abstainReason = null;
+ // Runtime backstop to the plan-time check: plan validation only sees
descriptor-carried
+ // bindings, so a setup that binds a prompt/tools/skills at the
instance level (or via a
+ // non-Java provider) is caught here. Same policy as a failed judge
call: FAIL is loud
+ // (a config error should not hide), IGNORE abstains so the default
model keeps answering.
+ String misconfigured = judgeSetupMisconfiguration(judgeModel, ctx);
+ if (misconfigured != null && errorStrategy !=
Agent.ErrorHandlingStrategy.IGNORE) {
+ throw new IllegalStateException(misconfigured);
+ }
+ if (misconfigured != null) {
+ abstainReason = misconfigured;
+ } else
+ try {
+ boolean[] truncated = new boolean[1];
+ List<ChatMessage> effective = effectiveJudgeMessages(router,
routingContext, ctx);
+ List<ChatMessage> judgeInput =
+ LlmJudgeRoutingExecutor.buildJudgeMessages(
+ strategy,
+ routingContext,
+ effective,
+
pinnedRenderedIndices(routingContext.getMessages(), effective),
+ truncated);
+ if (truncated[0]) {
+
judgeMetadata.put(LlmJudgeRoutingExecutor.CONTEXT_TRUNCATED_KEY, true);
+ }
+ ChatModelInvoker.ChatAttemptResult judgeResult =
+ ChatModelInvoker.chatWithRetries(
+ requestId,
+ judgeModel,
+ "judge:" + model,
+ judgeInput,
+ Map.of(),
+ null,
+ ctx,
+ errorStrategy,
+ numRetries,
+ retryWaitIntervalSec);
+ ChatModelAction.recordAttemptRetryStats(
+ ctx,
+ requestId,
+ judgeResult.chatModel,
+ judgeResult.retryCount,
+ judgeResult.totalRetryWaitSec);
+ ChatMessage reply = judgeResult.response;
+ // Same both-or-neither guard as the metrics reader of these
extraArgs keys
+ // (ChatModelAction#recordChatTokenMetrics): a half-populated
or non-Number pair
+ // must not leak into the durable decision metadata, so
metrics and routing
+ // metadata always agree about the same judge call.
Review Comment:
Does the last clause hold? `recordChatTokenMetrics` adds three conditions
this site does not: null metric group (`ChatModelAction.java:222`), non-empty
`model_name` (`:229-230`), both counts `> 0` (`:235`). A judge reply with
`promptTokens: 0` records no metric, but both keys still land in the metadata.
The `instanceof Number` half reads right. Worth trimming just the "always
agree" sentence?
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -691,6 +698,193 @@ private void checkNoRouterModelNameClash(ResourceProvider
provider) {
}
}
+ /**
+ * Static routing-strategy constraints fail at plan construction — never
per record. The
+ * strategy travels as a language-neutral type tag plus arguments, so
validation reads
+ * declaration data directly: no reflective instantiation, whose failure
modes previously let a
+ * misconfigured strategy skip validation entirely.
+ *
+ * <p>{@code LLM_JUDGE}: the judge chat model must be registered (a typo'd
name would otherwise
+ * fail-and-abstain on every request, silently disabling routing — cf.
{@link
+ * #checkNoRouterModelNameClash}), and its descriptor must bind no prompt,
tools, or skills — a
+ * bound prompt would prepend an (unfilled) task prompt ahead of the
verdict contract, bound
+ * tools divert the reply into tool calls, and skills inject both a
discovery prompt and tools,
+ * each silently breaking verdict parsing on every request.
+ *
+ * <p>{@code CUSTOM}: the executor class must exist, implement {@link
CustomRoutingExecutor},
+ * and expose a supported constructor — checked without instantiation, so
plan construction
+ * never runs user constructors (or their static initializers).
+ *
+ * <p>{@code RULE_BASED}: the rule map must compile ({@link
ModelRouter#compileRules} — the
+ * shared path, so diagnostics match the builder's) and every rule key
must name a declared
+ * candidate.
+ */
+ private void validateRoutingStrategies() {
+ if (resourceProviders == null) {
+ return;
+ }
+ Map<String, ResourceProvider> routers =
resourceProviders.get(ResourceType.MODEL_ROUTER);
+ if (routers == null) {
+ return;
+ }
+ Map<String, ResourceProvider> chatModels =
+ resourceProviders.getOrDefault(ResourceType.CHAT_MODEL,
Collections.emptyMap());
+ for (ResourceProvider provider : routers.values()) {
+ if (!(provider instanceof JavaResourceProvider)) {
+ continue;
+ }
+ ResourceDescriptor descriptor = ((JavaResourceProvider)
provider).getDescriptor();
+ if (descriptor == null || descriptor.getInitialArguments() ==
null) {
+ continue;
+ }
+ String typeTag =
descriptor.getArgument(ModelRouter.STRATEGY_TYPE_KEY);
+ if (typeTag == null) {
+ // Fail here, not per record on the TaskManager: ModelRouter's
constructor
+ // unconditionally rejects a descriptor without a strategy,
and a throwing
+ // construction is never cached, so it would re-throw on every
routed request.
+ throw new IllegalArgumentException(
+ String.format(
+ "Model router '%s' declares no routing
strategy ('%s' missing"
+ + " from its descriptor).",
+ provider.getName(),
ModelRouter.STRATEGY_TYPE_KEY));
+ }
+ // The declaration constructor re-validates the per-type argument
rules, so a
+ // structurally invalid configuration (e.g. a judge without a
judge model) fails
+ // plan construction with the same message as build().
+ RoutingStrategy strategy =
+ new RoutingStrategy(
+ RoutingStrategyType.fromTag(typeTag),
+ descriptor.getArgument(
+ ModelRouter.STRATEGY_ARGS_KEY,
Collections.emptyMap()),
+
descriptor.getArgument(ModelRouter.STRATEGY_EXECUTOR_CLASS_KEY));
+ switch (strategy.getType()) {
+ case LLM_JUDGE:
+ validateJudge(provider.getName(), strategy, chatModels);
+ break;
+ case CUSTOM:
+ validateCustomExecutor(provider.getName(), strategy);
+ break;
+ case RULE_BASED:
+ validateRuleKeys(
+ provider.getName(),
+ strategy,
+
descriptor.getArgument(ModelRouter.CANDIDATES_KEY));
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ /**
+ * Rule declarations are static constraints like the judge checks above:
the fluent builder
+ * rejects a bad one at build(), but a descriptor read back from a plan
(deserialized or
+ * hand-built) never went through the builder. Without this arm they would
surface only per
+ * record at request time — inside the durable call — where the IGNORE
error policy silently
+ * drops every matching record. {@link ModelRouter#compileRules} is the
single
+ * validation/compilation path (invalid patterns, non-String values, empty
keys), so those
+ * diagnostics are identical to the builder's; the key-vs-candidate check
mirrors build().
+ */
+ private static void validateRuleKeys(
+ String routerName, RoutingStrategy strategy, Object candidates) {
+ if (!(candidates instanceof List)) {
Review Comment:
This return also skips `compileRules`, and those pattern checks (empty key,
non-String value, bad regex at `ModelRouter.java:173-196`) never look at
`candidates`. The javadoc you added at `:718-720` promises them for
`RULE_BASED`.
What made me look twice is `:742-751` just above: a missing `strategy_type`
throws there, on the same "the constructor rejects it anyway" reasoning.
Reachability is low either way. Would hoisting the compile above the guard be
worth it?
Something like:
```java
var ruleKeys = ModelRouter.compileRules(strategy).keySet();
if (!(candidates instanceof List)) {
return;
}
for (String ruleKey : ruleKeys) { ... }
```
--
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]
