purushah commented on code in PR #1042:
URL: https://github.com/apache/flink-agents/pull/1042#discussion_r3917115984
##########
plan/src/main/java/org/apache/flink/agents/plan/actions/ModelRoutingResolver.java:
##########
@@ -87,20 +111,73 @@ public Class<RoutingDecision> getResultClass() {
@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.
+ // decision: a replayed run reports the original
strategy wall time — and
+ // the strategy is never re-executed on replay.
long start = System.nanoTime();
- RoutingDecision decision =
router.route(routingContext);
+ RoutingDecision decision = executePure(router,
strategy, routingContext);
return decision.withDecisionMs((System.nanoTime() -
start) / 1_000_000.0);
}
};
RoutingDecision decision = ctx.durableExecute(routeCallable);
- Double decisionMs = decision.getDecisionMs();
- FlinkAgentsMetricGroup actionMetrics = ctx.getActionMetricGroup();
- if (actionMetrics != null && decisionMs != null) {
-
actionMetrics.getHistogram("routingDecisionLatencyMs").update(Math.round(decisionMs));
+ recordDecisionLatency(ctx, decision);
+ return normalizeAndFinish(
+ requestId, model, router, decision,
ModelRoutingEvent.SOURCE_STRATEGY, ctx);
+ }
+
+ /** Executes the pure (engine-free) strategy types: built-in rules, or the
user's executor. */
+ private static RoutingDecision executePure(
+ ModelRouter router, RoutingStrategy strategy, RoutingContext
routingContext)
+ throws Exception {
+ switch (strategy.getType()) {
+ case RULE_BASED:
+ return executeRules(router, routingContext);
+ case CUSTOM:
+ return router.getCustomExecutor().route(strategy,
routingContext);
+ default:
+ throw new IllegalStateException(
+ "Unhandled routing strategy type: " +
strategy.getType());
}
+ }
+ /**
+ * Built-in keyword/regex rules: the first candidate whose pattern
(pre-compiled by the router)
+ * matches the most recent user message wins, in declaration order; no
match abstains so the
+ * router uses its default model.
+ */
+ private static RoutingDecision executeRules(ModelRouter router,
RoutingContext context) {
+ String text = context.lastUserMessage();
Review Comment:
Great catch — the restructure dropped exactly the test that pinned
last-vs-first. Will add a multi-turn case that runs the rule evaluator
end-to-end and asserts the match is against the latest user message.
##########
plan/src/main/java/org/apache/flink/agents/plan/AgentPlan.java:
##########
@@ -691,6 +698,156 @@ 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).
+ */
+ private void validateLlmJudgeReferences() {
+ 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;
+ default:
Review Comment:
You're right — a rule key naming a non-candidate is just as static as a
missing judge model, and the descriptor path skips the builder check. Will add
a RULE_BASED arm that validates rule keys against the candidates at plan
construction, with a test for the descriptor-built case.
--
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]
