diff --git a/backend/app/core/storage_utils.py b/backend/app/core/storage_utils.py index dcf4fe02d..27a67fccc 100644 --- a/backend/app/core/storage_utils.py +++ b/backend/app/core/storage_utils.py @@ -8,6 +8,7 @@ import json import logging import mimetypes +from collections.abc import Mapping, Sequence from datetime import datetime from io import BytesIO from pathlib import Path @@ -109,7 +110,7 @@ def upload_to_object_store( def upload_jsonl_to_object_store( storage: CloudStorage, - results: list[dict], + results: Sequence[Mapping[str, object]], filename: str, subdirectory: str, format: Literal["json", "jsonl"] = "jsonl", diff --git a/backend/app/crud/evaluations/core.py b/backend/app/crud/evaluations/core.py index 5fa886dbd..68da44f1f 100644 --- a/backend/app/crud/evaluations/core.py +++ b/backend/app/crud/evaluations/core.py @@ -1,9 +1,10 @@ import logging -from typing import Any +from collections.abc import Sequence +from typing import Any, cast from uuid import UUID from langfuse import Langfuse -from sqlmodel import Session, select +from sqlmodel import Session, col, select from app.core.cloud.storage import get_cloud_storage from app.core.db import engine @@ -11,7 +12,7 @@ from app.core.util import now from app.crud.config.version import ConfigVersionCrud from app.crud.evaluations.langfuse import fetch_trace_scores_from_langfuse -from app.crud.evaluations.score import DEFAULT_CATEGORY, EvaluationScore +from app.crud.evaluations.score import DEFAULT_CATEGORY, EvaluationScore, TraceData from app.models import EvaluationRun, EvaluationRunUpdate from app.models.evaluation import RunModeEnum from app.models.config.config import ConfigTag @@ -120,7 +121,7 @@ def list_evaluation_runs( limit: int = 50, offset: int = 0, dataset_id: int | None = None, -) -> list[EvaluationRun]: +) -> Sequence[EvaluationRun]: """ List all evaluation runs for an organization and project. @@ -146,7 +147,9 @@ def list_evaluation_runs( statement = statement.where(EvaluationRun.dataset_id == dataset_id) statement = ( - statement.order_by(EvaluationRun.inserted_at.desc()).limit(limit).offset(offset) + statement.order_by(col(EvaluationRun.inserted_at).desc()) + .limit(limit) + .offset(offset) ) runs = session.exec(statement).all() @@ -317,7 +320,7 @@ def get_or_fetch_score( logger.info( f"[get_or_fetch_score] Returning existing score | evaluation_id={eval_run.id}" ) - return eval_run.score + return cast(EvaluationScore, eval_run.score) logger.info( f"[get_or_fetch_score] Fetching score from Langfuse | " @@ -351,11 +354,12 @@ def get_or_fetch_score( "traces": langfuse_score.get("traces", []), } - # Update score column using existing helper + # Update score column using existing helper. + # TypedDict -> dict is invariant, so a cast is needed for the JSONB column. update_evaluation_run( session=session, eval_run=eval_run, - update=EvaluationRunUpdate(score=score), + update=EvaluationRunUpdate(score=cast(dict[str, object], score)), ) total_traces = len(score.get("traces", [])) @@ -372,7 +376,7 @@ def _upload_score_traces( session: Session, eval_run_id: int, project_id: int, - traces: list[dict[str, Any]], + traces: Sequence[TraceData], ) -> str | None: """Upload per-trace records to S3 for an evaluation run. @@ -415,7 +419,7 @@ def persist_score_traces( eval_run_id: int, organization_id: int, project_id: int, - traces: list[dict[str, Any]], + traces: Sequence[TraceData], ) -> EvaluationRun | None: """Persist the Q&A trace skeleton to S3 and record the ``score_trace_url`` pointer, WITHOUT touching the ``score`` column (keeps the run score-less @@ -500,9 +504,10 @@ def save_score( # IF TRACES DATA IS STORED IN S3 URL THEN HERE WE ARE JUST STORING THE SUMMARY SCORE # TODO: Evaluate whether this behaviour is needed or completely discard the storing data in db if score_trace_url: - db_score = {"summary_scores": summary_score} - if score.get("overall") is not None: - db_score["overall"] = score["overall"] + db_score: EvaluationScore = {"summary_scores": summary_score} + overall = score.get("overall") + if overall is not None: + db_score["overall"] = overall else: # fallback to store data in db if failed to store in s3 db_score = score @@ -511,7 +516,7 @@ def save_score( session=session, eval_run=eval_run, update=EvaluationRunUpdate( - score=db_score, + score=cast(dict[str, object], db_score), score_trace_url=score_trace_url or None, ), ) @@ -555,6 +560,8 @@ def group_traces_by_question_id( for trace in traces: question_id = trace.get("question_id") + if question_id is None: + continue if question_id not in groups: groups[question_id] = [] groups[question_id].append(trace) diff --git a/backend/app/crud/evaluations/cron_utils.py b/backend/app/crud/evaluations/cron_utils.py index 6cfc06a0d..6aa38dd11 100644 --- a/backend/app/crud/evaluations/cron_utils.py +++ b/backend/app/crud/evaluations/cron_utils.py @@ -11,7 +11,7 @@ from typing import Any from sqlalchemy import Integer -from sqlmodel import Session, select +from sqlmodel import Session, col, select from app.core.batch import ( BatchJobState, @@ -49,7 +49,7 @@ def fetch_processing_runs( statement = select(EvaluationRun).where( EvaluationRun.type == eval_type, EvaluationRun.status == "processing", - EvaluationRun.batch_job_id.is_not(None), + col(EvaluationRun.batch_job_id).is_not(None), ) return list(session.exec(statement).all()) diff --git a/backend/app/crud/evaluations/dataset.py b/backend/app/crud/evaluations/dataset.py index 80efae58c..f38de95bb 100644 --- a/backend/app/crud/evaluations/dataset.py +++ b/backend/app/crud/evaluations/dataset.py @@ -14,7 +14,7 @@ from fastapi import HTTPException from sqlalchemy import Integer, cast from sqlalchemy.exc import IntegrityError -from sqlmodel import Session, select +from sqlmodel import Session, col, select from app.core.cloud.storage import CloudStorage from app.core.config import settings @@ -228,7 +228,7 @@ def list_datasets( ) statement = ( - statement.order_by(EvaluationDataset.inserted_at.desc()) + statement.order_by(col(EvaluationDataset.inserted_at).desc()) .limit(limit) .offset(offset) ) diff --git a/backend/app/crud/evaluations/fast.py b/backend/app/crud/evaluations/fast.py index fa6690359..c1132dd89 100644 --- a/backend/app/crud/evaluations/fast.py +++ b/backend/app/crud/evaluations/fast.py @@ -19,7 +19,7 @@ import logging from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any +from typing import Any, cast import numpy as np import openai @@ -80,12 +80,14 @@ from app.crud.evaluations.score import ( COSINE_SCORE_COMMENT, COSINE_SCORE_NAME, + DEFAULT_CATEGORY, JUDGE_FAILED_REASON, UNSCOREABLE_EMBEDDING_FAILED, UNSCOREABLE_EMPTY_GROUND_TRUTH, UNSCOREABLE_EMPTY_OUTPUT, EvaluationScore, OverallSummary, + SummaryScore, TraceData, TraceScore, compute_overall_summary, @@ -599,11 +601,13 @@ def _merge_response_chunks( results: list[dict[str, Any]] = [] for chunk_index in sorted(chunk_job_by_index): + raw_output_url = chunk_job_by_index[chunk_index].raw_output_url + assert raw_output_url is not None # guaranteed by the filter above results.extend( _load_unit_from_s3( session=session, project_id=eval_run.project_id, - url=chunk_job_by_index[chunk_index].raw_output_url, + url=raw_output_url, ) ) @@ -881,7 +885,7 @@ def _attach_metric_scores( *, spec: JudgeMetricSpec, judge_results: dict[str, JudgeResult], - summary_scores: list[dict[str, Any]], + summary_scores: list[SummaryScore], ) -> None: """Append one metric's run-level summary score from the combined results. @@ -959,7 +963,7 @@ def _stage3_score_and_trace( similarities: list[float] = [] unscoreable: dict[str, str] = {} # {ref: reason} write_items: list[dict[str, Any]] = [] - summary_scores: list[dict[str, Any]] = [] + summary_scores: list[SummaryScore] = [] overall: OverallSummary | None = None # Resolved inside the judge block below, read again after the traces are built. config_prompt: str | None = None @@ -995,6 +999,7 @@ def _stage3_score_and_trace( else: unscoreable[ref] = UNSCOREABLE_EMBEDDING_FAILED continue + assert embedding_pair is not None # guaranteed by has_embeddings above cosine = calculate_cosine_similarity( embedding_pair["output_embedding"], embedding_pair["ground_truth_embedding"], @@ -1139,7 +1144,7 @@ def _stage3_score_and_trace( traces: list[TraceData] = [] for response in response_results: item_id = response["item_id"] - ref = item_id_to_ref.get(item_id, item_id) + ref = item_id_to_ref[item_id] if item_id in item_id_to_ref else item_id trace_scores: list[TraceScore] = [] # v2 carries no cosine score or placeholder — only the judge scores below. if not is_judge_run: @@ -1221,6 +1226,7 @@ def _stage3_score_and_trace( "llm_answer": response.get("generated_output", ""), "ground_truth_answer": response.get("ground_truth", ""), "question_id": response.get("question_id"), + "category": response.get("category") or DEFAULT_CATEGORY, "scores": trace_scores, } ) @@ -1414,7 +1420,7 @@ def run_fast_evaluation( ) if saved is not None: eval_run = saved - eval_run.score = score + eval_run.score = cast(dict[str, object], score) logger.info( f"[run_fast_evaluation] {log_prefix} Fast evaluation completed | " diff --git a/backend/app/crud/evaluations/langfuse.py b/backend/app/crud/evaluations/langfuse.py index 53f6f2473..9ec36a19f 100644 --- a/backend/app/crud/evaluations/langfuse.py +++ b/backend/app/crud/evaluations/langfuse.py @@ -13,6 +13,7 @@ from typing import Any from langfuse import Langfuse +from langfuse.api.commons.types.score_v1 import ScoreV1_Text from app.core.langfuse.langfuse import format_langfuse_error, set_trace_attributes from app.crud.evaluations.merge import compute_summary_scores @@ -280,6 +281,8 @@ def update_traces_with_cosine_scores( comment = f"Cannot compute: {reason}" else: value = score_item.get("cosine_similarity") + if value is None: + continue comment = COSINE_SCORE_COMMENT try: @@ -517,7 +520,7 @@ def _fetch_single_trace(trace_id: str) -> TraceData | None: "question": "", "llm_answer": "", "ground_truth_answer": "", - "question_id": "", + "question_id": None, "scores": [], } @@ -552,7 +555,11 @@ def _fetch_single_trace(trace_id: str) -> TraceData | None: if trace.scores: for score in trace.scores: score_name = score.name - score_value = score.value + score_value = ( + score.string_value + if isinstance(score, ScoreV1_Text) + else score.value + ) score_comment = score.comment # Get data_type from Langfuse score, default to NUMERIC data_type = getattr(score, "data_type", None) or "NUMERIC" diff --git a/backend/app/crud/evaluations/merge.py b/backend/app/crud/evaluations/merge.py index af90014e5..fa22d314c 100644 --- a/backend/app/crud/evaluations/merge.py +++ b/backend/app/crud/evaluations/merge.py @@ -9,7 +9,7 @@ import itertools import logging from collections import Counter -from typing import Any +from typing import cast import numpy as np @@ -18,6 +18,7 @@ COSINE_SCORE_NAME, DEFAULT_CATEGORY, EvaluationScore, + NumericSummaryScore, SummaryScore, TraceData, TraceScore, @@ -126,9 +127,11 @@ def apply_cosine_breakdown( breakdown = summarize_unscoreable(unscoreable) for entry in summary_scores: if entry.get("name") == COSINE_SCORE_NAME: + # Cosine is always a NUMERIC score; only that shape carries these fields. + numeric_entry = cast(NumericSummaryScore, entry) if total_items is not None: - entry["total_items"] = total_items - entry["unscoreable"] = breakdown + numeric_entry["total_items"] = total_items + numeric_entry["unscoreable"] = breakdown return summary_scores @@ -177,7 +180,7 @@ def _merge_single_trace(existing: TraceData, fresh: TraceData) -> TraceData: for fresh_score in fresh.get("scores", []): merged_scores_by_name[fresh_score["name"]] = fresh_score - merged: dict[str, Any] = { + merged: TraceData = { "trace_id": fresh.get("trace_id") or existing.get("trace_id", ""), "question": fresh.get("question") or existing.get("question", ""), "llm_answer": fresh.get("llm_answer") or existing.get("llm_answer", ""), @@ -205,6 +208,7 @@ def _reconcile_trace( or ``updated``. Exactly one of ``existing``/``fresh`` may be None, never both. """ if existing is None: + assert fresh is not None, "exactly one of existing/fresh must be set" return fresh, "added" if fresh is None: return existing, "reused" diff --git a/backend/app/crud/evaluations/processing.py b/backend/app/crud/evaluations/processing.py index 331a9b66f..42aad8b1b 100644 --- a/backend/app/crud/evaluations/processing.py +++ b/backend/app/crud/evaluations/processing.py @@ -13,7 +13,7 @@ import json import logging from collections import defaultdict -from typing import Any +from typing import Any, cast from fastapi import HTTPException from langfuse import Langfuse @@ -55,6 +55,7 @@ from app.crud.evaluations.score import ( COSINE_SCORE_COMMENT, COSINE_SCORE_NAME, + EvaluationScore, TraceData, ) from app.crud.job import get_batch_job, update_batch_job @@ -133,7 +134,7 @@ def _extract_batch_error_message( continue if error_counts: - top_error = max(error_counts, key=error_counts.get) + top_error = max(error_counts, key=lambda msg: error_counts[msg]) top_count = error_counts[top_error] total = sum(error_counts.values()) error_msg = f"{top_error} ({top_count}/{total} requests)" @@ -349,7 +350,10 @@ def build_trace_skeleton( """ traces: list[TraceData] = [] for result in results: - trace_id = trace_id_mapping.get(result.get("item_id")) + item_id = result.get("item_id") + if item_id is None: + continue + trace_id = trace_id_mapping.get(item_id) if not trace_id: continue traces.append( @@ -811,7 +815,10 @@ async def process_completed_embedding_batch( }, unscoreable=eval_run.unscoreable or {}, ) - full_score = {"summary_scores": summary_scores, "traces": traces} + full_score: EvaluationScore = { + "summary_scores": summary_scores, + "traces": traces, + } saved = save_score( eval_run_id=eval_run.id, organization_id=eval_run.organization_id, @@ -820,7 +827,7 @@ async def process_completed_embedding_batch( ) if saved is not None: eval_run = saved - eval_run.score = full_score + eval_run.score = cast(dict[str, Any], full_score) logger.info( f"[process_completed_embedding_batch] {log_prefix} Persisted " f"durable trace unit | traces={len(traces)}" @@ -1009,7 +1016,7 @@ async def check_and_process_evaluation( and status_result.get("error_file_id") ): error_msg = _extract_batch_error_message( - provider=provider, + provider=cast(OpenAIBatchProvider, provider), error_file_id=status_result["error_file_id"], batch_job=batch_job, session=session, diff --git a/backend/app/crud/evaluations/score.py b/backend/app/crud/evaluations/score.py index 2a40e2da0..23ea3fbd7 100644 --- a/backend/app/crud/evaluations/score.py +++ b/backend/app/crud/evaluations/score.py @@ -229,7 +229,7 @@ class TraceData(TypedDict): llm_answer: str question_id: int | None ground_truth_answer: str - category: str + category: NotRequired[str] scores: list[TraceScore] @@ -347,8 +347,13 @@ def compute_overall_summary( class EvaluationScore(TypedDict): - """Complete evaluation score data with traces and summary statistics.""" + """Complete evaluation score data with traces and summary statistics. + + `traces` is omitted from the DB-persisted summary-only variant (when + per-trace records are uploaded to S3 instead), so it's optional here. + """ summary_scores: list[SummaryScore] - traces: list[TraceData] + traces: NotRequired[list[TraceData]] overall: NotRequired[OverallSummary] + category_metrics: NotRequired[list[CategoryMetrics]] diff --git a/backend/app/models/evaluation.py b/backend/app/models/evaluation.py index 4935f4e60..0a44384ea 100644 --- a/backend/app/models/evaluation.py +++ b/backend/app/models/evaluation.py @@ -212,7 +212,7 @@ class EvaluationDataset(SQLModel, table=True): class EvaluationRun(SQLModel, table=True): """Database table for evaluation runs.""" - __tablename__ = "evaluation_run" + __tablename__ = "evaluation_run" # pyright: ignore[reportAssignmentType] __table_args__ = ( Index("idx_eval_run_status_org", "status", "organization_id"), Index("idx_eval_run_status_project", "status", "project_id"), diff --git a/backend/app/services/evaluations/batch_job.py b/backend/app/services/evaluations/batch_job.py index ec13045ee..146a99397 100644 --- a/backend/app/services/evaluations/batch_job.py +++ b/backend/app/services/evaluations/batch_job.py @@ -41,6 +41,7 @@ def execute_evaluation_batch_submission( ) if not run: return {"success": False, "error": "Run not found"} + confirmed_run = run try: config, error = resolve_evaluation_config( session=session, @@ -48,13 +49,21 @@ def execute_evaluation_batch_submission( config_version=config_version, project_id=project_id, ) - if error: + + def _fail(msg: str) -> dict: update_evaluation_run( session=session, - eval_run=run, - update=EvaluationRunUpdate(status="failed", error_message=error), + eval_run=confirmed_run, + update=EvaluationRunUpdate(status="failed", error_message=msg), ) - return {"success": False, "error": error} + return {"success": False, "error": msg} + + if error or config is None: + return _fail(error or "Config could not be resolved") + + provider = config.completion.provider + if provider is None: + return _fail("Config has no resolvable provider") langfuse = get_langfuse_client( session=session, org_id=organization_id, project_id=project_id @@ -64,7 +73,7 @@ def execute_evaluation_batch_submission( session=session, eval_run=run, params=config.completion.params, - provider=config.completion.provider, + provider=provider, ) return {"success": True, "batch_job_id": run.batch_job_id} except (Timeout, SoftTimeLimitExceeded): diff --git a/backend/app/services/evaluations/evaluation.py b/backend/app/services/evaluations/evaluation.py index 5f2f90bb2..a4dcb6ccf 100644 --- a/backend/app/services/evaluations/evaluation.py +++ b/backend/app/services/evaluations/evaluation.py @@ -1,7 +1,7 @@ """Evaluation run orchestration service.""" import logging -from typing import Any +from typing import cast from uuid import UUID from asgi_correlation_id import correlation_id @@ -124,7 +124,7 @@ def create_evaluation_run_or_409( def _compute_category_metrics( - traces: list[dict[str, Any]], + traces: list[TraceData], ) -> list[CategoryMetrics]: """Aggregate cosine + correctness scores per category across the trace list. @@ -183,17 +183,19 @@ def _mean(values: list[float]) -> float | None: return metrics -def _attach_category_metrics(score: dict[str, Any] | None) -> dict[str, Any] | None: +def _attach_category_metrics( + score: EvaluationScore | None, +) -> EvaluationScore | None: """Idempotently add `category_metrics` to a score dict in place. No-op when `score` is None or has no `traces`. Safe to call multiple times — recomputes from the current `traces` so cached scores that predate this feature get backfilled on read. """ - if not score or not isinstance(score, dict): + if not score: return score traces = score.get("traces") - if not isinstance(traces, list): + if traces is None: return score score["category_metrics"] = _compute_category_metrics(traces) return score @@ -277,7 +279,7 @@ def validate_and_start_batch_evaluation( config_version=config_version, project_id=project_id, ) - if error: + if error or config is None: raise HTTPException( status_code=400, detail=f"Failed to resolve config from stored config: {error}", @@ -375,7 +377,7 @@ def _load_cached_traces( storage=storage, url=eval_run.score_trace_url ) if traces is not None: - return traces, False + return cast(list[TraceData], traces), False logger.warning( f"[_load_cached_traces] Cached traces URL returned no data | " f"evaluation_id={eval_run.id} | url={eval_run.score_trace_url}" @@ -486,8 +488,8 @@ def get_evaluation_with_scores( total_items=eval_run.total_items, unscoreable=eval_run.unscoreable, ) - eval_run.score = _attach_category_metrics(cached_score) - if run_overall is not None: + eval_run.score = cast(dict[str, object], _attach_category_metrics(cached_score)) + if run_overall is not None and eval_run.score is not None: eval_run.score["overall"] = run_overall logger.info( f"[get_evaluation_with_scores] Served traces from cache | " @@ -574,7 +576,7 @@ def get_evaluation_with_scores( f"[get_evaluation_with_scores] Merged traces step-forward | " f"evaluation_id={evaluation_id} | cached={len(cached_traces)} | " f"fetched={len(langfuse_score.get('traces', []))} | " - f"merged={len(merged_score['traces'])} | reused={merge_stats['reused']} | " + f"merged={len(merged_score.get('traces', []))} | reused={merge_stats['reused']} | " f"updated={merge_stats['updated']} | added={merge_stats['added']}" ) @@ -586,6 +588,6 @@ def get_evaluation_with_scores( ) if eval_run: - eval_run.score = merged_score + eval_run.score = cast(dict[str, object], merged_score) return eval_run, None diff --git a/backend/app/services/evaluations/prompt_improvement.py b/backend/app/services/evaluations/prompt_improvement.py index 3ddeb3ba8..9163c5dbf 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -395,6 +395,9 @@ def execute_prompt_improvement( params = blob.get("completion", {}).get("params", {}) or {} current_instructions = params.get("instructions") or "" + if not run.score_trace_url: + raise RuntimeError("trace_download_failed: run has no score_trace_url") + storage = get_cloud_storage(session=session, project_id=project_id) traces = load_json_from_object_store( storage=storage, url=run.score_trace_url diff --git a/backend/app/tests/crud/evaluations/test_langfuse.py b/backend/app/tests/crud/evaluations/test_langfuse.py index 9a23ed513..9e62c1f54 100644 --- a/backend/app/tests/crud/evaluations/test_langfuse.py +++ b/backend/app/tests/crud/evaluations/test_langfuse.py @@ -451,6 +451,23 @@ def test_update_traces_with_cosine_scores_missing_trace_id(self) -> None: assert failed == [] assert mock_langfuse.create_score.call_count == 2 + def test_update_traces_with_cosine_scores_missing_value(self) -> None: + """Test that non-unscoreable items missing cosine_similarity are skipped.""" + mock_langfuse = MagicMock() + + per_item_scores = [ + {"trace_id": "trace_1", "cosine_similarity": 0.95}, + {"trace_id": "trace_2"}, # no cosine_similarity, not unscoreable + {"trace_id": "trace_3", "cosine_similarity": 0.92}, + ] + + failed = update_traces_with_cosine_scores( + langfuse=mock_langfuse, per_item_scores=per_item_scores + ) + + assert failed == [] + assert mock_langfuse.create_score.call_count == 2 + def test_update_traces_with_cosine_scores_reports_failure( self, ) -> None: diff --git a/backend/app/utils.py b/backend/app/utils.py index d0b6693a6..679ecb3cb 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -78,7 +78,7 @@ def failure_response( error: str | list, data: Optional[T] = None, metadata: Optional[Dict[str, Any]] = None, - ) -> "APIResponse[None]": + ) -> "APIResponse[T]": if isinstance(error, list): # to handle cases when error is a list of errors structured_errors = [] for err in error: