diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index b8e9c4275..2e43deac5 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -483,7 +483,7 @@ impl UpstreamHttpVersion { /// /// `first_*_ms` 是相对 `AttemptContext::timing_started_at()` 的请求级偏移， /// 与 Core 总耗时使用同一起点并包含选号与重试等待；其余字段是各阶段独立耗时 -/// 首字要求语义输出，缺失时保留 `None`，不使用首包或空结构帧代替 +/// 首字边界由 Provider 协议定义，缺失时保留 `None`，不使用首包代替 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ProviderResponseTimings { pub transport_decision_wait_ms: Option, diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs b/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs index 3a49aea17..325e5b657 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/dashboard.rs @@ -67,19 +67,19 @@ pub(crate) async fn request_metrics( as account_selection_wait_p99_ms, round((percentile_cont(0.10) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p10, + and latency_ms > 0)))::bigint as output_throughput_p10, round((percentile_cont(0.50) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p50, + and latency_ms > 0)))::bigint as output_throughput_p50, round((percentile_cont(0.90) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p90, + and latency_ms > 0)))::bigint as output_throughput_p90, count(capacity_total_slots)::bigint as capacity_sample_count, round(avg(capacity_used_slots::double precision / nullif(capacity_total_slots, 0)) * 10000)::bigint @@ -220,19 +220,19 @@ async fn request_metric_series_inner( as account_selection_wait_p99_ms, round((percentile_cont(0.10) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p10, + and latency_ms > 0)))::bigint as output_throughput_p10, round((percentile_cont(0.50) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p50, + and latency_ms > 0)))::bigint as output_throughput_p50, round((percentile_cont(0.90) within group ( order by output_tokens::double precision * 1000.0 - / greatest(latency_ms - first_token_ms, 1) + / nullif(latency_ms, 0) ) filter (where {fact} and output_tokens > 0 - and latency_ms > first_token_ms)))::bigint as output_throughput_p90, + and latency_ms > 0)))::bigint as output_throughput_p90, count(capacity_total_slots)::bigint as capacity_sample_count, round(avg(capacity_used_slots::double precision / nullif(capacity_total_slots, 0)) * 10000)::bigint diff --git a/backend/crates/gateway-store/tests/postgres/observability/mod.rs b/backend/crates/gateway-store/tests/postgres/observability/mod.rs index c82f3830a..e4aec4f9c 100644 --- a/backend/crates/gateway-store/tests/postgres/observability/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/observability/mod.rs @@ -79,6 +79,79 @@ fn postgres_admin_observability_adapter_implements_terminal_port() { assert_port::(); } +#[tokio::test] +async fn output_throughput_uses_full_duration_independently_of_first_token() { + let Some(database) = TestDatabase::create("output_throughput_full_duration").await else { + return; + }; + let now = Utc::now(); + seed_observability_facts(&database.pool, now).await.unwrap(); + let range = ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)) + .expect("observability range"); + let repository = observability_repository(&database.pool); + let store = admin_observability_store(&database.pool); + let admin_range = admin_observability::TimeRange::new(range.start, range.end).unwrap(); + + // 输出包含首字前的推理量；首字缺失或仅剩 1 ms 都不应改变整次请求速率 + for (first_token, latency, output, expected) in [ + (Some(17_799_i64), Some(19_216_i64), Some(605_i64), Some(31)), + (None, Some(19_216), Some(605), Some(31)), + (Some(19_215), Some(19_216), Some(605), Some(31)), + (Some(19_216), Some(19_216), Some(605), Some(31)), + (None, Some(0), Some(605), None), + (None, None, Some(605), None), + (None, Some(19_216), Some(0), None), + (None, Some(19_216), None, None), + ] { + sqlx::query( + "update model_requests + set output_tokens = $1, reasoning_tokens = 500, + first_token_ms = $2, latency_ms = $3 + where id = 'req_observe_success'", + ) + .bind(output) + .bind(first_token) + .bind(latency) + .execute(&database.pool) + .await + .unwrap(); + + let summary = repository + .usage_summary(range, UsageRecordFilter::default()) + .await + .expect("throughput summary"); + assert_eq!(summary.requests.output_throughput_p10, expected); + assert_eq!(summary.requests.output_throughput_p50, expected); + assert_eq!(summary.requests.output_throughput_p90, expected); + + let usage = store + .usage_trend(admin_range, admin_observability::UsageFilter::default()) + .await + .expect("usage throughput trend"); + let dashboard = store + .dashboard_trend(admin_range) + .await + .expect("dashboard throughput trend"); + for trend in [usage, dashboard] { + for value in [ + trend + .iter() + .find_map(|point| point.metrics.output_throughput_p10), + trend + .iter() + .find_map(|point| point.metrics.output_throughput_p50), + trend + .iter() + .find_map(|point| point.metrics.output_throughput_p90), + ] { + assert_eq!(value, expected); + } + } + } + + database.close().await; +} + #[tokio::test] async fn observability_preserves_and_filters_opaque_response_ids() { let Some(database) = TestDatabase::create("observability_opaque_response_id").await else { diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 11a92e1d0..6a0d26950 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -153,8 +153,8 @@ impl OpenAiResponseObservationState { started_at: Instant, ) -> bool { let mut changed = false; - // 复用协议层的语义输出判定，空结构帧不能代替首字 - if signals.semantic_output { + // 首字观测包含非前导结构事件，真实推理与正文仍分别计时 + if signals.output_start { changed |= insert_first_timing(&mut self.timings.first_token_ms, started_at); } if signals.reasoning_output { diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 99732dc2a..6a24d7db9 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -376,6 +376,7 @@ impl CodexCanonicalDecoder { fn merge_timing_signals(&mut self, signals: ResponseEventSignals) { self.timing_signals.protocol_progress |= signals.protocol_progress; + self.timing_signals.output_start |= signals.output_start; self.timing_signals.semantic_output |= signals.semantic_output; self.timing_signals.reasoning_output |= signals.reasoning_output; self.timing_signals.text_output |= signals.text_output; diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 79051508f..6d9a5ac79 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -212,11 +212,12 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir /// 单个 Responses 事件对计时系统提供的稳定语义信号 /// -/// `protocol_progress` 只说明上游仍在工作，不能替代首字；其余字段分别标记 -/// 客户端可消费的语义输出、reasoning 输出与正文输出 +/// `output_start` 用于首字观测，包含非前导结构事件；`semantic_output` 仍要求 +/// 实际内容，供重试与交付边界使用，不能与首字观测互换 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ResponseEventSignals { pub protocol_progress: bool, + pub output_start: bool, pub semantic_output: bool, pub reasoning_output: bool, pub text_output: bool, @@ -224,11 +225,22 @@ pub struct ResponseEventSignals { /// 从已解析的 Responses 事件提取计时语义 /// -/// 生命周期与空结构帧不算首字，文本、推理、工具参数、图片结果或工具执行 -/// 才算语义输出；终态帧只在携带语义内容时计入 +/// 首字采用首个非前导、非心跳、非失败事件，包含结构事件 +/// 语义输出仍要求文本、推理、工具参数、图片结果或工具执行 pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> ResponseEventSignals { let mut signals = ResponseEventSignals { protocol_progress: !matches!(event_type, Some("response.failed" | "error")), + output_start: event_type.is_some_and(|event_type| { + !matches!( + event_type, + "response.created" + | "response.in_progress" + | "keepalive" + | "codex.rate_limits" + | "response.failed" + | "error" + ) + }), ..ResponseEventSignals::default() }; match event_type { diff --git a/backend/crates/providers/openai/tests/provider/contract/timing.rs b/backend/crates/providers/openai/tests/provider/contract/timing.rs index cf8bab362..c8f9381e3 100644 --- a/backend/crates/providers/openai/tests/provider/contract/timing.rs +++ b/backend/crates/providers/openai/tests/provider/contract/timing.rs @@ -94,9 +94,9 @@ async fn first_sse_chunk_records_content_on_the_request_clock_for_every_attempt( } #[tokio::test] -async fn structural_frames_wait_for_semantic_output_on_http_and_websocket() { +async fn structural_frames_start_ttft_before_content_on_http_and_websocket() { for websocket in [false, true] { - for (output, has_output) in [ + for (output, has_content) in [ ( json!({"type":"response.output_text.delta","output_index":0,"content_index":0,"delta":"hello"}), true, @@ -120,6 +120,8 @@ async fn structural_frames_wait_for_semantic_output_on_http_and_websocket() { ] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; + let is_text = output["type"] == "response.output_text.delta"; + let is_reasoning = output["type"] == "response.reasoning_summary_text.delta"; let created = json!({"type":"response.created","response":{"id":"resp_timing","model":"gpt-5.4"}}); let added = json!({"type":"response.output_item.added","output_index":0,"item":{"type":"message","content":[]}}); let completed = json!({"type":"response.completed","response":{"id":"resp_timing","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}); @@ -160,15 +162,36 @@ async fn structural_frames_wait_for_semantic_output_on_http_and_websocket() { let mut stream = provider_with_base_url(&store, base_url) .execute( planned_request("openai", operation), - timed_context(Instant::now(), 1), + timed_context(Instant::now() - Duration::from_secs(3), 2), ) .await .unwrap(); - let first = first_upstream_timings(&mut stream).await; - assert_eq!(first.first_token_ms, None); + let first = timeout(Duration::from_secs(5), async { + while let Some(event) = stream.next().await { + if let Some(observation) = event.unwrap().response_observation() + && observation.timings().first_token_ms.is_some() + { + return observation.timings(); + } + } + panic!("missing structural first-token observation"); + }) + .await + .expect("structural first token before releasing content"); + assert!(first.first_token_ms.is_some_and(|value| value >= 3_000)); + assert_eq!(first.first_text_ms, None); + assert_eq!(first.first_reasoning_ms, None); release.send(()).unwrap(); let final_timings = finish_timings(&mut stream, first).await; - assert_eq!(final_timings.first_token_ms.is_some(), has_output); + assert_eq!(final_timings.first_token_ms, first.first_token_ms); + assert_eq!( + final_timings.first_text_ms.is_some(), + has_content && is_text + ); + assert_eq!( + final_timings.first_reasoning_ms.is_some(), + has_content && is_reasoning + ); server.await.unwrap(); } } diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index d6f3308ba..8594375b1 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/backend/crates/providers/openai/tests/transport/canonical.rs @@ -94,6 +94,7 @@ fn raw_sse_passthrough_should_drop_rate_limit_control_without_timing_side_effect assert!(events.is_empty()); assert!(!signals.protocol_progress); + assert!(!signals.output_start); assert!(!signals.semantic_output); } diff --git a/backend/crates/providers/openai/tests/transport/protocol/responses.rs b/backend/crates/providers/openai/tests/transport/protocol/responses.rs index a63b2a11a..d88e046b1 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/responses.rs @@ -89,16 +89,43 @@ fn semantic_output_should_ignore_structural_empty_and_terminal_frames() { } #[test] -fn response_signals_should_not_count_structural_output_item_added() { +fn response_signals_should_count_structural_start_without_semantic_output() { let signals = response_event_signals( Some("response.output_item.added"), &json!({"item": {"type": "message", "content": []}}), ); assert!(signals.protocol_progress); + assert!(signals.output_start); assert!(!signals.semantic_output); } +#[test] +fn first_token_should_follow_non_preamble_events_without_reclassifying_content() { + for event in [ + "response.created", + "response.in_progress", + "keepalive", + "codex.rate_limits", + "response.failed", + "error", + ] { + let signals = response_event_signals(Some(event), &json!({"type": event})); + assert!(!signals.output_start, "not a first token: {event}"); + } + for event in [ + "response.output_item.added", + "response.content_part.added", + "response.reasoning_summary_part.added", + "response.output_text.delta", + "response.completed", + ] { + let signals = response_event_signals(Some(event), &json!({"type": event})); + assert!(signals.output_start, "first non-preamble event: {event}"); + assert!(!signals.semantic_output, "no content: {event}"); + } +} + #[test] fn response_signals_should_count_text_inside_a_completed_output_item() { let signals = response_event_signals( diff --git a/docs/api.md b/docs/api.md index 0276b96d5..49afe31d2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1762,13 +1762,15 @@ OpenAI 优先采用服务端 `openai-model` / `x-openai-model` 报告（流内 不能仅凭客户端的同名 metadata 或输出 Token 为零排除普通推理；其他 Provider 不套用该规则 `latencyMs` 从模型执行会话开始计到终结，包含账号选择、重试及流交付等待，不包含此前的入口解析、路由和准入。 -`firstTokenLatencyMs` 与它使用同一计时起点，表示首个语义输出到达网关的时间；文本、推理和工具输出均可触发， -空增量、空结构帧和无输出的终态帧不算首字。`latencyDetails` 的首事件、首推理和首正文时间也使用请求级起点， +`firstTokenLatencyMs` 与它使用同一计时起点，首字边界由 Provider 协议定义。 +OpenAI Responses 采用首个非前导输出事件，包含 `response.output_item.added` 等结构事件， +跳过 `response.created`、`response.in_progress`、心跳、额度控制和失败事件；xAI 采用首个语义输出。 +`latencyDetails` 的首事件、首推理和首正文时间也使用请求级起点，首推理与首正文仍要求实际内容， 连接、响应头等传输阶段耗时独立计量，不能直接相加作为总耗时 -列表与性能统计的输出速率为 `outputTokens × 1000 / (latencyMs − firstTokenLatencyMs)`， -只在输出 Token 为正、首字已采集且总耗时大于首字时间时计算。该值是网关观测到的平均输出速率， -生成区间仍包含流传输与交付等待，不表示模型内部的纯解码速度 +列表与性能统计的输出速率为 `outputTokens × 1000 / latencyMs`，只在输出 Token 和总耗时为正时计算， +不依赖首字是否采集。输出 Token 保留上游用量口径，OpenAI 的输出已包含推理 Token，不重复相加或扣除。 +该值表示完整请求期间的平均输出速率，包含选号、重试、推理与流交付等待，不表示模型内部的纯解码速度 ### 诊断与恢复关联 diff --git a/frontend/src/components/usage/shared/presenter.ts b/frontend/src/components/usage/shared/presenter.ts index ff57630f8..76f27ba46 100644 --- a/frontend/src/components/usage/shared/presenter.ts +++ b/frontend/src/components/usage/shared/presenter.ts @@ -87,10 +87,10 @@ export function usagePerformanceDetails(record: UsagePerformanceRecord) { const firstTokenMs = usageFirstTokenMs(record) const totalMs = durationValue(record.latencyMs) const outputTokens = record.tokenDetails?.outputTokens - // 与性能统计使用同一吞吐口径，首个生命周期事件不能代替首字 + // 输出包含推理 Token，使用完整请求耗时，避免扣除推理等待后高估速率 const throughput = typeof outputTokens === 'number' && Number.isFinite(outputTokens) && outputTokens > 0 - && firstTokenMs !== null && totalMs !== null && totalMs > firstTokenMs - ? outputTokens * 1000 / (totalMs - firstTokenMs) + && totalMs !== null && totalMs > 0 + ? outputTokens * 1000 / totalMs : null return {