diff --git a/backend/app/alembic/versions/086_add_latency_to_evaluation_run.py b/backend/app/alembic/versions/086_add_latency_to_evaluation_run.py new file mode 100644 index 000000000..768334bab --- /dev/null +++ b/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") diff --git a/backend/app/api/docs/evaluation/v2/create_evaluation.md b/backend/app/api/docs/evaluation/v2/create_evaluation.md index 2d92d0f15..066bcbd1c 100644 --- a/backend/app/api/docs/evaluation/v2/create_evaluation.md +++ b/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 diff --git a/backend/app/crud/evaluations/fast.py b/backend/app/crud/evaluations/fast.py index ffade467e..07f5a6da2 100644 --- a/backend/app/crud/evaluations/fast.py +++ b/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 / @@ -28,6 +28,7 @@ """ import logging +import time from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field @@ -94,6 +95,11 @@ ) from app.crud.evaluations.fast_traces import build_trace_records from app.crud.evaluations.judge import METRIC_REGISTRY, JudgeMetricSpec, JudgeResult +from app.crud.evaluations.latency import ( + LATENCY_MS_DECIMALS, + MS_PER_SECOND, + attach_latency, +) from app.crud.evaluations.judge_stage import ( build_metric_summary_scores, judge_rows, @@ -240,12 +246,16 @@ def failed_result(generated_output: str) -> ResponseResult: return failed_result("ERROR: missing question in dataset item") try: + started = time.perf_counter() result = _execute_llm_call_for_question( config=config, question=question, project_id=project_id, organization_id=organization_id, ) + latency_ms = round( + (time.perf_counter() - started) * MS_PER_SECOND, LATENCY_MS_DECIMALS + ) except Exception as exc: logger.warning( f"[_llm_call_for_item] Item failed | item_id={item_id} | error={exc}", @@ -265,6 +275,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 +297,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 +314,25 @@ def _embedding_call_for_pair( return build_embedding_failure(item_id, "empty output or ground_truth") try: + started = time.perf_counter() response = _create_embedding( openai_client, model=embedding_model, output_text=output_text, ground_truth=ground_truth, ) + latency_ms = round( + (time.perf_counter() - started) * MS_PER_SECOND, LATENCY_MS_DECIMALS + ) 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 +663,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 +694,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 +771,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 +853,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 +932,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 +941,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 +1071,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, diff --git a/backend/app/crud/evaluations/fast_results.py b/backend/app/crud/evaluations/fast_results.py index 44664860b..ccd8b33fa 100644 --- a/backend/app/crud/evaluations/fast_results.py +++ b/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, } diff --git a/backend/app/crud/evaluations/judge.py b/backend/app/crud/evaluations/judge.py index e6752250e..a7fac0d9d 100644 --- a/backend/app/crud/evaluations/judge.py +++ b/backend/app/crud/evaluations/judge.py @@ -9,6 +9,7 @@ import json import logging +import time from dataclasses import dataclass from enum import Enum from typing import Any @@ -25,6 +26,7 @@ KNOWLEDGE_BASE_JUDGE_PROMPT, PROMPT_JUDGE_PROMPT, ) +from app.crud.evaluations.latency import LATENCY_MS_DECIMALS, MS_PER_SECOND 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 +140,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 +307,11 @@ def judge_row( } try: + started = time.perf_counter() response = _create_judge_response(openai_client, params) + latency_ms = round( + (time.perf_counter() - started) * MS_PER_SECOND, LATENCY_MS_DECIMALS + ) 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 +334,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) diff --git a/backend/app/crud/evaluations/latency.py b/backend/app/crud/evaluations/latency.py new file mode 100644 index 000000000..d7ec32c9c --- /dev/null +++ b/backend/app/crud/evaluations/latency.py @@ -0,0 +1,139 @@ +"""Average OpenAI call latency for fast evaluation runs. +""" + +import logging +from collections.abc import Iterable, Mapping +from typing import Any + +from app.models import EvaluationRun + +logger = logging.getLogger(__name__) + +MS_PER_SECOND = 1000.0 +LATENCY_MS_DECIMALS = 1 + +STAGE_RESPONSE = "response" +STAGE_EMBEDDING = "embedding" +STAGE_JUDGE = "judge" + +AVG_LATENCY_MS_KEY = "avg_latency_ms" +CALL_COUNT_KEY = "call_count" +# Unrounded running total; internal to the rollup, never persisted. +_SUM_MS_KEY = "sum_ms" + + +def _stage_entry(rows: Iterable[Mapping[str, Any]]) -> dict[str, float] | None: + """Unrounded `{sum_ms, call_count}` over successful timed rows; None if none.""" + sum_ms = 0.0 + count = 0 + for row in rows: + if row.get("failed"): + continue + latency_ms = row.get("latency_ms") + if latency_ms is None: + continue + sum_ms += float(latency_ms) + count += 1 + + if count == 0: + return None + return {_SUM_MS_KEY: sum_ms, CALL_COUNT_KEY: count} + + +def _emit_entry(sum_ms: float, count: int) -> dict[str, float]: + return { + AVG_LATENCY_MS_KEY: round(sum_ms / count, LATENCY_MS_DECIMALS), + CALL_COUNT_KEY: count, + } + + +def _preserved_entry(entry: Mapping[str, Any] | None) -> dict[str, float] | None: + """Carry a persisted stage entry forward, with an approximate sum for the overall.""" + if not entry: + return None + count = int(entry[CALL_COUNT_KEY]) + if count == 0: + return None + avg_latency_ms = float(entry[AVG_LATENCY_MS_KEY]) + # Keep the stored avg so re-emitting doesn't drift it through float re-division. + return { + _SUM_MS_KEY: avg_latency_ms * count, + CALL_COUNT_KEY: count, + AVG_LATENCY_MS_KEY: avg_latency_ms, + } + + +def _build_latency_dict( + response: dict[str, float] | None, + embedding: dict[str, float] | None, + judge: dict[str, float] | None, +) -> dict[str, Any] | None: + """Combine present stages into the `eval_run.latency` payload; None if none.""" + stages: dict[str, dict[str, float] | None] = { + STAGE_RESPONSE: response, + STAGE_EMBEDDING: embedding, + STAGE_JUDGE: judge, + } + + latency: dict[str, Any] = {} + total_ms = 0.0 + total_count = 0 + for stage, entry in stages.items(): + if entry is None: + continue + stage_sum = entry[_SUM_MS_KEY] + stage_count = int(entry[CALL_COUNT_KEY]) + if AVG_LATENCY_MS_KEY in entry: + latency[stage] = { + AVG_LATENCY_MS_KEY: entry[AVG_LATENCY_MS_KEY], + CALL_COUNT_KEY: stage_count, + } + else: + latency[stage] = _emit_entry(stage_sum, stage_count) + total_ms += stage_sum + total_count += stage_count + + if total_count == 0: + return None + latency.update(_emit_entry(total_ms, total_count)) + return latency + + +def attach_latency( + *, + eval_run: EvaluationRun, + log_prefix: str, + response_results: list[dict[str, Any]] | None = None, + embedding_results: list[dict[str, Any]] | None = None, + judge_results: list[dict[str, Any]] | None = None, +) -> None: + """Recompute latency for the passed stage(s) onto `eval_run.latency`, never raising.""" + try: + existing = eval_run.latency or {} + passed: dict[str, list[dict[str, Any]] | None] = { + STAGE_RESPONSE: response_results, + STAGE_EMBEDDING: embedding_results, + STAGE_JUDGE: judge_results, + } + + entries: dict[str, dict[str, float] | None] = {} + for stage, rows in passed.items(): + if rows is None: + entries[stage] = _preserved_entry(existing.get(stage)) + continue + entry = _stage_entry(rows) + entries[stage] = entry + + latency = _build_latency_dict( + response=entries[STAGE_RESPONSE], + embedding=entries[STAGE_EMBEDDING], + judge=entries[STAGE_JUDGE], + ) + if latency is None: + return + # Reassign wholesale: in-place mutation of a JSONB dict isn't flushed. + eval_run.latency = latency + except Exception as latency_err: + logger.warning( + f"[attach_latency] {log_prefix} Failed to compute latency | {latency_err}" + ) diff --git a/backend/app/models/evaluation.py b/backend/app/models/evaluation.py index 0a44384ea..20cd69323 100644 --- a/backend/app/models/evaluation.py +++ b/backend/app/models/evaluation.py @@ -445,6 +445,16 @@ class EvaluationRun(SQLModel, table=True): description="Cost breakdown by stage (response, embedding) with token counts and USD", ) + latency: dict[str, Any] | None = SQLField( + default=None, + sa_column=Column( + JSONB, + nullable=True, + comment="Avg OpenAI call latency per stage (fast runs only; NULL for batch)", + ), + description="Avg OpenAI call latency per stage (response, embedding, judge) with call counts; fast runs only", + ) + # Error message field error_message: str | None = SQLField( default=None, @@ -530,6 +540,7 @@ class EvaluationRunUpdate(SQLModel): unscoreable: dict[str, Any] | None = None is_score_updated: bool | None = None cost: dict[str, Any] | None = None + latency: dict[str, Any] | None = None embedding_batch_job_id: int | None = None is_judge_run: bool | None = None callback_url: str | None = None @@ -557,6 +568,7 @@ class EvaluationRunPublic(SQLModel): is_score_updated: bool | None = None is_judge_run: bool | None = None cost: dict[str, Any] | None + latency: dict[str, Any] | None = None error_message: str | None organization_id: int project_id: int diff --git a/backend/app/tests/api/routes/test_evaluation_fast.py b/backend/app/tests/api/routes/test_evaluation_fast.py index 3fdf49d36..67187727b 100644 --- a/backend/app/tests/api/routes/test_evaluation_fast.py +++ b/backend/app/tests/api/routes/test_evaluation_fast.py @@ -3,6 +3,11 @@ Covers FR-1 through FR-15 from the Fast Evaluation SRD plus the chunk/aggregate split of the responses stage. External boundaries (OpenAI, Langfuse, S3, Celery dispatch) are mocked; the DB is real (`db` fixture). + +Per-row timing is pinned at the worker level in `test_fast_llm_call.py` by +patching `app.crud.evaluations.fast.time`, which swaps the whole `time` module +object seen by `fast.py`; tests here never patch the clock because the thread +pool interleaves `perf_counter` reads across rows. """ import random @@ -139,6 +144,7 @@ def _make_fast_run( total_items: int = 0, run_name: str | None = None, batch_job_id: int | None = None, + latency: dict[str, Any] | None = None, ) -> EvaluationRun: """Persist a fast-mode EvaluationRun with a real dataset + config.""" dataset = _make_fast_eligible_dataset(db=db, user_api_key=user_api_key) @@ -153,6 +159,7 @@ def _make_fast_run( run_mode=RunModeEnum.FAST.value, total_items=total_items, batch_job_id=batch_job_id, + latency=latency, organization_id=user_api_key.organization_id, project_id=user_api_key.project_id, ) @@ -180,6 +187,7 @@ def _resp_result( *, failed: bool = False, question_id: Any = None, + latency_ms: float | None = None, ) -> dict[str, Any]: """A Stage-1 per-item result in the shape a chunk partial persists.""" return { @@ -191,6 +199,7 @@ def _resp_result( "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, "question_id": question_id, "failed": failed, + "latency_ms": latency_ms, } @@ -425,6 +434,46 @@ def test_fr15_get_evaluation_returns_run_mode( assert resp.status_code == 200 assert resp.json()["data"]["run_mode"] == "fast" + def test_get_evaluation_exposes_latency( + self, + client: TestClient, + user_api_key_header: dict[str, str], + db: Session, + user_api_key: TestAuthContext, + ): + eval_run = _make_fast_run( + db=db, + user_api_key=user_api_key, + status="completed", + latency={"avg_latency_ms": 250.0, "call_count": 4}, + ) + + resp = client.get( + f"/api/v1/evaluations/{eval_run.id}", headers=user_api_key_header + ) + assert resp.status_code == 200 + assert resp.json()["data"]["latency"] == { + "avg_latency_ms": 250.0, + "call_count": 4, + } + + def test_get_evaluation_latency_is_null_for_untimed_run( + self, + client: TestClient, + user_api_key_header: dict[str, str], + db: Session, + user_api_key: TestAuthContext, + ): + eval_run = _make_fast_run(db=db, user_api_key=user_api_key) + + resp = client.get( + f"/api/v1/evaluations/{eval_run.id}", headers=user_api_key_header + ) + assert resp.status_code == 200 + data = resp.json()["data"] + assert "latency" in data + assert data["latency"] is None + # Dataset listing eligibility filter (FR-5) @@ -534,6 +583,38 @@ def test_writes_chunk_job_and_partial_unit( assert job.raw_output_url == f"s3://bucket/responses_{eval_run.id}_0.json" assert len(_s3_store[job.raw_output_url]) == 2 + def test_every_stored_row_carries_latency_ms( + self, + db: Session, + user_api_key: TestAuthContext, + _s3_store, + ): + eval_run = _make_fast_run(db=db, user_api_key=user_api_key) + items = [_dataset_item("item-ok", "Q-ok"), _dataset_item("item-bad", "Q-bad")] + + def _execute(*, query, **_): + if query.input.content.value == "Q-bad": + return BlockResult(error="invalid request") + return _ok_block_result() + + with patch("app.crud.evaluations.fast.execute_llm_call", side_effect=_execute): + run_response_chunk( + session=db, + eval_run=eval_run, + config_blob=_kaapi_config_blob(), + dataset_items_slice=items, + chunk_index=0, + ) + + job = get_chunk_job(session=db, eval_run_id=eval_run.id, chunk_index=0) + rows = {r["item_id"]: r for r in _s3_store[job.raw_output_url]} + assert all("latency_ms" in r for r in rows.values()) + assert rows["item-ok"]["failed"] is False + assert isinstance(rows["item-ok"]["latency_ms"], float) + assert rows["item-ok"]["latency_ms"] >= 0 + assert rows["item-bad"]["failed"] is True + assert rows["item-bad"]["latency_ms"] is None + def test_reads_the_batch_job_model_from_native_dict_params( self, db: Session, @@ -706,6 +787,27 @@ def test_retry_reloads_canonical_when_batch_job_id_set( ).all() assert len(canonical_rows) == 1 + def test_latency_ms_survives_merge_and_retry_reload( + self, + db: Session, + user_api_key: TestAuthContext, + _s3_store, + ): + eval_run = _make_fast_run(db=db, user_api_key=user_api_key) + _seed_response_chunk( + db=db, + eval_run=eval_run, + chunk_index=0, + results=[_resp_result("item-a", "Qa", latency_ms=123.4)], + store=_s3_store, + ) + + merged_run, merged = _merge_response_chunks(session=db, eval_run=eval_run) + _, reloaded = _merge_response_chunks(session=db, eval_run=merged_run) + + assert merged[0]["latency_ms"] == 123.4 + assert reloaded[0]["latency_ms"] == 123.4 + # Stage skipping on retry (FR-7 — embeddings) @@ -744,6 +846,7 @@ def test_fr7_stage2_skips_when_embedding_batch_job_id_already_set( "ground_truth_embedding": [1.0, 0.0], "usage": {"prompt_tokens": 5, "total_tokens": 5}, "failed": False, + "latency_ms": 87.5, } ] @@ -765,6 +868,7 @@ def test_fr7_stage2_skips_when_embedding_batch_job_id_already_set( ) assert results == cached + assert results[0]["latency_ms"] == 87.5 fake_openai.embeddings.create.assert_not_called() @@ -979,6 +1083,47 @@ def _fake_save_score(*, eval_run_id, score, **_): assert sample["scores"][0]["name"] == "Cosine Similarity" assert sample["scores"][0]["value"] == pytest.approx(1.0, abs=0.01) + def test_pipeline_persists_run_latency( + self, + db: Session, + user_api_key: TestAuthContext, + _fast_pipeline_mocks, + ): + eval_run = _make_fast_run(db=db, user_api_key=user_api_key, status="pending") + _seed_response_chunk( + db=db, + eval_run=eval_run, + chunk_index=0, + results=[ + _resp_result("item-1", "Q1", "A1", question_id=1, latency_ms=100.0), + _resp_result("item-2", "Q2", "A2", question_id=2, latency_ms=300.0), + ], + store=_fast_pipeline_mocks.store, + ) + fake_openai = MagicMock() + fake_openai.embeddings.create.return_value = _fake_embedding_response() + + with patch( + "app.crud.evaluations.fast.save_score", + side_effect=_persist_score_into(db), + ): + result = run_fast_evaluation( + session=db, + openai_client=fake_openai, + langfuse=MagicMock(), + eval_run=eval_run, + ) + + db.expire_all() + run = db.get(EvaluationRun, result.id) + assert run.latency["response"] == {"avg_latency_ms": 200.0, "call_count": 2} + # Embedding rows are timed against the real clock; only the count is stable. + assert run.latency["embedding"]["call_count"] == 2 + assert run.latency["call_count"] == 4 + assert "judge" not in run.latency + # attach_latency runs unmocked beside the mocked attach_cost. + assert _fast_pipeline_mocks.attach_cost.call_count == 2 + # Aggregate-level failure threshold (FR-14) diff --git a/backend/app/tests/celery/test_notifications.py b/backend/app/tests/celery/test_notifications.py index 7ff62a095..82896106a 100644 --- a/backend/app/tests/celery/test_notifications.py +++ b/backend/app/tests/celery/test_notifications.py @@ -391,6 +391,10 @@ def test_completed_run_delivers_slim_success_payload( run_mode="fast", callback_url=self._HOOK, ) + # Populated so the leak check below can't pass merely because it's NULL. + run.latency = {"avg_latency_ms": 250.0, "call_count": 4} + patched_session.add(run) + patched_session.commit() with ( patch.object( @@ -417,7 +421,14 @@ def test_completed_run_delivers_slim_success_payload( assert data["id"] == run.id assert data["status"] == "completed" assert data["run_mode"] == "fast" - for leaked in ("score", "traces", "cost", "callback_url", "error_message"): + for leaked in ( + "score", + "traces", + "cost", + "latency", + "callback_url", + "error_message", + ): assert leaked not in data def test_failed_run_delivers_slim_failure_payload( diff --git a/backend/app/tests/crud/evaluations/test_completion_trigger.py b/backend/app/tests/crud/evaluations/test_completion_trigger.py index ad66488a0..c2a218c56 100644 --- a/backend/app/tests/crud/evaluations/test_completion_trigger.py +++ b/backend/app/tests/crud/evaluations/test_completion_trigger.py @@ -17,12 +17,15 @@ from unittest.mock import MagicMock, patch import pytest +from sqlmodel import Session from app.crud.evaluations.core import ( TERMINAL_EVAL_STATUSES, update_evaluation_run, ) from app.models import EvaluationRun, EvaluationRunUpdate +from app.tests.utils.auth import TestAuthContext +from app.tests.utils.test_data import create_test_evaluation_run def _stub_eval_run( @@ -163,3 +166,33 @@ def test_transition_with_callback_url_enqueues_both( update=EvaluationRunUpdate(status=status, **extra), ) callback_delay.assert_called_once_with(eval_run.id) + + +class TestLatencyUpdate: + def test_latency_only_update_persists_without_enqueue( + self, db: Session, user_api_key: TestAuthContext + ): + # EvaluationRunUpdate drops unknown kwargs, so a missing field would be a + # silent no-op rather than an error. + eval_run = create_test_evaluation_run( + db, + organization_id=user_api_key.organization_id, + project_id=user_api_key.project_id, + ) + + with patch( + "app.crud.evaluations.core._enqueue_eval_completion_notification" + ) as enq: + update_evaluation_run( + session=db, + eval_run=eval_run, + update=EvaluationRunUpdate( + latency={"avg_latency_ms": 1.0, "call_count": 1} + ), + ) + + enq.assert_not_called() + db.expire_all() + persisted = db.get(EvaluationRun, eval_run.id) + assert persisted.latency == {"avg_latency_ms": 1.0, "call_count": 1} + assert persisted.status == "processing" diff --git a/backend/app/tests/crud/evaluations/test_fast_judge.py b/backend/app/tests/crud/evaluations/test_fast_judge.py index fb7dcb728..739a410fa 100644 --- a/backend/app/tests/crud/evaluations/test_fast_judge.py +++ b/backend/app/tests/crud/evaluations/test_fast_judge.py @@ -17,6 +17,7 @@ `_create_judge_response`), S3, `save_score`, model/cost resolution. DB is real. """ +import itertools import json from collections.abc import Iterator from contextlib import ExitStack @@ -144,7 +145,11 @@ def _make_run( def _resp_result( - item_id: str, question: str, ground_truth: str = "golden" + item_id: str, + question: str, + ground_truth: str = "golden", + *, + latency_ms: float | None = None, ) -> dict[str, Any]: return { "item_id": item_id, @@ -155,6 +160,7 @@ def _resp_result( "usage": {"input_tokens": 5, "output_tokens": 5, "total_tokens": 10}, "question_id": item_id, "failed": False, + "latency_ms": latency_ms, } @@ -541,6 +547,64 @@ def test_judge_cost_stage_tracked( # cost_usd from the mocked estimate: (0.001 + 0.002) rounded. assert stage["cost_usd"] == pytest.approx(0.003, abs=1e-9) + def test_judge_latency_stage_tracked( + self, db: Session, user_api_key: TestAuthContext, _s3_store + ): + eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True) + # One row only: with more, the judge pool interleaves perf_counter reads. + _seed_chunk( + db=db, + eval_run=eval_run, + results=[_resp_result("item-1", "Q1", "golden-1", latency_ms=100.0)], + store=_s3_store, + ) + + with patch("app.crud.evaluations.judge.time") as mock_time: + mock_time.perf_counter.side_effect = itertools.chain( + [1.0, 1.25], itertools.repeat(1.25) + ) + result, _ = _run_pipeline( + db=db, + eval_run=eval_run, + judge_side_effect=lambda _p: _judge_response(4, "close"), + ) + + db.expire_all() + run = db.get(EvaluationRun, result.id) + assert run.latency == { + "response": {"avg_latency_ms": 100.0, "call_count": 1}, + "judge": {"avg_latency_ms": 250.0, "call_count": 1}, + "avg_latency_ms": 175.0, + "call_count": 2, + } + + def test_malformed_judge_reply_is_not_a_timed_call( + self, db: Session, user_api_key: TestAuthContext, _s3_store + ): + eval_run = _make_run(db=db, user_api_key=user_api_key, is_judge_run=True) + _seed_chunk( + db=db, + eval_run=eval_run, + results=[ + _resp_result("item-good", "Q-good", "golden-good"), + _resp_result("item-bad", "Q-bad", "golden-bad"), + ], + store=_s3_store, + ) + + def _judge(params): + if "Q-bad" in params["input"]: + return _raw_judge_response("totally not json") + return _judge_response(5, "correct") + + result, _ = _run_pipeline(db=db, eval_run=eval_run, judge_side_effect=_judge) + + db.expire_all() + run = db.get(EvaluationRun, result.id) + assert run.latency["judge"]["call_count"] == 1 + # Seeded response rows carry no latency_ms (pre-deploy shape): stage omitted. + assert "response" not in run.latency + BOT_INSTRUCTIONS = "You are a farming helpline bot. Always answer in Hindi." @@ -875,6 +939,7 @@ def _judge(params): assert GROUND_TRUTH_SCORE_NAME not in summary_names assert PROMPT_SCORE_NAME not in summary_names assert COSINE_SCORE_NAME in summary_names + assert "judge" not in (run.latency or {}) class TestRunOverallSummary: @@ -1120,6 +1185,7 @@ def test_success_with_chunks_returns_serializable_plain_dicts(self): {"score": 0.91, "text": "chunk A", "filename": "doc.pdf"}, {"score": 0.42, "text": "chunk B", "filename": None}, ] + assert isinstance(result["latency_ms"], float) json.dumps(result) # the S3 unit must stay JSON-serializable def test_success_without_hits_returns_empty_chunks(self): @@ -1135,6 +1201,7 @@ def test_error_path_has_no_chunks(self): assert result["failed"] is True assert result["retrieved_chunks"] is None + assert result["latency_ms"] is None class TestKnowledgeBaseScoring: diff --git a/backend/app/tests/crud/evaluations/test_fast_llm_call.py b/backend/app/tests/crud/evaluations/test_fast_llm_call.py index e82769c6a..9da9c6847 100644 --- a/backend/app/tests/crud/evaluations/test_fast_llm_call.py +++ b/backend/app/tests/crud/evaluations/test_fast_llm_call.py @@ -1,13 +1,17 @@ -"""Fast-eval generation against `execute_llm_call`.""" +"""Fast-eval per-item workers: generation via `execute_llm_call`, embedding pairs.""" +import itertools from collections.abc import Iterator +from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock, patch from uuid import UUID, uuid4 +import openai import pytest from app.crud.evaluations.fast import ( + _embedding_call_for_pair, _execute_llm_call_for_question, _llm_call_for_item, ) @@ -203,6 +207,155 @@ def test_rephrased_row_does_not_report_the_canned_reply_as_llm_input( assert result["output_from_llm"] is None +@pytest.fixture +def fast_clock() -> Iterator[MagicMock]: + """1.5 s between the two reads, open-ended so an extra read never StopIterations. + + Patches `fast.time` (the whole module object), never global perf_counter: + tenacity's own timing uses it. + """ + with patch("app.crud.evaluations.fast.time") as mock_time: + mock_time.perf_counter.side_effect = itertools.chain( + [0.0, 1.5], itertools.repeat(1.5) + ) + yield mock_time + + +class TestResponseLatency: + def test_successful_call_records_wall_time_in_ms( + self, fast_clock: MagicMock + ) -> None: + result, _ = _run_item( + BlockResult(response=text_llm_call_response("answer"), usage=USAGE) + ) + + assert result["failed"] is False + assert result["latency_ms"] == 1500.0 + + def test_raised_exception_has_no_latency(self, fast_clock: MagicMock) -> None: + result, _ = _run_item(RuntimeError("boom")) + + assert result["failed"] is True + assert result["latency_ms"] is None + + def test_provider_error_has_no_latency(self, fast_clock: MagicMock) -> None: + result, _ = _run_item(BlockResult(error="provider 503", usage=USAGE)) + + assert result["failed"] is True + assert result["latency_ms"] is None + + def test_retry_exhausted_row_has_no_latency( + self, fast_clock: MagicMock, no_backoff: list[float] + ) -> None: + with patch( + "app.crud.evaluations.fast.execute_llm_call", + return_value=BlockResult(error="provider 503", retryable=True), + ) as mock_execute: + result = _llm_call_for_item( + config=_config(), + project_id=PROJECT_ID, + organization_id=ORG_ID, + item=_item(), + ) + + assert mock_execute.call_count == 3 + assert result["failed"] is True + assert result["latency_ms"] is None + + def test_missing_question_has_no_latency(self, fast_clock: MagicMock) -> None: + result, _ = _run_item( + BlockResult(response=text_llm_call_response()), item=_item(question=None) + ) + + assert result["latency_ms"] is None + + @pytest.mark.parametrize( + "block", + [ + BlockResult(guardrail_outcome="blocked", error="uli", usage=USAGE), + BlockResult( + guardrail_outcome="rephrased", + response=text_llm_call_response("please rephrase"), + usage=USAGE, + ), + ], + ids=["blocked", "rephrased"], + ) + def test_guardrail_short_circuit_is_not_a_timed_call( + self, fast_clock: MagicMock, block: BlockResult + ) -> None: + # Scored (failed=False) but may never have reached the LLM. + result, _ = _run_item(block) + + assert result["failed"] is False + assert result["latency_ms"] is None + + def test_guardrail_applied_row_is_still_timed(self, fast_clock: MagicMock) -> None: + result, _ = _run_item( + BlockResult( + response=text_llm_call_response("answer"), + usage=USAGE, + metadata={"input_guardrail": {"input_to_llm": "redacted q"}}, + ) + ) + + assert result["guardrail"] == "applied" + assert result["latency_ms"] == 1500.0 + + +class TestEmbeddingLatency: + @staticmethod + def _client( + *, data: list[tuple[int, list[float]]] | None = None, error=None + ) -> MagicMock: + client = MagicMock() + if error is not None: + client.embeddings.create.side_effect = error + else: + client.embeddings.create.return_value = SimpleNamespace( + data=[SimpleNamespace(index=i, embedding=v) for i, v in data or []], + usage=SimpleNamespace(prompt_tokens=4, total_tokens=4), + ) + return client + + def _embed(self, client: MagicMock, output_text: str = "out") -> dict[str, Any]: + return _embedding_call_for_pair( + openai_client=client, + embedding_model="text-embedding-3-small", + item_id="item-1", + output_text=output_text, + ground_truth="golden", + ) + + def test_successful_pair_records_wall_time_in_ms( + self, fast_clock: MagicMock + ) -> None: + result = self._embed(self._client(data=[(0, [1.0]), (1, [2.0])])) + + assert result["failed"] is False + assert result["latency_ms"] == 1500.0 + + def test_openai_error_has_no_latency(self, fast_clock: MagicMock) -> None: + # Base OpenAIError is not in the retryable set, so no backoff runs. + result = self._embed(self._client(error=openai.OpenAIError("bad request"))) + + assert result["failed"] is True + assert result["latency_ms"] is None + + def test_short_payload_has_no_latency(self, fast_clock: MagicMock) -> None: + result = self._embed(self._client(data=[(0, [1.0])])) + + assert result["failed"] is True + assert result["latency_ms"] is None + + def test_empty_side_is_not_called_or_timed(self, fast_clock: MagicMock) -> None: + client = self._client(data=[(0, [1.0]), (1, [2.0])]) + result = self._embed(client, output_text="") + + assert client.embeddings.create.call_count == 0 + assert result["latency_ms"] is None + + class TestRowIsolation: def test_missing_question_fails_the_row_without_calling_the_llm(self) -> None: result, mock_call = _run_item( diff --git a/backend/app/tests/crud/evaluations/test_fast_results.py b/backend/app/tests/crud/evaluations/test_fast_results.py index a0fde65c1..ea4b80634 100644 --- a/backend/app/tests/crud/evaluations/test_fast_results.py +++ b/backend/app/tests/crud/evaluations/test_fast_results.py @@ -45,6 +45,7 @@ def test_carries_every_key_the_s3_unit_needs(self) -> None: "guardrail", "input_to_llm", "output_from_llm", + "latency_ms", } def test_optional_fields_default_to_none(self) -> None: @@ -62,6 +63,19 @@ def test_optional_fields_default_to_none(self) -> None: assert result["guardrail"] is None assert result["input_to_llm"] is None assert result["output_from_llm"] is None + assert result["latency_ms"] is None + + def test_carries_latency_ms(self) -> None: + result = build_response_result( + item_id="i", + question="q", + ground_truth="gt", + question_id=1, + generated_output="out", + failed=False, + latency_ms=12.3, + ) + assert result["latency_ms"] == 12.3 class TestBuildEmbeddingFailure: @@ -72,6 +86,11 @@ def test_marks_failed_with_both_vectors_absent(self) -> None: assert result["ground_truth_embedding"] is None assert result["error"] == "empty output or ground_truth" + def test_has_null_latency_ms(self) -> None: + result = build_embedding_failure("item_1_0", "boom") + assert "latency_ms" in result + assert result["latency_ms"] is None + class TestExtractUsage: def test_reads_requested_counters(self) -> None: @@ -123,6 +142,21 @@ def test_short_payload_is_a_failure(self) -> None: assert result["failed"] is True assert "expected 2 embeddings" in result["error"] + def test_carries_latency_ms(self) -> None: + result = parse_embedding_pair( + item_id="i", + response=self._response([(0, [1.0]), (1, [2.0])]), + latency_ms=45.6, + ) + assert result["latency_ms"] == 45.6 + + def test_short_payload_drops_latency_ms(self) -> None: + # The call returned but produced nothing usable, so it isn't a timed call. + result = parse_embedding_pair( + item_id="i", response=self._response([(0, [1.0])]), latency_ms=45.6 + ) + assert result["latency_ms"] is None + class TestSumUsage: def test_sums_requested_keys_across_results(self) -> None: diff --git a/backend/app/tests/crud/evaluations/test_judge.py b/backend/app/tests/crud/evaluations/test_judge.py index 518e3c00c..2333702c6 100644 --- a/backend/app/tests/crud/evaluations/test_judge.py +++ b/backend/app/tests/crud/evaluations/test_judge.py @@ -8,6 +8,7 @@ mocked at `_create_judge_response`; parsing/prompt-composition helpers are pure. """ +import itertools import json from dataclasses import replace from types import SimpleNamespace @@ -354,6 +355,26 @@ def test_returns_metric_score_and_usage(self, db: Session) -> None: "total_tokens": 23, } + def test_returns_latency_ms(self, db: Session) -> None: + # Patch the module's `time`, never global perf_counter: tenacity uses it. + with patch( + "app.crud.evaluations.judge._create_judge_response", + return_value=_judge_response( + {"ground_truth": {"score": 5, "reasoning": "same meaning"}} + ), + ), patch("app.crud.evaluations.judge.time") as mock_time: + mock_time.perf_counter.side_effect = itertools.chain( + [1.0, 1.35], itertools.repeat(1.35) + ) + result = judge_row( + openai_client=SimpleNamespace(), + base_params=self._base_params(db), + metrics=_ALL_METRICS, + inputs=self._gt_inputs(), + ) + + assert result.latency_ms == pytest.approx(350.0) + def test_chunkless_row_judges_ground_truth_only(self, db: Session) -> None: captured: dict = {} diff --git a/backend/app/tests/crud/evaluations/test_latency.py b/backend/app/tests/crud/evaluations/test_latency.py new file mode 100644 index 000000000..a218d62fc --- /dev/null +++ b/backend/app/tests/crud/evaluations/test_latency.py @@ -0,0 +1,241 @@ +"""Run-level latency rollup (`attach_latency`) onto `eval_run.latency`. + +Pure in-memory: `attach_latency` takes no session, so a transient +`EvaluationRun` is enough; persistence is covered by the fast-pipeline tests. +""" + +import logging + +import pytest + +from app.crud.evaluations.latency import attach_latency +from app.models import EvaluationRun + +LOG_PREFIX = "[test]" + + +def _run(latency: dict | None = None) -> EvaluationRun: + return EvaluationRun(latency=latency) + + +class TestStageAverage: + def test_response_stage_mean_and_overall(self) -> None: + run = _run() + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + response_results=[{"latency_ms": 100.0}, {"latency_ms": 300.0}], + ) + assert run.latency == { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + + def test_failed_and_untimed_rows_are_excluded(self) -> None: + run = _run() + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + response_results=[ + {"latency_ms": 100.0}, + {"failed": True, "latency_ms": 900.0}, + {}, + {"failed": False, "latency_ms": None}, + ], + ) + assert run.latency == { + "response": {"avg_latency_ms": 100.0, "call_count": 1}, + "avg_latency_ms": 100.0, + "call_count": 1, + } + + def test_embedding_stage_read_from_embedding_result_dicts(self) -> None: + run = _run() + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + embedding_results=[ + { + "item_id": "item_0_0", + "output_embedding": [0.1], + "ground_truth_embedding": [0.2], + "usage": {"prompt_tokens": 4, "total_tokens": 4}, + "failed": False, + "latency_ms": 40.0, + }, + { + "item_id": "item_1_0", + "output_embedding": None, + "ground_truth_embedding": None, + "usage": {}, + "failed": True, + "error": "boom", + "latency_ms": None, + }, + ], + ) + assert run.latency == { + "embedding": {"avg_latency_ms": 40.0, "call_count": 1}, + "avg_latency_ms": 40.0, + "call_count": 1, + } + + def test_rounds_to_one_decimal(self) -> None: + run = _run() + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + response_results=[{"latency_ms": 100.04}, {"latency_ms": 100.06}], + ) + assert run.latency["response"]["avg_latency_ms"] == 100.1 + + +class TestNoQualifyingRows: + @pytest.mark.parametrize( + "rows", + [[], [{"failed": True, "latency_ms": 50.0}], [{}, {"latency_ms": None}]], + ) + def test_latency_stays_null(self, rows: list[dict]) -> None: + run = _run() + attach_latency(eval_run=run, log_prefix=LOG_PREFIX, response_results=rows) + assert run.latency is None + + def test_no_lists_passed_leaves_null(self) -> None: + run = _run() + attach_latency(eval_run=run, log_prefix=LOG_PREFIX) + assert run.latency is None + + +class TestOverall: + def test_overall_is_call_weighted_not_mean_of_means(self) -> None: + run = _run() + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + response_results=[{"latency_ms": 100.0}, {"latency_ms": 300.0}], + judge_results=[{"latency_ms": 500.0}], + ) + # A mean of stage means would give 350.0. + assert run.latency == { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "judge": {"avg_latency_ms": 500.0, "call_count": 1}, + "avg_latency_ms": 300.0, + "call_count": 3, + } + + def test_overall_uses_unrounded_stage_sums(self) -> None: + run = _run() + # (1.04 + 1.48) / 2 = 1.26 -> 1.3. Averaging the rounded stage values + # instead would give (1.0 + 1.5) / 2 = 1.25, and (1.0 + 1.48) / 2 = 1.24 -> 1.2. + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + response_results=[{"latency_ms": 1.04}], + embedding_results=[{"latency_ms": 1.48}], + ) + assert run.latency == { + "response": {"avg_latency_ms": 1.0, "call_count": 1}, + "embedding": {"avg_latency_ms": 1.5, "call_count": 1}, + "avg_latency_ms": 1.3, + "call_count": 2, + } + + +class TestPartialReattach: + def test_judge_only_preserves_response_and_recomputes_overall(self) -> None: + run = _run( + latency={ + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + ) + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + judge_results=[{"latency_ms": 500.0}], + ) + assert run.latency == { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "judge": {"avg_latency_ms": 500.0, "call_count": 1}, + "avg_latency_ms": 300.0, + "call_count": 3, + } + + def test_preserved_stage_avg_is_emitted_verbatim(self) -> None: + # 33.33 has more decimals than the rollup emits: a preserved stage must + # not be re-rounded, only fed into the overall as avg * count. + run = _run( + latency={ + "response": {"avg_latency_ms": 33.33, "call_count": 3}, + "avg_latency_ms": 33.33, + "call_count": 3, + } + ) + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + judge_results=[{"latency_ms": 100.0}], + ) + # (33.33 * 3 + 100.0) / 4 = 49.9975 -> 50.0 + assert run.latency == { + "response": {"avg_latency_ms": 33.33, "call_count": 3}, + "judge": {"avg_latency_ms": 100.0, "call_count": 1}, + "avg_latency_ms": 50.0, + "call_count": 4, + } + + def test_rerun_with_same_rows_is_idempotent(self) -> None: + rows = [{"latency_ms": 100.0}, {"latency_ms": 300.0}] + run = _run() + attach_latency(eval_run=run, log_prefix=LOG_PREFIX, response_results=rows) + attach_latency(eval_run=run, log_prefix=LOG_PREFIX, judge_results=[]) + attach_latency(eval_run=run, log_prefix=LOG_PREFIX, response_results=rows) + assert run.latency == { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + + def test_reassigns_rather_than_mutating_in_place(self) -> None: + # In-place mutation of a JSONB dict isn't flushed by SQLAlchemy. + before = { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + run = _run(latency=before) + attach_latency( + eval_run=run, log_prefix=LOG_PREFIX, judge_results=[{"latency_ms": 500.0}] + ) + assert run.latency is not before + assert "judge" not in before + + +class TestNeverRaises: + def test_bad_row_logs_warning_and_leaves_latency_unchanged( + self, caplog: pytest.LogCaptureFixture + ) -> None: + prior = { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + run = _run(latency=prior) + with caplog.at_level(logging.WARNING): + attach_latency( + eval_run=run, + log_prefix=LOG_PREFIX, + judge_results=[{"latency_ms": "x"}], + ) + assert run.latency == { + "response": {"avg_latency_ms": 200.0, "call_count": 2}, + "avg_latency_ms": 200.0, + "call_count": 2, + } + assert any( + r.levelno == logging.WARNING + and r.getMessage().startswith("[attach_latency]") + for r in caplog.records + ) diff --git a/backend/app/tests/crud/evaluations/test_processing.py b/backend/app/tests/crud/evaluations/test_processing.py index a874b7f2f..13555a7dc 100644 --- a/backend/app/tests/crud/evaluations/test_processing.py +++ b/backend/app/tests/crud/evaluations/test_processing.py @@ -595,6 +595,8 @@ async def test_process_completed_evaluation_success( assert result.cost["total_cost_usd"] == response_cost["cost_usd"] # Embedding cost is added later by process_completed_embedding_batch. assert "embedding" not in result.cost + # Batch has no per-call timing; no stand-in latency is ever written. + assert result.latency is None @pytest.mark.asyncio @patch("app.crud.evaluations.processing.persist_score_traces") @@ -902,6 +904,7 @@ async def test_process_completed_embedding_batch_success( assert result.cost["total_cost_usd"] == pytest.approx( 0.000375 + embedding_cost["cost_usd"] ) + assert result.latency is None @pytest.mark.asyncio @patch("app.crud.evaluations.processing.save_score") diff --git a/backend/app/tests/utils/test_data.py b/backend/app/tests/utils/test_data.py index accd824b0..02275504e 100644 --- a/backend/app/tests/utils/test_data.py +++ b/backend/app/tests/utils/test_data.py @@ -20,6 +20,7 @@ Credential, CredsCreate, EvaluationDataset, + EvaluationRun, FineTuning, FineTuningJobCreate, ModelEvaluation, @@ -460,3 +461,30 @@ def create_test_evaluation_dataset( db.commit() db.refresh(dataset) return dataset + + +def create_test_evaluation_run( + db: Session, + organization_id: int, + project_id: int, + status: str = "processing", +) -> EvaluationRun: + """Persist an EvaluationRun backed by a real dataset and stored config.""" + dataset = create_test_evaluation_dataset( + db=db, organization_id=organization_id, project_id=project_id + ) + config = create_test_config(db, project_id=project_id) + eval_run = EvaluationRun( + run_name=f"test_run_{random_lower_string()}", + dataset_name=dataset.name, + dataset_id=dataset.id, + config_id=config.id, + config_version=1, + status=status, + organization_id=organization_id, + project_id=project_id, + ) + db.add(eval_run) + db.commit() + db.refresh(eval_run) + return eval_run diff --git a/docs/architecture/kaapi-evaluations-ARCHITECTURE.md b/docs/architecture/kaapi-evaluations-ARCHITECTURE.md index 51d85f13b..280b2010b 100644 --- a/docs/architecture/kaapi-evaluations-ARCHITECTURE.md +++ b/docs/architecture/kaapi-evaluations-ARCHITECTURE.md @@ -146,6 +146,7 @@ backend/app/ │ │ ├── langfuse.py dataset run · traces · fetch trace scores │ │ ├── score.py TypedDicts for score shapes │ │ ├── cost.py token → USD against model_config +│ │ ├── latency.py avg OpenAI call latency per stage (fast only) │ │ ├── cron.py process_all_pending_evaluations (entrypoint) │ │ ├── cron_utils.py shared STT/TTS poll loop + helpers │ │ └── dataset.py dataset CRUD + CSV S3 helpers @@ -193,7 +194,7 @@ Everything below the run diverges per family. ```mermaid flowchart TD EDS["evaluation_dataset\n(type, name, langfuse_dataset_id,\nobject_store_url, dataset_metadata)"] - ER["evaluation_run\n(type, status, config_id+version,\nproviders, batch_job_id,\nembedding_batch_job_id, score, cost)"] + ER["evaluation_run\n(type, status, config_id+version,\nproviders, batch_job_id,\nembedding_batch_job_id, score, cost,\nlatency)"] EDS -->|"dataset_id"| ER ER -->|type=text| TX["scored via Langfuse traces\n(no Kaapi results table)"] @@ -212,7 +213,8 @@ flowchart TD advances (`pending → processing → completed | failed`). Holds the config reference (text: `config_id + config_version`), `providers` (STT/TTS model list), `batch_job_id` + `embedding_batch_job_id`, the aggregate `score` JSONB, a - per-stage `cost` JSONB, and `score_trace_url` (S3 cache of per-trace scores). + per-stage `cost` JSONB, a per-stage `latency` JSONB (fast runs only), and + `score_trace_url` (S3 cache of per-trace scores). - **`batch_job`** — one row per provider batch submission. `config` JSONB carries everything needed to reconnect a batch to its run (`evaluation_run_id`, `stt_provider`/`tts_provider`, `gemini_audio_files`, @@ -620,6 +622,7 @@ erDiagram int embedding_batch_job_id FK "text only" jsonb score "summary_scores (+ traces cache)" jsonb cost "response + embedding USD" + jsonb latency "fast only: avg OpenAI call ms per stage" string score_trace_url "S3 cache of per-trace scores" } stt_sample { @@ -662,6 +665,14 @@ erDiagram (response + embedding), priced against `global.model_config` at **batch** rates ([crud/evaluations/cost.py](../../backend/app/crud/evaluations/cost.py)); if cost computation fails, it is ignored and never blocks completion. +- **Latency** (`evaluation_run.latency`, fast text runs only; NULL for batch) is + the average wall time of successful OpenAI calls per stage (response, + embedding, judge) plus a call-weighted overall `avg_latency_ms` and + `call_count` ([crud/evaluations/latency.py](../../backend/app/crud/evaluations/latency.py)). + Timed in the worker thread around the retry-wrapped call, so retries and + backoff of eventually-successful rows are included. Kept separate from `cost` + so neither failure can drop the other; like cost, a failure never blocks + completion. - **Completion notification** — transitioning a _text_ run to a terminal status enqueues `send_eval_completion_notification` (email to project members). STT and TTS runs do **not** currently emit this notification. @@ -708,7 +719,7 @@ built for processing many requests at once). The accepted downsides: resolution. So an eval does **not** run the same code that production will run; "passes eval" does not guarantee "behaves identically in prod". 2. **Slow feedback loop.** Batch completion happens within a window, with no - guaranteed timing; there's no latency signal and no quick way to iterate on a + guaranteed timing; batch has no per-call latency signal and no quick way to iterate on a prompt. ### 12.2 Fast evals (planned — design not finalised) @@ -729,7 +740,8 @@ how thorough the scoring is: the user must be able to _see results fast_ and iterate, instead of waiting out a batch window. Benefits, because it actually calls `/llm/call`: -- a real **latency** signal (not possible with batch), +- a real **latency** signal (not possible with batch) — shipped as + `evaluation_run.latency` (per-stage avg OpenAI call latency), - **matches production** — same code path, so failures that only happen in production show up during the eval, and the difference described in 12.1 goes away, diff --git a/docs/wiki/modules/evaluations.md b/docs/wiki/modules/evaluations.md index c39f05de9..f2bee6d88 100644 --- a/docs/wiki/modules/evaluations.md +++ b/docs/wiki/modules/evaluations.md @@ -23,7 +23,7 @@ All paths relative to `backend/app/`. | `batch_job` (BatchJob) | `models/batch_job.py` | | `evaluation_iteration_run` (EvaluationIterationRun) | `models/evaluation_iteration.py` — `last_dispatched_at` is the cron's in-flight guard, stamped at kickoff and on every cron resume; NULL only on rows predating migration 084 | -Key `EvaluationRun` JSONB fields: `score` (per-trace `scores` + `summary_scores`), `per_item_scores`, `cost` (per-stage), `unscoreable`. References `config_id`, `batch_job_id`. `callback_url` (nullable): optional webhook set by the v2 run trigger; on terminal transition Kaapi POSTs a slim `APIResponse` snapshot to it (see Async). Not exposed on `EvaluationRunPublic` (no leak). `duplication_factor` (nullable): optional per-run override of the dataset's stored factor, set by the v2 run trigger for S3-only (Langfuse-free) datasets only — rejected with `422` for Langfuse-backed datasets whose items are already physically duplicated; persisted on the run so the fan-out sizing, the chunk re-load (`execute_fast_evaluation_chunk`), and the ai_summary repetition math all use the same effective factor. +Key `EvaluationRun` JSONB fields: `score` (per-trace `scores` + `summary_scores`), `per_item_scores`, `cost` (per-stage), `latency`, `unscoreable`. `latency` (nullable, fast runs only; NULL for batch, STT/TTS, pre-deploy runs and fast runs with no successful call): `{response|embedding|judge: {avg_latency_ms, call_count}, avg_latency_ms, call_count}`; stages with no successful timed call are omitted, v1 has no `judge` key, v2 no `embedding` key, and the top-level average is call-weighted from unrounded per-row sums. Measured as wall time in the worker thread around the retry-wrapped call (`_execute_llm_call_for_question`, `_create_embedding`, `_create_judge_response`), so SDK retries and Kaapi tenacity backoff of eventually-successful rows are included; only `failed=False` rows count (plus guardrail blocked/rephrased response rows are excluded, since they may never reach the LLM). Per-row `latency_ms` is persisted in the response/embedding S3 units, but judge per-row latency is in-memory only, so the judge stage is recomputed on every Stage 3 redelivery; an empty judge stage on redelivery preserves the prior entry (same as cost). The Claude AI-summary call is not counted; duplicated items count once per call; `call_count` can be below the successful-row count for runs redelivered across the deploy (pre-deploy units lack `latency_ms`). References `config_id`, `batch_job_id`. `callback_url` (nullable): optional webhook set by the v2 run trigger; on terminal transition Kaapi POSTs a slim `APIResponse` snapshot to it (see Async). Not exposed on `EvaluationRunPublic` (no leak). `duplication_factor` (nullable): optional per-run override of the dataset's stored factor, set by the v2 run trigger for S3-only (Langfuse-free) datasets only — rejected with `422` for Langfuse-backed datasets whose items are already physically duplicated; persisted on the run so the fan-out sizing, the chunk re-load (`execute_fast_evaluation_chunk`), and the ai_summary repetition math all use the same effective factor. v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native judging + Langfuse skip). All judge metrics are trace-only — per-row score + reasoning live in the `score_trace_url` trace unit (the native source of truth); there is no per-metric backup column. Judging is system-config only — always the fallback model (`gpt-5-mini`) + built-in prompts, no per-run config. Judge metrics score on an **integer 0–5 stepped scale** (the LLM returns integers 0–5; `crud/evaluations/judge.py::_parse_metric_score` enforces it) with English-only reasoning; scores are stored raw (0–5), so the API structure is unchanged but the value range is 0–5 (cosine on the v1 path stays 0–1). Each numeric judge-metric trace score also carries a `verdict` band (`crud/evaluations/score.py`: `VerdictEnum` + `verdict_from_score`, 0–5 cutoffs 2/4 → 0–1 Needs Improvement, 2–3 Needs Refinement, 4–5 Good), set in the `crud/evaluations/fast.py` trace-build loop; cosine and unscoreable/`N/A` entries carry none. `score.overall` (`OverallSummary`, `crud/evaluations/score.py`): run-level weighted rollup for judge runs, computed by `compute_overall_summary` from each metric's `avg` + `METRIC_REGISTRY` weight (renormalized over metrics that actually scored ≥1 row) — `overall_score`, `verdict`, per-metric `breakdown` (score/weight/delta/verdict), plus `ai_summary` (best-effort natural-language note, `crud/evaluations/summary.py`, Anthropic call via the platform-owned `settings.ANTHROPIC_API_KEY` (same key as prompt improvement, no per-project credential), model = `settings.EVAL_SUMMARY_MODEL`; `None` on any failure, never fails the run). Persisted on `run_fast_evaluation`'s final `EvaluationRun.score` write and re-attached verbatim by `services/evaluations/evaluation.py::get_evaluation_with_scores` on every cache/resync path (trace merging never recomputes it). @@ -32,7 +32,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud ## Services / CRUD - `services/evaluations/` — `evaluation.py`, `dataset.py` (`upload_dataset`; `use_langfuse=False` is the v2 Langfuse-free upload), `fast.py` (`validate_fast_evaluation_inputs` extracted for reuse by both the direct eval-start path and the iteration loop), `batch_job.py`, `validators.py`, `prompt_improvement.py`, `iteration.py` (`validate_and_start_evaluation_iteration`, `compute_round_scores`), `iteration_graph.py` (the LangGraph `StateGraph`: nodes + graph wiring only), `iteration_state.py` (`EvaluationIterationState`, `advance_round_state` — the pure per-round bookkeeping/stop arithmetic, `build_iteration_report`), `iteration_checkpointer.py` (`get_evaluation_iteration_checkpointer`, psycopg pool) - `services/stt_evaluations/`, `services/tts_evaluations/` -- `crud/evaluations/` — `core.py`, `batch.py`, `fast.py` (orchestration + S3/OpenAI IO only; the pieces below carry the rest), `fast_results.py` (pure per-item S3 unit shapes, usage sums, failure threshold), `fast_chunks.py` (`batch_job` bookkeeping + chunk queries/cleanup), `fast_cosine.py` (pure v1 cosine scoring), `fast_traces.py` (pure `TraceData` building), `judge_stage.py` (run-level v2 judge: `resolve_config_prompt`, `judge_rows` pool, per-metric summary rollup), `judge_prompts.py` (rubric text, moved out of `score.py`), `retry.py` (shared OpenAI retry policy), `judge.py` (`METRIC_REGISTRY` + combined judge call; `ground_truth`, `prompt`, and `knowledge_base` metrics, applied per-row by which required inputs the row carries, each spec carrying a `weight` for the overall rollup), `score.py` (`VerdictEnum`/`verdict_from_score`, `OverallSummary`/`compute_overall_summary`), `summary.py` (`generate_run_ai_summary` — one-shot Anthropic `messages.create` call via `ClaudeProvider`, plain-text output — deliberately NOT a `json_schema` `output_config`, since a quotation mark in the prose reads as the JSON string terminator and silently truncates the note; `stop_reason != STOP_REASON_COMPLETE` (the shared `end_turn` constant in `services/llm/providers/claude.py`) is logged as a truncation warning; prompt carries every trace's raw per-question scores + judge rationale, golden/generated answers, and the evaluated config, each trace keyed on `question_id` (the 1-based dataset row number from `merge.py`, cited back as "Question N") rather than the Langfuse `trace_id`, and returns a severity-ranked diagnostic note, not just a qualitative band summary), `embeddings.py`, `cost.py`, `langfuse.py`, `merge.py`, `processing.py`, `cron.py`, `iteration.py` (thin-row CRUD for the iteration loop) +- `crud/evaluations/` — `core.py`, `batch.py`, `fast.py` (orchestration + S3/OpenAI IO only; the pieces below carry the rest), `fast_results.py` (pure per-item S3 unit shapes incl. fast-only `latency_ms`, usage sums, failure threshold), `fast_chunks.py` (`batch_job` bookkeeping + chunk queries/cleanup), `fast_cosine.py` (pure v1 cosine scoring), `fast_traces.py` (pure `TraceData` building), `judge_stage.py` (run-level v2 judge: `resolve_config_prompt`, `judge_rows` pool, per-metric summary rollup), `judge_prompts.py` (rubric text, moved out of `score.py`), `retry.py` (shared OpenAI retry policy), `judge.py` (`METRIC_REGISTRY` + combined judge call; `ground_truth`, `prompt`, and `knowledge_base` metrics, applied per-row by which required inputs the row carries, each spec carrying a `weight` for the overall rollup), `score.py` (`VerdictEnum`/`verdict_from_score`, `OverallSummary`/`compute_overall_summary`), `summary.py` (`generate_run_ai_summary` — one-shot Anthropic `messages.create` call via `ClaudeProvider`, plain-text output — deliberately NOT a `json_schema` `output_config`, since a quotation mark in the prose reads as the JSON string terminator and silently truncates the note; `stop_reason != STOP_REASON_COMPLETE` (the shared `end_turn` constant in `services/llm/providers/claude.py`) is logged as a truncation warning; prompt carries every trace's raw per-question scores + judge rationale, golden/generated answers, and the evaluated config, each trace keyed on `question_id` (the 1-based dataset row number from `merge.py`, cited back as "Question N") rather than the Langfuse `trace_id`, and returns a severity-ranked diagnostic note, not just a qualitative band summary), `embeddings.py`, `cost.py`, `latency.py` (`attach_latency` — per-stage avg OpenAI call latency onto `evaluation_run.latency`, never raises), `langfuse.py`, `merge.py`, `processing.py`, `cron.py`, `iteration.py` (thin-row CRUD for the iteration loop) - `core/batch/` — shared provider batch clients: `openai.py`, `gemini.py`, `anthropic.py`, `polling.py`, `operations.py` ## Async @@ -42,7 +42,7 @@ v2 judge field on `EvaluationRun`: `is_judge_run` (bool marker gating native jud - v2 prompt iteration reuses the same job/Celery/config-version machinery (`services/evaluations/prompt_improvement.py`) and branches on `run.is_judge_run`: the v2 route calls `start_prompt_improvement_job(..., require_judge_run=True)` (a non-judge run → `422 not_a_judge_run`), and the worker drafts from the three-metric judge trace (`_draft_improved_prompt(is_judge_run=True)`, using each metric's score + reasoning) and delivers a `PromptRecommendationJobPublic` callback carrying `recommendation_type` (`Literal["prompt"]`, `models/evaluation.py`; widens to a union when knowledge-base / model recommendations land). Non-judge (v1) runs keep the default `_draft_improved_prompt` brief + `PromptImprovementJobPublic` path unchanged. - Evaluation iteration loop (`POST /evaluations/iterations`) chains fast-eval + v2 prompt improvement into a self-driving cycle via LangGraph (`services/evaluations/iteration_graph.py`): `start_eval_node` → `wait_eval_node` → (conditional) `start_improve_node` → `wait_improve_node` → loops back to `start_eval_node`, or → `finalize_node`. Stop-score = mean(`Adherence to Ground Truth`, `Adherence to Prompt`) from `EvaluationRun.score["summary_scores"]` (`compute_round_scores`); `Adherence to Knowledge Base` is recorded per round for visibility only, never gates stopping. Stops on 3 consecutive rounds with `