fix(metrics): align TTFT and throughput with sub2api

Count non-preamble Responses events for first-token timing while preserving
semantic output boundaries and request-level timing origins. Calculate
output throughput over the complete request duration in usage rows and
aggregate queries.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-07 03:52:59 +08:00
co-authored by Codex
parent 119032c067
commit 9db8528758
11 changed files with 173 additions and 34 deletions
+1 -1
View File
@@ -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<u64>,
@@ -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
@@ -79,6 +79,79 @@ fn postgres_admin_observability_adapter_implements_terminal_port() {
assert_port::<PgAdminObservabilityStore>();
}
#[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 {
@@ -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 {
@@ -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;
@@ -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 {
@@ -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();
}
}
@@ -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);
}
@@ -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(
+7 -5
View File
@@ -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,不重复相加或扣除。
该值表示完整请求期间的平均输出速率,包含选号、重试、推理与流交付等待,不表示模型内部的纯解码速度
### 诊断与恢复关联
@@ -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 {