i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-01 17:00:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(adaptive): 完善自适应 RPM 学习并将 Pool 调度状态与健康分解耦

Browse Source 
- orchestration 新增 AdaptiveSuccess 效果，在成功回报路径上根据利用率窗口扩张 learned_rpm_limit
- 429 路径改用 429_observation/adjustment 记录以及基于历史的置信度评估，新增 last_rpm_peak 边界字段
- Pool 调度状态不再因 health_score 低或熔断而降级/拦截，前端同步移除相关按钮与文案兜底
- 新增前端 poolTrace 工具（附测试）承接原 HorizontalRequestTimeline 内的候选合并逻辑
This commit is contained in:
fawney19
2026-04-20 00:27:48 +08:00
parent 1f74e660de
commit c1b9d94c84
16 changed files with 1776 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
apps/aether-gateway/src/execution_runtime/stream/execution.rs
View File

@@ -67,9 +67,10 @@ use crate::execution_runtime::{
use crate::execution_runtime::{MAX_STREAM_PREFETCH_BYTES, MAX_STREAM_PREFETCH_FRAMES};
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect,
LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
LocalHealthSuccessEffect, LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect,
LocalAttemptFailureEffect, LocalExecutionEffect, LocalExecutionEffectContext,
LocalHealthFailureEffect, LocalHealthSuccessEffect, LocalOAuthInvalidationEffect,
LocalPoolErrorEffect,
};
use crate::request_candidate_runtime::{
ensure_execution_request_candidate_slot, record_local_request_candidate_status,
@@ -1668,6 +1669,15 @@ async fn execute_stream_from_frame_stream(
LocalExecutionEffect::HealthSuccess(LocalHealthSuccessEffect),
)
.await;
apply_local_execution_effect(
&state_for_report,
LocalExecutionEffectContext {
plan: &plan_for_report,
report_context: report_context_owned.as_ref(),
},
LocalExecutionEffect::AdaptiveSuccess(LocalAdaptiveSuccessEffect),
)
.await;
apply_local_execution_effect(
&state_for_report,
LocalExecutionEffectContext {

16
apps/aether-gateway/src/execution_runtime/sync/execution.rs
View File

@@ -36,9 +36,10 @@ use crate::execution_runtime::{
};
use crate::log_ids::short_request_id;
use crate::orchestration::{
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect,
LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
LocalHealthSuccessEffect, LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect,
LocalAttemptFailureEffect, LocalExecutionEffect, LocalExecutionEffectContext,
LocalHealthFailureEffect, LocalHealthSuccessEffect, LocalOAuthInvalidationEffect,
LocalPoolErrorEffect,
};
use crate::request_candidate_runtime::{
ensure_execution_request_candidate_slot, record_local_request_candidate_status,
@@ -467,6 +468,15 @@ pub(crate) async fn execute_execution_runtime_sync(
LocalExecutionEffect::HealthSuccess(LocalHealthSuccessEffect),
)
.await;
apply_local_execution_effect(
state,
LocalExecutionEffectContext {
plan: &plan,
report_context: report_context.as_ref(),
},
LocalExecutionEffect::AdaptiveSuccess(LocalAdaptiveSuccessEffect),
)
.await;
apply_local_execution_effect(
state,
LocalExecutionEffectContext {

34
apps/aether-gateway/src/handlers/admin/provider/pool_admin/payloads.rs
View File

@@ -672,8 +672,6 @@ fn admin_pool_scheduling_payload(
key: &StoredProviderCatalogKey,
cooldown_reason: Option<&str>,
cooldown_ttl_seconds: Option<u64>,
health_score: f64,
circuit_breaker_open: bool,
account_blocked: bool,
account_status_code: Option<&str>,
account_status_label: Option<&str>,
@@ -741,36 +739,6 @@ fn admin_pool_scheduling_payload(
})],
);
}
if circuit_breaker_open {
return (
"degraded".to_string(),
"circuit_breaker".to_string(),
"熔断中".to_string(),
vec![json!({
"code": "circuit_breaker",
"label": "熔断中",
"blocking": true,
"source": "health",
"ttl_seconds": serde_json::Value::Null,
"detail": serde_json::Value::Null,
})],
);
}
if health_score < 0.5 {
return (
"degraded".to_string(),
"health_low".to_string(),
"健康度较低".to_string(),
vec![json!({
"code": "health_low",
"label": "健康度较低",
"blocking": false,
"source": "health",
"ttl_seconds": serde_json::Value::Null,
"detail": serde_json::Value::Null,
})],
);
}
(
"available".to_string(),
"available".to_string(),
@@ -857,8 +825,6 @@ pub(super) fn build_admin_pool_key_payload(
key,
cooldown_reason.as_deref(),
cooldown_ttl_seconds,
health_score,
circuit_breaker_open,
account_status_blocked,
account_status_code.as_deref(),
account_status_label.as_deref(),

3
apps/aether-gateway/src/handlers/admin/request/system/adaptive.rs
View File

@@ -139,6 +139,9 @@ impl<'a> AdminAppState<'a> {
key.last_429_at_unix_secs = None;
key.last_429_type = None;
key.adjustment_history = None;
key.utilization_samples = None;
key.last_probe_increase_at_unix_secs = None;
key.last_rpm_peak = None;
let Some(updated) = self.update_provider_catalog_key(&key).await? else {
return Ok(admin_adaptive_key_not_found_response(key_id));
};

928
apps/aether-gateway/src/orchestration/adaptive.rs
View File

File diff suppressed because it is too large Load Diff

215
apps/aether-gateway/src/orchestration/effects.rs
View File

@@ -2,7 +2,9 @@ use std::collections::BTreeMap;
use aether_admin::provider::quota as admin_provider_quota_pure;
use aether_contracts::{ExecutionPlan, ExecutionTelemetry};
use aether_scheduler_core::build_scheduler_affinity_cache_key_for_api_key_id;
use aether_scheduler_core::{
build_scheduler_affinity_cache_key_for_api_key_id, count_recent_rpm_requests_for_provider_key,
};
use aether_usage_runtime::{
build_stream_terminal_usage_outcome, build_sync_terminal_usage_outcome,
GatewayStreamReportRequest, GatewaySyncReportRequest, TerminalUsageOutcome,
@@ -11,8 +13,9 @@ use serde_json::Value;
use tracing::warn;
use super::{
local_failover_error_message, project_local_adaptive_rate_limit, project_local_failure_health,
project_local_success_health, LocalFailoverClassification,
local_failover_error_message, project_local_adaptive_rate_limit,
project_local_adaptive_success, project_local_failure_health, project_local_success_health,
LocalFailoverClassification,
};
use crate::ai_pipeline::extract_pool_sticky_session_token;
use crate::clock::current_unix_secs;
@@ -59,6 +62,9 @@ pub(crate) struct LocalHealthFailureEffect {
#[derive(Debug, Clone, Copy)]
pub(crate) struct LocalHealthSuccessEffect;
#[derive(Debug, Clone, Copy)]
pub(crate) struct LocalAdaptiveSuccessEffect;
#[derive(Debug, Clone, Copy)]
pub(crate) struct LocalOAuthInvalidationEffect<'a> {
pub(crate) status_code: u16,
@@ -71,6 +77,7 @@ pub(crate) enum LocalExecutionEffect<'a> {
AdaptiveRateLimit(LocalAdaptiveRateLimitEffect<'a>),
HealthFailure(LocalHealthFailureEffect),
HealthSuccess(LocalHealthSuccessEffect),
AdaptiveSuccess(LocalAdaptiveSuccessEffect),
OauthInvalidation(LocalOAuthInvalidationEffect<'a>),
PoolSuccessSync {
payload: &'a GatewaySyncReportRequest,
@@ -88,6 +95,8 @@ struct PoolFeedbackContext {
sticky_session_token: Option<String>,
}
const ADAPTIVE_RPM_RECENT_CANDIDATE_LIMIT: usize = 512;
pub(crate) async fn apply_local_execution_effect(
state: &AppState,
context: LocalExecutionEffectContext<'_>,
@@ -106,6 +115,9 @@ pub(crate) async fn apply_local_execution_effect(
LocalExecutionEffect::HealthSuccess(effect) => {
record_health_success_effect(state, context, effect).await;
}
LocalExecutionEffect::AdaptiveSuccess(effect) => {
record_adaptive_success_effect(state, context, effect).await;
}
LocalExecutionEffect::OauthInvalidation(effect) => {
record_oauth_invalidation_effect(state, context, effect).await;
}
@@ -256,6 +268,7 @@ async fn record_adaptive_rate_limit_effect(
context: LocalExecutionEffectContext<'_>,
effect: LocalAdaptiveRateLimitEffect<'_>,
) {
let observed_at_unix_secs = current_unix_secs();
let Some(current_key) = state
.read_provider_catalog_keys_by_ids(std::slice::from_ref(&context.plan.key_id))
.await
@@ -264,22 +277,39 @@ async fn record_adaptive_rate_limit_effect(
else {
return;
};
let current_rpm = state
.read_recent_request_candidates(ADAPTIVE_RPM_RECENT_CANDIDATE_LIMIT)
.await
.ok()
.map(|recent_candidates| {
count_recent_rpm_requests_for_provider_key(
&recent_candidates,
&context.plan.key_id,
observed_at_unix_secs,
) as u32
});
let Some(projection) = project_local_adaptive_rate_limit(
&current_key,
effect.classification,
effect.status_code,
current_rpm,
effect.headers,
current_unix_secs(),
observed_at_unix_secs,
) else {
return;
};
let mut updated_key = current_key.clone();
updated_key.rpm_429_count = Some(projection.rpm_429_count);
updated_key.rpm_429_count = Some(projection.rpm_429_count).filter(|value| *value > 0);
updated_key.learned_rpm_limit = projection.learned_rpm_limit;
updated_key.last_429_at_unix_secs = Some(projection.last_429_at_unix_secs);
updated_key.last_429_type = Some(projection.last_429_type);
updated_key.adjustment_history = projection.adjustment_history;
updated_key.utilization_samples = projection.utilization_samples;
updated_key.last_probe_increase_at_unix_secs = projection.last_probe_increase_at_unix_secs;
updated_key.last_rpm_peak = projection.last_rpm_peak;
updated_key.status_snapshot = Some(projection.status_snapshot);
updated_key.updated_at_unix_secs = Some(current_unix_secs());
updated_key.updated_at_unix_secs = Some(observed_at_unix_secs);
if let Err(err) = state.update_provider_catalog_key(&updated_key).await {
warn!(
@@ -289,6 +319,54 @@ async fn record_adaptive_rate_limit_effect(
}
}
async fn record_adaptive_success_effect(
state: &AppState,
context: LocalExecutionEffectContext<'_>,
_effect: LocalAdaptiveSuccessEffect,
) {
let observed_at_unix_secs = current_unix_secs();
let Some(current_key) = state
.read_provider_catalog_keys_by_ids(std::slice::from_ref(&context.plan.key_id))
.await
.ok()
.and_then(|mut keys| keys.drain(..).next())
else {
return;
};
let Some(recent_candidates) = state
.read_recent_request_candidates(ADAPTIVE_RPM_RECENT_CANDIDATE_LIMIT)
.await
.ok()
else {
return;
};
let current_rpm = count_recent_rpm_requests_for_provider_key(
&recent_candidates,
&context.plan.key_id,
observed_at_unix_secs,
) as u32;
let Some(projection) =
project_local_adaptive_success(&current_key, current_rpm, observed_at_unix_secs)
else {
return;
};
let mut updated_key = current_key.clone();
updated_key.learned_rpm_limit = projection.learned_rpm_limit;
updated_key.adjustment_history = projection.adjustment_history;
updated_key.utilization_samples = projection.utilization_samples;
updated_key.last_probe_increase_at_unix_secs = projection.last_probe_increase_at_unix_secs;
updated_key.status_snapshot = Some(projection.status_snapshot);
updated_key.updated_at_unix_secs = Some(observed_at_unix_secs);
if let Err(err) = state.update_provider_catalog_key(&updated_key).await {
warn!(
"gateway orchestration effects: failed to persist adaptive success projection for provider {} endpoint {} key {}: {:?}",
context.plan.provider_id, context.plan.endpoint_id, context.plan.key_id, err
);
}
}
async fn record_health_failure_effect(
state: &AppState,
context: LocalExecutionEffectContext<'_>,
@@ -540,7 +618,11 @@ mod tests {
use aether_contracts::{ExecutionPlan, RequestBody};
use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY};
use aether_data::repository::candidates::InMemoryRequestCandidateRepository;
use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
use aether_data_contracts::repository::candidates::{
RequestCandidateStatus, StoredRequestCandidate,
};
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
};
@@ -548,9 +630,9 @@ mod tests {
use super::{
apply_local_execution_effect, local_candidate_failure_should_record_pool_error,
LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect, LocalExecutionEffect,
LocalExecutionEffectContext, LocalHealthFailureEffect, LocalHealthSuccessEffect,
LocalOAuthInvalidationEffect,
LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect, LocalAttemptFailureEffect,
LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
LocalHealthSuccessEffect, LocalOAuthInvalidationEffect,
};
use crate::data::GatewayDataState;
use crate::orchestration::LocalFailoverClassification;
@@ -791,6 +873,26 @@ mod tests {
)
}
fn adaptive_state_with_request_candidates(
key: StoredProviderCatalogKey,
request_candidates: Vec<StoredRequestCandidate>,
) -> AppState {
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_health_provider()],
vec![sample_health_endpoint()],
vec![key],
));
let request_candidates =
Arc::new(InMemoryRequestCandidateRepository::seed(request_candidates));
AppState::new()
.expect("gateway state should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(provider_catalog)
.with_request_candidate_reader(request_candidates)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
)
}
fn fixed_limit_state() -> AppState {
let mut key = sample_health_key();
key.rpm_limit = Some(24);
@@ -1168,7 +1270,7 @@ mod tests {
.status_snapshot
.as_ref()
.and_then(|value| value.get("learning_confidence")),
Some(&json!(0.283))
Some(&json!(0.3))
);
assert_eq!(
stored_key
@@ -1212,4 +1314,97 @@ mod tests {
assert_eq!(stored_key.last_429_at_unix_secs, None);
assert_eq!(stored_key.last_429_type, None);
}
#[tokio::test]
async fn adaptive_success_effect_expands_limit_from_recent_rpm_usage() {
let now_unix_secs = chrono::Utc::now().timestamp().max(0) as u64;
let mut key = sample_adaptive_key();
key.learned_rpm_limit = Some(20);
key.last_rpm_peak = Some(25);
key.last_429_at_unix_secs = Some(now_unix_secs.saturating_sub(600));
key.adjustment_history = Some(json!([
{
"timestamp": "2026-04-19T00:00:00Z",
"old_limit": 0,
"new_limit": 20,
"reason": "rpm_429",
"confidence": 0.8
}
]));
key.utilization_samples = Some(json!([
{"ts": now_unix_secs.saturating_sub(40), "util": 0.90},
{"ts": now_unix_secs.saturating_sub(30), "util": 0.95},
{"ts": now_unix_secs.saturating_sub(20), "util": 0.85},
{"ts": now_unix_secs.saturating_sub(10), "util": 0.80}
]));
let state = adaptive_state_with_request_candidates(
key,
vec![StoredRequestCandidate::new(
"candidate-1".to_string(),
"req-1".to_string(),
None,
None,
None,
None,
0,
0,
Some("prov-1".to_string()),
Some("ep-1".to_string()),
Some("key-1".to_string()),
RequestCandidateStatus::Success,
None,
false,
Some(200),
None,
None,
Some(10),
Some(19),
None,
None,
i64::try_from(now_unix_secs.saturating_sub(30) * 1000)
.expect("candidate created_at should fit i64"),
Some(
i64::try_from(now_unix_secs.saturating_sub(30) * 1000)
.expect("candidate started_at should fit i64"),
),
Some(
i64::try_from(now_unix_secs.saturating_sub(29) * 1000)
.expect("candidate finished_at should fit i64"),
),
)
.expect("request candidate should build")],
);
let plan = sample_plan();
apply_local_execution_effect(
&state,
LocalExecutionEffectContext {
plan: &plan,
report_context: None,
},
LocalExecutionEffect::AdaptiveSuccess(LocalAdaptiveSuccessEffect),
)
.await;
let stored_key = state
.read_provider_catalog_keys_by_ids(std::slice::from_ref(&plan.key_id))
.await
.expect("provider catalog keys should load")
.into_iter()
.next()
.expect("stored key should exist");
assert_eq!(stored_key.learned_rpm_limit, Some(25));
assert_eq!(stored_key.utilization_samples, Some(json!([])));
assert_eq!(
stored_key
.adjustment_history
.as_ref()
.and_then(Value::as_array)
.and_then(|items| items.last())
.and_then(Value::as_object)
.and_then(|record| record.get("reason"))
.and_then(Value::as_str),
Some("high_utilization")
);
}
}

10
apps/aether-gateway/src/orchestration/mod.rs
View File

@@ -12,7 +12,8 @@ mod recovery;
mod report_effects;
pub(crate) use self::adaptive::{
project_local_adaptive_rate_limit, LocalAdaptiveRateLimitProjection,
project_local_adaptive_rate_limit, project_local_adaptive_success,
LocalAdaptiveRateLimitProjection, LocalAdaptiveSuccessProjection,
};
pub(crate) use self::attempt::{
attempt_identity_from_report_context, build_local_attempt_identities,
@@ -24,9 +25,10 @@ pub(crate) use self::classifier::{
LocalFailoverInput,
};
pub(crate) use self::effects::{
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAttemptFailureEffect,
LocalExecutionEffect, LocalExecutionEffectContext, LocalHealthFailureEffect,
LocalHealthSuccessEffect, LocalOAuthInvalidationEffect, LocalPoolErrorEffect,
apply_local_execution_effect, LocalAdaptiveRateLimitEffect, LocalAdaptiveSuccessEffect,
LocalAttemptFailureEffect, LocalExecutionEffect, LocalExecutionEffectContext,
LocalHealthFailureEffect, LocalHealthSuccessEffect, LocalOAuthInvalidationEffect,
LocalPoolErrorEffect,
};
pub(crate) use self::health::{project_local_failure_health, project_local_success_health};
pub(crate) use self::policy::{

87
apps/aether-gateway/src/tests/control/admin/pool.rs
View File

@@ -761,6 +761,93 @@ async fn gateway_marks_account_blocked_pool_key_in_list_keys_response() {
);
}
#[tokio::test]
async fn gateway_ignores_health_signals_in_pool_scheduling_status() {
let mut provider = sample_provider("provider-openai", "openai", 10).with_transport_fields(
true,
false,
true,
None,
None,
None,
None,
None,
Some(json!({
"pool_advanced": {
"enabled": true
}
})),
);
provider.provider_type = "openai".to_string();
let mut circuit_key = sample_key(
"key-openai-circuit",
"provider-openai",
"openai:chat",
"sk-circuit",
);
circuit_key.name = "circuit-open".to_string();
circuit_key.circuit_breaker_by_format = Some(json!({
"openai:chat": {
"open": true
}
}));
let mut low_health_key = sample_key(
"key-openai-health",
"provider-openai",
"openai:chat",
"sk-health",
);
low_health_key.name = "low-health".to_string();
low_health_key.health_by_format = Some(json!({
"openai:chat": {
"health_score": 0.2
}
}));
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![circuit_key, low_health_key],
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
));
let response = local_admin_pool_response(
&state,
http::Method::GET,
"/api/admin/pool/provider-openai/keys?page=1&page_size=50&status=all",
None,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
let keys = payload["keys"].as_array().expect("keys should be array");
assert_eq!(keys.len(), 2);
assert_eq!(keys[0]["key_name"], json!("circuit-open"));
assert_eq!(keys[0]["circuit_breaker_open"], json!(true));
assert_eq!(keys[0]["scheduling_status"], json!("available"));
assert_eq!(keys[0]["scheduling_reason"], json!("available"));
assert_eq!(keys[0]["scheduling_label"], json!("可用"));
assert_eq!(keys[1]["key_name"], json!("low-health"));
assert_eq!(keys[1]["health_score"], json!(0.2));
assert_eq!(keys[1]["scheduling_status"], json!("available"));
assert_eq!(keys[1]["scheduling_reason"], json!("available"));
assert_eq!(keys[1]["scheduling_label"], json!("可用"));
}
#[tokio::test]
async fn gateway_handles_admin_pool_list_keys_with_quota_compatibility_fields() {
let upstream_hits = Arc::new(Mutex::new(0usize));