From 9d777dabfc79987e5832de906c751e29ce1d7eb3 Mon Sep 17 00:00:00 2001 From: AkhileshNegi Date: Mon, 10 Aug 2026 09:58:49 +0530 Subject: [PATCH] updated typechecks --- backend/app/core/storage_utils.py | 5 ++-- backend/app/crud/evaluations/core.py | 28 +++++++++++-------- backend/app/crud/evaluations/cron_utils.py | 4 +-- backend/app/crud/evaluations/dataset.py | 4 +-- backend/app/crud/evaluations/fast.py | 18 ++++++++---- backend/app/crud/evaluations/langfuse.py | 15 ++++++++-- backend/app/crud/evaluations/merge.py | 12 +++++--- backend/app/crud/evaluations/processing.py | 19 +++++++++---- backend/app/crud/evaluations/score.py | 2 +- backend/app/crud/evaluations/summary.py | 7 ++--- backend/app/models/evaluation.py | 2 +- backend/app/services/evaluations/batch_job.py | 19 +++++++++---- .../app/services/evaluations/evaluation.py | 14 +++++----- .../evaluations/prompt_improvement.py | 3 ++ backend/app/utils.py | 2 +- 15 files changed, 99 insertions(+), 55 deletions(-) diff --git a/backend/app/core/storage_utils.py b/backend/app/core/storage_utils.py index dcf4fe02d..67b644ac0 100644 --- a/backend/app/core/storage_utils.py +++ b/backend/app/core/storage_utils.py @@ -8,10 +8,11 @@ import json import logging import mimetypes +from collections.abc import Mapping, Sequence from datetime import datetime from io import BytesIO from pathlib import Path -from typing import Literal +from typing import Any, Literal from urllib.parse import unquote, urlparse from uuid import UUID @@ -109,7 +110,7 @@ def upload_to_object_store( def upload_jsonl_to_object_store( storage: CloudStorage, - results: list[dict], + results: Sequence[Mapping[str, Any]], 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 9a2a0e5e9..1cf694366 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 Mapping, 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 @@ -119,7 +120,7 @@ def list_evaluation_runs( project_id: int, limit: int = 50, offset: int = 0, -) -> list[EvaluationRun]: +) -> Sequence[EvaluationRun]: """ List all evaluation runs for an organization and project. @@ -138,7 +139,7 @@ def list_evaluation_runs( .where(EvaluationRun.organization_id == organization_id) .where(EvaluationRun.project_id == project_id) .where(EvaluationRun.type == EvaluationType.TEXT.value) - .order_by(EvaluationRun.inserted_at.desc()) + .order_by(col(EvaluationRun.inserted_at).desc()) .limit(limit) .offset(offset) ) @@ -301,7 +302,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 | " @@ -339,7 +340,7 @@ def get_or_fetch_score( update_evaluation_run( session=session, eval_run=eval_run, - update=EvaluationRunUpdate(score=score), + update=EvaluationRunUpdate(score=cast(dict[str, Any], score)), ) total_traces = len(score.get("traces", [])) @@ -356,7 +357,7 @@ def _upload_score_traces( session: Session, eval_run_id: int, project_id: int, - traces: list[dict[str, Any]], + traces: Sequence[Mapping[str, Any]], ) -> str | None: """Upload per-trace records to S3 for an evaluation run. @@ -399,7 +400,7 @@ def persist_score_traces( eval_run_id: int, organization_id: int, project_id: int, - traces: list[dict[str, Any]], + traces: Sequence[Mapping[str, Any]], ) -> 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 @@ -484,12 +485,13 @@ 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: dict[str, Any] = {"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 + db_score = cast(dict[str, Any], score) update_evaluation_run( session=session, @@ -539,6 +541,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 4876ddc90..b7caa25de 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 if is_judge_run: @@ -993,6 +997,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"], @@ -1171,7 +1176,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: @@ -1253,6 +1258,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, } ) @@ -1409,7 +1415,7 @@ def run_fast_evaluation( ) if saved is not None: eval_run = saved - eval_run.score = score + eval_run.score = cast(dict[str, Any], 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..71a062d40 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,12 @@ def update_traces_with_cosine_scores( comment = f"Cannot compute: {reason}" else: value = score_item.get("cosine_similarity") + if value is None: + logger.warning( + "[update_traces_with_cosine_scores] " + f"Score item missing cosine_similarity, skipping | trace_id={trace_id}" + ) + continue comment = COSINE_SCORE_COMMENT try: @@ -517,7 +524,7 @@ def _fetch_single_trace(trace_id: str) -> TraceData | None: "question": "", "llm_answer": "", "ground_truth_answer": "", - "question_id": "", + "question_id": None, "scores": [], } @@ -552,7 +559,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..5018cbdad 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] diff --git a/backend/app/crud/evaluations/summary.py b/backend/app/crud/evaluations/summary.py index 08fea05fa..f4ca5d144 100644 --- a/backend/app/crud/evaluations/summary.py +++ b/backend/app/crud/evaluations/summary.py @@ -2,7 +2,6 @@ """ import logging -from typing import Any import openai from openai import OpenAI @@ -10,7 +9,7 @@ from app.crud.evaluations.judge import JudgeMetricEnum from app.crud.evaluations.response_parsing import extract_response_text -from app.crud.evaluations.score import OverallSummary +from app.crud.evaluations.score import OverallSummary, SummaryScore from app.services.llm.mappers import map_kaapi_to_openai_params logger = logging.getLogger(__name__) @@ -66,7 +65,7 @@ def _format_overall_for_prompt( *, overall: OverallSummary, run_name: str, - summary_scores: list[dict[str, Any]], + summary_scores: list[SummaryScore], duplication_factor: int, ) -> str: """Compact qualitative brief for the summary model — bands, not raw scores. @@ -106,7 +105,7 @@ def generate_run_ai_summary( model: str, overall: OverallSummary, run_name: str, - summary_scores: list[dict[str, Any]], + summary_scores: list[SummaryScore], duplication_factor: int, ) -> str | None: """Best-effort one-shot natural-language note on the run's overall quality.""" diff --git a/backend/app/models/evaluation.py b/backend/app/models/evaluation.py index e7d47819c..c45db4e52 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..631c19380 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 Any, cast from uuid import UUID from asgi_correlation_id import correlation_id @@ -277,7 +277,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 +375,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 +486,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 = _attach_category_metrics(cast(dict[str, Any], 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 | " @@ -565,7 +565,7 @@ def get_evaluation_with_scores( # Recompute `category_metrics` from the merged trace set so the per-category # rollup stays in sync with the new traces, not just the cached ones. # `_attach_category_metrics` mutates in place and is idempotent. - _attach_category_metrics(merged_score) + _attach_category_metrics(cast(dict[str, Any], merged_score)) if run_overall is not None: merged_score["overall"] = run_overall @@ -586,6 +586,6 @@ def get_evaluation_with_scores( ) if eval_run: - eval_run.score = merged_score + eval_run.score = cast(dict[str, Any], 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 532864841..5711959fb 100644 --- a/backend/app/services/evaluations/prompt_improvement.py +++ b/backend/app/services/evaluations/prompt_improvement.py @@ -410,6 +410,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/utils.py b/backend/app/utils.py index 65c40431c..69e1ea965 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -74,7 +74,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: