purushah opened a new pull request, #1330:
URL: https://github.com/apache/flink-agents/pull/1330
Linked issue: #1325
### Purpose of change
A `MODEL_ROUTER` can now route with a registered decision model.
`Strategies.decision("dm")` asks the `DECISION_MODEL` resource one choice
question (options: the candidates; criteria: their `describe(...)` texts) and
routes on the returned choice. The choice's probability becomes the routing
`score`; `withMinProbability(p)` abstains to the default model below `p`.
Rules, the LLM judge and custom executors are unchanged.
The LLM judge returns a candidate name and nothing else, so a router cannot
tell a confident selection from a coin flip or decline a weak one. Decision
models return typed answers with probabilities, so the strategy reuses them
instead of parsing a verdict out of free text.
#### Runtime flow
1. `ModelRoutingResolver` dispatches `DECISION_MODEL` to
`DecisionModelRoutingExecutor`, which declares internal durable calls and so
runs before the `route:<router>` record, like the judge.
2. The executor resolves the setup, renders the conversation to text with
the judge's rendering (anchor candidate's bound prompt, roles, tool calls and
results, optional `max_context_chars` cap, prompt-argument fallback; media
dropped and counted) and builds one `ChoiceQuestion` named `model`.
3. `DecisionModelInvoker.decideWithRetries` (new, shared with the built-in
decision action) runs `setup.decide` inside `durableExecute` under
`decide:<router>` with the configured retry budget, execution reports and
`model_resource` token metrics; retries are accumulated on the routed request
like the judge's.
4. The answer maps to a `RoutingDecision` (table below) with metadata
`decision_model`, `decision_probabilities`, optional `provider_confidence`,
token counts, `decision_context_truncated`, `decision_media_dropped`, and
`decision_choice` on abstain. The resolver persists, normalizes and emits
`ModelRoutingEvent` (source `decision_model`); the chat path is unchanged.
#### Key decisions
- The score is the provider's probability for its choice, not the maximum of
the map; the framework does not normalize distributions.
- `provider_confidence` is recorded but not thresholded; not every provider
reports it.
- Every candidate must carry a description, enforced by one
`RoutingCandidateValidator` helper from the constructor, `build()` and plan
validation; plan validation also requires an explicitly declared setup
connection to be registered.
- The call bypasses the built-in decision action (durable calls cannot nest
and its flat `decide` id would collide) but shares its retry loop through
`DecisionModelInvoker`, which the action now uses too.
### Behavioral Semantics
#### Interaction decisions
| Answer | `min_probability` | Result |
|---|---|---|
| choice, probability ≥ min or no min | any | route to choice, `score` =
probability, source `decision_model` |
| choice, probability < min | set | abstain to default; reason names
probability and threshold; `decision_choice` recorded |
| choice naming a non-candidate | any | abstain to default (defensive) |
| refusal | any | abstain to default, reason `decision model refused[:
reason]` |
| predicate or score answer | any | abstain to default, reason `decision
answer was not a choice` |
| request carries no text and no prompt arguments | any | abstain to default
without calling the provider |
| provider error after retries | any | request fails (`Decision model
'<name>' failed: …`); no default, no fallback |
#### Behavioral contracts
1. A choice routes to its candidate with the reported probability as score.
2. A choice below `withMinProbability` abstains to the default; choice and
probability map stay in metadata.
3. A refusal or non-choice answer abstains to the default; so does a request
with no text and no prompt arguments, without a provider call.
4. The call is durable under `decide:<router>`: on replay the stored answer
is reused, the provider not called.
5. Provider failures use the retry budget; exhausting it fails the request,
no candidate fallback.
6. Fewer than two candidates, an undescribed or blank-described candidate,
an unregistered decision model, a declared connection that is blank or
unregistered, a declared blank model, or `min_probability` outside `[0, 1]`:
rejected at declaration or plan construction.
7. Tool rounds, candidate fallback and the other strategies behave as before.
#### Failure behavior
Unresolvable decision model, a request that cannot be rendered, or an answer
that cannot be read: `RoutingFailure` (failed `ChatResponseEvent`).
Durable-state or recovery failures during the decision call propagate to the
runner. Provider exception or null result: retried per `MAX_RETRIES` /
`RETRY_WAIT_INTERVAL`, then `RoutingFailure`; cancellation propagates, never
persisted. Persistence or recovery errors propagate to the runner, as for the
judge.
### Tests
| Contract | Tests (E = `DecisionModelRoutingExecutorTest`, C =
`ChatModelActionRoutingTest`) |
|---|---|
| 1 | E `choiceSelects…AsScore`,
`scoreIsTheChosenCandidateProbabilityNotTheMaximum`; C
`decisionModelChoiceRoutesToNamedCandidate` |
| 2 | E `probabilityBelowMinimumAbstains…`,
`probabilityAtTheMinimumIsAccepted`; C
`decisionBelowMinProbabilityAbstainsToDefault` |
| 3 | E `refusalAbstains…`, `nonChoiceAnswerAbstains`,
`nonCandidateChoiceAbstains`, `emptyRequestAbstainsWithoutCallingTheProvider`;
C `decisionRefusalAbstainsToDefault` |
| 4 | C `decisionReplayReusesTheStoredAnswerWithoutCallingTheProvider`,
`asyncDecisionCallRunsAsFlatSiblingBeforeTheRouteRecord` |
| 5 | C `decisionModelUsesTheRetryBudget…`,
`decisionModelFailureFailsTheRequest…`,
`unresolvableDecisionModelFailsTheRequest`,
`durableStateFailureDuringTheDecisionCallPropagatesToTheRunner`; E
`unrenderableRequestIsARoutingFailure` |
| 6 | `RoutingTest` decision tests, `AgentPlanDecisionModelValidationTest`
(9 tests) |
| 7 | existing routing suites; `AgentPlanCrossLanguageTest` decision-router
snapshot; Python `test_python_can_deserialize_java_plan_with_decision_router` |
```
mvn -pl api,plan test # api 836 pass; plan 690 run, only
AgentPlanDeclareMCPServerTest fails (talks to a local port-8000 server; fails
identically on unmodified main)
uv run pytest flink_agents/plan/tests/test_agent_plan_cross_language.py #
21 passed, 1 skipped
mvn -pl e2e-test/...-integration,e2e-test/...-resource-cross-language test
# 71 pass; cross-language: 2 Python-bridge tests fail with 'No module named
flink_agents' from the system python3, unrelated to routing
```
End to end on a standalone Flink 2.3.0 cluster with the new
`examples/openai/OpenAiDecisionRoutingExample` (gpt-4o-mini and gpt-4o
candidates, OpenAI Decisions `gpt-6-luna` via a native Java
`BaseDecisionModelConnection`): job finished in 3.6 s; output
`final_model=small … score=0.99` for the greeting and `final_model=big …
score=0.87` for the SQL request; no TaskManager errors.
Not verified: Python-side routing (routers remain Java-only), a decision
model reached through the Python bridge in a running job, and cancellation
during the decision call's retry backoff.
<details>
<summary>Implementation invariants and supporting detail</summary>
- `decide:<router>` is distinct from the built-in decision action's flat
`decide` id; they belong to different actions and cannot collide in one journal.
- `DecisionOutcome` (result or error text) is the durable record,
JSON-serializable like the chat invoker's `InvocationOutcome`.
- Rendering helpers in `LlmJudgeRoutingExecutor` became package-private and
are shared; behavior unchanged, still pinned by `LlmJudgeRoutingExecutorTest`.
- `ModelRouter.CANDIDATE_DESCRIPTIONS_KEY` names the existing
`candidate_descriptions` argument; the serialized plan is unchanged.
- `withMaxContextChars` validation is shared by the judge and decision
declarations.
- A media-only, empty or SYSTEM-only conversation contributes no request
text, so the prompt arguments become the decision input
(`mediaOnlyRequestFallsBackToPromptArgs`,
`promptArgsAreTheInputWhenTheConversationCarriesNoRequestText`).
- `DecisionModelActionTest` (32 tests) pins the built-in action's behavior
across the move to `DecisionModelInvoker`.
- Snapshot
`e2e-test/cross-language-agent-plan-snapshots/java/agent_plan_with_decision_router.json`
generated with `-Dregenerate.snapshots=true`.
</details>
### API
New: `RoutingStrategyType.DECISION_MODEL` (tag `decision_model`),
`Strategies.decision(String)` / `decision(String, String)`,
`RoutingStrategy.withMinProbability(double)`, `ARG_DECISION_MODEL` /
`ARG_INSTRUCTIONS` / `ARG_MIN_PROBABILITY`,
`ModelRoutingEvent.SOURCE_DECISION_MODEL`,
`ModelRouter.CANDIDATE_DESCRIPTIONS_KEY`,
`RoutingCandidateValidator.requireDescriptionsForDecisionStrategy`.
`withMaxContextChars` also accepts decision declarations. No existing signature
changes; existing routers, plans and persisted decisions are unaffected.
Routers remain Java-only (#1062).
### Documentation
- [ ] `doc-needed`
- [ ] `doc-not-needed`
- [x] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: Claude Code 2.1.309 (Claude Fable 5.1)
--
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]
