Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

59 changes: 39 additions & 20 deletions crates/libsy-llm-client/src/run.rs
Original file line number Diff line number Diff line change
Expand Up @@ -62,17 +62,17 @@ pub async fn run(
.and_then(|outcome| outcome.response.as_ref())
.and_then(Response::served_model);
emit_routing_observations(&observer, &routing_observations, answered_model);
let outcome = outcome?;
let mut outcome = outcome?;
let overhead = run_started.elapsed();
metrics::record_routing_overhead(&algorithm_name, overhead);

let selected_model_id = outcome.selected_model_id;
let (result, answer_duration) = if let Some(response) = outcome.response {
let selected_model_id = outcome.selected_model_id.clone();
let (result, answer_duration) = if let Some(response) = outcome.response.take() {
(Ok(response), None)
} else {
let mut models = Vec::with_capacity(1 + outcome.fallback_models.len());
models.push(selected_model_id.clone());
models.extend(outcome.fallback_models);
models.extend(outcome.fallback_models.iter().cloned());
let answer_started = Instant::now();
let observe = |observation| {
if let Some(observer) = &observer {
Expand All @@ -82,8 +82,8 @@ pub async fn run(
let result = call_first_available(
&clients,
&algorithm_name,
&outcome.request,
&models,
move |target| outcome.request_for(target),
&observe,
)
.await;
Expand Down Expand Up @@ -142,8 +142,8 @@ async fn serve(
let result = call_first_available(
&clients,
&call.algorithm,
&call.request,
&call.models,
|target| call.request_for(target),
&observe,
)
.await;
Expand All @@ -154,12 +154,12 @@ async fn serve(
async fn call_first_available(
clients: &ClientRouter,
algorithm: &str,
request: &Request,
models: &[ModelId],
request_for: impl Fn(&ModelId) -> Result<Request> + Send,
observe: &(dyn Fn(LlmCallObservation) + Send + Sync),
) -> Result<Response> {
for (index, target) in models.iter().enumerate() {
let request = request_for(request, target);
let request = request_for(target)?;
match call_one(
clients,
target,
Expand Down Expand Up @@ -298,13 +298,6 @@ fn fallback_reason(error: &LibsyError) -> Option<RoutingFallbackReason> {
}
}

/// Clone a request and stamp the candidate model that should receive it.
fn request_for(request: &Request, target: &ModelId) -> Request {
let mut request = request.clone();
request.llm_request.model = Some(target.to_string());
request
}

/// Resolves a routed call's selected model to the client that serves it.
///
/// An algorithm routes among named targets; which provider each target lives on is the
Expand Down Expand Up @@ -379,10 +372,10 @@ mod tests {
use async_trait::async_trait;
use futures::StreamExt;
use http::StatusCode;
use switchyard_libsy::{Driver, RoutingOutcome};
use switchyard_libsy::{Driver, RoutingOutcome, TargetPrompts, with_target_prompts};
use switchyard_protocol::{
LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, completion_text, text_request,
text_response,
ContentBlock, LlmResponse, LlmResponseChunk, LlmResponseStreamEvent, completion_text,
text_request, text_response,
};
use wiremock::matchers::method;
use wiremock::{Mock, MockServer, ResponseTemplate};
Expand Down Expand Up @@ -429,7 +422,7 @@ mod tests {
request: Request,
) -> Result<RoutingOutcome> {
let response = driver
.call_model(request.clone(), vec![self.model.clone()])
.call_answer_model(request.clone(), self.model.clone())
.await?;
Ok(RoutingOutcome::answered(
self.model.clone(),
Expand All @@ -449,6 +442,7 @@ mod tests {

struct CandidateClient {
calls: Mutex<Vec<ModelId>>,
prompts: Mutex<Vec<Vec<String>>>,
first: FirstOutcome,
}

Expand All @@ -457,6 +451,18 @@ mod tests {
async fn call(&self, request: Request) -> std::result::Result<Response, LlmClientError> {
let model = request.model_id().unwrap_or_default();
self.calls.lock().push(model.clone());
self.prompts.lock().push(
request
.llm_request
.instructions
.iter()
.flat_map(|instruction| &instruction.content)
.filter_map(|block| match block {
ContentBlock::Text { text } => Some(text.clone()),
_ => None,
})
.collect(),
);
if model == "weak" {
return match self.first {
FirstOutcome::ContextWindow => Err(LlmClientError::ContextWindowExceeded {
Expand Down Expand Up @@ -521,11 +527,16 @@ mod tests {
) -> (Arc<CandidateClient>, Result<(ModelId, Response)>) {
let client = Arc::new(CandidateClient {
calls: Mutex::new(Vec::new()),
prompts: Mutex::new(Vec::new()),
first,
});
let algorithm = Arc::new(CandidateAlgorithm {
let inner: Arc<dyn Algorithm> = Arc::new(CandidateAlgorithm {
models: vec!["weak".into(), "strong".into()],
});
let prompts = TargetPrompts::default()
.with("weak", "weak prompt")
.with("strong", "strong prompt");
let algorithm = with_target_prompts(inner, prompts);
let result = run(
algorithm,
ClientRouter::single(client.clone()),
Expand All @@ -540,6 +551,7 @@ mod tests {
async fn answered_outcome_does_not_make_a_second_model_call() -> Result<()> {
let client = Arc::new(CandidateClient {
calls: Mutex::new(Vec::new()),
prompts: Mutex::new(Vec::new()),
first: FirstOutcome::StreamSuccess,
});
let observations = Arc::new(Mutex::new(Vec::new()));
Expand Down Expand Up @@ -621,6 +633,13 @@ mod tests {
&*client.calls.lock(),
&[ModelId::from("weak"), "strong".into()]
);
assert_eq!(
&*client.prompts.lock(),
&[
vec!["weak prompt".to_string()],
vec!["strong prompt".to_string()]
]
);
assert_eq!(
response
.llm_response
Expand Down
1 change: 1 addition & 0 deletions crates/libsy/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ parking_lot.workspace = true
rand.workspace = true
regex.workspace = true
switchyard-protocol.workspace = true
switchyard-translation.workspace = true
thiserror.workspace = true
tokio.workspace = true
tokio-stream = "0.1"
Expand Down
9 changes: 9 additions & 0 deletions crates/libsy/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,15 @@ fallbacks, rewritten request, and an optional response already produced while ro
makes no network calls itself — `switchyard-llm-client`'s `run` is a ready-made consumer that
drives the stream and performs the terminal answer call, retries, and fallback over HTTP.

[`RoutingOutcome`]'s `request` field is ready for the selected answer target. A custom host
trying the selected target or a fallback should call [`RoutingOutcome::request_for`]; that
prepares the candidate's model and any prompt configured with [`with_target_prompts`] as one
operation.

Routing-time [`CallModel`] requests are likewise ready for their first candidate. Hosts trying
a later classifier or judge candidate should use [`CallModel::request_for`] so exact provider
bodies receive the candidate model together with the normalized request.

The provider-neutral [`Request`], [`Response`], [`Usage`], and [`LlmResponse`]
contracts come from `switchyard-protocol`.

Expand Down
2 changes: 1 addition & 1 deletion crates/libsy/src/algorithms/advisor_gate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -334,7 +334,7 @@ impl AdvisorGate {
// Gated phase: generate the turn once, fully buffered, so the gate
// can inspect it before the client sees anything.
let response = driver
.call_model(request.clone(), vec![self.executor.clone()])
.call_answer_model(request.clone(), self.executor.clone())
.await?;
let turn = buffer_turn(self.executor.as_str(), response).await?;

Expand Down
21 changes: 20 additions & 1 deletion crates/libsy/src/algorithms/advisor_gate/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ use switchyard_protocol::{
use super::transcript::{NO_TEXT_PLACEHOLDER, TRUNCATION_MARKER, middle_drop};
use super::*;
use crate::core::testing::{reply, test_drive};
use crate::{TargetPrompts, with_target_prompts};

const EXECUTOR: &str = "executor";
const ADVISOR: &str = "advisor";
Expand Down Expand Up @@ -268,7 +269,12 @@ async fn tool_call_turn_replays_without_review() {
#[tokio::test]
async fn approved_terminal_turn_returns_buffered_body() {
let script = Script::new();
let gate = gate(AdvisorGateConfig::default());
let gate = with_target_prompts(
gate(AdvisorGateConfig::default()),
TargetPrompts::default()
.with(EXECUTOR, "executor prompt")
.with(ADVISOR, "answer-only advisor prompt"),
);
let serve = script.serve("APPROVE", |_| reply("all done"));
let (selected_model, response) = test_drive(gate, task_request(), serve)
.await
Expand All @@ -279,6 +285,19 @@ async fn approved_terminal_turn_returns_buffered_body() {
);
assert_eq!(completion_text(&agg_of(response).await), "all done");
assert_eq!(selected_model, EXECUTOR);
let executor = script.call(0);
assert_eq!(
executor.llm_request.instructions[0].content,
vec![ContentBlock::Text {
text: "executor prompt".to_string(),
}]
);
let advisor = script.call(1);
assert!(!advisor.llm_request.instructions.iter().any(|instruction| {
instruction.content.iter().any(|block| {
matches!(block, ContentBlock::Text { text } if text == "answer-only advisor prompt")
})
}));
}

#[tokio::test]
Expand Down
40 changes: 36 additions & 4 deletions crates/libsy/src/algorithms/llm_class.rs
Original file line number Diff line number Diff line change
Expand Up @@ -522,7 +522,7 @@ impl Classifier<State> for EscalationClassifier {
"escalation classifier selected efficient tier"
);
let efficient_response = match driver
.call_model(request.clone(), vec![self.efficient.clone()])
.call_answer_model(request.clone(), self.efficient.clone())
.await
{
Ok(r) => r,
Expand Down Expand Up @@ -1777,6 +1777,16 @@ mod tests {
}
}

/// Reports whether a request contains `expected` as an instruction text block.
fn has_instruction(request: &Request, expected: &str) -> bool {
request
.llm_request
.instructions
.iter()
.flat_map(|instruction| &instruction.content)
.any(|block| matches!(block, ContentBlock::Text { text } if text == expected))
}

/// Returns a stream that emits partial content before failing during aggregation.
fn streamed_then_error(error: LlmClientError) -> Response {
Response {
Expand Down Expand Up @@ -1814,17 +1824,39 @@ mod tests {
// Judge: no escalation. Expect the efficient response to be returned directly.
let judge = Queue::new([r#"{"escalate":false,"reason":"progressing"}"#]);
let model = Queue::new(["efficient answer"]);
let router = escalation_router()?;
let replies = queued(model, judge);
let prompted = Arc::new(Mutex::new(Vec::new()));
let recorded = Arc::clone(&prompted);
let serve = move |target: ModelId, request: Request| {
let expected = if target == "judge" {
"answer-only judge prompt"
} else {
"efficient prompt"
};
recorded
.lock()
.push((target.clone(), has_instruction(&request, expected)));
replies.serve(target, request)
};
let router = crate::with_target_prompts(
escalation_router()?,
crate::TargetPrompts::default()
.with("efficient", "efficient prompt")
.with("judge", "answer-only judge prompt"),
);

let (selected_model, response) =
test_drive(router, classify_request(), queued(model, judge)).await?;
let (selected_model, response) = test_drive(router, classify_request(), serve).await?;

// The efficient model is the serving target, and the response comes from its call.
assert_eq!(selected_model, "efficient");
assert_eq!(
response.llm_response.as_agg().map(completion_text),
Some("efficient answer".to_string())
);
assert_eq!(
&*prompted.lock(),
&[(ModelId::from("efficient"), true), ("judge".into(), false)]
);
Ok(())
}

Expand Down
Loading