bge_m3_embedding_server/handler/
sparse.rs1use std::sync::Arc;
18use std::time::Instant;
19
20use axum::{Json, extract::State, http::HeaderMap};
21
22use super::common::{check_ready, collect_x_headers, validate_input};
23use crate::error::AppError;
24use crate::models::{SparseEmbeddingData, SparseRequest, SparseResponse, SparseValues};
25use crate::state::AppState;
26
27#[allow(clippy::cast_possible_truncation)]
40#[tracing::instrument(
41 skip(state, req, headers),
42 fields(
43 batch_size,
44 chunks,
45 max_chunk_seq,
46 tokenize_ms,
47 inference_ms,
48 queue_wait_ms,
49 total_ms,
50 )
51)]
52pub async fn sparse_embeddings(
53 State(state): State<Arc<AppState>>,
54 headers: HeaderMap,
55 Json(req): Json<SparseRequest>,
56) -> Result<Json<SparseResponse>, AppError> {
57 check_ready(&state)?;
58 let x_headers = collect_x_headers(&headers);
59 let texts = req.input.0;
60 validate_input(&texts, state.max_batch)?;
61 let batch_size = texts.len();
62 tracing::Span::current().record("batch_size", batch_size);
63
64 let t0 = Instant::now();
65
66 let _permit = Arc::clone(&state.request_permits)
67 .acquire_owned()
68 .await
69 .expect("request semaphore is never closed");
70
71 let queue_wait_ms = u64::try_from(t0.elapsed().as_millis()).unwrap_or(u64::MAX);
72
73 let (embeddings, embed_stats) = state.pool.sparse(texts).await?;
74
75 let total_ms = u64::try_from(t0.elapsed().as_millis()).unwrap_or(u64::MAX);
76 tracing::Span::current()
77 .record("chunks", embed_stats.chunks)
78 .record("max_chunk_seq", embed_stats.max_chunk_seq)
79 .record("tokenize_ms", embed_stats.tokenize_ms)
80 .record("inference_ms", embed_stats.inference_ms)
81 .record("queue_wait_ms", queue_wait_ms)
82 .record("total_ms", total_ms);
83 let x_headers_val =
84 (!x_headers.is_empty()).then(|| serde_json::to_string(&x_headers).unwrap_or_default());
85 tracing::info!(
86 route = "sparse",
87 batch_size,
88 chunks = embed_stats.chunks,
89 max_chunk_seq = embed_stats.max_chunk_seq,
90 total_token_positions = embed_stats.total_token_positions,
91 seq_len_min = embed_stats.seq_len_min,
92 seq_len_max = embed_stats.seq_len_max,
93 seq_len_mean = embed_stats.seq_len_mean,
94 seq_len_p95 = embed_stats.seq_len_p95,
95 tokenize_ms = embed_stats.tokenize_ms,
96 inference_ms = embed_stats.inference_ms,
97 queue_wait_ms,
98 total_ms,
99 x_headers = x_headers_val,
100 "embedding request complete"
101 );
102
103 let data = embeddings
104 .into_iter()
105 .enumerate()
106 .map(|(index, emb)| SparseEmbeddingData {
107 index,
108 sparse_values: SparseValues {
109 indices: emb.indices.iter().map(|i| *i as u32).collect(),
110 values: emb.values,
111 },
112 })
113 .collect();
114
115 Ok(Json(SparseResponse { data }))
116}