feat(libsy): add EscalationRouter algorithm - #172
Conversation
Signed-off-by: ayushag <ayushag@nvidia.com>
Signed-off-by: ayushag <ayushag@nvidia.com>
Signed-off-by: ayushag <ayushag@nvidia.com>
| fn is_pinned(&self, session_id: &str) -> bool { | ||
| self.pins.lock().contains_key(session_id) | ||
| } | ||
|
|
||
| fn pin(&self, session_id: &str) { | ||
| let mut pins = self.pins.lock(); | ||
| if pins.len() >= MAX_PINS { | ||
| if let Some(evicted) = pins.keys().next().cloned() { | ||
| pins.remove(&evicted); | ||
| } | ||
| } | ||
| pins.insert(session_id.to_string(), ()); | ||
| } |
There was a problem hiding this comment.
we should not be re-inventing pinning here. Try to use AffinityRouter stuff in here if possible.
There was a problem hiding this comment.
Done — replaced the custom pins map with AffinityRouter::new().with_latch_only([capable_name]). Pinning now goes through the same mechanism as the rest of the codebase.
| fn load_judge_config() -> Result<JudgeConfig> { | ||
| let response_schema: Value = | ||
| serde_json::from_str(SCHEMA_TEMPLATE).map_err(|error| LibsyError::AlgorithmError { | ||
| message: format!("escalation response schema is invalid: {error}"), | ||
| })?; | ||
| let prompt_schema = response_schema | ||
| .pointer("/json_schema/schema") | ||
| .ok_or_else(|| LibsyError::AlgorithmError { | ||
| message: "escalation response schema has no json_schema.schema".to_string(), | ||
| })?; | ||
| let prompt_schema = serde_json::to_string_pretty(prompt_schema).map_err(|error| { | ||
| LibsyError::AlgorithmError { | ||
| message: format!("escalation prompt schema could not be rendered: {error}"), | ||
| } | ||
| })?; | ||
| Ok(JudgeConfig { | ||
| system_prompt: PROMPT_TEMPLATE.replace("{{RESPONSE_SCHEMA}}", &prompt_schema), | ||
| response_schema: Some(response_schema), | ||
| }) | ||
| } |
There was a problem hiding this comment.
All this seems to follow the same pattern and duplicate code from llm_class.rs -- can we uniffy ?? Only diff is in the run loop I believe how the latch behaviour work -- for that may be you should extend the AffinityRouter for EscalationAffinity ... ??
There was a problem hiding this comment.
Done — extracted load_judge_config(prompt, schema) into util/llm_judge.rs so both classifiers share it. For the run-loop difference (the judge needs the efficient model's response, not just the original request), added minimal EscalationJudge/EscalationPolicy types that plug into JudgeClassifier — latch logic stays in AffinityRouter, judge plumbing in the existing abstraction.
| impl Decision for EscalationDecision { | ||
| fn selected_model(&self) -> &str { | ||
| &self.model | ||
| } | ||
|
|
||
| fn routing_tier(&self) -> Option<&'static str> { | ||
| Some(self.tier) | ||
| } | ||
|
|
||
| fn reasoning(&self) -> Option<&str> { | ||
| Some(self.reason) | ||
| } | ||
|
|
||
| fn as_any(&self) -> &dyn std::any::Any { | ||
| self | ||
| } | ||
| } |
There was a problem hiding this comment.
why do we need this ? can't we use Decision directly ??
There was a problem hiding this comment.
Removed — JudgeClassifier uses its own internal JudgeDecision. No custom struct needed.
| async fn consult_judge( | ||
| &self, | ||
| ctx: Context, | ||
| driver: &Driver, | ||
| request: &Request, | ||
| efficient_text: &str, | ||
| ) -> bool { | ||
| let judge_model = self.judge_target.semantic_name.as_str(); | ||
| let warn = |error: &dyn std::fmt::Display| { | ||
| tracing::warn!( | ||
| target: "libsy", | ||
| judge_model, | ||
| error = %error, | ||
| "escalation judge unavailable; skipping escalation" | ||
| ); | ||
| }; | ||
|
|
||
| let judge_request = self.build_judge_request(request, efficient_text); | ||
| let judge_decision: Arc<dyn Decision> = Arc::new(JudgeCallDecision { | ||
| model: self.judge_target.semantic_name.clone(), | ||
| }); | ||
|
|
||
| let response = match driver | ||
| .call_llm_target(ctx, &self.judge_target, judge_request, judge_decision) | ||
| .await | ||
| { | ||
| Ok(r) => r, | ||
| Err(e) => { | ||
| warn(&e); | ||
| return false; | ||
| } | ||
| }; | ||
|
|
||
| let agg = match response.llm_response.into_agg().await { | ||
| Ok(a) => a, | ||
| Err(e) => { | ||
| warn(&e); | ||
| return false; | ||
| } | ||
| }; | ||
|
|
||
| let text = completion_text(&agg); | ||
| match parse_verdict(text.trim()) { | ||
| Some(v) => v.should_escalate, | ||
| None => { | ||
| tracing::warn!( | ||
| target: "libsy", | ||
| judge_model, | ||
| "escalation judge verdict did not parse; skipping escalation" | ||
| ); |
There was a problem hiding this comment.
The JudgeClassifier abstraction is generic enough to make a llm call using judge inputs and give u a output ? Why do we need to implement this again ??
There was a problem hiding this comment.
Agreed, removed. Added EscalationJudge (implements Judge, reads the efficient response from state.extra) and EscalationPolicy (implements JudgePolicy, maps verdict to classification). JudgeClassifier handles the call.
67b9ac4 to
58a2da2
Compare
Signed-off-by: ayushag <ayushag@nvidia.com>
| let candidate = state | ||
| .extra | ||
| .get(CANDIDATE_KEY) | ||
| .and_then(|v| { | ||
| if let StateValue::String(s) = v { | ||
| Some(s.as_str()) | ||
| } else { | ||
| None | ||
| } | ||
| }) | ||
| .unwrap_or(""); |
There was a problem hiding this comment.
what does a candidate here refer to ?
There was a problem hiding this comment.
candidate is the efficient model's response text — stored in state.extra under CANDIDATE_KEY just before the judge call. EscalationJudge::build_request reads it to construct the prompt for the judge: "here's what the efficient model replied — was it sufficient?" Without it the judge would only see the original request and couldn't evaluate actual output quality.
| } | ||
|
|
||
| #[async_trait] | ||
| impl Algorithm<SharedState> for EscalationRouter { |
There was a problem hiding this comment.
@linj-glitch Can this be implemented the FallThrough pattern ? We are implementing all the algos using that
There was a problem hiding this comment.
Agreed it's the right long-term direction — Greg flagged the same thing. The blocker is that FallThrough currently has no Event::Response: processors only see Event::Request and Event::Decision, so there's no hook to feed the efficient model's reply into a classifier mid-run. Once Processor gets a response event, escalation folds in naturally. Greg said "we can do that later," so keeping it as a standalone algo for now and we can refactor when the event is there.
Signed-off-by: Lin Jia <linj@nvidia.com>
Signed-off-by: Lin Jia <linj@nvidia.com>
3b34ba2 to
135864e
Compare
…ix API migration
Replace the generic 42-line prompt with the validated 179-line agentic trajectory
judge from switchyard/lib/processors/prompts/escalation_judge.md. Simplify schema
to {escalate, reason}. Migrate EscalationRouter off the removed SharedState/generic
Algorithm API: manage session state internally via Mutex<HashMap>, delegate stateful
logic to a plain async fn (execute) to avoid MutexGuard lifetime conflicts with
async_trait's 'static bound from AffinityRouter's S: 'static impl constraint.
Signed-off-by: Lin Jia <linj@nvidia.com>
…ionRouter Matches the pattern FallThrough::route already uses: dispatch through dyn Classifier<State>/dyn Processor<State> so async_trait doesn't propagate the S: 'static bound from AffinityRouter's impl. Replaces the scratch-State workaround and passes real session state through. Signed-off-by: Lin Jia <linj@nvidia.com>
Signed-off-by: Lin Jia <linj@nvidia.com>
…lation-router Signed-off-by: Lin Jia <linj@nvidia.com> # Conflicts: # crates/libsy/README.md # crates/libsy/examples/research_agent.rs # crates/libsy/examples/research_agent_core.rs # crates/libsy/src/algorithms/llm_class.rs # crates/libsy/src/algorithms/util/affinity.rs # crates/libsy/src/lib.rs # crates/libsy/tests/observability.rs # crates/switchyard-server/src/config.rs
WalkthroughThis PR adds a new ChangesEscalation Router Feature
Estimated code review effort: 4 (Complex) | ~60 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
crates/libsy/src/algorithms/escalation.rs (2)
327-350: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe escalation path re-implements
call_tier. Lines 346-349 duplicate thedriver.info+call_llm_targetsequence because the decision has to exist earlier for the affinity latch. Threading a prebuilt decision throughcall_tierkeeps the publish/serve sequence in one place.♻️ Sketch
- async fn call_tier( - &self, - ctx: Context, - driver: &Driver, - request: Request, - target: &LlmTarget, - tier: &'static str, - reason: &'static str, - ) -> Result<Response> { - let decision: Arc<dyn Decision> = Arc::new(EscalationDecision { - model: target.semantic_name.clone(), - tier, - reason, - }); + async fn serve( + &self, + ctx: Context, + driver: &Driver, + request: Request, + target: &LlmTarget, + decision: Arc<dyn Decision>, + ) -> Result<Response> { driver.info(ctx.clone(), decision.clone()).await?; driver.call_llm_target(ctx, target, request, decision).await }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libsy/src/algorithms/escalation.rs` around lines 327 - 350, Refactor the escalation flow around call_tier to accept an optional prebuilt Decision, allowing the affinity latch to use it before serving while keeping driver.info and call_llm_target centralized in call_tier. Update the escalation caller to construct and pass the capable-model decision, and remove its duplicated publish/serve sequence while preserving existing behavior for calls without a prebuilt decision.
83-95: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider surfacing the judge's
reasoninstead of discarding it.
EscalationVerdict::reasonis parsed then dropped (#[allow(dead_code)]), andEscalationDecision::reasonis a&'static str, so the trace never records why the judge escalated. SinceJudgePolicy::to_classificationonly returns aClassification, the cheapest fix is atracing::info!of the reason insideEscalationPolicy::to_classification; wideningEscalationDecision::reasontoCow<'static, str>would let it reach the trace.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libsy/src/algorithms/escalation.rs` around lines 83 - 95, Update EscalationPolicy::to_classification to surface the parsed EscalationVerdict::reason via tracing::info! before returning the Classification. Remove the unnecessary dead-code allowance on reason, and keep EscalationDecision unchanged unless the implementation is instead extended to carry the reason into tracing.crates/libsy/src/algorithms/util/llm_judge.rs (1)
174-201: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider a direct unit test for
load_judge_config's error branches.Now that this is a shared, multi-caller utility, it's only exercised indirectly through
LlmTaskClassifier's tests (happy path). A couple of focused tests here (invalid schema JSON, missing/json_schema/schemapointer) would pin down theLibsyError::AlgorithmErrorcontract independent of any one caller.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/libsy/src/algorithms/util/llm_judge.rs` around lines 174 - 201, Add focused unit tests for the shared load_judge_config function covering invalid schema JSON and a schema missing the /json_schema/schema pointer. Assert both cases return LibsyError::AlgorithmError with the expected messages, while keeping the existing successful configuration behavior unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/libsy/src/algorithms/escalation.rs`:
- Around line 460-483: Update the local assemble closure in the conversation
assembly flow to construct the header using the current window length on every
invocation, rather than capturing the pre-loop header. Ensure each call after
window.remove(0) reports the actual number of messages included while preserving
the existing turn, total-count, anchors, and truncation behavior.
In `@crates/libsy/src/prompts/escalation/prompt.md`:
- Line 14: Update every escalation hold example in the prompt, including the
examples near the referenced lines, so each {"escalate": false} response also
includes a concise reason field. Keep the existing escalation behavior and
wording otherwise unchanged, and ensure no hold example omits the required
reason.
---
Nitpick comments:
In `@crates/libsy/src/algorithms/escalation.rs`:
- Around line 327-350: Refactor the escalation flow around call_tier to accept
an optional prebuilt Decision, allowing the affinity latch to use it before
serving while keeping driver.info and call_llm_target centralized in call_tier.
Update the escalation caller to construct and pass the capable-model decision,
and remove its duplicated publish/serve sequence while preserving existing
behavior for calls without a prebuilt decision.
- Around line 83-95: Update EscalationPolicy::to_classification to surface the
parsed EscalationVerdict::reason via tracing::info! before returning the
Classification. Remove the unnecessary dead-code allowance on reason, and keep
EscalationDecision unchanged unless the implementation is instead extended to
carry the reason into tracing.
In `@crates/libsy/src/algorithms/util/llm_judge.rs`:
- Around line 174-201: Add focused unit tests for the shared load_judge_config
function covering invalid schema JSON and a schema missing the
/json_schema/schema pointer. Assert both cases return LibsyError::AlgorithmError
with the expected messages, while keeping the existing successful configuration
behavior unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 18a50d5d-1568-4c23-b149-1960647d06aa
📒 Files selected for processing (8)
crates/libsy/src/algorithms.rscrates/libsy/src/algorithms/escalation.rscrates/libsy/src/algorithms/llm_class.rscrates/libsy/src/algorithms/util.rscrates/libsy/src/algorithms/util/llm_judge.rscrates/libsy/src/lib.rscrates/libsy/src/prompts/escalation/prompt.mdcrates/libsy/src/prompts/escalation/schema.json
| let header = format!( | ||
| "Conversation turn {turn}; showing the last {} of {} messages after the task framing.", | ||
| window.len(), | ||
| messages.len(), | ||
| ); | ||
| let assemble = |window: &[String]| { | ||
| std::iter::once(header.as_str()) | ||
| .chain(anchors.iter().map(String::as_str)) | ||
| .chain(window.iter().map(String::as_str)) | ||
| .collect::<Vec<_>>() | ||
| .join("\n") | ||
| }; | ||
|
|
||
| let mut text = assemble(&window); | ||
| while text.chars().count() > settings.max_request_chars && !window.is_empty() { | ||
| window.remove(0); | ||
| text = assemble(&window); | ||
| } | ||
| if text.chars().count() > settings.max_request_chars { | ||
| let keep = settings | ||
| .max_request_chars | ||
| .saturating_sub(TRUNCATION_SUFFIX.chars().count() + 1); | ||
| text = text.chars().take(keep).collect::<String>() + TRUNCATION_SUFFIX; | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Header message count goes stale after the char-cap drop loop.
header is computed from window.len() before the loop at Lines 474-477 drops the oldest window lines. Whenever max_request_chars forces a drop, the transcript tells the judge it is seeing the last N messages while only N - dropped are present — precisely the "pace" signal the doc comment says the header exists to convey. Recompute it inside assemble.
🐛 Proposed fix
- let header = format!(
- "Conversation turn {turn}; showing the last {} of {} messages after the task framing.",
- window.len(),
- messages.len(),
- );
- let assemble = |window: &[String]| {
- std::iter::once(header.as_str())
- .chain(anchors.iter().map(String::as_str))
- .chain(window.iter().map(String::as_str))
- .collect::<Vec<_>>()
- .join("\n")
- };
+ let total = messages.len();
+ let assemble = |window: &[String]| {
+ let header = format!(
+ "Conversation turn {turn}; showing the last {} of {total} messages after the task framing.",
+ window.len(),
+ );
+ std::iter::once(header)
+ .chain(anchors.iter().cloned())
+ .chain(window.iter().cloned())
+ .collect::<Vec<_>>()
+ .join("\n")
+ };📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let header = format!( | |
| "Conversation turn {turn}; showing the last {} of {} messages after the task framing.", | |
| window.len(), | |
| messages.len(), | |
| ); | |
| let assemble = |window: &[String]| { | |
| std::iter::once(header.as_str()) | |
| .chain(anchors.iter().map(String::as_str)) | |
| .chain(window.iter().map(String::as_str)) | |
| .collect::<Vec<_>>() | |
| .join("\n") | |
| }; | |
| let mut text = assemble(&window); | |
| while text.chars().count() > settings.max_request_chars && !window.is_empty() { | |
| window.remove(0); | |
| text = assemble(&window); | |
| } | |
| if text.chars().count() > settings.max_request_chars { | |
| let keep = settings | |
| .max_request_chars | |
| .saturating_sub(TRUNCATION_SUFFIX.chars().count() + 1); | |
| text = text.chars().take(keep).collect::<String>() + TRUNCATION_SUFFIX; | |
| } | |
| let total = messages.len(); | |
| let assemble = |window: &[String]| { | |
| let header = format!( | |
| "Conversation turn {turn}; showing the last {} of {total} messages after the task framing.", | |
| window.len(), | |
| ); | |
| std::iter::once(header) | |
| .chain(anchors.iter().cloned()) | |
| .chain(window.iter().cloned()) | |
| .collect::<Vec<_>>() | |
| .join("\n") | |
| }; | |
| let mut text = assemble(&window); | |
| while text.chars().count() > settings.max_request_chars && !window.is_empty() { | |
| window.remove(0); | |
| text = assemble(&window); | |
| } | |
| if text.chars().count() > settings.max_request_chars { | |
| let keep = settings | |
| .max_request_chars | |
| .saturating_sub(TRUNCATION_SUFFIX.chars().count() + 1); | |
| text = text.chars().take(keep).collect::<String>() + TRUNCATION_SUFFIX; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/libsy/src/algorithms/escalation.rs` around lines 460 - 483, Update the
local assemble closure in the conversation assembly flow to construct the header
using the current window length on every invocation, rather than capturing the
pre-loop header. Ensure each call after window.remove(0) reports the actual
number of messages included while preserving the existing turn, total-count,
anchors, and truncation behavior.
|
|
||
| Escalation is one-way for the rest of the task and expensive. Escalate | ||
| only on a clear PATTERN of trouble, never on a single failed command. | ||
| When the evidence is thin or ambiguous, return {"escalate": false}. |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Hold examples omit the required reason field.
Lines 14, 141, 163 and 165 show {"escalate": false} with no reason, but schema.json lists reason in required and EscalationVerdict.reason is a non-optional String with deny_unknown_fields. On providers where the response schema is advisory rather than enforced, a model copying these examples produces a reply that fails to parse and burns a judge call before falling open. Add a short reason to every example.
📝 Example fix
-When the evidence is thin or ambiguous, return {"escalate": false}.
+When the evidence is thin or ambiguous, return
+{"escalate": false, "reason": "evidence too thin to justify escalation"}.Also applies to: 138-173
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/libsy/src/prompts/escalation/prompt.md` at line 14, Update every
escalation hold example in the prompt, including the examples near the
referenced lines, so each {"escalate": false} response also includes a concise
reason field. Keep the existing escalation behavior and wording otherwise
unchanged, and ensure no hold example omits the required reason.
…s (SWITCH-1054) Signed-off-by: Lin Jia <linj@nvidia.com>
…ration Signed-off-by: Lin Jia <linj@nvidia.com>
Signed-off-by: Lin Jia <linj@nvidia.com>
Signed-off-by: Lin Jia <linj@nvidia.com>
Signed-off-by: Lin Jia <linj@nvidia.com>
…mpletion is empty Signed-off-by: Lin Jia <linj@nvidia.com>
| /// Completion budget for one judge reply, covering any reasoning the judge model emits | ||
| /// alongside the verdict. | ||
| /// | ||
| /// A runaway guard, not a budget: output tokens cost what they generate, so a generous cap is | ||
| /// nearly free, while a tight one truncates mid-reasoning into unparseable JSON and fails the | ||
| /// judge open on every call. Sized well above the verdict itself — the benchmarked judge's | ||
| /// `reason` ran ~40 tokens at the median and ~136 at its longest — leaving the remainder as | ||
| /// reasoning headroom. |
There was a problem hiding this comment.
do we need so many lines of comments for 1 variable ? Can we just do it either 1 or 2 lines
3 participants
Implements SWITCH-1054. Adds EscalationRouter to libsy: calls the efficient model first, runs a judge against its response, and escalates to the capable model if quality is insufficient. Escalated sessions are pinned to the capable model for subsequent turns.
Summary by CodeRabbit
New Features
Bug Fixes
Documentation