Repository navigation
[api][plan][runtime][examples] Add pluggable in-chat model routing (MODEL_ROUTER) - #964
Conversation
|
Thanks @weiqingy for the sharp review — all eight comments are addressed in the follow-up commits with new tests. The |
fd3cd5d to
7dade1a
Compare
wenjin272
left a comment
There was a problem hiding this comment.
Thanks for pushing this design forward and for addressing the earlier feedback. I have two structural suggestions around tool-round state and the organization of ChatModelAction.
|
Thanks @wenjin272 — both suggestions implemented in the follow-up commits. |
weiqingy
left a comment
There was a problem hiding this comment.
Thanks for addressing the comments. While reviewing the reordered code, a few inline comments below.
|
Thanks @weiqingy for another careful pass — all seven addressed: the Python enum member is in (mixed jobs deserialize; full Python routing stays the follow-up), the three test gaps are covered, the |
…ODEL_ROUTER) Adds select-not-delegate model routing: a MODEL_ROUTER resource picks a concrete chat model per request, then ChatModelAction runs the normal chat path against it — so the selected call stays a first-class chat with tokens attributed to the real backend, and the decision itself is recorded as an observability-only ModelRoutingEvent. Implements the design agreed in discussion apache#897; supersedes the delegating approach from apache#852. API: - ModelRouter resource + fluent builder: candidates, per-candidate describe(name, description) surfaced to strategies via RoutingCandidate.getDescription(), defaultModel (must be a candidate), fallback flag; strategy travels as class name + args so the plan stays serializable. - RoutingStrategy SPI (one method: RoutingContext -> RoutingDecision), Strategies factories (no magic-string dispatch), built-in RuleBasedRoutingStrategy (regex over the most recent user message, evaluated in map iteration order — pass a LinkedHashMap when precedence matters; no match abstains). RoutingContext carries the initial request id for correlation and deterministic per-request policies, and deliberately exposes no chat API: strategies cannot make hidden model calls (framework-managed LLM-judge routing is a documented follow-up per apache#897). Per-candidate metadata is deferred until a strategy can consume it. YAML routing support is intentionally left out of v1 to keep the first Java API/runtime change small; it and Python support are follow-ups. - RoutingDecision (selected model or abstain, plus reason / score / metadata; snake_case JSON; invariants enforced on the JSON construction path used by durable replay; framework-stamped decision_ms persists the strategy wall time with the decision). - ModelRoutingEvent carrying router, candidates, selected model, decision source (strategy / default / fallback), fallback_enabled, strategy metadata, and decision_ms. Runtime semantics in ChatModelAction: - Router detection via non-throwing RunnerContext.hasResource, which covers both registered providers and cache-injected resources. A name registered as both chat model and router is rejected at the addResource call site (Agent / AgentsExecutionEnvironment) with an AgentPlan backstop. - The routing decision executes as its own durable call ("route:<requestId>:<router>"); latency is measured inside the call, so recovery replays the persisted decision with its original decision_ms instead of re-running a possibly non-deterministic strategy. - Route once per reasoning loop: tool rounds reuse the selected model and carry the routing metadata onto the final response. - Fallback layered on retries: the selected model gets its full retry budget (durable id "chat:<router>:<candidate>"), then remaining candidates in declaration order; fallback outcomes emit a second ModelRoutingEvent and are recorded on the response. - Abstain resolves to the default model; a non-candidate selection fails clearly. Non-router requests keep the legacy "chat" durable id and are byte-for-byte unaffected. - Decision latency also feeds the routingDecisionLatencyMs histogram. Examples: rule-based routing pipelines (Ollama + OpenAI variants). Tests: 20 API routing tests (strategy/router/decision/registration validation), 9 ChatModelAction routing tests (rule match/abstain, builder validation, JSON round-trips, routing/fallback/tool-round semantics, event contents), and a ResourceCache case covering cached-only resource visibility.
…overy "route:<requestId>:<router>" embedded the ChatRequestEvent's random UUID, which is regenerated when Flink rolls back and re-processes after recovery — so durable-store lookups for routing decisions never hit, and a non-deterministic strategy silently re-ran instead of replaying (measured in kill/restore trials: 0/138 decisions replayed). Per-request uniqueness already comes from the action-state store's (key, sequence number, event, action) scoping; the call id must stay deterministic, like the chat call ids. Now "route:<router>".
- ModelRouter.build() validates rule keys against candidates, so a typo'd rule fails at the registration call site instead of throwing per record - resolveRouter failures honor error-handling-strategy (IGNORE drops the request with a warning instead of killing the job) - fallback exhaustion chains each candidate's error via addSuppressed, logs per-candidate errors, and warns with the tried-candidate list - an unresolvable candidate counts as that candidate failing (lookup moved inside the attempt conversion), so fallback and IGNORE still apply - RoutingContext deep-copies messages: a strategy can no longer mutate the prompt that is actually sent - RuleBasedRoutingStrategy rejects null/non-String rule values instead of compiling the literal pattern "null" - durability javadoc (ChatModelAction, RoutingDecision) qualified: replay requires a configured action-state store - compatibility note on per-attempt retry metrics recording Tests: exhaustion test asserts the suppressed chain with distinct markers; new tests for unresolvable-candidate fallback, strategy failure under IGNORE/default, build-time rule-key validation, null/non-String rule values, and RoutingContext deep-copy.
…ig key in javadoc - RoutingContext.deepCopy tolerated only the constructor's non-null toolCalls default, but Jackson's setToolCalls stores null as-is, so a message deserialized from JSON with an explicit "tool_calls": null NPE'd the copy. Null now re-normalizes through ChatMessage's constructor: strategies always see a non-null (empty) list. - The durability javadoc cited a non-existent config key (agent.action-state-store.backend); the actual option is actionStateStoreBackend (AgentConfigOptions.ACTION_STATE_STORE_BACKEND).
The constant existed but was missing from the allConstants() map, so condition expressions like "type == EventType.ModelRoutingEvent" could not resolve it. Covered in EventTypeTest and by a compiled condition-expression test.
…ackage The submit-examples E2E job submits every class directly under org.apache.flink.agents.examples against a keyless local cluster, so an example that requires OPENAI_API_KEY fails CI by construction. Moving it to the openai subpackage keeps it in the repo and runnable locally while excluding it from auto-discovery (which matches top-level classes only). The Ollama-based ModelRoutingExample remains the CI-exercised routing example.
…rying it Per wenjin272's review: the model_routing block is observability-only and needed exactly once, on the loop's final response. It is now stored once in an initial-request-keyed context when a routed response starts a tool loop, attached when the loop produces its final response, and removed on every loop exit (including IGNORE-dropped exhaustion). This removes the RoutingSelection.carried state, the per-round stamp/read/copy cycle, and observability metadata from intermediate ChatMessage extraArgs (which previously persisted in the conversation history for the whole loop). Test: tool-round test now also asserts the intermediate message is unstamped and the parked context is consumed.
…elAction Per wenjin272's review: ChatModelAction had grown to ~1000 lines across several responsibilities. Routing-cohesive pieces move to package-private classes: - ResolvedModelRoute: the resolution outcome (candidates, fallback policy, decision facts) plus attempt order, routed durable-call ids, and the model_routing response block — the former RoutingSelection inner class and its free-function helpers, now methods on the type they describe. - ModelRoutingResolver: strategy execution inside the durable route call, decision normalization, and ModelRoutingEvent emission. - ChatModelInvoker: one-candidate invocation with durable-call + retry machinery (chatWithRetries, ChatAttemptResult, ChatAttemptFailed). ChatModelAction keeps event orchestration, the fallback loop, tool-loop state, and retry-stats accounting. recordChatTokenMetrics and generateStructuredOutput stay put to avoid churning code (and tests) that main owns. No behavior change; full plan suite green.
… tests, RoutingContext boundary javadoc Per weiqingy's second review: - Python ResourceType gains MODEL_ROUTER so a Java plan containing a router deserializes on the Python side: mixed jobs (Java router + Python actions) must not fail at operator open with a ValidationError. Full Python routing remains a follow-up. Cross-language round-trip test added. - Routed durable-call ids are now asserted (route:<router>, chat:<router>:<candidate>, distinct per candidate in the fallback test) — the format recovery depends on, and which changed once already. - retryBudgetRunsBeforeFallback pins the retry-before-fallback ordering the class javadoc guarantees (selected model's retry succeeds; fallback candidate never resolved). FakeRunnerContext gains withRetryBudget. - AgentPlanRoutingBackstopTest exercises the cross-registry name-clash path that only AgentPlan's backstop check catches (addResourcesIfAbsent merges bypass both per-call checks). - RoutingContext javadoc states the one-level-deep isolation boundary explicitly: defensive copies make accidental top-level mutation harmless; nested values are shared; the SPI forbids mutation. api 366 / plan 275 / python plan-tests 128 green.
3b9bdc8 to
cd95d5f
Compare
weiqingy
left a comment
There was a problem hiding this comment.
Thanks for addressing the comments. Two small things inline, neither blocking. Otherwise, LGTM.
| throw new IllegalStateException("ModelRouter requires a strategy(...)."); | ||
| } | ||
| // Rule keys are candidate names; validate here, where both lists are in hand, so a | ||
| // typo fails at the registration call site instead of throwing per record at runtime. |
There was a problem hiding this comment.
The comment here says rule keys are validated at build time so a typo fails at the registration call site instead of throwing per record at runtime. The rule values have the same shape and no check: Pattern.compile runs only in the strategy constructor (RuleBasedRoutingStrategy.java:62), reached from ModelRouter:88 when the resource is provided, and ResourceCache caches only after provide() returns (ResourceCache.java:144). A PatternSyntaxException therefore never caches and re-throws per routed request: under the default FAIL the job dies on the first one, under ignore requests through that router are dropped with a WARN while the job stays RUNNING. Would compiling the patterns in this same block move that to build() too?
There was a problem hiding this comment.
Good catch on the caching consequence — fixed: build() now compiles every rule pattern (and rejects null/non-String values) in the same block as the key check, so a malformed regex fails at the registration call site. Two new builder tests cover it.
| Map<String, Object> routing = new LinkedHashMap<>(); | ||
| routing.put("router", this.requestedModel); | ||
| routing.put("selected_model", this.selectedModel); | ||
| routing.put("initial_selected_model", this.selectedModel); |
There was a problem hiding this comment.
nit: this and selected_model above are both this.selectedModel, and initial_selected_model has no other reference in the PR. Nothing reads it and no test asserts it. The fallback winner is already final_model on the next line. Removing a key from a shipped payload is a compat change later. Is the second key meant to differ from the first on some path, or can it go?
There was a problem hiding this comment.
You are right — it duplicated selected_model on every path (fallback winners are final_model). Removed before it ships and becomes a compat constraint.
…ial_selected_model payload key Per weiqingy's post-LGTM pass: - ModelRouter.build() now compiles each rule pattern (and rejects null/ non-String values) so a malformed regex fails at the registration call site; an invalid pattern is never cached by the resource cache, so it previously re-threw per routed request (job death under FAIL, silent per-request drops under IGNORE). Two new builder tests. - initial_selected_model duplicated selected_model byte-for-byte on every path (fallback winners live in final_model); removed before the payload ships so it never becomes a compatibility constraint.
weiqingy
left a comment
There was a problem hiding this comment.
Both addressed at e95f20bd. No remaining comments from me.
|
Thanks for taking this on @purushah. I noticed that this feature has the design discussion (#897) and implementation PR (#964), but no corresponding issue. Could you please create one summarizing the agreed scope, link both the discussion and this PR? This will make the feature easier to track in the project history and release notes. |
|
Thanks @wenjin272 and @weiqingy for the careful reviews and the merge — the feature came out much stronger. A heads-up on what's next: the |
Adopts the declaration/executor split from the PR apache#1042 review (wenjin272), with two amendments keeping apache#964's guarantees structural, and folds in the remaining review items from both reviewers. Architecture (review thread on LlmJudgeRoutingStrategy.java:84): - RoutingStrategy is now an immutable, serializable declaration: RoutingStrategyType (RULE_BASED | LLM_JUDGE | CUSTOM) + arguments (+ executorClass for CUSTOM). No route() exists at the API layer; the throwing UnsupportedOperationException class is gone. - Plan JSON encodes the strategy as a language-neutral tag (strategy_type: "llm_judge") + strategy_args — no Java FQCNs for built-ins, unblocking Python parity (apache#1062). Java-produced snapshot committed; the Python cross-language test now deserializes it verbatim. - Execution lives in Plan: rule matching in the resolver over patterns precompiled by the router; judge prompt/verdict logic in LlmJudgeRoutingExecutor (package-private pure helpers); dispatch by type. - Amendment 1: the resolver owns the durable route: boundary for every strategy type — no executor can skip persistence; replay never re-invokes the strategy or the judge. - Amendment 2: user-defined executors implement CustomRoutingExecutor with the data-only RoutingContext (v1's select-not-delegate fence, verbatim); the engine-facing signature stays plan-internal. Review fixes: - isCancellation() replaces containsInterrupt(): thread-state + explicit cancellation types only; a bare InterruptedIOException (Okio timeout shape) follows the normal FAIL/IGNORE policy. Regression test included. - Judge input is now full-fidelity: complete message list plus the rendered request (anchor candidate's bound prompt + args), with an opt-in Strategies.llm(...).withMaxContextChars(n) cap (SYSTEM + newest pinned, newest-first fill, judge_context_truncated flagged in metadata). - Judge setup constraints (no prompt/tools/skills) validate at plan construction; judgeSetupMisconfiguration() removed from the request path. - Strategy argument validation lives in the declaration constructor, so a bad-args judge fails Strategies.llm(...)/plan construction — the reflective instantiateIfLlmJudge() bypass is structurally gone. - decision_source is no longer duplicated into decision metadata (it contradicted the top-level key on judge+fallback). - FakeRunnerContext gained a seedable durable store; new tests cover the replay path (stored abstain -> current default; decision_ms survives) and judge token metadata (extraArgs-driven). Generated-by: Claude Code 2.1.239 (Claude Fable 5)
Adds docs/development/model_routing: a guide page covering router declaration, the rule-based, LLM-judge and custom strategies, default model and fallback with the error-handling policy table, durability and replay, and observability; plus a Reference subpage with the routing event and response fields, metrics, validation stages, and descriptor keys. Cross-links from Chat Models, Workflow Agent, and Monitoring (new rows for the routingDecisionLatencyMs histogram and the _model_routing_event type). Sibling page weights shift by one so the page sits after Chat Models; this also resolves the pre-existing MCP/Skills weight tie. Covers the doc-needed label on apache#964 and apache#1042.
Linked issue: #897 (design discussion). Supersedes #852.
Purpose of change
Today an agent must name a concrete chat model in each
ChatRequestEvent; any routing logic lives in user code and is not first-class in the runtime. Most requests are easy and a small model handles them fine, but a minority really need the strong model — so a fixed choice either overpays on the easy majority or under-serves the hard ones. This PR makes per-request routing a framework capability: an agent sends its request to a router, and the runtime picks one of several candidate models.My first attempt at this (#852) made the router a special
ChatModelSetupthat called the chosen backend itself. Review rightly pointed out that this breaks the two things the framework is good at: token metrics collapse onto the router instead of the real backend, and the actual model call disappears from the EventLog. The redesign discussed in #897 fixes this by making the router select instead of delegate: the router only returns a model name, andChatModelActionruns the normal chat path against that model. A routed request produces the same events and the same per-model metrics as if the agent had named the chosen model directly.From the user side it stays simple:
A few behaviors were worth getting right, and they came out of the #897 discussion:
"route:<router>"— deterministic across recovery re-processing; per-request uniqueness comes from the store's key/sequence/event scoping), so on recovery the persisted decision is replayed instead of re-running the strategy. Decision latency is measured inside that call, so a replayed run reports the originaldecision_ms.ModelRoutingEventso the event log shows which model actually answered.decision_source=default); a strategy that returns a non-candidate name fails loudly, because that is a bug, not a runtime condition.ModelRoutingEventis observability-only. It has no built-in consumer and does not drive dispatch — removing every listener does not change which model runs.Since routers and chat models share the
ChatRequestEventnamespace, one name must not be registered as both; this is validated at theaddResourcecall site (with anAgentPlanbackstop) so the failure points at the user's own line.Per the discussion, LLM-as-judge routing is deliberately not in this PR.
RoutingContextexposes no chat API, so a strategy cannot make hidden model calls; the framework-managed observable judge (the judge call running on the normal durable/metered chat path) is the agreed follow-up. YAML support is intentionally left out of v1 to keep the first Java API/runtime change small; it and Python support are follow-ups.Tests
RoutingTest,RoutingResourceValidationTest): rule matching and abstain over the latest user message, builder and registration validation, candidate descriptions reaching strategies,RoutingDecisioninvariants and snake_case JSON round-trips,decision_mssurviving replay deserialization, event attribute normalization.ChatModelActionRoutingTestintegration tests against a scripted fake chat model: routing to the matched candidate, abstain-to-default, invalid candidate failing clearly, non-router requests keeping the legacy"chat"durable id unchanged, fallback across candidates (including exhaustion), and route-once semantics across a tool round with metadata carried to the final response.ResourceCacheTestcase coveringhasResourcevisibility of resources inserted directly into the cache (no provider).Verified locally (JDK 17):
mvn spotless:check -pl api,plan,runtime,examplesmvn test -pl api,plan— 312 + 196 tests, 0 failuresmvn test -pl runtime -Dtest=ResourceCacheTestmvn compile -pl examplesAPI
New public surface, all additive:
ResourceType.MODEL_ROUTERorg.apache.flink.agents.api.chat.model.routing:ModelRouter(+ builder),RoutingStrategy,RoutingContext,RoutingDecision,RoutingCandidate,RoutingStrategyDescriptor,Strategies,RuleBasedRoutingStrategyModelRoutingEventRunnerContext.hasResource(name, type)(default method, returns false)Compatibility impact
A request naming a plain chat model takes the unchanged success path, including the legacy durable call id, and the default
error-handling-strategy(FAIL) is unaffected. Three deliberate behavior changes on non-default failure paths:retry: retry metrics are recorded per attempt (previously once with cumulative totals on the final response), so requests that ultimately fail now contribute their retry counts and each attempt lands on the model that served it. Totals over a completed request are unchanged.retry: the retry WARN log now emits underChatModelInvoker's logger (wasChatModelAction) — external log alerting keyed to the old logger name needs updating.ignore: a request naming a missing chat-model resource is warned and dropped (consistent with other request failures underignore) instead of failing the job with a raw lookup exception.Documentation
doc-neededdoc-not-neededdoc-includedThe Java examples show usage, but this adds public API, so proper docs are warranted — happy to do a docs follow-up PR once the API settles in review.