You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "purushah (via GitHub)" <gi...@apache.org> on 2026/06/15 20:37:16 UTC
[PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
purushah opened a new pull request, #852:
URL: https://github.com/apache/flink-agents/pull/852
## What is the purpose of the change
Adds a **drop-in chat model that routes each request to the best underlying model**, then delegates to it. The router is a `CHAT_MODEL` resource, so an agent points at it by name with **no change to the runtime, events, or agent definition**.
This is the *in-chat* selector (which LLM serves a single `chat()` call). A DataStream-level content-based agent router (branching records across agent operators) is a separate, follow-up concern.
## Brief change log
- **`RoutingStrategy`** — pluggable selection SPI (`request -> candidate name`). Selection is a *pure* concern; returning `null` means "abstain / no opinion".
- **`ChatModelRouter`** — orchestrates select → (optional cache) → validate → delegate. A strategy that abstains (`null`) or names a non-candidate is a **routing miss** and degrades to the configured `default` candidate (validated at construction; defaults to the first candidate) rather than failing the request.
- **`FallbackPolicy`** — optional: try remaining candidates on error.
- **`CachingStrategy`** — optional bounded-LRU memoization of the decision per conversation, so an expensive strategy (e.g. an LLM judge) runs **once per conversation**, not once per tool-call round. Abstentions (`null`) are never cached.
- Built-in strategies:
- **`RuleBasedRoutingStrategy`** — deterministic keyword/regex rules + default.
- **`LlmRoutingStrategy`** — a small "judge" model picks the candidate from each candidate's name/description (RouteLLM-style). Distinguishes a **transient judge failure** (abstain → retried next round, uncached) from an **unparseable reply** (deterministic default). Parses by whole-token match (no substring mis-routing, e.g. `gpt-4o-mini` won't match a `gpt-4` candidate).
- **Bring-your-own** strategies are first-class: implement `RoutingStrategy` and reference it by fully-qualified class name; loaded via the thread context classloader (cluster-safe). ML/learned routing is supported the same way.
- Adds `LlmRoutingAgentExample` and unit tests.
## Verifying this change
This change adds tests and can be verified as follows:
- Unit tests under `api/.../chat/model/routing/` covering rule selection, judge parsing (whole-token match), stickiness across tool-call rounds, fallback, caching (incl. abstain-not-cached), routing-miss degrade-to-default, and bring-your-own loading. All pass; `spotless:check` clean (JDK 17).
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): **no**
- The public API: **yes** — adds the `org.apache.flink.agents.api.chat.model.routing` package (additive; no existing API changed).
- The serializers: **no**
- The runtime per-record code paths: **no** (router is a `CHAT_MODEL` resource resolved by name)
- Anything that affects deployment or recovery: **no** — preserves exactly-once / keyed-state / checkpoint semantics (no new operator, no nested invocation).
## Security note
An LLM/ML routing decision is a **hint, not an authority** — the user's message is sent to the judge model, so a routing decision is susceptible to prompt injection. Cost/privilege/safety controls must not be gated solely on it. This is documented on `LlmRoutingStrategy`.
## Documentation
- New public package is documented via javadoc on each type. Built-in strategies, the abstain/routing-miss contract, and the bring-your-own extension point are described on the SPI.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494562978
##########
python/flink_agents/api/resource.py:
##########
@@ -356,5 +356,17 @@ class Java:
# Milvus
MILVUS_VECTOR_STORE = "org.apache.flink.agents.integrations.vectorstores.milvus.MilvusVectorStore"
+ class RoutingStrategy:
Review Comment:
Yes — the intended Python story for this PR is intentionally narrow: Python agents can reference the Java router resource by name using the mirrored FQCN constants. A native Python `ChatModelRouter` or Python-side `RoutingStrategy` SPI is out of scope for this PR and can be considered separately if/when Python-native routing becomes needed. I added the constants only as a bridge to the Java resource, not as a full Python routing surface.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494566456
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/CachingStrategy.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * A {@link RoutingStrategy} decorator that memoizes the wrapped strategy's decision per
+ * conversation (keyed on {@link RoutingContext#firstUserMessage()}), so an expensive selection —
+ * e.g. an LLM judge — runs once per conversation rather than on every tool-call round.
+ *
+ * <p>The cache is a <b>bounded LRU</b> with real eviction (oldest entries are dropped past the
+ * capacity), so it never grows without bound and never silently stops caching. Empty keys (requests
+ * with no user message) are not cached, to avoid coupling unrelated empty-prompt conversations. A
+ * {@code null} decision (the strategy abstaining — e.g. a transient LLM-judge failure) is likewise
+ * not cached, so the strategy is re-consulted on the next round rather than pinned to a fallback.
+ * Thread-safe for the async execution pool.
+ */
+public final class CachingStrategy implements RoutingStrategy {
+
+ /** Default cache capacity if none is specified. */
+ public static final int DEFAULT_MAX_ENTRIES = 1024;
+
+ private final RoutingStrategy delegate;
+ private final Map<String, String> cache;
+
+ public CachingStrategy(RoutingStrategy delegate) {
+ this(delegate, DEFAULT_MAX_ENTRIES);
+ }
+
+ public CachingStrategy(RoutingStrategy delegate, int maxEntries) {
+ if (delegate == null) {
+ throw new IllegalArgumentException("delegate strategy must not be null");
+ }
+ if (maxEntries <= 0) {
+ throw new IllegalArgumentException("maxEntries must be positive: " + maxEntries);
+ }
+ this.delegate = delegate;
+ this.cache = Collections.synchronizedMap(new LruMap(maxEntries));
+ }
+
+ @Override
+ public String route(RoutingContext context) throws Exception {
+ String key = context.firstUserMessage();
+ if (key.isEmpty()) {
+ // Don't cache empty keys: every empty-prompt conversation would otherwise share one
+ // decision. Recompute each time instead.
+ return delegate.route(context);
+ }
+ String cached = cache.get(key);
Review Comment:
Updated here as well. `LlmRoutingStrategy` now says the judge “typically” runs once per conversation when wrapped in `CachingStrategy`, and points readers to `CachingStrategy` for the concurrent first-touch caveat.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-4886258770
Hi @purushah ,
Sorry for the late response. I was busy preparing the 0.3 release, Flink Forward Asia, and catching up on a few other works, and only had the chance to carefully look into this PR today.
First of all, thank you again for the contribution! I think this is a very useful and interesting feature, and the implementation quality is also very high. It’s clear that you’ve put a lot of thought and effort into both the design and the implementation.
That said, for a feature of this importance and complexity, I would generally prefer to discuss the overall design first (e.g., through a GitHub Discussion) before moving into a full implementation. After going through the PR, I realized that most of my questions are actually about the design rather than the implementation itself. If we can align on these design questions first, we can hopefully avoid unnecessary rework later.
---
Here are my current thoughts.
### 1. Should ChatModelRouter be modeled as a special kind of ChatModelSetup?
I’m not entirely convinced yet.
The biggest advantage of the current design is exactly what you pointed out: it requires almost no framework changes, and the router can be used just like any other ChatModelSetup, which makes the implementation straightforward.
However, I also see some potential drawbacks.
- The first one is observability. Today, EventLog is our primary observability mechanism. When using a normal ChatModelSetup, every model invocation is represented by ChatRequestEvent and ChatResponseEvent, so the model inputs and outputs are fully captured. With ChatModelRouter, especially when using an LLMRoutingStrategy, the model invocation used to determine which candidate should be selected is no longer represented in the EventLog. As a result, users cannot observe the routing decision process itself.
- Another concern is metrics. Today, many model-related metrics (for example, input/output token usage) are naturally aggregated at the ChatModelSetup level. If a single ChatModelSetup internally invokes multiple different models, those metrics will all be mixed together, making it difficult to understand which model consumed which tokens. This could make cost analysis and cost control more difficult.
One alternative direction (just as an example—I haven’t fully thought through the design yet) would be to treat model routing as a framework capability rather than a specialized ChatModelSetup, for example by introducing a dedicated ChatModelRouter resource type. My intuition is that this might naturally address some of the observability and metrics concerns mentioned above. That said, this is only an initial thought, not a concrete proposal.
### 2. The boundary between Public API and Internal Implementation doesn’t seem very clear yet
This was another impression I had while reading through the PR.
At the moment, most of the newly introduced concepts and implementations live inside the API module, so it’s difficult to immediately distinguish:
* Which concepts are intended to be part of the public API that users should interact with.
* Which ones are merely implementation details used internally by the framework.
Ideally, I think the API module should contain only the minimal set of interfaces that users actually need to interact with. This would help us maintain API compatibility more easily across future releases while still giving us enough flexibility to evolve the internal implementation.
### 3. What is the intended user experience of this feature?
I also think there is still room to discuss the user-facing API design.
In other words, it may be helpful to first step back and think about how users are expected to use this feature end-to-end, and then evaluate whether the API naturally supports that workflow.
For example:
* Is a RulBasedRoutingStrategy actually necessary? I’m not entirely sure that keyword-based routing has enough real-world demand to justify exposing it as part of the public API.
* Should caching be exposed as a general configuration option instead of a separate strategy? The current design, where CachingStrategy delegates to another strategy, looks like a perfectly reasonable internal implementation. However, from a user’s perspective, it might be more natural to simply enable or disable caching through configuration instead of explicitly wrapping one strategy with another.
* Both RoutingStrategy and FallbackPolicy are currently exposed as public interfaces. Do we really expect users to implement these interfaces directly? Or would it be sufficient for most users if we simply provided a set of built-in implementations and let users choose among them through configuration?
I'm not saying that I’m against the current design. In fact, I don’t have firm answers to the above questions either. What I'm trying to say is that, in order to answer these questions, we probably should have a more fundamental design discussion first:
* How do we expect users to use this feature?
* Which decisions should users make explicitly?
* Which decisions can be handled by the framework?
* For the decisions that do need to be exposed, are we exposing them through an API that is as simple and intuitive as possible?
---
Overall, I think this is a valuable feature, and I really appreciate the amount of work you’ve put into this PR.
My suggestion would be to treat this PR as a prototype for now rather than moving directly into detailed code review.
As a next step, I think it would be helpful to first flesh out the overall design—either by opening a GitHub Discussion or by adding a design proposal to this PR—especially around the following questions:
* What should the public API look like?
* What should remain internal implementation details?
* What is the intended user experience of this feature?
Once we have aligned on those aspects, I believe the implementation details will become much easier to converge on, and hopefully we don't need to rework this PR a lot.
WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-4899564216
Thanks for the detailed review, @xintongsong — and no worries on the timing!
I think this is a great idea. I agree it's better to align on the overall design first before going deeper on the implementation. Let's keep this PR open and use it as a prototype/reference for the discussion. I'll follow up here shortly with a short design proposal covering the points you raised. Thanks again for the thoughtful feedback!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494564490
##########
api/src/test/java/org/apache/flink/agents/api/chat/model/routing/RoutingTestSupport.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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 org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** Shared fakes for the routing tests. */
+final class RoutingTestSupport {
+
+ private RoutingTestSupport() {}
+
+ static ResourceDescriptor emptyDescriptor(Class<?> clazz) {
+ return new ResourceDescriptor(clazz.getName(), Collections.emptyMap());
+ }
+
+ /** A chat model that records the messages it received and answers with its own tag. */
+ static final class RecordingModel extends BaseChatModelSetup {
+ final String tag;
+ int callCount = 0;
+ List<ChatMessage> lastMessages;
+
+ RecordingModel(String tag) {
+ super(emptyDescriptor(RecordingModel.class), null);
+ this.tag = tag;
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ this.lastMessages = new ArrayList<>(messages);
+ return new ChatMessage(MessageRole.ASSISTANT, "handled-by:" + tag);
+ }
+ }
+
+ /** A chat model that always fails — used to exercise fallback behavior. */
+ static final class FailingModel extends BaseChatModelSetup {
+ int callCount = 0;
+
+ FailingModel() {
+ super(emptyDescriptor(FailingModel.class), null);
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ throw new RuntimeException("boom from failing model");
+ }
+ }
+
+ /** A chat model that returns a scripted reply — used as an LLM-as-router judge. */
+ static final class ScriptedJudge extends BaseChatModelSetup {
+ final String reply;
+ int callCount = 0;
+ List<ChatMessage> lastMessages;
+
+ ScriptedJudge(String reply) {
+ super(emptyDescriptor(ScriptedJudge.class), null);
+ this.reply = reply;
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ this.lastMessages = new ArrayList<>(messages);
+ return new ChatMessage(MessageRole.ASSISTANT, reply);
+ }
+ }
+
+ /** A {@link ResourceContext} backed by a fixed name → resource map. */
+ static ResourceContext context(Map<String, Resource> byName) {
+ return ResourceContext.fromGetResource(
Review Comment:
Agreed that this was worth pinning in a test. I added `testCandidateOpenedBeforeChat()` using an opening `ResourceContext` that mirrors the runtime `ResourceCache` behavior, plus a negative `testCandidateNotOpenedFails()` case to prove the invariant is load-bearing. I also added `OpenRequiringModel` in test support to behave like a real chat model whose backend connection is initialized in `open()`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494568307
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/ChatModelRouter.java:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A pluggable LLM router that selects, per request, which underlying chat model should serve it.
+ *
+ * <p>The router is a drop-in {@link BaseChatModelSetup}: it reports {@link ResourceType#CHAT_MODEL}
+ * and is resolved by the built-in chat action exactly like any other model, so an agent points at
+ * the router by name and nothing else in the framework needs to change. Concerns are layered:
+ * selection ({@link RoutingStrategy}) decides the model, optional {@link CachingStrategy} memoizes
+ * that decision per conversation, and {@link FallbackPolicy} decides what to try if the chosen
+ * model errors. The router then delegates to the chosen model's own {@code chat(...)}, preserving
+ * its prompt, tools, parameters, and token metrics.
+ *
+ * <h2>Configuration ({@link ResourceDescriptor} arguments)</h2>
+ *
+ * <ul>
+ * <li>{@code candidates} (required) — routable models; each a {@link RoutingCandidate}, a {@link
+ * String} name, or a {@link Map} with {@code name}/{@code description}/{@code metadata}.
+ * Every name must reference a registered {@code CHAT_MODEL} resource.
+ * <li>{@code strategy} (required) — a {@link ResourceDescriptor} naming the {@link
+ * RoutingStrategy} implementation and its args (use {@link
+ * org.apache.flink.agents.api.resource.ResourceName.RoutingStrategy} for built-ins). The
+ * class needs a public {@code (ResourceDescriptor, ResourceContext)} constructor (see {@link
+ * AbstractRoutingStrategy}); it is loaded with the thread context classloader so user classes
+ * resolve on a cluster.
+ * <li>{@code fallback} (optional, default {@code false}) — when {@code true}, a failing model
+ * falls back to the remaining candidates in declaration order.
+ * <li>{@code cache} (optional, default {@code true}) — memoize the selection per conversation.
+ * <li>{@code cache_size} (optional, default {@link CachingStrategy#DEFAULT_MAX_ENTRIES}) — LRU
+ * capacity when caching is enabled.
+ * <li>{@code default} (optional) — the candidate used when the strategy abstains or names a
+ * non-candidate (a routing miss). Must be one of {@code candidates}; defaults to the first
+ * candidate.
+ * </ul>
+ *
+ * <p><b>Graceful degrade:</b> if the strategy returns {@code null} ("no opinion", e.g. a transient
+ * LLM-judge failure) or a name that is not a configured candidate, the router treats it as a
+ * routing miss and serves the request from {@code default} (then the fallback order) rather than
+ * failing.
+ *
+ * <p><b>Bash/skill tool args (v1 scope):</b> the built-in chat action injects bash allowlists and
+ * skill directories from the resource resolved by the agent's model name — i.e. this router — not
+ * the chosen backend. So configure {@code allowed_commands} / {@code allowed_script_dirs} / {@code
+ * skills} <b>on the router</b>; per-candidate skills/allowlists are not supported in v1.
+ *
+ * <p><b>Metrics note (v1):</b> retry metrics recorded by the built-in chat action are grouped under
+ * this router's connection label ({@code "router"}), not the backend model actually used.
+ * Per-backend attribution is a documented follow-up.
+ */
+public class ChatModelRouter extends BaseChatModelSetup {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChatModelRouter.class);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ public static final String ARG_CANDIDATES = "candidates";
+ public static final String ARG_STRATEGY = "strategy";
+ public static final String ARG_FALLBACK = "fallback";
+ public static final String ARG_CACHE = "cache";
+ public static final String ARG_CACHE_SIZE = "cache_size";
+ public static final String ARG_DEFAULT = "default";
+
+ private final List<RoutingCandidate> candidates;
+ private final Set<String> candidateNames;
+ private final RoutingStrategy strategy;
+ private final FallbackPolicy fallbackPolicy;
+ private final boolean fallbackEnabled;
+ private final String defaultCandidate;
+
+ @SuppressWarnings("unchecked")
+ public ChatModelRouter(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+ super(descriptor, resourceContext);
+
+ List<Object> rawCandidates = descriptor.getArgument(ARG_CANDIDATES);
+ if (rawCandidates == null || rawCandidates.isEmpty()) {
+ throw new IllegalArgumentException(
+ "ChatModelRouter requires a non-empty 'candidates' argument.");
+ }
+ List<RoutingCandidate> parsed = new ArrayList<>(rawCandidates.size());
+ Set<String> names = new LinkedHashSet<>();
+ for (Object spec : rawCandidates) {
+ RoutingCandidate candidate = RoutingCandidate.from(spec);
+ parsed.add(candidate);
+ names.add(candidate.getName());
+ }
+ this.candidates = Collections.unmodifiableList(parsed);
+ this.candidateNames = Collections.unmodifiableSet(names);
+
+ ResourceDescriptor strategyDescriptor =
+ toResourceDescriptor(descriptor.getArgument(ARG_STRATEGY));
+ RoutingStrategy base = instantiateStrategy(strategyDescriptor, resourceContext);
+
+ boolean cache = descriptor.getArgument(ARG_CACHE, Boolean.TRUE);
+ if (cache) {
+ int cacheSize =
+ descriptor.getArgument(ARG_CACHE_SIZE, CachingStrategy.DEFAULT_MAX_ENTRIES);
+ this.strategy = new CachingStrategy(base, cacheSize);
+ } else {
+ this.strategy = base;
+ }
+
+ this.fallbackEnabled = descriptor.getArgument(ARG_FALLBACK, Boolean.FALSE);
+ this.fallbackPolicy =
+ fallbackEnabled ? FallbackPolicy.remainingCandidates() : FallbackPolicy.none();
+
+ // Default candidate used on a routing miss (strategy abstains / names a non-candidate).
+ // Validated at construction so a typo is a clear config error, not a per-request failure.
+ String configuredDefault = descriptor.getArgument(ARG_DEFAULT);
+ if (configuredDefault != null && !candidateNames.contains(configuredDefault)) {
+ throw new IllegalArgumentException(
+ "ChatModelRouter 'default' '"
+ + configuredDefault
+ + "' is not one of the configured candidates "
+ + candidateNames);
+ }
+ this.defaultCandidate =
+ configuredDefault != null ? configuredDefault : candidates.get(0).getName();
+ }
+
+ /**
+ * The router has no connection of its own to resolve (it delegates to candidate models, each of
+ * which resolves its own). Override to skip the base class's connection resolution.
+ */
+ @Override
+ public void open() {
Review Comment:
Good point. The stronger “runs once per conversation” wording was overstating the guarantee. I softened this to “typically runs once per conversation” and documented that this is best-effort memoization: a concurrent first-touch on the same key can double-compute, but the synchronized backing map avoids corruption and last-writer-wins is benign, so I kept the implementation lock-free.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-5175343471
Closing this in favor of #964. The design discussion in #897 concluded that the router should select a model rather than call it, with routing owned by `ChatModelAction` — a ground-up redesign of this PR's approach, so a fresh PR with a clean diff made more sense than force-pushing over this one. Thanks @wenjin272 @weiqingy for the reviews here and on #897; they shaped the new design directly.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-4847448527
Thanks for the follow-ups, @purushah. My comments are resolved. I'll leave the final call to the maintainers.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3424600676
##########
api/src/test/java/org/apache/flink/agents/api/chat/model/routing/RoutingTestSupport.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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 org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** Shared fakes for the routing tests. */
+final class RoutingTestSupport {
+
+ private RoutingTestSupport() {}
+
+ static ResourceDescriptor emptyDescriptor(Class<?> clazz) {
+ return new ResourceDescriptor(clazz.getName(), Collections.emptyMap());
+ }
+
+ /** A chat model that records the messages it received and answers with its own tag. */
+ static final class RecordingModel extends BaseChatModelSetup {
+ final String tag;
+ int callCount = 0;
+ List<ChatMessage> lastMessages;
+
+ RecordingModel(String tag) {
+ super(emptyDescriptor(RecordingModel.class), null);
+ this.tag = tag;
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ this.lastMessages = new ArrayList<>(messages);
+ return new ChatMessage(MessageRole.ASSISTANT, "handled-by:" + tag);
+ }
+ }
+
+ /** A chat model that always fails — used to exercise fallback behavior. */
+ static final class FailingModel extends BaseChatModelSetup {
+ int callCount = 0;
+
+ FailingModel() {
+ super(emptyDescriptor(FailingModel.class), null);
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ throw new RuntimeException("boom from failing model");
+ }
+ }
+
+ /** A chat model that returns a scripted reply — used as an LLM-as-router judge. */
+ static final class ScriptedJudge extends BaseChatModelSetup {
+ final String reply;
+ int callCount = 0;
+ List<ChatMessage> lastMessages;
+
+ ScriptedJudge(String reply) {
+ super(emptyDescriptor(ScriptedJudge.class), null);
+ this.reply = reply;
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ Map<String, Object> promptArgs,
+ Map<String, Object> modelParams) {
+ this.callCount++;
+ this.lastMessages = new ArrayList<>(messages);
+ return new ChatMessage(MessageRole.ASSISTANT, reply);
+ }
+ }
+
+ /** A {@link ResourceContext} backed by a fixed name → resource map. */
+ static ResourceContext context(Map<String, Resource> byName) {
+ return ResourceContext.fromGetResource(
Review Comment:
The fakes resolve through `ResourceContext.fromGetResource(...)`, which returns the registered resource directly — unlike the runtime `ResourceCache.getResource()`, which calls `open()` on the resolved resource before handing it back (`ResourceCache.java:125`). The upshot is that the load-bearing production invariant — a routed candidate gets `open()`-ed, so its `connection` is non-null before `chat()` — isn't exercised by any unit test; it rests on the `ResourceCache` path plus the e2e example. That's a structural property of resolving fakes without a real cache, not a defect in this PR. Does the e2e `LlmRoutingAgentExample` exercise the real-candidate (`open()`-then-`chat()`) path in CI, so the invariant is covered somewhere? If it doesn't, would a single router test backed by a real `ResourceCache` and a candidate that actually needs `open()` be worth adding, to pin the one fact the whole drop-in design depends on?
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/RoutingCandidate.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A routable target of a {@link ChatModelRouter}.
+ *
+ * <p>A candidate names a chat-model setup that was registered as a {@link
+ * org.apache.flink.agents.api.resource.ResourceType#CHAT_MODEL} resource, together with an optional
+ * human-readable {@code description} (consumed by LLM-as-router strategies to reason about which
+ * model fits a request) and free-form {@code metadata} (e.g. {@code cost}, {@code tags},
+ * capabilities) that rule-based or custom strategies can match against.
+ */
+public class RoutingCandidate {
+
+ private final String name;
+ private final String description;
+ private final Map<String, Object> metadata;
+
+ public RoutingCandidate(String name, String description, Map<String, Object> metadata) {
+ this.name = Objects.requireNonNull(name, "candidate name must not be null");
Review Comment:
The ctor null-checks `name` but lets an empty string through, and in `LlmRoutingStrategy.parseChoice` the whole-token regex for an empty quoted name over-matches almost any boundary → mis-route. It takes a pathological config (an empty candidate name) to hit, so low priority — a one-line `name.isEmpty()` guard alongside the `requireNonNull` closes it cleanly.
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/CachingStrategy.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * A {@link RoutingStrategy} decorator that memoizes the wrapped strategy's decision per
+ * conversation (keyed on {@link RoutingContext#firstUserMessage()}), so an expensive selection —
+ * e.g. an LLM judge — runs once per conversation rather than on every tool-call round.
+ *
+ * <p>The cache is a <b>bounded LRU</b> with real eviction (oldest entries are dropped past the
+ * capacity), so it never grows without bound and never silently stops caching. Empty keys (requests
+ * with no user message) are not cached, to avoid coupling unrelated empty-prompt conversations. A
+ * {@code null} decision (the strategy abstaining — e.g. a transient LLM-judge failure) is likewise
+ * not cached, so the strategy is re-consulted on the next round rather than pinned to a fallback.
+ * Thread-safe for the async execution pool.
+ */
+public final class CachingStrategy implements RoutingStrategy {
+
+ /** Default cache capacity if none is specified. */
+ public static final int DEFAULT_MAX_ENTRIES = 1024;
+
+ private final RoutingStrategy delegate;
+ private final Map<String, String> cache;
+
+ public CachingStrategy(RoutingStrategy delegate) {
+ this(delegate, DEFAULT_MAX_ENTRIES);
+ }
+
+ public CachingStrategy(RoutingStrategy delegate, int maxEntries) {
+ if (delegate == null) {
+ throw new IllegalArgumentException("delegate strategy must not be null");
+ }
+ if (maxEntries <= 0) {
+ throw new IllegalArgumentException("maxEntries must be positive: " + maxEntries);
+ }
+ this.delegate = delegate;
+ this.cache = Collections.synchronizedMap(new LruMap(maxEntries));
+ }
+
+ @Override
+ public String route(RoutingContext context) throws Exception {
+ String key = context.firstUserMessage();
+ if (key.isEmpty()) {
+ // Don't cache empty keys: every empty-prompt conversation would otherwise share one
+ // decision. Recompute each time instead.
+ return delegate.route(context);
+ }
+ String cached = cache.get(key);
Review Comment:
The check-then-act here is non-atomic: `cache.get(key)` … `delegate.route(context)` … `cache.put(key, chosen)`. Two async-pool threads handling the same conversation key on its first touch can both miss the cache and both invoke `delegate.route()` — i.e. two judge-model calls for the same conversation. The map itself is `synchronizedMap`, so there's no corruption, and the redundant compute is benign (last-writer-wins on the same key), so I wouldn't add locking for it — the cost isn't worth it. My only question is about the wording: the class Javadoc here and the `LlmRoutingStrategy` Javadoc ("the judge runs once per conversation") read as a hard guarantee, where this is really best-effort memoization. Would it be worth softening to "typically once per conversation" (or a one-line note that a concurrent first-touch can double-compute), so the contract matches the behavior? Or is the stronger phrasing intentional?
##########
python/flink_agents/api/resource.py:
##########
@@ -356,5 +356,17 @@ class Java:
# Milvus
MILVUS_VECTOR_STORE = "org.apache.flink.agents.integrations.vectorstores.milvus.MilvusVectorStore"
+ class RoutingStrategy:
Review Comment:
This is the only Python surface the feature gets — the mirrored `RoutingStrategy.Java.{RULE_BASED, LLM}` FQCN constants, with no Python `ChatModelRouter`, no Python `RoutingStrategy` SPI, and no Python tests. CLAUDE.md asks for Java/Python parity, though that's scoped to "when changing shared logic," and this PR adds a new Java resource resolved by name rather than touching shared logic. The mirrored constants read as a deliberate bridge — a Python agent points a `CHAT_MODEL` at the Java router by name. Is that the intended Python story for v1 ("reference the Java router by name," with a native Python SPI as a possible follow-up), or is Python routing out of scope for now? Asking mainly so the intent is on the record — the constants imply Python users are expected to reach this, so it'd help to state how far that's meant to go.
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/ChatModelRouter.java:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Constructor;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A pluggable LLM router that selects, per request, which underlying chat model should serve it.
+ *
+ * <p>The router is a drop-in {@link BaseChatModelSetup}: it reports {@link ResourceType#CHAT_MODEL}
+ * and is resolved by the built-in chat action exactly like any other model, so an agent points at
+ * the router by name and nothing else in the framework needs to change. Concerns are layered:
+ * selection ({@link RoutingStrategy}) decides the model, optional {@link CachingStrategy} memoizes
+ * that decision per conversation, and {@link FallbackPolicy} decides what to try if the chosen
+ * model errors. The router then delegates to the chosen model's own {@code chat(...)}, preserving
+ * its prompt, tools, parameters, and token metrics.
+ *
+ * <h2>Configuration ({@link ResourceDescriptor} arguments)</h2>
+ *
+ * <ul>
+ * <li>{@code candidates} (required) — routable models; each a {@link RoutingCandidate}, a {@link
+ * String} name, or a {@link Map} with {@code name}/{@code description}/{@code metadata}.
+ * Every name must reference a registered {@code CHAT_MODEL} resource.
+ * <li>{@code strategy} (required) — a {@link ResourceDescriptor} naming the {@link
+ * RoutingStrategy} implementation and its args (use {@link
+ * org.apache.flink.agents.api.resource.ResourceName.RoutingStrategy} for built-ins). The
+ * class needs a public {@code (ResourceDescriptor, ResourceContext)} constructor (see {@link
+ * AbstractRoutingStrategy}); it is loaded with the thread context classloader so user classes
+ * resolve on a cluster.
+ * <li>{@code fallback} (optional, default {@code false}) — when {@code true}, a failing model
+ * falls back to the remaining candidates in declaration order.
+ * <li>{@code cache} (optional, default {@code true}) — memoize the selection per conversation.
+ * <li>{@code cache_size} (optional, default {@link CachingStrategy#DEFAULT_MAX_ENTRIES}) — LRU
+ * capacity when caching is enabled.
+ * <li>{@code default} (optional) — the candidate used when the strategy abstains or names a
+ * non-candidate (a routing miss). Must be one of {@code candidates}; defaults to the first
+ * candidate.
+ * </ul>
+ *
+ * <p><b>Graceful degrade:</b> if the strategy returns {@code null} ("no opinion", e.g. a transient
+ * LLM-judge failure) or a name that is not a configured candidate, the router treats it as a
+ * routing miss and serves the request from {@code default} (then the fallback order) rather than
+ * failing.
+ *
+ * <p><b>Bash/skill tool args (v1 scope):</b> the built-in chat action injects bash allowlists and
+ * skill directories from the resource resolved by the agent's model name — i.e. this router — not
+ * the chosen backend. So configure {@code allowed_commands} / {@code allowed_script_dirs} / {@code
+ * skills} <b>on the router</b>; per-candidate skills/allowlists are not supported in v1.
+ *
+ * <p><b>Metrics note (v1):</b> retry metrics recorded by the built-in chat action are grouped under
+ * this router's connection label ({@code "router"}), not the backend model actually used.
+ * Per-backend attribution is a documented follow-up.
+ */
+public class ChatModelRouter extends BaseChatModelSetup {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ChatModelRouter.class);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ public static final String ARG_CANDIDATES = "candidates";
+ public static final String ARG_STRATEGY = "strategy";
+ public static final String ARG_FALLBACK = "fallback";
+ public static final String ARG_CACHE = "cache";
+ public static final String ARG_CACHE_SIZE = "cache_size";
+ public static final String ARG_DEFAULT = "default";
+
+ private final List<RoutingCandidate> candidates;
+ private final Set<String> candidateNames;
+ private final RoutingStrategy strategy;
+ private final FallbackPolicy fallbackPolicy;
+ private final boolean fallbackEnabled;
+ private final String defaultCandidate;
+
+ @SuppressWarnings("unchecked")
+ public ChatModelRouter(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+ super(descriptor, resourceContext);
+
+ List<Object> rawCandidates = descriptor.getArgument(ARG_CANDIDATES);
+ if (rawCandidates == null || rawCandidates.isEmpty()) {
+ throw new IllegalArgumentException(
+ "ChatModelRouter requires a non-empty 'candidates' argument.");
+ }
+ List<RoutingCandidate> parsed = new ArrayList<>(rawCandidates.size());
+ Set<String> names = new LinkedHashSet<>();
+ for (Object spec : rawCandidates) {
+ RoutingCandidate candidate = RoutingCandidate.from(spec);
+ parsed.add(candidate);
+ names.add(candidate.getName());
+ }
+ this.candidates = Collections.unmodifiableList(parsed);
+ this.candidateNames = Collections.unmodifiableSet(names);
+
+ ResourceDescriptor strategyDescriptor =
+ toResourceDescriptor(descriptor.getArgument(ARG_STRATEGY));
+ RoutingStrategy base = instantiateStrategy(strategyDescriptor, resourceContext);
+
+ boolean cache = descriptor.getArgument(ARG_CACHE, Boolean.TRUE);
+ if (cache) {
+ int cacheSize =
+ descriptor.getArgument(ARG_CACHE_SIZE, CachingStrategy.DEFAULT_MAX_ENTRIES);
+ this.strategy = new CachingStrategy(base, cacheSize);
+ } else {
+ this.strategy = base;
+ }
+
+ this.fallbackEnabled = descriptor.getArgument(ARG_FALLBACK, Boolean.FALSE);
+ this.fallbackPolicy =
+ fallbackEnabled ? FallbackPolicy.remainingCandidates() : FallbackPolicy.none();
+
+ // Default candidate used on a routing miss (strategy abstains / names a non-candidate).
+ // Validated at construction so a typo is a clear config error, not a per-request failure.
+ String configuredDefault = descriptor.getArgument(ARG_DEFAULT);
+ if (configuredDefault != null && !candidateNames.contains(configuredDefault)) {
+ throw new IllegalArgumentException(
+ "ChatModelRouter 'default' '"
+ + configuredDefault
+ + "' is not one of the configured candidates "
+ + candidateNames);
+ }
+ this.defaultCandidate =
+ configuredDefault != null ? configuredDefault : candidates.get(0).getName();
+ }
+
+ /**
+ * The router has no connection of its own to resolve (it delegates to candidate models, each of
+ * which resolves its own). Override to skip the base class's connection resolution.
+ */
+ @Override
+ public void open() {
Review Comment:
Worth calling out the design here: the three-way split into selection / resilience / caching (`RoutingStrategy` / `FallbackPolicy` / `CachingStrategy`) lands the right seams, and the drop-in `CHAT_MODEL` story holds up against the real `ChatModelAction` + `ResourceCache` path. One non-obvious correctness point stood out at this `open()`: the no-op is safe precisely *because* a routed candidate is lazily `open()`-ed by `ResourceCache.getResource()` when it's resolved, so its `connection` is non-null before `chat()` runs. Documenting that invariant in the Javadoc rather than leaving it implicit is exactly right — it's the kind of thing that bites a future maintainer if it only lives in someone's head.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494554119
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/RoutingCandidate.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A routable target of a {@link ChatModelRouter}.
+ *
+ * <p>A candidate names a chat-model setup that was registered as a {@link
+ * org.apache.flink.agents.api.resource.ResourceType#CHAT_MODEL} resource, together with an optional
+ * human-readable {@code description} (consumed by LLM-as-router strategies to reason about which
+ * model fits a request) and free-form {@code metadata} (e.g. {@code cost}, {@code tags},
+ * capabilities) that rule-based or custom strategies can match against.
+ */
+public class RoutingCandidate {
+
+ private final String name;
+ private final String description;
+ private final Map<String, Object> metadata;
+
+ public RoutingCandidate(String name, String description, Map<String, Object> metadata) {
+ this.name = Objects.requireNonNull(name, "candidate name must not be null");
Review Comment:
Added RoutingCandidateTest covering null, empty, and valid names.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-4960643341
Opened a discussion to align on the design before going further here: #897
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on code in PR #852:
URL: https://github.com/apache/flink-agents/pull/852#discussion_r3494554119
##########
api/src/main/java/org/apache/flink/agents/api/chat/model/routing/RoutingCandidate.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A routable target of a {@link ChatModelRouter}.
+ *
+ * <p>A candidate names a chat-model setup that was registered as a {@link
+ * org.apache.flink.agents.api.resource.ResourceType#CHAT_MODEL} resource, together with an optional
+ * human-readable {@code description} (consumed by LLM-as-router strategies to reason about which
+ * model fits a request) and free-form {@code metadata} (e.g. {@code cost}, {@code tags},
+ * capabilities) that rule-based or custom strategies can match against.
+ */
+public class RoutingCandidate {
+
+ private final String name;
+ private final String description;
+ private final Map<String, Object> metadata;
+
+ public RoutingCandidate(String name, String description, Map<String, Object> metadata) {
+ this.name = Objects.requireNonNull(name, "candidate name must not be null");
Review Comment:
Good catch. I added an explicit empty-name guard in `RoutingCandidate`. Empty names now fail fast with `IllegalArgumentException`, since they have no valid resource to resolve and can make the whole-token parser over-match boundaries. I also added `RoutingCandidateTest` coverage for null, empty, and valid names.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah commented on PR #852:
URL: https://github.com/apache/flink-agents/pull/852#issuecomment-4857196264
Thanks @weiqingy — really appreciate you taking the time to go through the design and call out the subtle cases. Your comments were very helpful, especially around the open() invariant and the caching wording. Glad the follow-ups addressed them.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy) [flink-agents]
Posted by "purushah (via GitHub)" <gi...@apache.org>.
purushah closed pull request #852: [api][routing] Pluggable in-chat LLM routing (ChatModelRouter + RoutingStrategy)
URL: https://github.com/apache/flink-agents/pull/852
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org