feat: metrics for non-streaming

Author: tedzhouhk

lib/llm/src/http/service/openai.rs

@@ -290,7 +290,13 @@ async fn completions(
 
         Ok(sse_stream.into_response())
     } else {
-        // TODO: report ISL/OSL for non-streaming requests
+        // Process the stream to collect metrics for non-streaming requests
+        let stream = stream.map(move |response| {
+            // Process metrics but return the original response for aggregation
+            process_metrics_only(&response, &mut response_collector);
+            response
+        });
+
         let response = NvCreateCompletionResponse::from_annotated_stream(stream)
             .await
             .map_err(|e| {
@@ -515,7 +521,13 @@ async fn chat_completions(
 
         Ok(sse_stream.into_response())
     } else {
-        // TODO: report ISL/OSL for non-streaming requests
+        // Process the stream to collect metrics for non-streaming requests
+        let stream = stream.map(move |response| {
+            // Process metrics but return the original response for aggregation
+            process_metrics_only(&response, &mut response_collector);
+            response
+        });
+
         let response = NvCreateChatCompletionResponse::from_annotated_stream(stream)
             .await
             .map_err(|e| {
@@ -911,6 +923,17 @@ impl<T> From<Annotated<T>> for EventConverter<T> {
     }
 }
 
+fn process_metrics_only<T>(
+    annotated: &Annotated<T>,
+    response_collector: &mut ResponseMetricCollector,
+) {
+    // update metrics
+    if let Ok(Some(metrics)) = LLMMetricAnnotation::from_annotation(annotated) {
+        response_collector.observe_current_osl(metrics.output_tokens);
+        response_collector.observe_response(metrics.input_tokens, metrics.chunk_tokens);
+    }
+}
+
 fn process_event_converter<T: Serialize>(
     annotated: EventConverter<T>,
     response_collector: &mut ResponseMetricCollector,