mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
fix: add provider/model failover to streaming LLM calls (#2022)
This commit is contained in:
Sathvik Gilakamsetty
@@ -1,5 +1,5 @@
|
||||
use super::traits::{
|
||||
ChatMessage, ChatRequest, ChatResponse, StreamChunk, StreamOptions, StreamResult,
|
||||
ChatMessage, ChatRequest, ChatResponse, StreamChunk, StreamError, StreamOptions, StreamResult,
|
||||
};
|
||||
use super::Provider;
|
||||
use async_trait::async_trait;
|
||||
@@ -59,6 +59,32 @@ fn is_non_retryable(err: &anyhow::Error) -> bool {
|
||||
|| msg_lower.contains("invalid"))
|
||||
}
|
||||
|
||||
/// Classify a StreamError without losing type information.
|
||||
/// Inspects the inner reqwest::Error status directly for Http variants.
|
||||
fn is_stream_error_non_retryable(err: &StreamError) -> bool {
|
||||
match err {
|
||||
StreamError::Http(reqwest_err) => {
|
||||
if let Some(status) = reqwest_err.status() {
|
||||
let code = status.as_u16();
|
||||
// Client errors except 429 (rate limit) and 408 (timeout) are non-retryable
|
||||
return status.is_client_error() && code != 429 && code != 408;
|
||||
}
|
||||
false
|
||||
}
|
||||
StreamError::Provider(msg) => {
|
||||
let lower = msg.to_lowercase();
|
||||
lower.contains("invalid api key")
|
||||
|| lower.contains("unauthorized")
|
||||
|| lower.contains("forbidden")
|
||||
|| lower.contains("model")
|
||||
&& (lower.contains("not found") || lower.contains("unsupported"))
|
||||
}
|
||||
// JSON/SSE parse errors and IO errors are generally non-retryable
|
||||
StreamError::Json(_) | StreamError::InvalidSse(_) => true,
|
||||
StreamError::Io(_) => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_context_window_exceeded(err: &anyhow::Error) -> bool {
|
||||
let lower = err.to_string().to_lowercase();
|
||||
let hints = [
|
||||
@@ -924,63 +950,126 @@ impl Provider for ReliableProvider {
|
||||
temperature: f64,
|
||||
options: StreamOptions,
|
||||
) -> stream::BoxStream<'static, StreamResult<StreamChunk>> {
|
||||
// Try each provider/model combination for streaming
|
||||
// For streaming, we use the first provider that supports it and has streaming enabled
|
||||
for (provider_name, provider) in &self.providers {
|
||||
if !provider.supports_streaming() || !options.enabled {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Clone provider data for the stream
|
||||
let provider_clone = provider_name.clone();
|
||||
|
||||
// Try the first model in the chain for streaming
|
||||
let current_model = match self.model_chain(model).first() {
|
||||
Some(m) => m.to_string(),
|
||||
None => model.to_string(),
|
||||
};
|
||||
|
||||
// For streaming, we attempt once and propagate errors
|
||||
// The caller can retry the entire request if needed
|
||||
let stream = provider.stream_chat_with_system(
|
||||
system_prompt,
|
||||
message,
|
||||
¤t_model,
|
||||
temperature,
|
||||
options,
|
||||
);
|
||||
|
||||
// Use a channel to bridge the stream with logging
|
||||
let (tx, rx) = tokio::sync::mpsc::channel::<StreamResult<StreamChunk>>(100);
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut stream = stream;
|
||||
while let Some(chunk) = stream.next().await {
|
||||
if let Err(ref e) = chunk {
|
||||
tracing::warn!(
|
||||
provider = provider_clone,
|
||||
model = current_model,
|
||||
"Streaming error: {e}"
|
||||
);
|
||||
}
|
||||
if tx.send(chunk).await.is_err() {
|
||||
break; // Receiver dropped
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Convert channel receiver to stream
|
||||
return stream::unfold(rx, |mut rx| async move {
|
||||
rx.recv().await.map(|chunk| (chunk, rx))
|
||||
if !options.enabled {
|
||||
return stream::once(async move {
|
||||
Err(super::traits::StreamError::Provider(
|
||||
"Streaming disabled".to_string(),
|
||||
))
|
||||
})
|
||||
.boxed();
|
||||
}
|
||||
|
||||
// No streaming support available
|
||||
stream::once(async move {
|
||||
Err(super::traits::StreamError::Provider(
|
||||
"No provider supports streaming".to_string(),
|
||||
))
|
||||
// Collect streaming-capable providers
|
||||
let streaming_providers: Vec<_> = self
|
||||
.providers
|
||||
.iter()
|
||||
.filter(|(_, p)| p.supports_streaming())
|
||||
.collect();
|
||||
|
||||
if streaming_providers.is_empty() {
|
||||
return stream::once(async move {
|
||||
Err(super::traits::StreamError::Provider(
|
||||
"No provider supports streaming".to_string(),
|
||||
))
|
||||
})
|
||||
.boxed();
|
||||
}
|
||||
|
||||
// Build model chain and provider info for the spawned task
|
||||
let models = self.model_chain(model);
|
||||
let model_chain: Vec<String> = models.into_iter().map(|m| m.to_string()).collect();
|
||||
let base_backoff_ms = self.base_backoff_ms;
|
||||
|
||||
// Collect provider streams lazily inside the task — we need owned data
|
||||
// Provider trait is object-safe, so we call stream_chat_with_system per attempt
|
||||
// We need to pre-create all possible streams since Provider is behind &self
|
||||
// Instead, collect the streams for each provider+model combo upfront
|
||||
let mut candidate_streams: Vec<(
|
||||
String,
|
||||
String,
|
||||
stream::BoxStream<'static, StreamResult<StreamChunk>>,
|
||||
)> = Vec::new();
|
||||
for current_model in &model_chain {
|
||||
for (provider_name, provider) in &streaming_providers {
|
||||
let s = provider.stream_chat_with_system(
|
||||
system_prompt,
|
||||
message,
|
||||
current_model,
|
||||
temperature,
|
||||
options,
|
||||
);
|
||||
candidate_streams.push(((*provider_name).clone(), current_model.clone(), s));
|
||||
}
|
||||
}
|
||||
|
||||
let (tx, rx) = tokio::sync::mpsc::channel::<StreamResult<StreamChunk>>(100);
|
||||
let max_retries = self.max_retries;
|
||||
|
||||
tokio::spawn(async move {
|
||||
for (provider_name, current_model, mut candidate_stream) in candidate_streams {
|
||||
let mut backoff_ms = base_backoff_ms;
|
||||
let mut attempts = 0u32;
|
||||
|
||||
loop {
|
||||
match candidate_stream.next().await {
|
||||
Some(Ok(chunk)) => {
|
||||
// First chunk succeeded — commit to this stream
|
||||
if tx.send(Ok(chunk)).await.is_err() {
|
||||
return;
|
||||
}
|
||||
// Forward remaining chunks
|
||||
while let Some(chunk) = candidate_stream.next().await {
|
||||
if tx.send(chunk).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
return; // Done successfully
|
||||
}
|
||||
Some(Err(ref e)) => {
|
||||
let non_retryable = is_stream_error_non_retryable(e);
|
||||
|
||||
tracing::warn!(
|
||||
provider = provider_name,
|
||||
model = current_model,
|
||||
attempt = attempts + 1,
|
||||
error = %e,
|
||||
"Streaming failed{}", if non_retryable { " (non-retryable)" } else { "" }
|
||||
);
|
||||
|
||||
if non_retryable || attempts >= max_retries {
|
||||
break; // Move to next candidate
|
||||
}
|
||||
|
||||
attempts += 1;
|
||||
tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
|
||||
backoff_ms = (backoff_ms.saturating_mul(2)).min(10_000);
|
||||
// Continue inner loop — stream may yield more items
|
||||
}
|
||||
None => {
|
||||
// Stream exhausted without success
|
||||
if attempts == 0 {
|
||||
tracing::warn!(
|
||||
provider = provider_name,
|
||||
model = current_model,
|
||||
"Stream returned empty"
|
||||
);
|
||||
}
|
||||
break; // Move to next candidate
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// All providers/models exhausted
|
||||
let _ = tx
|
||||
.send(Err(super::traits::StreamError::Provider(
|
||||
"All streaming providers/models failed".to_string(),
|
||||
)))
|
||||
.await;
|
||||
});
|
||||
|
||||
stream::unfold(rx, |mut rx| async move {
|
||||
rx.recv().await.map(|chunk| (chunk, rx))
|
||||
})
|
||||
.boxed()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user