feat(evals): Add latency summary feature (#1252)

Co-authored-by: Akhilesh Negi <akhileshnegi.an3@gmail.com>

Author: Ayush8923

backend/app/alembic/versions/086_add_latency_to_evaluation_run.py

@@ -0,0 +1,31 @@
+"""Add latency to evaluation_run
+
+Revision ID: 086
+Revises: 085
+Create Date: 2026-10-14 00:00:00.000000
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "086"
+down_revision = "085"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "evaluation_run",
+        sa.Column(
+            "latency",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment="Avg OpenAI call latency per stage (fast runs only; NULL for batch)",
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("evaluation_run", "latency")

backend/app/api/docs/evaluation/v2/create_evaluation.md

@@ -10,7 +10,8 @@ assistant's configured instructions; applies only when the run resolves a config
 prompt), and **Adherence to Knowledge Base** (groundedness of the answer against
 the retrieved chunks; applies only to rows that retrieved chunks). Per-row scores +
 reasoning are stored natively by Kaapi in the `score_trace_url` trace unit; the
-single `judge` cost stage covers the one combined call. v2 runs compute **no cosine
+single `judge` cost stage covers the one combined call. The run also records the
+per-stage average OpenAI call latency (response, judge) on `latency`. v2 runs compute **no cosine
 similarity** and do **not** touch Langfuse.
 
 Judging is system-config only: the judge always uses the configured model
@@ -37,7 +38,7 @@ reaches a terminal state. A single POST fires on **both** completion and failure
 carrying an `APIResponse` body — `success: true` for a completed run, `success:
 false` with the `error` message for a failed one; either way `data` is a slim run
 snapshot (`id`, `run_name`, `dataset_name`, `status`, `run_mode`, `inserted_at`,
-`updated_at` — no `score`/`cost`; fetch those from the run's own endpoint). When a
+`updated_at` — no `score`/`cost`/`latency`; fetch those from the run's own endpoint). When a
 `webhook_secret` credential is configured for the project the request is signed
 (`X-Webhook-Signature` / `X-Webhook-Timestamp`). Delivery is best-effort and
 at-least-once, so the receiver should be idempotent. The URL is rejected with

backend/app/crud/evaluations/fast.py

@@ -6,7 +6,7 @@
 
     Stage 1 — Responses unit:   evaluation_run.batch_job_id
     Stage 2 — Embeddings unit:  evaluation_run.embedding_batch_job_id
-    Stage 3 — Score + trace + cost (no marker; each step is idempotent)
+    Stage 3 — Score + trace + cost + latency (no marker; each step is idempotent)
     Stage 4 — Persist score unit (summary + per-trace) via the shared save_score
               helper, so the cached trace unit (score_trace_url) exists before
               the run is advertised as complete and the read path (trace view /
@@ -100,6 +100,7 @@
     resolve_config_prompt,
     select_judgeable_rows,
 )
+from app.crud.evaluations.latency import attach_latency, elapsed_ms, start_timer
 from app.crud.evaluations.langfuse import (
     create_langfuse_dataset_run,
     update_traces_with_cosine_scores,
@@ -240,12 +241,14 @@ def failed_result(generated_output: str) -> ResponseResult:
         return failed_result("ERROR: missing question in dataset item")
 
     try:
+        started = start_timer()
         result = _execute_llm_call_for_question(
             config=config,
             question=question,
             project_id=project_id,
             organization_id=organization_id,
         )
+        latency_ms = elapsed_ms(started)
     except Exception as exc:
         logger.warning(
             f"[_llm_call_for_item] Item failed | item_id={item_id} | error={exc}",
@@ -265,6 +268,8 @@ def failed_result(generated_output: str) -> ResponseResult:
         return failed_result(f"ERROR: {result.error}")
 
     guardrail, input_to_llm, output_from_llm = _guardrail_fields(result)
+    if result.guardrail_outcome is not None:
+        latency_ms = None
 
     # Tokens are billed even on an output block, so usage is read off every outcome.
     return build_response_result(
@@ -285,6 +290,7 @@ def failed_result(generated_output: str) -> ResponseResult:
         retrieved_chunks=extract_file_search_chunks(
             result.response.provider_raw_response if result.response else None
         ),
+        latency_ms=latency_ms,
     )
 
 
@@ -301,19 +307,23 @@ def _embedding_call_for_pair(
         return build_embedding_failure(item_id, "empty output or ground_truth")
 
     try:
+        started = start_timer()
         response = _create_embedding(
             openai_client,
             model=embedding_model,
             output_text=output_text,
             ground_truth=ground_truth,
         )
+        latency_ms = elapsed_ms(started)
     except openai.OpenAIError as exc:
         logger.warning(
             f"[_embedding_call_for_pair] Item failed | item_id={item_id} | error={exc}"
         )
         return build_embedding_failure(item_id, str(exc))
 
-    return parse_embedding_pair(item_id=item_id, response=response)
+    return parse_embedding_pair(
+        item_id=item_id, response=response, latency_ms=latency_ms
+    )
 
 
 def _upload_unit_to_s3(
@@ -644,7 +654,7 @@ def _attach_stage_costs(
     response_results: list[dict[str, Any]],
     embedding_results: list[dict[str, Any]] | None,
 ) -> None:
-    """Attach the response- and embedding-stage costs (idempotent per stage)."""
+    """Attach the response- and embedding-stage costs and latency (idempotent per stage)."""
     if response_results:
         attach_cost(
             session=session,
@@ -675,6 +685,13 @@ def _attach_stage_costs(
             embedding_raw_results=embedding_raw,
         )
 
+    attach_latency(
+        eval_run=eval_run,
+        log_prefix=log_prefix,
+        response_results=response_results or None,
+        embedding_results=embedding_results or None,
+    )
+
 
 def _score_cosine_path(
     *,
@@ -745,14 +762,22 @@ def _score_judge_path(
 
     # One combined judge call means all metric tokens land in single cost stage.
     if outcome.judge_results and judge_model:
+        judge_usage_rows: list[dict[str, Any]] = []
+        for result in outcome.judge_results.values():
+            judge_usage_rows.append(
+                {"usage": result.usage, "latency_ms": result.latency_ms}
+            )
         attach_cost(
             session=session,
             eval_run=eval_run,
             log_prefix=log_prefix,
             judge_model=judge_model,
-            judge_results=[
-                {"usage": result.usage} for result in outcome.judge_results.values()
-            ],
+            judge_results=judge_usage_rows,
+        )
+        attach_latency(
+            eval_run=eval_run,
+            log_prefix=log_prefix,
+            judge_results=judge_usage_rows,
         )
     return outcome
 
@@ -819,7 +844,7 @@ def _stage3_score_and_trace(
     embedding_results: list[EmbeddingResult] | None,
     log_prefix: str,
 ) -> tuple[EvaluationRun, EvaluationScore, list[dict[str, Any]]]:
-    """Stage 3 — cosine (v1) or judge (v2), create traces, attach costs. Idempotent.
+    """Stage 3 — cosine (v1) or judge (v2), create traces, attach costs + latency. Idempotent.
 
     Returns the run, the full score unit (summary_scores + per-trace records in the
     batch path's shape), and the Langfuse `write_items` (empty for v2). Everything
@@ -898,7 +923,7 @@ def attach_costs() -> None:
         else None
     )
 
-    # Persist cost + unscoreable here; the score unit (summary + traces) is persisted
+    # Persist cost + latency + unscoreable here; the score unit (summary + traces) is persisted
     # by the caller via save_score, before the completed transition, so it lands in
     # S3 like the batch path. This write must stay committed: save_score opens a
     # second session on the same row.
@@ -907,6 +932,7 @@ def attach_costs() -> None:
         eval_run=eval_run,
         update=EvaluationRunUpdate(
             cost=eval_run.cost,
+            latency=eval_run.latency,
             unscoreable=eval_run.unscoreable,
         ),
     )
@@ -1036,7 +1062,7 @@ def run_fast_evaluation(
             f"Score unit not persisted; run row is gone | eval_run_id={eval_run.id}"
         )
 
-    # Stage 5 — mark completed. Cost was persisted in Stage 3, score in Stage 4,
+    # Stage 5 — mark completed. Cost + latency were persisted in Stage 3, score in Stage 4,
     # so there's never a completed + NULL-score window.
     eval_run = update_evaluation_run(
         session=session,

backend/app/crud/evaluations/fast_results.py

@@ -2,7 +2,8 @@
 
 Pure builders and aggregation helpers: no OpenAI, DB or storage access. The
 dicts produced here are the units uploaded to S3, so they must stay
-JSON-serializable and match the batch path's shape.
+JSON-serializable. They are a superset of the batch path's shape; `latency_ms`
+is fast-only.
 """
 
 from collections.abc import Mapping
@@ -39,6 +40,7 @@ class ResponseResult(TypedDict, total=False):
     guardrail: str | None
     input_to_llm: str | None
     output_from_llm: str | None
+    latency_ms: float | None
 
 
 class EmbeddingResult(TypedDict, total=False):
@@ -50,6 +52,7 @@ class EmbeddingResult(TypedDict, total=False):
     usage: dict[str, int]
     failed: bool
     error: str
+    latency_ms: float | None
 
 
 RESPONSE_USAGE_KEYS: tuple[str, ...] = ("input_tokens", "output_tokens", "total_tokens")
@@ -70,6 +73,7 @@ def build_response_result(
     guardrail: str | None = None,
     input_to_llm: str | None = None,
     output_from_llm: str | None = None,
+    latency_ms: float | None = None,
 ) -> ResponseResult:
     """One Stage-1 per-item result, in the batch path's shape."""
     return {
@@ -85,6 +89,7 @@ def build_response_result(
         "guardrail": guardrail,
         "input_to_llm": input_to_llm,
         "output_from_llm": output_from_llm,
+        "latency_ms": latency_ms,
     }
 
 
@@ -103,6 +108,7 @@ def build_embedding_failure(item_id: str, error: str) -> EmbeddingResult:
         "usage": {},
         "failed": True,
         "error": error,
+        "latency_ms": None,
     }
 
 
@@ -111,7 +117,9 @@ def extract_usage(usage_obj: Any, keys: tuple[str, ...]) -> dict[str, int]:
     return {key: int(field_value(usage_obj, key, 0) or 0) for key in keys}
 
 
-def parse_embedding_pair(*, item_id: str, response: Any) -> EmbeddingResult:
+def parse_embedding_pair(
+    *, item_id: str, response: Any, latency_ms: float | None = None
+) -> EmbeddingResult:
     """Unpack an embeddings response into one Stage-2 per-pair result.
 
     The request embeds `[output_text, ground_truth]`, so index 0 is the generated
@@ -139,6 +147,7 @@ def parse_embedding_pair(*, item_id: str, response: Any) -> EmbeddingResult:
         "ground_truth_embedding": ground_truth_embedding,
         "usage": extract_usage(field_value(response, "usage"), EMBEDDING_USAGE_KEYS),
         "failed": output_embedding is None or ground_truth_embedding is None,
+        "latency_ms": latency_ms,
     }
 
 

backend/app/crud/evaluations/judge.py

@@ -25,6 +25,7 @@
     KNOWLEDGE_BASE_JUDGE_PROMPT,
     PROMPT_JUDGE_PROMPT,
 )
+from app.crud.evaluations.latency import elapsed_ms, start_timer
 from app.crud.evaluations.response_parsing import extract_response_text
 from app.crud.evaluations.retry import retry_openai_call
 from app.crud.evaluations.score import (
@@ -138,6 +139,7 @@ class JudgeResult:
 
     metrics: dict[JudgeMetricEnum, MetricScore]
     usage: dict[str, int]
+    latency_ms: float | None = None
 
 
 def _required_input_union(metrics: list[JudgeMetricSpec]) -> list[JudgeInputEnum]:
@@ -304,7 +306,9 @@ def judge_row(
     }
 
     try:
+        started = start_timer()
         response = _create_judge_response(openai_client, params)
+        latency_ms = elapsed_ms(started)
     except openai.OpenAIError as exc:
         status = getattr(exc, "status_code", None)
         # 5xx is provider-side (alert-worthy); 4xx/None is caller/Kaapi-side noise.
@@ -327,4 +331,4 @@ def judge_row(
     }
 
     metric_scores = _parse_judge_output(extract_response_text(response), applicable)
-    return JudgeResult(metrics=metric_scores, usage=usage)
+    return JudgeResult(metrics=metric_scores, usage=usage, latency_ms=latency_ms)