feat: add Responses API inference metrics

Signed-off-by: Major Hayden <major@redhat.com>

Author: major

src/app/endpoints/responses.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import json
+import time
 from collections.abc import AsyncIterator, Sequence
 from datetime import UTC, datetime
 from typing import Annotated, Any, Final, NoReturn, Optional, cast
@@ -39,6 +40,7 @@
 from configuration import configuration
 from constants import ENDPOINT_PATH_RESPONSES, SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER
 from log import get_logger
+from metrics import recording
 from models.api.responses.constants import UNAUTHORIZED_OPENAPI_EXAMPLES_WITH_MCP_OAUTH
 from models.api.responses.error import (
     ConflictResponse,
@@ -630,6 +632,34 @@ async def responses_endpoint_handler(
     )
 
 
+def _record_response_inference_result(
+    model_id: str,
+    endpoint_path: str,
+    result: str,
+    duration: float,
+    record_failure: bool = False,
+) -> None:
+    """Record inference result metrics for a Responses API call.
+
+    Extracts the provider and model from the composite model identifier and
+    records the inference duration histogram. Optionally records a failure
+    counter increment.
+
+    Args:
+        model_id: Composite model identifier in ``provider/model`` format.
+        endpoint_path: API endpoint path for metric labeling.
+        result: Result label such as ``success`` or ``failure``.
+        duration: Inference call duration in seconds.
+        record_failure: When True, also increment the LLM failure counter.
+    """
+    provider, model = extract_provider_and_model_from_model_id(model_id)
+    if record_failure:
+        recording.record_llm_failure(provider, model, endpoint_path)
+    recording.record_llm_inference_duration(
+        provider, model, endpoint_path, result, duration
+    )
+
+
 async def handle_streaming_response(
     original_request: ResponsesRequest,
     api_params: ResponsesApiParams,
@@ -658,6 +688,7 @@ async def handle_streaming_response(
             context.moderation_result.message,
         )
     else:
+        inference_start_time = time.monotonic()
         try:
             response = await context.client.responses.create(
                 **api_params.model_dump(exclude_none=True)
@@ -668,13 +699,21 @@ async def handle_streaming_response(
                 api_params=api_params,
                 context=context,
                 turn_summary=turn_summary,
+                inference_start_time=inference_start_time,
             )
         except (
             RuntimeError,
             APIConnectionError,
             LLSApiStatusError,
             OpenAIAPIStatusError,
         ) as e:
+            _record_response_inference_result(
+                api_params.model,
+                context.endpoint_path,
+                "failure",
+                time.monotonic() - inference_start_time,
+                record_failure=True,
+            )
             _raise_response_api_http_exception(e, api_params, context)
 
     return StreamingResponse(
@@ -928,6 +967,7 @@ async def response_generator(
     api_params: ResponsesApiParams,
     context: ResponsesContext,
     turn_summary: TurnSummary,
+    inference_start_time: float,
 ) -> AsyncIterator[str]:
     """Generate SSE-formatted streaming response with LCORE-enriched events.
 
@@ -937,6 +977,7 @@ async def response_generator(
         api_params: ResponsesApiParams
         context: Responses context
         turn_summary: TurnSummary to populate during streaming
+        inference_start_time: Monotonic timestamp taken before the inference call.
     Yields:
         SSE-formatted strings for streaming events, ending with [DONE]
     """
@@ -947,76 +988,102 @@ async def response_generator(
     configured_mcp_labels = {s.name for s in configuration.mcp_servers}
     # Track output indices of server-deployed MCP calls to filter their events
     server_mcp_output_indices: set[int] = set()
+    inference_metric_recorded = False
 
-    async for chunk in stream:
-        logger.debug("Processing streaming chunk, type: %s", chunk.type)
+    try:
+        async for chunk in stream:
+            logger.debug("Processing streaming chunk, type: %s", chunk.type)
 
-        # Filter out streaming events for server-deployed MCP tools.
-        # These are handled internally by LCS and should not be forwarded
-        # to clients that don't understand the mcp_call item type.
-        if _should_filter_mcp_chunk(
-            chunk, configured_mcp_labels, server_mcp_output_indices
-        ):
-            continue
+            # Filter out streaming events for server-deployed MCP tools.
+            # These are handled internally by LCS and should not be forwarded
+            # to clients that don't understand the mcp_call item type.
+            if _should_filter_mcp_chunk(
+                chunk, configured_mcp_labels, server_mcp_output_indices
+            ):
+                continue
 
-        chunk_dict = chunk.model_dump(exclude_none=True, by_alias=True)
+            chunk_dict = chunk.model_dump(exclude_none=True, by_alias=True)
 
-        # Create own sequence number for chunks to maintain order
-        chunk_dict["sequence_number"] = sequence_number
-        sequence_number += 1
+            # Create own sequence number for chunks to maintain order
+            chunk_dict["sequence_number"] = sequence_number
+            sequence_number += 1
 
-        if "response" in chunk_dict:
-            chunk_dict["response"]["conversation"] = normalize_conversation_id(
-                api_params.conversation
-            )
-            _sanitize_response_dict(
-                chunk_dict["response"],
-                configured_mcp_labels,
-                original_request,
-            )
-            tools = chunk_dict["response"].get("tools")
-            if tools is not None:
-                chunk_dict["response"]["tools"] = (
-                    translate_vector_store_ids_to_user_facing(
-                        tools,
-                        configuration.rag_id_mapping,
+            if "response" in chunk_dict:
+                chunk_dict["response"]["conversation"] = normalize_conversation_id(
+                    api_params.conversation
+                )
+                _sanitize_response_dict(
+                    chunk_dict["response"],
+                    configured_mcp_labels,
+                    original_request,
+                )
+                tools = chunk_dict["response"].get("tools")
+                if tools is not None:
+                    chunk_dict["response"]["tools"] = (
+                        translate_vector_store_ids_to_user_facing(
+                            tools,
+                            configuration.rag_id_mapping,
+                        )
                     )
+            # Intermediate response - no quota consumption and text yet
+            if chunk.type == "response.in_progress":
+                chunk_dict["response"]["available_quotas"] = {}
+                chunk_dict["response"]["output_text"] = ""
+
+            # Handle completion, incomplete, and failed events
+            if chunk.type in (
+                "response.completed",
+                "response.incomplete",
+                "response.failed",
+            ):
+                latest_response_object = cast(
+                    OpenAIResponseObject, cast(Any, chunk).response
                 )
-        # Intermediate response - no quota consumption and text yet
-        if chunk.type == "response.in_progress":
-            chunk_dict["response"]["available_quotas"] = {}
-            chunk_dict["response"]["output_text"] = ""
-
-        # Handle completion, incomplete, and failed events - only quota handling here
-        if chunk.type in (
-            "response.completed",
-            "response.incomplete",
-            "response.failed",
-        ):
-            latest_response_object = cast(
-                OpenAIResponseObject, cast(Any, chunk).response
-            )
 
-            # Extract and consume tokens if any were used
-            turn_summary.token_usage = extract_token_usage(
-                latest_response_object.usage, api_params.model, context.endpoint_path
-            )
-            consume_query_tokens(
-                user_id=context.auth[0],
-                model_id=api_params.model,
-                token_usage=turn_summary.token_usage,
-            )
+                # Extract and consume tokens if any were used
+                turn_summary.token_usage = extract_token_usage(
+                    latest_response_object.usage,
+                    api_params.model,
+                    context.endpoint_path,
+                )
+                consume_query_tokens(
+                    user_id=context.auth[0],
+                    model_id=api_params.model,
+                    token_usage=turn_summary.token_usage,
+                )
 
-            # Get available quotas after token consumption
-            chunk_dict["response"]["available_quotas"] = get_available_quotas(
-                quota_limiters=configuration.quota_limiters, user_id=context.auth[0]
-            )
-            turn_summary.llm_response = extract_text_from_response_items(
-                latest_response_object.output
+                # Get available quotas after token consumption
+                chunk_dict["response"]["available_quotas"] = get_available_quotas(
+                    quota_limiters=configuration.quota_limiters,
+                    user_id=context.auth[0],
+                )
+                turn_summary.llm_response = extract_text_from_response_items(
+                    latest_response_object.output
+                )
+                chunk_dict["response"]["output_text"] = turn_summary.llm_response
+
+                # Record inference duration metric for terminal events
+                result = "failure" if chunk.type == "response.failed" else "success"
+                _record_response_inference_result(
+                    api_params.model,
+                    context.endpoint_path,
+                    result,
+                    time.monotonic() - inference_start_time,
+                    record_failure=(result == "failure"),
+                )
+                inference_metric_recorded = True
+
+            yield f"event: {chunk.type or 'error'}\ndata: {json.dumps(chunk_dict)}\n\n"
+    except Exception:
+        if not inference_metric_recorded:
+            _record_response_inference_result(
+                api_params.model,
+                context.endpoint_path,
+                "failure",
+                time.monotonic() - inference_start_time,
+                record_failure=True,
             )
-            chunk_dict["response"]["output_text"] = turn_summary.llm_response
-
-        yield f"event: {chunk.type or 'error'}\ndata: {json.dumps(chunk_dict)}\n\n"
+        raise
 
     # Extract response metadata from final response object
     if latest_response_object:
@@ -1108,13 +1175,22 @@ async def handle_non_streaming_response(
         await _persist_blocked_response_turn(api_params, context)
         _queue_blocked_response_event(api_params, context, output_text)
     else:
+        inference_start_time = time.monotonic()
+        inference_metric_recorded = False
         try:
             api_response = cast(
                 OpenAIResponseObject,
                 await context.client.responses.create(
                     **api_params.model_dump(exclude_none=True)
                 ),
             )
+            _record_response_inference_result(
+                api_params.model,
+                context.endpoint_path,
+                "success",
+                time.monotonic() - inference_start_time,
+            )
+            inference_metric_recorded = True
             token_usage = extract_token_usage(
                 api_response.usage, api_params.model, context.endpoint_path
             )
@@ -1138,6 +1214,14 @@ async def handle_non_streaming_response(
             LLSApiStatusError,
             OpenAIAPIStatusError,
         ) as e:
+            if not inference_metric_recorded:
+                _record_response_inference_result(
+                    api_params.model,
+                    context.endpoint_path,
+                    "failure",
+                    time.monotonic() - inference_start_time,
+                    record_failure=True,
+                )
             _raise_response_api_http_exception(e, api_params, context)
 
     # Get available quotas

src/metrics/recording.py

@@ -7,6 +7,7 @@
 
 from collections.abc import Iterator
 from contextlib import contextmanager
+from typing import Final
 
 import metrics
 from log import get_logger
@@ -111,6 +112,32 @@ def record_llm_token_usage(
         logger.warning("Failed to update token metrics", exc_info=True)
 
 
+LLM_INFERENCE_RESULT_SUCCESS: Final[str] = "success"
+LLM_INFERENCE_RESULT_FAILURE: Final[str] = "failure"
+ALLOWED_LLM_INFERENCE_RESULTS: Final[frozenset[str]] = frozenset(
+    {LLM_INFERENCE_RESULT_SUCCESS, LLM_INFERENCE_RESULT_FAILURE}
+)
+
+
+def normalize_llm_inference_result(result: str) -> str:
+    """Clamp an inference result string to the bounded label set.
+
+    Unknown or unexpected values are mapped to ``failure`` so that the
+    Prometheus label cardinality stays bounded.
+
+    Args:
+        result: Raw result label from the caller.
+
+    Returns:
+        A value guaranteed to be in ``ALLOWED_LLM_INFERENCE_RESULTS``.
+    """
+    return (
+        result
+        if result in ALLOWED_LLM_INFERENCE_RESULTS
+        else LLM_INFERENCE_RESULT_FAILURE
+    )
+
+
 def record_llm_inference_duration(
     provider: str, model: str, endpoint_path: str, result: str, duration: float
 ) -> None:
@@ -123,9 +150,10 @@ def record_llm_inference_duration(
         result: Bounded result label, such as ``success`` or ``failure``.
         duration: Inference call duration in seconds.
     """
+    bounded_result = normalize_llm_inference_result(result)
     try:
         metrics.llm_inference_duration_seconds.labels(
-            provider, model, endpoint_path, result
+            provider, model, endpoint_path, bounded_result
         ).observe(duration)
     except (AttributeError, TypeError, ValueError):
         logger.warning("Failed to update LLM inference duration metric", exc_info=True)