From 9a7dd6ef65c58a5f8443cd940b28c5a9448a49ac Mon Sep 17 00:00:00 2001 From: Prashant Vasudevan <71649489+vprashrex@users.noreply.github.com> Date: Fri, 3 Jul 2026 09:25:53 +0530 Subject: [PATCH 1/3] Refactor Langfuse integration to support tracing opt-out - Updated functions to accept `Langfuse | None` to handle cases where tracing is disabled. - Modified dataset upload and evaluation processing to gracefully handle scenarios without Langfuse. - Introduced `get_tracing_client` to manage Langfuse client initialization based on project settings. - Enhanced error handling and logging for tracing-related operations. - Updated tests to cover new behavior for tracing opt-out scenarios and ensure proper functionality. --- .../versions/071_add_settings_to_project.py | 39 ++++++ .../app/api/docs/projects/update_settings.md | 12 ++ backend/app/api/routes/project.py | 32 ++++- backend/app/api/routes/responses.py | 5 +- backend/app/api/routes/threads.py | 10 +- backend/app/crud/__init__.py | 1 + backend/app/crud/credentials.py | 29 ++++ backend/app/crud/evaluations/batch.py | 97 ++++++++++++- backend/app/crud/evaluations/core.py | 11 +- backend/app/crud/evaluations/embeddings.py | 16 +-- backend/app/crud/evaluations/fast.py | 72 +++++----- backend/app/crud/evaluations/langfuse.py | 18 ++- backend/app/crud/evaluations/processing.py | 41 +++--- backend/app/crud/project.py | 28 ++++ backend/app/models/__init__.py | 1 + backend/app/models/project.py | 27 +++- backend/app/services/evaluations/batch_job.py | 8 +- backend/app/services/evaluations/dataset.py | 55 ++++---- .../app/services/evaluations/evaluation.py | 34 ++++- backend/app/services/evaluations/fast.py | 10 +- backend/app/services/llm/jobs.py | 8 +- backend/app/services/response/response.py | 5 +- .../app/tests/api/routes/test_evaluation.py | 42 +++--- .../tests/api/routes/test_evaluation_fast.py | 4 +- .../app/tests/crud/evaluations/test_batch.py | 128 ++++++++++++++++++ .../tests/crud/evaluations/test_embeddings.py | 13 +- .../tests/crud/evaluations/test_processing.py | 8 +- .../evaluations/test_evaluation_service_s3.py | 97 ++++++++++++- backend/app/utils.py | 39 +++++- 29 files changed, 725 insertions(+), 165 deletions(-) create mode 100644 backend/app/alembic/versions/071_add_settings_to_project.py create mode 100644 backend/app/api/docs/projects/update_settings.md create mode 100644 backend/app/tests/crud/evaluations/test_batch.py diff --git a/backend/app/alembic/versions/071_add_settings_to_project.py b/backend/app/alembic/versions/071_add_settings_to_project.py new file mode 100644 index 000000000..f7dcab5af --- /dev/null +++ b/backend/app/alembic/versions/071_add_settings_to_project.py @@ -0,0 +1,39 @@ +"""add settings jsonb column to project + +Revision ID: 071 +Revises: 070 +Create Date: 2026-06-17 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +# revision identifiers, used by Alembic. +revision = "071" +down_revision = "070" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "project", + sa.Column( + "settings", + postgresql.JSONB(), + nullable=False, + server_default=sa.text("'{}'::jsonb"), + comment=( + "Project-level settings (JSONB). Keys: 'tracing' (bool) — " + "Langfuse tracing opt-in, off by default to conserve Langfuse " + "rate-limit/credit budget. Gates tracing for both the response " + "path and evaluations (which fall back to cosine-only scoring)." + ), + ), + ) + + +def downgrade(): + op.drop_column("project", "settings") diff --git a/backend/app/api/docs/projects/update_settings.md b/backend/app/api/docs/projects/update_settings.md new file mode 100644 index 000000000..dc541228f --- /dev/null +++ b/backend/app/api/docs/projects/update_settings.md @@ -0,0 +1,12 @@ +Update project-level settings. + +Patches the `settings` JSONB of the project bound to the authenticating organization +API key. Only the keys provided in the request body are changed; existing keys are kept. + +**Settings** + +- `tracing` (bool): enable/disable Langfuse tracing for this project. Off by default to + conserve the org's Langfuse rate-limit/credit budget. Gates tracing for both the + response path and evaluations; when off, evaluations fall back to cosine-only scoring. + +**Scope:** requires an organization API key with project access. diff --git a/backend/app/api/routes/project.py b/backend/app/api/routes/project.py index 9f01e9c3f..e64e492c7 100644 --- a/backend/app/api/routes/project.py +++ b/backend/app/api/routes/project.py @@ -4,7 +4,7 @@ from sqlalchemy import func from sqlmodel import select -from app.api.deps import SessionDep +from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission from app.crud.organization import get_organization_by_id, validate_organization from app.crud.project import ( @@ -14,6 +14,7 @@ get_projects_by_organization, hard_delete_project, soft_delete_project, + update_project_settings, ) from app.crud.user_project import ( deactivate_users_without_projects, @@ -25,6 +26,7 @@ Project, ProjectCreate, ProjectPublic, + ProjectSettingsUpdate, ProjectUpdate, ) from app.utils import APIResponse, load_description @@ -79,6 +81,34 @@ def create_new_project(*, session: SessionDep, project_in: ProjectCreate): return APIResponse.success_response(project) +@router.patch( + "/settings", + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], + response_model=APIResponse[ProjectPublic], + description=load_description("projects/update_settings.md"), +) +def update_project_settings_route( + *, + session: SessionDep, + auth_context: AuthContextDep, + settings_in: ProjectSettingsUpdate, +): + settings_patch = settings_in.model_dump(exclude_unset=True) + if not settings_patch: + raise HTTPException(status_code=400, detail="No settings provided") + + project = update_project_settings( + session=session, + project_id=auth_context.project.id, + settings_patch=settings_patch, + ) + logger.info( + f"[update_project_settings_route] Settings updated | project_id={project.id}, " + f"keys={list(settings_patch.keys())}" + ) + return APIResponse.success_response(project) + + @router.get( "/{project_id}", dependencies=[Depends(require_permission(Permission.SUPERUSER))], diff --git a/backend/app/api/routes/responses.py b/backend/app/api/routes/responses.py index 4d97f6194..cb0e62f75 100644 --- a/backend/app/api/routes/responses.py +++ b/backend/app/api/routes/responses.py @@ -7,7 +7,7 @@ from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission from app.core.langfuse.langfuse import LangfuseTracer -from app.crud.credentials import get_provider_credential +from app.crud.credentials import get_tracing_credential from app.models import ( CallbackResponse, Diagnostics, @@ -97,10 +97,9 @@ async def responses_sync( }, ) - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=organization_id, - provider="langfuse", project_id=project_id, ) tracer = LangfuseTracer( diff --git a/backend/app/api/routes/threads.py b/backend/app/api/routes/threads.py index 3bdd8619c..b8eff4cfb 100644 --- a/backend/app/api/routes/threads.py +++ b/backend/app/api/routes/threads.py @@ -9,11 +9,11 @@ from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission -from app.core import logging, settings +from app.core import logging from app.models import OpenAIThreadCreate from app.crud import upsert_thread_result, get_thread_result from app.utils import APIResponse, mask_string -from app.crud.credentials import get_provider_credential +from app.crud.credentials import get_provider_credential, get_tracing_credential from app.core.util import configure_openai from app.core.langfuse.langfuse import LangfuseTracer @@ -312,10 +312,9 @@ async def threads( error="OpenAI API key not configured for this organization." ) - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=_current_user.organization_.id, - provider="langfuse", project_id=request.get("project_id"), ) @@ -387,10 +386,9 @@ async def threads_sync( ) # Get Langfuse credentials - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=_current_user.organization_.id, - provider="langfuse", project_id=request.get("project_id"), ) diff --git a/backend/app/crud/__init__.py b/backend/app/crud/__init__.py index 45954e15d..0aae1a7cd 100644 --- a/backend/app/crud/__init__.py +++ b/backend/app/crud/__init__.py @@ -36,6 +36,7 @@ update_creds_for_org, remove_creds_for_org, get_provider_credential, + get_tracing_credential, remove_provider_credential, ) diff --git a/backend/app/crud/credentials.py b/backend/app/crud/credentials.py index 1d23ff587..0e0151e6f 100644 --- a/backend/app/crud/credentials.py +++ b/backend/app/crud/credentials.py @@ -9,6 +9,7 @@ from app.core.providers import validate_provider, validate_provider_credentials from app.core.security import decrypt_credentials, encrypt_credentials from app.core.util import now +from app.crud.project import get_project_by_id from app.models import Credential, CredsCreate, CredsUpdate @@ -167,6 +168,34 @@ def get_provider_credential( return None +def get_tracing_credential( + *, + session: Session, + org_id: int, + project_id: int | str, +) -> dict[str, Any] | None: + """Return langfuse credentials only when the project opted into tracing. + + Tracing is gated by the project's `settings["tracing"]` flag (off by + default). When disabled, returns None so LangfuseTracer / + observe_llm_execution degrade to a no-op and evaluations (via + get_tracing_client) fall back to cosine-only scoring. + """ + project = get_project_by_id(session=session, project_id=int(project_id)) + if not project or not (project.settings or {}).get("tracing", False): + logger.info( + f"[get_tracing_credential] Tracing disabled | project_id={project_id}" + ) + return None + + return get_provider_credential( + session=session, + org_id=org_id, + project_id=int(project_id), + provider="langfuse", + ) + + def get_providers(*, session: Session, org_id: int, project_id: int) -> list[str]: """Returns a list of all active providers for which credentials are stored.""" creds = get_creds_by_org(session=session, org_id=org_id, project_id=project_id) diff --git a/backend/app/crud/evaluations/batch.py b/backend/app/crud/evaluations/batch.py index 0ce20a4c2..c325e2417 100644 --- a/backend/app/crud/evaluations/batch.py +++ b/backend/app/crud/evaluations/batch.py @@ -70,6 +70,97 @@ def fetch_dataset_items(langfuse: Langfuse, dataset_name: str) -> list[dict[str, return items +def reconcile_tracing_client( + session: Session, + eval_run: EvaluationRun, + langfuse: Langfuse | None, +) -> Langfuse | None: + """Return the live client only if the run's dataset is Langfuse-backed, else + None, so an opt-out dataset is never traced even if the flag is later on.""" + if langfuse is None: + return None + + from app.crud.evaluations.dataset import get_dataset_by_id + + dataset = get_dataset_by_id( + session=session, + dataset_id=eval_run.dataset_id, + organization_id=eval_run.organization_id, + project_id=eval_run.project_id, + ) + return langfuse if (dataset and dataset.langfuse_dataset_id) else None + + +def load_evaluation_dataset_items( + session: Session, + eval_run: EvaluationRun, + langfuse: Langfuse | None, +) -> list[dict[str, Any]]: + """Load dataset items, sourced by the dataset's backing (Langfuse if + langfuse_dataset_id is set, else object store) so ids stay stable across an + eval's build and process even if the tracing flag is toggled in between.""" + # Lazy imports avoid a crud -> services -> crud import cycle. + from app.core.cloud import get_cloud_storage + from app.crud.evaluations.dataset import ( + download_csv_from_object_store, + get_dataset_by_id, + ) + from app.crud.evaluations.score import DEFAULT_CATEGORY + from app.services.evaluations.validators import parse_csv_items + + dataset = get_dataset_by_id( + session=session, + dataset_id=eval_run.dataset_id, + organization_id=eval_run.organization_id, + project_id=eval_run.project_id, + ) + if not dataset: + raise ValueError( + f"Cannot source dataset items for eval_run {eval_run.id}: dataset " + f"{eval_run.dataset_id} not found" + ) + + # Langfuse only when tracing is on AND the dataset is Langfuse-backed; + # otherwise fall back to object store (opt-out, or a Langfuse-backed dataset + # run with tracing off). + if langfuse is not None and dataset.langfuse_dataset_id: + return fetch_dataset_items( + langfuse=langfuse, dataset_name=eval_run.dataset_name + ) + + if not dataset.object_store_url: + raise ValueError( + f"Cannot source dataset items for eval_run {eval_run.id}: dataset " + f"{eval_run.dataset_id} has no object-store backing" + ) + + storage = get_cloud_storage(session=session, project_id=eval_run.project_id) + csv_content = download_csv_from_object_store( + storage=storage, object_store_url=dataset.object_store_url + ) + original_items = parse_csv_items(csv_content) + # JSONB may hand back a str/float; coerce and floor at 1 to keep range() safe. + duplication_factor = max( + 1, int((dataset.dataset_metadata or {}).get("duplication_factor", 1)) + ) + + items: list[dict[str, Any]] = [] + for row_idx, item in enumerate(original_items): + for dup_idx in range(duplication_factor): + items.append( + { + "id": f"item_{row_idx}_{dup_idx}", + "input": {"question": item["question"]}, + "expected_output": {"answer": item["answer"]}, + "metadata": { + "category": item.get("category") or DEFAULT_CATEGORY, + "question_id": f"item_{row_idx}", + }, + } + ) + return items + + def build_openai_evaluation_jsonl( dataset_items: list[dict[str, Any]], openai_params: dict[str, Any] ) -> list[dict[str, Any]]: @@ -153,7 +244,7 @@ def build_google_evaluation_jsonl( def start_evaluation_batch( - langfuse: Langfuse, + langfuse: Langfuse | None, session: Session, eval_run: EvaluationRun, params: dict[str, Any], @@ -176,8 +267,8 @@ def start_evaluation_batch( logger.info( f"[start_evaluation_batch] Starting evaluation batch | run={eval_run.run_name} | provider={provider}" ) - dataset_items = fetch_dataset_items( - langfuse=langfuse, dataset_name=eval_run.dataset_name + dataset_items = load_evaluation_dataset_items( + session=session, eval_run=eval_run, langfuse=langfuse ) base_provider = provider.replace("-native", "") diff --git a/backend/app/crud/evaluations/core.py b/backend/app/crud/evaluations/core.py index ed15f720e..8002d28b2 100644 --- a/backend/app/crud/evaluations/core.py +++ b/backend/app/crud/evaluations/core.py @@ -271,7 +271,7 @@ def _enqueue_eval_completion_notification(eval_run: EvaluationRun) -> None: def get_or_fetch_score( session: Session, eval_run: EvaluationRun, - langfuse: Langfuse, + langfuse: Langfuse | None, force_refetch: bool = False, ) -> EvaluationScore: """ @@ -295,6 +295,15 @@ def get_or_fetch_score( ValueError: If the run is not found in Langfuse Exception: If Langfuse API calls fail """ + # Tracing disabled: no Langfuse traces exist, so return whatever + # cosine-based summary_scores were already computed. + if langfuse is None: + logger.info( + f"[get_or_fetch_score] Tracing off; returning cosine-only score | " + f"evaluation_id={eval_run.id}" + ) + return eval_run.score or {"summary_scores": [], "traces": []} + # Check if score already exists with traces has_traces = eval_run.score is not None and "traces" in eval_run.score if not force_refetch and has_traces: diff --git a/backend/app/crud/evaluations/embeddings.py b/backend/app/crud/evaluations/embeddings.py index 5d1458271..414e37ca8 100644 --- a/backend/app/crud/evaluations/embeddings.py +++ b/backend/app/crud/evaluations/embeddings.py @@ -105,26 +105,18 @@ def build_embedding_jsonl( logger.warning("Skipping result with no item_id") continue - # Get trace_id from mapping - trace_id = trace_id_mapping.get(item_id) - if not trace_id: - logger.warning(f"Skipping item {item_id} - no trace_id found") - skipped.append( - {"item_id": item_id, "trace_id": None, "reason": "missing_trace_id"} - ) - continue + # ref = trace_id when traced, else item_id (opt-out). + ref = trace_id_mapping.get(item_id) or item_id # Empty output/ground_truth can't be embedded; record the reason. if not generated_output or not ground_truth: reason = "empty_output" if not generated_output else "empty_ground_truth" logger.warning(f"Skipping item {item_id} - {reason}") - skipped.append({"item_id": item_id, "trace_id": trace_id, "reason": reason}) + skipped.append({"item_id": item_id, "trace_id": ref, "reason": reason}) continue - # Build the batch request object for Embeddings API - # Use trace_id as BATCH_KEY for direct score updates batch_request = { - BATCH_KEY: trace_id, + BATCH_KEY: ref, "method": "POST", "url": "/v1/embeddings", "body": { diff --git a/backend/app/crud/evaluations/fast.py b/backend/app/crud/evaluations/fast.py index 120da1eb1..79a7149a7 100644 --- a/backend/app/crud/evaluations/fast.py +++ b/backend/app/crud/evaluations/fast.py @@ -40,7 +40,10 @@ load_json_from_object_store, upload_jsonl_to_object_store, ) -from app.crud.evaluations.batch import fetch_dataset_items +from app.crud.evaluations.batch import ( + load_evaluation_dataset_items, + reconcile_tracing_client, +) from app.crud.evaluations.core import ( resolve_model_from_config, save_score, @@ -601,7 +604,7 @@ def _stage3_score_and_trace( *, session: Session, eval_run: EvaluationRun, - langfuse: Langfuse, + langfuse: Langfuse | None, response_results: list[dict[str, Any]], embedding_results: list[dict[str, Any]], log_prefix: str, @@ -634,15 +637,16 @@ def _stage3_score_and_trace( # Per-item cosine scores, keyed by trace_id (Langfuse) and item_id (persisted # records). Items with a trace but no computable score are flagged unscoreable # so they're kept out of avg/std/total_pairs. + # ref = trace_id when traced, else item_id, so cosine persists on opt-out. per_item_scores: list[dict[str, Any]] = [] item_id_to_score: dict[str, float] = {} + item_id_to_ref: dict[str, str] = {} similarities: list[float] = [] - unscoreable: dict[str, str] = {} # {trace_id: reason} + unscoreable: dict[str, str] = {} # {ref: reason} for response in response_results: item_id = response["item_id"] - trace_id = trace_id_mapping.get(item_id) - if not trace_id: - continue + ref = trace_id_mapping.get(item_id) or item_id + item_id_to_ref[item_id] = ref embedding_pair = item_id_to_pair.get(item_id) has_embeddings = ( embedding_pair is not None @@ -652,11 +656,11 @@ def _stage3_score_and_trace( if not has_embeddings: # Classify why this item cannot be scored, for the UI flag. if not response.get("generated_output"): - unscoreable[trace_id] = "empty_output" + unscoreable[ref] = "empty_output" elif not response.get("ground_truth"): - unscoreable[trace_id] = "empty_ground_truth" + unscoreable[ref] = "empty_ground_truth" else: - unscoreable[trace_id] = "embedding_failed" + unscoreable[ref] = "embedding_failed" continue cosine = calculate_cosine_similarity( embedding_pair["output_embedding"], @@ -664,21 +668,27 @@ def _stage3_score_and_trace( ) similarities.append(cosine) item_id_to_score[item_id] = cosine - per_item_scores.append({"trace_id": trace_id, "cosine_similarity": cosine}) + per_item_scores.append( + { + "trace_id": trace_id_mapping.get(item_id), + "cosine_similarity": cosine, + } + ) - # Langfuse write list (cosine + 0-scores for unscoreable items); written - # after completion in run_fast_evaluation. + # Langfuse write list, filtered to real trace_ids (empty on opt-out). unscoreable_writes = [ - {"trace_id": trace_id, "unscoreable": True, "reason": reason} - for trace_id, reason in unscoreable.items() + {"trace_id": trace_id_mapping[item_id], "unscoreable": True, "reason": reason} + for item_id, ref in item_id_to_ref.items() + if item_id in trace_id_mapping and (reason := unscoreable.get(ref)) is not None ] - write_items = per_item_scores + unscoreable_writes + scored_writes = [w for w in per_item_scores if w["trace_id"] is not None] + write_items = scored_writes + unscoreable_writes - # Durable source of truth, persisted by the commit below. + # Durable source of truth, persisted by the commit below. Keyed by ref + # (trace_id when traced, item_id otherwise) so the read path is consistent. eval_run.per_item_scores = { - trace_id_mapping[item_id]: round(float(score), 6) + item_id_to_ref[item_id]: round(float(score), 6) for item_id, score in item_id_to_score.items() - if item_id in trace_id_mapping } eval_run.unscoreable = unscoreable or None @@ -740,15 +750,11 @@ def _stage3_score_and_trace( embedding_raw_results=embedding_raw, ) - # Build the per-trace records in the same shape the batch path persists (via - # fetch_trace_scores_from_langfuse). One record per response that has a - # Langfuse trace; the cosine score is attached when its embedding succeeded. + # Per-item UI records, keyed by ref, so results render on opt-out too. traces: list[TraceData] = [] for response in response_results: item_id = response["item_id"] - trace_id = trace_id_mapping.get(item_id) - if not trace_id: - continue + ref = item_id_to_ref.get(item_id, item_id) trace_scores: list[TraceScore] = [] cosine = item_id_to_score.get(item_id) if cosine is not None: @@ -760,20 +766,20 @@ def _stage3_score_and_trace( "comment": COSINE_SCORE_COMMENT, } ) - elif trace_id in unscoreable: + elif ref in unscoreable: # Placeholder 0-score, excluded from summary stats via the marker. trace_scores.append( { "name": COSINE_SCORE_NAME, "value": 0, "data_type": "NUMERIC", - "comment": f"Cannot compute: {unscoreable[trace_id]}", + "comment": f"Cannot compute: {unscoreable[ref]}", "unscoreable": True, } ) traces.append( { - "trace_id": trace_id, + "trace_id": ref, "question": response.get("question", ""), "llm_answer": response.get("generated_output", ""), "ground_truth_answer": response.get("ground_truth", ""), @@ -801,7 +807,7 @@ def run_fast_evaluation( *, session: Session, openai_client: OpenAI, - langfuse: Langfuse, + langfuse: Langfuse | None, eval_run: EvaluationRun, config: TextLLMParams, ) -> EvaluationRun: @@ -824,8 +830,12 @@ def run_fast_evaluation( update=EvaluationRunUpdate(status="processing"), ) - dataset_items = fetch_dataset_items( - langfuse=langfuse, dataset_name=eval_run.dataset_name + langfuse = reconcile_tracing_client( + session=session, eval_run=eval_run, langfuse=langfuse + ) + + dataset_items = load_evaluation_dataset_items( + session=session, eval_run=eval_run, langfuse=langfuse ) if not dataset_items: raise ValueError( @@ -877,7 +887,7 @@ def run_fast_evaluation( # batch path). is_score_updated tracks the outcome so a cron can retry the # gap from per_item_scores. is_score_updated = True - if write_items: + if langfuse is not None and write_items: try: failed_trace_ids = update_traces_with_cosine_scores( langfuse=langfuse, per_item_scores=write_items diff --git a/backend/app/crud/evaluations/langfuse.py b/backend/app/crud/evaluations/langfuse.py index 9a1d3ca08..9b092ce91 100644 --- a/backend/app/crud/evaluations/langfuse.py +++ b/backend/app/crud/evaluations/langfuse.py @@ -45,7 +45,7 @@ def _write_trace_score( def create_langfuse_dataset_run( - langfuse: Langfuse, + langfuse: Langfuse | None, dataset_name: str, run_name: str, results: list[dict[str, Any]], @@ -87,11 +87,15 @@ def create_langfuse_dataset_run( model: Model name used for evaluation (for cost calculation by Langfuse) Returns: - dict[str, str]: Mapping of item_id to Langfuse trace_id + dict[str, str]: Mapping of item_id to Langfuse trace_id. Empty when + tracing is disabled (langfuse is None). Raises: Exception: If Langfuse operations fail """ + if langfuse is None: + return {} + logger.info( f"[create_langfuse_dataset_run] Creating Langfuse dataset run | " f"run_name={run_name} | dataset={dataset_name} | items={len(results)}" @@ -269,11 +273,11 @@ def update_traces_with_cosine_scores( def upload_dataset_to_langfuse( - langfuse: Langfuse, + langfuse: Langfuse | None, items: list[dict[str, str]], dataset_name: str, duplication_factor: int, -) -> tuple[str, int]: +) -> tuple[str | None, int]: """ Upload a dataset to Langfuse from pre-parsed items. @@ -284,11 +288,15 @@ def upload_dataset_to_langfuse( duplication_factor: Number of times to duplicate each item Returns: - Tuple of (langfuse_dataset_id, total_items_uploaded) + Tuple of (langfuse_dataset_id, total_items_uploaded). Returns + (None, 0) when tracing is disabled (langfuse is None). Raises: Exception: If Langfuse operations fail """ + if langfuse is None: + return None, 0 + logger.info( f"[upload_dataset_to_langfuse] Uploading dataset to Langfuse | " f"dataset={dataset_name} | items={len(items)} | " diff --git a/backend/app/crud/evaluations/processing.py b/backend/app/crud/evaluations/processing.py index 331a9b66f..36b5988a7 100644 --- a/backend/app/crud/evaluations/processing.py +++ b/backend/app/crud/evaluations/processing.py @@ -32,7 +32,10 @@ from app.core.batch.gemini import BatchJobState, extract_text_from_response_dict from app.core.cloud.storage import get_cloud_storage from app.core.storage_utils import load_json_from_object_store -from app.crud.evaluations.batch import fetch_dataset_items +from app.crud.evaluations.batch import ( + load_evaluation_dataset_items, + reconcile_tracing_client, +) from app.crud.evaluations.core import ( persist_score_traces, resolve_model_from_config, @@ -61,7 +64,7 @@ from app.models import EvaluationRun, EvaluationRunUpdate from app.models.batch_job import BatchJob, BatchJobUpdate from app.models.evaluation import RunModeEnum -from app.utils import get_langfuse_client, get_openai_client +from app.utils import get_openai_client, get_tracing_client logger = logging.getLogger(__name__) @@ -340,21 +343,17 @@ def build_trace_skeleton( results: list[dict[str, Any]], trace_id_mapping: dict[str, str], ) -> list[TraceData]: - """ - Build per-trace records (Q&A keyed by Langfuse trace_id, no scores yet) from - parsed evaluation results. Persisted at the response stage so the - embedding-completion step can attach cosine scores and write a complete trace - unit, making cosine display independent of Langfuse. Items without a trace_id - are skipped. - """ + """Build per-item Q&A records (no scores yet), keyed by ref (trace_id when + traced, else item_id) so the results view renders on opt-out too.""" traces: list[TraceData] = [] for result in results: - trace_id = trace_id_mapping.get(result.get("item_id")) - if not trace_id: + item_id = result.get("item_id") + ref = trace_id_mapping.get(item_id) or item_id + if not ref: continue traces.append( { - "trace_id": trace_id, + "trace_id": ref, "question": result.get("question", ""), "llm_answer": result.get("generated_output", ""), "ground_truth_answer": result.get("ground_truth", ""), @@ -369,7 +368,7 @@ async def process_completed_evaluation( eval_run: EvaluationRun, session: Session, openai_client: OpenAI, - langfuse: Langfuse, + langfuse: Langfuse | None, ) -> EvaluationRun: """ Process a completed evaluation batch. @@ -438,9 +437,10 @@ async def process_completed_evaluation( f"[process_completed_evaluation] {log_prefix} Fetching dataset items | dataset={eval_run.dataset_name}" ) dataset_items = await asyncio.to_thread( - fetch_dataset_items, + load_evaluation_dataset_items, + session=session, + eval_run=eval_run, langfuse=langfuse, - dataset_name=eval_run.dataset_name, ) # Step 4: Parse evaluation results @@ -631,7 +631,7 @@ async def process_completed_embedding_batch( eval_run: EvaluationRun, session: Session, openai_client: OpenAI, - langfuse: Langfuse, + langfuse: Langfuse | None, ) -> EvaluationRun: """ Process a completed embedding batch and calculate similarity scores. @@ -767,7 +767,7 @@ async def process_completed_embedding_batch( write_items = per_item_scores + unscoreable_writes # False if any write fails, so a cron can retry the gap from per_item_scores. is_score_updated = True - if write_items: + if langfuse is not None and write_items: try: failed_trace_ids = update_traces_with_cosine_scores( langfuse=langfuse, @@ -858,7 +858,7 @@ async def check_and_process_evaluation( eval_run: EvaluationRun, session: Session, openai_client: OpenAI, - langfuse: Langfuse, + langfuse: Langfuse | None, ) -> dict[str, Any]: """ Check evaluation batch status and process if completed. @@ -887,6 +887,9 @@ async def check_and_process_evaluation( """ log_prefix = f"[org={eval_run.organization_id}][project={eval_run.project_id}][eval={eval_run.id}]" previous_status = eval_run.status + langfuse = reconcile_tracing_client( + session=session, eval_run=eval_run, langfuse=langfuse + ) try: # Check if we need to process embedding batch first @@ -1192,7 +1195,7 @@ async def poll_all_pending_evaluations(session: Session) -> dict[str, Any]: org_id=org_id, project_id=project_id, ) - langfuse = get_langfuse_client( + langfuse = get_tracing_client( session=session, org_id=org_id, project_id=project_id, diff --git a/backend/app/crud/project.py b/backend/app/crud/project.py index be47e24cf..6611d232b 100644 --- a/backend/app/crud/project.py +++ b/backend/app/crud/project.py @@ -1,4 +1,5 @@ import logging +from typing import Any, List, Optional from fastapi import HTTPException from sqlmodel import Session, select @@ -46,6 +47,33 @@ def get_project_by_id(*, session: Session, project_id: int) -> Project | None: return session.get(Project, project_id) +def update_project_settings( + *, session: Session, project_id: int, settings_patch: dict[str, Any] +) -> Project: + """Merge `settings_patch` into the project's `settings` JSONB column. + + Only keys present in `settings_patch` are changed; existing keys are kept. + Reassigns a new dict so SQLAlchemy detects the JSONB mutation. + """ + project = get_project_by_id(session=session, project_id=project_id) + if not project: + logger.warning( + f"[update_project_settings] Project not found | project_id={project_id}" + ) + raise HTTPException(404, "Project not found") + + project.settings = {**(project.settings or {}), **settings_patch} + project.updated_at = now() + session.add(project) + session.commit() + session.refresh(project) + logger.info( + f"[update_project_settings] Settings updated | project_id={project_id}, " + f"keys={list(settings_patch.keys())}" + ) + return project + + def get_project_by_name( *, session: Session, project_name: str, organization_id: int ) -> Project | None: diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index f604d6175..f855398bb 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -180,6 +180,7 @@ ProjectCreate, ProjectPublic, ProjectsPublic, + ProjectSettingsUpdate, ProjectUpdate, ) from .response import ( diff --git a/backend/app/models/project.py b/backend/app/models/project.py index 03b8d0f46..b286ace49 100644 --- a/backend/app/models/project.py +++ b/backend/app/models/project.py @@ -1,7 +1,9 @@ from datetime import datetime -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Any, Optional from uuid import UUID, uuid4 +from sqlalchemy import Column +from sqlalchemy.dialects.postgresql import JSONB from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint from app.core.util import now @@ -47,6 +49,15 @@ class ProjectUpdate(SQLModel): is_active: bool | None = Field(default=None) +# Properties to receive via API when patching project settings. +# Each field maps to a key inside the project's `settings` JSONB column. +class ProjectSettingsUpdate(SQLModel): + tracing: bool | None = Field( + default=None, + description="Enable/disable Langfuse tracing for this project.", + ) + + # Database model for Project class Project(ProjectBase, table=True): """Database model for projects.""" @@ -66,6 +77,19 @@ class Project(ProjectBase, table=True): unique=True, sa_column_kwargs={"comment": "Unique UUID used for cloud storage path"}, ) + settings: dict[str, Any] = Field( + default_factory=dict, + sa_column=Column( + JSONB, + nullable=False, + comment=( + "Project-level settings (JSONB). Keys: 'tracing' (bool) — " + "Langfuse tracing opt-in, off by default to conserve Langfuse " + "rate-limit/credit budget. Gates tracing for both the response " + "path and evaluations (which fall back to cosine-only scoring)." + ), + ), + ) # Foreign keys organization_id: int = Field( @@ -111,6 +135,7 @@ class Project(ProjectBase, table=True): class ProjectPublic(ProjectBase): id: int organization_id: int + settings: dict[str, Any] = {} inserted_at: datetime updated_at: datetime diff --git a/backend/app/services/evaluations/batch_job.py b/backend/app/services/evaluations/batch_job.py index ec13045ee..e97570ffe 100644 --- a/backend/app/services/evaluations/batch_job.py +++ b/backend/app/services/evaluations/batch_job.py @@ -11,9 +11,10 @@ resolve_evaluation_config, start_evaluation_batch, ) +from app.crud.evaluations.batch import reconcile_tracing_client from app.crud.evaluations.core import update_evaluation_run from app.models.evaluation import EvaluationRunUpdate -from app.utils import get_langfuse_client +from app.utils import get_tracing_client logger = logging.getLogger(__name__) @@ -56,9 +57,12 @@ def execute_evaluation_batch_submission( ) return {"success": False, "error": error} - langfuse = get_langfuse_client( + langfuse = get_tracing_client( session=session, org_id=organization_id, project_id=project_id ) + langfuse = reconcile_tracing_client( + session=session, eval_run=run, langfuse=langfuse + ) run = start_evaluation_batch( langfuse=langfuse, session=session, diff --git a/backend/app/services/evaluations/dataset.py b/backend/app/services/evaluations/dataset.py index fe0e89249..c73e36e0e 100644 --- a/backend/app/services/evaluations/dataset.py +++ b/backend/app/services/evaluations/dataset.py @@ -16,7 +16,7 @@ parse_csv_items, sanitize_dataset_name, ) -from app.utils import get_langfuse_client +from app.utils import get_tracing_client logger = logging.getLogger(__name__) @@ -107,35 +107,36 @@ def upload_dataset( ) object_store_url = None - # Step 4: Upload to Langfuse + # Step 4: Upload to Langfuse when tracing is enabled. On opt-out the dataset + # lives only in object store; evaluations source items from there. langfuse_dataset_id = None - try: - langfuse = get_langfuse_client( - session=session, - org_id=organization_id, - project_id=project_id, - ) - - langfuse_dataset_id, _ = upload_dataset_to_langfuse( - langfuse=langfuse, - items=original_items, - dataset_name=dataset_name, - duplication_factor=duplication_factor, - ) + langfuse = get_tracing_client( + session=session, + org_id=organization_id, + project_id=project_id, + ) + if langfuse is not None: + try: + langfuse_dataset_id, _ = upload_dataset_to_langfuse( + langfuse=langfuse, + items=original_items, + dataset_name=dataset_name, + duplication_factor=duplication_factor, + ) - logger.info( - f"[upload_dataset] Successfully uploaded dataset to Langfuse | " - f"dataset={dataset_name} | id={langfuse_dataset_id}" - ) + logger.info( + f"[upload_dataset] Successfully uploaded dataset to Langfuse | " + f"dataset={dataset_name} | id={langfuse_dataset_id}" + ) - except Exception as e: - logger.error( - f"[upload_dataset] Failed to upload dataset to Langfuse | {e}", - exc_info=True, - ) - raise HTTPException( - status_code=500, detail=f"Failed to upload dataset to Langfuse: {e}" - ) + except Exception as e: + logger.error( + f"[upload_dataset] Failed to upload dataset to Langfuse | {e}", + exc_info=True, + ) + raise HTTPException( + status_code=500, detail=f"Failed to upload dataset to Langfuse: {e}" + ) # Step 5: Store metadata in database metadata = { diff --git a/backend/app/services/evaluations/evaluation.py b/backend/app/services/evaluations/evaluation.py index c01bfbeb9..c8bfaec8d 100644 --- a/backend/app/services/evaluations/evaluation.py +++ b/backend/app/services/evaluations/evaluation.py @@ -29,7 +29,7 @@ from app.models.evaluation import EvaluationRun, EvaluationRunUpdate, RunModeEnum from app.models.llm.constants import CompletionType from app.services.llm.providers import LLMProvider -from app.utils import get_langfuse_client +from app.utils import get_tracing_client logger = logging.getLogger(__name__) @@ -263,11 +263,11 @@ def validate_and_start_batch_evaluation( f"langfuse_id={dataset.langfuse_dataset_id}" ) - if not dataset.langfuse_dataset_id: + if not dataset.langfuse_dataset_id and not dataset.object_store_url: raise HTTPException( status_code=400, - detail=f"Dataset {dataset_id} does not have a Langfuse dataset ID. " - "Please ensure Langfuse credentials were configured when the dataset was created.", + detail=f"Dataset {dataset_id} has no Langfuse nor object-store " + "backing; cannot run evaluation.", ) # Step 2: Resolve config from stored config management @@ -505,12 +505,36 @@ def get_evaluation_with_scores( ) # Fetch fresh scores from Langfuse (first sync, or resync). - langfuse = get_langfuse_client( + langfuse = get_tracing_client( session=session, org_id=organization_id, project_id=project_id, ) + # Opt-out: resync needs Langfuse (400); a normal read serves durable cosine. + if langfuse is None: + if resync_score: + raise HTTPException( + status_code=400, + detail="Tracing is disabled for this project; cannot resync " + "scores from Langfuse.", + ) + cosine_score, _ = merge_scores_step_forward( + existing_score={ + "summary_scores": (eval_run.score or {}).get("summary_scores", []), + "traces": cached_traces or [], + }, + fresh_score={"summary_scores": [], "traces": []}, + per_item_scores=eval_run.per_item_scores, + ) + apply_cosine_breakdown( + cosine_score["summary_scores"], + total_items=eval_run.total_items, + unscoreable=eval_run.unscoreable, + ) + eval_run.score = _attach_category_metrics(cosine_score) + return eval_run, None + dataset_name = eval_run.dataset_name run_name = eval_run.run_name eval_run_id = eval_run.id diff --git a/backend/app/services/evaluations/fast.py b/backend/app/services/evaluations/fast.py index e8fd2e650..9a318be6c 100644 --- a/backend/app/services/evaluations/fast.py +++ b/backend/app/services/evaluations/fast.py @@ -26,7 +26,7 @@ from app.models.llm.request import TextLLMParams from app.services.evaluations.evaluation import create_evaluation_run_or_409 from app.services.llm.providers import LLMProvider -from app.utils import get_langfuse_client, get_openai_client +from app.utils import get_openai_client, get_tracing_client logger = logging.getLogger(__name__) @@ -86,12 +86,12 @@ def validate_and_start_fast_evaluation( "organization/project" ), ) - if not dataset.langfuse_dataset_id: + if not dataset.langfuse_dataset_id and not dataset.object_store_url: raise HTTPException( status_code=400, detail=( - f"Dataset {dataset_id} has no Langfuse dataset id; cannot run " - "evaluation." + f"Dataset {dataset_id} has no Langfuse nor object-store backing; " + "cannot run evaluation." ), ) @@ -244,7 +244,7 @@ def execute_fast_evaluation(*, eval_run_id: int) -> None: org_id=eval_run.organization_id, project_id=eval_run.project_id, ) - langfuse_client = get_langfuse_client( + langfuse_client = get_tracing_client( session=session, org_id=eval_run.organization_id, project_id=eval_run.project_id, diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index b9cb25b5a..67049246d 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -29,7 +29,7 @@ suppress_http_instrumentation, ) from app.crud.config import ConfigVersionCrud -from app.crud.credentials import get_provider_credential +from app.crud.credentials import get_provider_credential, get_tracing_credential from app.crud.model_config import validate_blob_model_or_raise from app.crud.jobs import JobCrud from app.crud.llm import ( @@ -1273,11 +1273,10 @@ def execute_job( job_id=job_uuid, job_update=JobUpdate(status=JobStatus.PROCESSING) ) - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=organization_id, project_id=project_id, - provider="langfuse", ) result = execute_llm_call( @@ -1459,11 +1458,10 @@ def execute_chain_job( f"chain_id={chain_uuid}, job_id={job_uuid}" ) - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=organization_id, project_id=project_id, - provider="langfuse", ) context = ChainContext( diff --git a/backend/app/services/response/response.py b/backend/app/services/response/response.py index 2c021649d..43f95cafb 100644 --- a/backend/app/services/response/response.py +++ b/backend/app/services/response/response.py @@ -12,7 +12,7 @@ from app.crud import ( JobCrud, get_assistant_by_id, - get_provider_credential, + get_tracing_credential, create_conversation, get_ancestor_id_from_response, get_conversation_by_ancestor_id, @@ -236,10 +236,9 @@ def process_response( except HTTPException as e: return _fail_job(job_id, str(e.detail)) - langfuse_credentials = get_provider_credential( + langfuse_credentials = get_tracing_credential( session=session, org_id=organization_id, - provider="langfuse", project_id=project_id, ) diff --git a/backend/app/tests/api/routes/test_evaluation.py b/backend/app/tests/api/routes/test_evaluation.py index 7653f6182..d16c780ef 100644 --- a/backend/app/tests/api/routes/test_evaluation.py +++ b/backend/app/tests/api/routes/test_evaluation.py @@ -79,7 +79,7 @@ def test_upload_dataset_valid_csv( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -163,7 +163,7 @@ def test_upload_dataset_empty_rows( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -211,7 +211,7 @@ def test_upload_with_default_duplication( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -255,7 +255,7 @@ def test_upload_with_custom_duplication( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -300,7 +300,7 @@ def test_upload_with_description( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -404,7 +404,7 @@ def test_upload_with_duplication_factor_boundary_minimum( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, patch( - "app.services.evaluations.dataset.get_langfuse_client" + "app.services.evaluations.dataset.get_tracing_client" ) as mock_get_langfuse_client, patch( "app.services.evaluations.dataset.upload_dataset_to_langfuse" @@ -439,22 +439,28 @@ def test_upload_with_duplication_factor_boundary_minimum( class TestDatasetUploadErrors: """Test error handling.""" - def test_upload_langfuse_configuration_fails( + def test_upload_succeeds_without_langfuse_when_tracing_off( self, client: TestClient, user_api_key_header: dict[str, str], valid_csv_content: str, ) -> None: - """Test when Langfuse client configuration fails.""" + """With tracing opt-out (get_tracing_client returns None), the dataset is + created from object store only; no Langfuse upload, no error.""" with ( patch("app.core.cloud.get_cloud_storage") as _mock_storage, patch( "app.services.evaluations.dataset.upload_csv_to_object_store" ) as mock_store_upload, - patch("app.crud.credentials.get_provider_credential") as mock_get_cred, + patch( + "app.services.evaluations.dataset.get_tracing_client" + ) as mock_get_tracing_client, + patch( + "app.services.evaluations.dataset.upload_dataset_to_langfuse" + ) as mock_langfuse_upload, ): mock_store_upload.return_value = "s3://bucket/datasets/test_dataset.csv" - mock_get_cred.return_value = None + mock_get_tracing_client.return_value = None filename, file_obj = create_csv_file(valid_csv_content) @@ -468,17 +474,11 @@ def test_upload_langfuse_configuration_fails( headers=user_api_key_header, ) - # Accept either 400 (credentials not configured) or 500 (configuration/auth fails) - assert response.status_code in [400, 500] - response_data = response.json() - error_str = response_data.get( - "detail", response_data.get("message", str(response_data)) - ) - assert ( - "langfuse" in error_str.lower() - or "credential" in error_str.lower() - or "unauthorized" in error_str.lower() - ) + assert response.status_code == 200, response.text + data = response.json()["data"] + assert data["langfuse_dataset_id"] is None + assert data["object_store_url"] == "s3://bucket/datasets/test_dataset.csv" + mock_langfuse_upload.assert_not_called() def test_upload_invalid_csv_format( self, client: TestClient, user_api_key_header: dict[str, str] diff --git a/backend/app/tests/api/routes/test_evaluation_fast.py b/backend/app/tests/api/routes/test_evaluation_fast.py index 132d8fe64..613d64f20 100644 --- a/backend/app/tests/api/routes/test_evaluation_fast.py +++ b/backend/app/tests/api/routes/test_evaluation_fast.py @@ -564,7 +564,9 @@ def _fast_pipeline_mocks(): not need a model_config row. """ with ( - patch("app.crud.evaluations.fast.fetch_dataset_items") as mock_fetch_items, + patch( + "app.crud.evaluations.fast.load_evaluation_dataset_items" + ) as mock_fetch_items, patch( "app.crud.evaluations.fast._upload_unit_to_s3", side_effect=lambda **kw: f"s3://bucket/{kw['filename']}", diff --git a/backend/app/tests/crud/evaluations/test_batch.py b/backend/app/tests/crud/evaluations/test_batch.py new file mode 100644 index 000000000..d70f47d8d --- /dev/null +++ b/backend/app/tests/crud/evaluations/test_batch.py @@ -0,0 +1,128 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from app.crud.evaluations.batch import load_evaluation_dataset_items + + +def _eval_run() -> SimpleNamespace: + return SimpleNamespace( + id=1, + dataset_id=7, + dataset_name="ds", + organization_id=1, + project_id=1, + ) + + +class TestLoadEvaluationDatasetItems: + def test_langfuse_backed_dataset_delegates_to_langfuse(self) -> None: + """A Langfuse-backed dataset (langfuse_dataset_id set) reads from Langfuse.""" + langfuse = MagicMock() + expected = [{"id": "lf_1", "input": {"question": "q"}}] + dataset = SimpleNamespace( + id=7, + langfuse_dataset_id="lf_ds_1", + object_store_url="s3://bucket/ds.csv", + dataset_metadata={}, + ) + + with ( + patch( + "app.crud.evaluations.dataset.get_dataset_by_id", + return_value=dataset, + ), + patch( + "app.crud.evaluations.batch.fetch_dataset_items", + return_value=expected, + ) as mock_fetch, + ): + items = load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=langfuse + ) + + mock_fetch.assert_called_once_with(langfuse=langfuse, dataset_name="ds") + assert items == expected + + def test_langfuse_backed_but_tracing_off_falls_back_to_object_store(self) -> None: + """A Langfuse-backed dataset run with tracing off sources from object + store (cosine-only) rather than failing.""" + csv_bytes = b"question,answer\nq1,a1\n" + dataset = SimpleNamespace( + id=7, + langfuse_dataset_id="lf_ds_1", + object_store_url="s3://bucket/ds.csv", + dataset_metadata={}, + ) + + with ( + patch( + "app.crud.evaluations.dataset.get_dataset_by_id", + return_value=dataset, + ), + patch( + "app.crud.evaluations.dataset.download_csv_from_object_store", + return_value=csv_bytes, + ), + patch("app.core.cloud.get_cloud_storage", return_value=MagicMock()), + ): + items = load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=None + ) + + assert [i["id"] for i in items] == ["item_0_0"] + + def test_object_store_backed_sources_from_object_store(self) -> None: + """An object-store-backed dataset (no langfuse_dataset_id) sources items + from the CSV with deterministic ids and applied duplication — regardless + of whether a live Langfuse client is present.""" + csv_bytes = b"question,answer\nq1,a1\nq2,a2\n" + dataset = SimpleNamespace( + id=7, + langfuse_dataset_id=None, + object_store_url="s3://bucket/ds.csv", + dataset_metadata={"duplication_factor": 2}, + ) + + with ( + patch( + "app.crud.evaluations.dataset.get_dataset_by_id", + return_value=dataset, + ), + patch( + "app.crud.evaluations.dataset.download_csv_from_object_store", + return_value=csv_bytes, + ), + patch("app.core.cloud.get_cloud_storage", return_value=MagicMock()), + ): + # Even with a live client, an object-store-backed dataset ignores it + # for DATA so ids stay stable. + items = load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=MagicMock() + ) + + assert [i["id"] for i in items] == [ + "item_0_0", + "item_0_1", + "item_1_0", + "item_1_1", + ] + assert items[0]["input"] == {"question": "q1"} + assert items[0]["expected_output"] == {"answer": "a1"} + assert items[2]["input"] == {"question": "q2"} + + def test_object_store_backed_without_url_raises(self) -> None: + """No Langfuse and no object-store URL cannot source items.""" + dataset = SimpleNamespace( + id=7, langfuse_dataset_id=None, object_store_url=None, dataset_metadata={} + ) + + with patch( + "app.crud.evaluations.dataset.get_dataset_by_id", return_value=dataset + ): + try: + load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=None + ) + raise AssertionError("expected ValueError") + except ValueError as e: + assert "object-store" in str(e) diff --git a/backend/app/tests/crud/evaluations/test_embeddings.py b/backend/app/tests/crud/evaluations/test_embeddings.py index 9336abdce..d01e42412 100644 --- a/backend/app/tests/crud/evaluations/test_embeddings.py +++ b/backend/app/tests/crud/evaluations/test_embeddings.py @@ -132,8 +132,9 @@ def test_build_embedding_jsonl_missing_item_id(self) -> None: # Item with no item_id is dropped silently (cannot be keyed at all). assert skipped == [] - def test_build_embedding_jsonl_missing_trace_id(self) -> None: - """Items whose item_id has no mapped trace_id are reported as skipped.""" + def test_build_embedding_jsonl_falls_back_to_item_id(self) -> None: + """Items with no mapped trace_id use item_id as the key (opt-out), so + they are still embedded rather than skipped.""" results = [ { "item_id": "item_1", @@ -153,11 +154,9 @@ def test_build_embedding_jsonl_missing_trace_id(self) -> None: jsonl_data, skipped = build_embedding_jsonl(results, trace_id_mapping) - assert len(jsonl_data) == 1 - assert jsonl_data[0]["custom_id"] == "trace_2" - assert skipped == [ - {"item_id": "item_1", "trace_id": None, "reason": "missing_trace_id"} - ] + assert len(jsonl_data) == 2 + assert {j["custom_id"] for j in jsonl_data} == {"item_1", "trace_2"} + assert skipped == [] class TestParseEmbeddingResults: diff --git a/backend/app/tests/crud/evaluations/test_processing.py b/backend/app/tests/crud/evaluations/test_processing.py index a874b7f2f..aecf9ffb8 100644 --- a/backend/app/tests/crud/evaluations/test_processing.py +++ b/backend/app/tests/crud/evaluations/test_processing.py @@ -515,7 +515,7 @@ def eval_run_with_batch(self, db: Session, test_dataset) -> EvaluationRun: @pytest.mark.asyncio @patch("app.crud.evaluations.processing.download_batch_results") - @patch("app.crud.evaluations.processing.fetch_dataset_items") + @patch("app.crud.evaluations.processing.load_evaluation_dataset_items") @patch("app.crud.evaluations.processing.create_langfuse_dataset_run") @patch("app.crud.evaluations.processing.start_embedding_batch") @patch("app.crud.evaluations.processing.upload_batch_results_to_object_store") @@ -599,7 +599,7 @@ async def test_process_completed_evaluation_success( @pytest.mark.asyncio @patch("app.crud.evaluations.processing.persist_score_traces") @patch("app.crud.evaluations.processing.download_batch_results") - @patch("app.crud.evaluations.processing.fetch_dataset_items") + @patch("app.crud.evaluations.processing.load_evaluation_dataset_items") @patch("app.crud.evaluations.processing.create_langfuse_dataset_run") @patch("app.crud.evaluations.processing.start_embedding_batch") @patch("app.crud.evaluations.processing.upload_batch_results_to_object_store") @@ -667,7 +667,7 @@ async def test_process_completed_evaluation_persists_trace_skeleton( @pytest.mark.asyncio @patch("app.crud.evaluations.processing.download_batch_results") - @patch("app.crud.evaluations.processing.fetch_dataset_items") + @patch("app.crud.evaluations.processing.load_evaluation_dataset_items") async def test_process_completed_evaluation_no_results( self, mock_fetch_dataset, @@ -1676,7 +1676,7 @@ async def test_poll_all_pending_evaluations_no_pending( @pytest.mark.asyncio @patch("app.crud.evaluations.processing.check_and_process_evaluation") @patch("app.crud.evaluations.processing.get_openai_client") - @patch("app.crud.evaluations.processing.get_langfuse_client") + @patch("app.crud.evaluations.processing.get_tracing_client") async def test_poll_all_pending_evaluations_with_runs( self, mock_langfuse_client, diff --git a/backend/app/tests/services/evaluations/test_evaluation_service_s3.py b/backend/app/tests/services/evaluations/test_evaluation_service_s3.py index 722318ac2..44a9a98fe 100644 --- a/backend/app/tests/services/evaluations/test_evaluation_service_s3.py +++ b/backend/app/tests/services/evaluations/test_evaluation_service_s3.py @@ -113,7 +113,7 @@ def test_returns_db_traces_when_no_s3_url( @patch("app.services.evaluations.evaluation.save_score") @patch("app.services.evaluations.evaluation.fetch_trace_scores_from_langfuse") - @patch("app.services.evaluations.evaluation.get_langfuse_client") + @patch("app.services.evaluations.evaluation.get_tracing_client") @patch("app.services.evaluations.evaluation.get_evaluation_run_by_id") @patch("app.services.evaluations.evaluation.load_json_from_object_store") @patch("app.services.evaluations.evaluation.get_cloud_storage") @@ -166,7 +166,100 @@ def test_resync_merges_cache_with_langfuse( @patch("app.services.evaluations.evaluation.save_score") @patch("app.services.evaluations.evaluation.fetch_trace_scores_from_langfuse") - @patch("app.services.evaluations.evaluation.get_langfuse_client") + @patch("app.services.evaluations.evaluation.get_tracing_client") + @patch("app.services.evaluations.evaluation.get_evaluation_run_by_id") + @patch("app.services.evaluations.evaluation.load_json_from_object_store") + @patch("app.services.evaluations.evaluation.get_cloud_storage") + def test_resync_opt_out_raises( + self, + mock_get_storage: MagicMock, + mock_load: MagicMock, + mock_get_eval: MagicMock, + mock_get_langfuse: MagicMock, + mock_fetch_langfuse: MagicMock, + mock_save_score: MagicMock, + eval_run_factory: Callable[..., MagicMock], + ) -> None: + """resync=True with tracing off returns 400 (nothing to resync from).""" + eval_run = eval_run_factory( + id=101, + status="completed", + score={"summary_scores": [{"name": "cosine_similarity", "avg": 0.8}]}, + score_trace_url="s3://bucket/traces.json", + dataset_name="test_dataset", + run_name="test_run", + ) + eval_run.per_item_scores = {"item_0_0": 0.8} + eval_run.total_items = 1 + eval_run.unscoreable = None + mock_get_eval.return_value = eval_run + mock_get_storage.return_value = MagicMock() + mock_load.return_value = [{"trace_id": "item_0_0", "scores": []}] + mock_get_langfuse.return_value = None + + with pytest.raises(HTTPException) as exc: + get_evaluation_with_scores( + session=MagicMock(), + evaluation_id=101, + organization_id=1, + project_id=1, + get_trace_info=True, + resync_score=True, + ) + + assert exc.value.status_code == 400 + mock_fetch_langfuse.assert_not_called() + mock_save_score.assert_not_called() + + @patch("app.services.evaluations.evaluation.save_score") + @patch("app.services.evaluations.evaluation.fetch_trace_scores_from_langfuse") + @patch("app.services.evaluations.evaluation.get_tracing_client") + @patch("app.services.evaluations.evaluation.get_evaluation_run_by_id") + @patch("app.services.evaluations.evaluation.load_json_from_object_store") + @patch("app.services.evaluations.evaluation.get_cloud_storage") + def test_read_opt_out_serves_cosine_without_langfuse( + self, + mock_get_storage: MagicMock, + mock_load: MagicMock, + mock_get_eval: MagicMock, + mock_get_langfuse: MagicMock, + mock_fetch_langfuse: MagicMock, + mock_save_score: MagicMock, + eval_run_factory: Callable[..., MagicMock], + ) -> None: + """Non-resync read with tracing off serves durable cosine from cache.""" + eval_run = eval_run_factory( + id=102, + status="completed", + score={"summary_scores": [{"name": "cosine_similarity", "avg": 0.8}]}, + score_trace_url=None, + dataset_name="test_dataset", + run_name="test_run", + ) + eval_run.per_item_scores = {"item_0_0": 0.8} + eval_run.total_items = 1 + eval_run.unscoreable = None + mock_get_eval.return_value = eval_run + mock_get_storage.return_value = MagicMock() + mock_load.return_value = None + mock_get_langfuse.return_value = None + + result, error = get_evaluation_with_scores( + session=MagicMock(), + evaluation_id=102, + organization_id=1, + project_id=1, + get_trace_info=True, + resync_score=False, + ) + + assert error is None + mock_fetch_langfuse.assert_not_called() + assert result.score["summary_scores"][0]["name"] == "cosine_similarity" + + @patch("app.services.evaluations.evaluation.save_score") + @patch("app.services.evaluations.evaluation.fetch_trace_scores_from_langfuse") + @patch("app.services.evaluations.evaluation.get_tracing_client") @patch("app.services.evaluations.evaluation.get_evaluation_run_by_id") @patch("app.services.evaluations.evaluation.load_json_from_object_store") @patch("app.services.evaluations.evaluation.get_cloud_storage") diff --git a/backend/app/utils.py b/backend/app/utils.py index f3f2fb248..c9b5a0a62 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -31,7 +31,7 @@ from app.core import security from app.core.audio_utils import AudioRef from app.core.config import settings -from app.crud.credentials import get_provider_credential +from app.crud.credentials import get_provider_credential, get_tracing_credential from app.models.llm.request import ( TextInput, AudioInput, @@ -387,6 +387,43 @@ def get_langfuse_client(session: Session, org_id: int, project_id: int) -> Langf ) +def get_tracing_client( + session: Session, org_id: int, project_id: int +) -> Langfuse | None: + """Return the Langfuse client when the project opted into tracing, else None + (never raises), so evaluations degrade to cosine-only instead of failing.""" + credentials = get_tracing_credential( + session=session, + org_id=org_id, + project_id=project_id, + ) + + if not credentials or not all( + key in credentials for key in ["public_key", "secret_key", "host"] + ): + logger.info( + f"[get_tracing_client] Tracing off or credentials missing; " + f"skipping Langfuse | project_id: {project_id}" + ) + return None + + try: + return Langfuse( + public_key=credentials["public_key"], + secret_key=credentials["secret_key"], + host=credentials["host"], + timeout=60, + ) + except Exception as e: + logger.warning( + f"[get_tracing_client] Failed to configure Langfuse client; " + f"continuing without tracing | project_id: {project_id} | " + f"error: {str(e)}", + exc_info=True, + ) + return None + + def handle_openai_error(e: openai.OpenAIError) -> str: if hasattr(e, "body") and isinstance(e.body, dict) and "message" in e.body: return e.body["message"] From 0afe5fb6eb62fece7962bb62cb7430f0d698f937 Mon Sep 17 00:00:00 2001 From: Prashant Vasudevan <71649489+vprashrex@users.noreply.github.com> Date: Fri, 3 Jul 2026 12:50:30 +0530 Subject: [PATCH 2/3] feat(langfuse): Refactor tracing client integration and improve dataset loading logic --- backend/app/api/routes/project.py | 2 +- backend/app/crud/credentials.py | 15 ++- backend/app/crud/evaluations/batch.py | 55 ++++----- backend/app/crud/evaluations/fast.py | 4 +- backend/app/crud/evaluations/processing.py | 13 +- backend/app/crud/project.py | 5 +- backend/app/models/project.py | 1 - backend/app/services/evaluations/batch_job.py | 4 +- .../app/tests/crud/evaluations/test_batch.py | 111 +++++++----------- 9 files changed, 93 insertions(+), 117 deletions(-) diff --git a/backend/app/api/routes/project.py b/backend/app/api/routes/project.py index e64e492c7..95c25674b 100644 --- a/backend/app/api/routes/project.py +++ b/backend/app/api/routes/project.py @@ -92,7 +92,7 @@ def update_project_settings_route( session: SessionDep, auth_context: AuthContextDep, settings_in: ProjectSettingsUpdate, -): +) -> APIResponse[ProjectPublic]: settings_patch = settings_in.model_dump(exclude_unset=True) if not settings_patch: raise HTTPException(status_code=400, detail="No settings provided") diff --git a/backend/app/crud/credentials.py b/backend/app/crud/credentials.py index 0e0151e6f..915dbb0be 100644 --- a/backend/app/crud/credentials.py +++ b/backend/app/crud/credentials.py @@ -181,17 +181,24 @@ def get_tracing_credential( observe_llm_execution degrade to a no-op and evaluations (via get_tracing_client) fall back to cosine-only scoring. """ - project = get_project_by_id(session=session, project_id=int(project_id)) - if not project or not (project.settings or {}).get("tracing", False): + try: + pid = int(project_id) + except (TypeError, ValueError): logger.info( - f"[get_tracing_credential] Tracing disabled | project_id={project_id}" + f"[get_tracing_credential] Invalid project_id; tracing off | " + f"project_id={project_id}" ) return None + project = get_project_by_id(session=session, project_id=pid) + if not project or not (project.settings or {}).get("tracing", False): + logger.info(f"[get_tracing_credential] Tracing disabled | project_id={pid}") + return None + return get_provider_credential( session=session, org_id=org_id, - project_id=int(project_id), + project_id=pid, provider="langfuse", ) diff --git a/backend/app/crud/evaluations/batch.py b/backend/app/crud/evaluations/batch.py index c325e2417..271178245 100644 --- a/backend/app/crud/evaluations/batch.py +++ b/backend/app/crud/evaluations/batch.py @@ -20,6 +20,12 @@ start_batch_job, ) from app.core.batch.client import GeminiClient +from app.core.cloud import get_cloud_storage +from app.crud.evaluations.dataset import ( + download_csv_from_object_store, + get_dataset_by_id, +) +from app.crud.evaluations.score import DEFAULT_CATEGORY from app.models import EvaluationRun from app.models.batch_job import BatchJobType from app.services.llm.mappers import ( @@ -70,7 +76,7 @@ def fetch_dataset_items(langfuse: Langfuse, dataset_name: str) -> list[dict[str, return items -def reconcile_tracing_client( +def use_langfuse_client( session: Session, eval_run: EvaluationRun, langfuse: Langfuse | None, @@ -80,8 +86,6 @@ def reconcile_tracing_client( if langfuse is None: return None - from app.crud.evaluations.dataset import get_dataset_by_id - dataset = get_dataset_by_id( session=session, dataset_id=eval_run.dataset_id, @@ -96,18 +100,8 @@ def load_evaluation_dataset_items( eval_run: EvaluationRun, langfuse: Langfuse | None, ) -> list[dict[str, Any]]: - """Load dataset items, sourced by the dataset's backing (Langfuse if - langfuse_dataset_id is set, else object store) so ids stay stable across an - eval's build and process even if the tracing flag is toggled in between.""" - # Lazy imports avoid a crud -> services -> crud import cycle. - from app.core.cloud import get_cloud_storage - from app.crud.evaluations.dataset import ( - download_csv_from_object_store, - get_dataset_by_id, - ) - from app.crud.evaluations.score import DEFAULT_CATEGORY - from app.services.evaluations.validators import parse_csv_items - + """Load dataset items from Langfuse when a (reconciled) client is present, + else from the dataset's object-store CSV.""" dataset = get_dataset_by_id( session=session, dataset_id=eval_run.dataset_id, @@ -115,31 +109,28 @@ def load_evaluation_dataset_items( project_id=eval_run.project_id, ) if not dataset: - raise ValueError( - f"Cannot source dataset items for eval_run {eval_run.id}: dataset " - f"{eval_run.dataset_id} not found" - ) + raise ValueError(f"Dataset {eval_run.dataset_id} not found") - # Langfuse only when tracing is on AND the dataset is Langfuse-backed; - # otherwise fall back to object store (opt-out, or a Langfuse-backed dataset - # run with tracing off). - if langfuse is not None and dataset.langfuse_dataset_id: - return fetch_dataset_items( - langfuse=langfuse, dataset_name=eval_run.dataset_name - ) + if langfuse is not None: + return fetch_dataset_items(langfuse=langfuse, dataset_name=eval_run.dataset_name) + + return _load_items_from_object_store(session=session, dataset=dataset) + + +def _load_items_from_object_store( + session: Session, dataset: Any +) -> list[dict[str, Any]]: + """Load items from the dataset's object-store CSV with deterministic ids.""" + from app.services.evaluations.validators import parse_csv_items if not dataset.object_store_url: - raise ValueError( - f"Cannot source dataset items for eval_run {eval_run.id}: dataset " - f"{eval_run.dataset_id} has no object-store backing" - ) + raise ValueError(f"Dataset {dataset.id} has no object-store backing") - storage = get_cloud_storage(session=session, project_id=eval_run.project_id) + storage = get_cloud_storage(session=session, project_id=dataset.project_id) csv_content = download_csv_from_object_store( storage=storage, object_store_url=dataset.object_store_url ) original_items = parse_csv_items(csv_content) - # JSONB may hand back a str/float; coerce and floor at 1 to keep range() safe. duplication_factor = max( 1, int((dataset.dataset_metadata or {}).get("duplication_factor", 1)) ) diff --git a/backend/app/crud/evaluations/fast.py b/backend/app/crud/evaluations/fast.py index 79a7149a7..a1e0335e0 100644 --- a/backend/app/crud/evaluations/fast.py +++ b/backend/app/crud/evaluations/fast.py @@ -42,7 +42,7 @@ ) from app.crud.evaluations.batch import ( load_evaluation_dataset_items, - reconcile_tracing_client, + use_langfuse_client, ) from app.crud.evaluations.core import ( resolve_model_from_config, @@ -830,7 +830,7 @@ def run_fast_evaluation( update=EvaluationRunUpdate(status="processing"), ) - langfuse = reconcile_tracing_client( + langfuse = use_langfuse_client( session=session, eval_run=eval_run, langfuse=langfuse ) diff --git a/backend/app/crud/evaluations/processing.py b/backend/app/crud/evaluations/processing.py index 36b5988a7..b346405ed 100644 --- a/backend/app/crud/evaluations/processing.py +++ b/backend/app/crud/evaluations/processing.py @@ -34,7 +34,7 @@ from app.core.storage_utils import load_json_from_object_store from app.crud.evaluations.batch import ( load_evaluation_dataset_items, - reconcile_tracing_client, + use_langfuse_client, ) from app.crud.evaluations.core import ( persist_score_traces, @@ -343,8 +343,13 @@ def build_trace_skeleton( results: list[dict[str, Any]], trace_id_mapping: dict[str, str], ) -> list[TraceData]: - """Build per-item Q&A records (no scores yet), keyed by ref (trace_id when - traced, else item_id) so the results view renders on opt-out too.""" + """ + Build per-trace records (Q&A keyed by Langfuse trace_id, no scores yet) from + parsed evaluation results. Persisted at the response stage so the + embedding-completion step can attach cosine scores and write a complete trace + unit, making cosine display independent of Langfuse. Items without a trace_id + are skipped. + """ traces: list[TraceData] = [] for result in results: item_id = result.get("item_id") @@ -887,7 +892,7 @@ async def check_and_process_evaluation( """ log_prefix = f"[org={eval_run.organization_id}][project={eval_run.project_id}][eval={eval_run.id}]" previous_status = eval_run.status - langfuse = reconcile_tracing_client( + langfuse = use_langfuse_client( session=session, eval_run=eval_run, langfuse=langfuse ) diff --git a/backend/app/crud/project.py b/backend/app/crud/project.py index 6611d232b..1c5332799 100644 --- a/backend/app/crud/project.py +++ b/backend/app/crud/project.py @@ -55,7 +55,10 @@ def update_project_settings( Only keys present in `settings_patch` are changed; existing keys are kept. Reassigns a new dict so SQLAlchemy detects the JSONB mutation. """ - project = get_project_by_id(session=session, project_id=project_id) + # Lock the row so concurrent patches merge serially, not lost-update. + project = session.exec( + select(Project).where(Project.id == project_id).with_for_update() + ).first() if not project: logger.warning( f"[update_project_settings] Project not found | project_id={project_id}" diff --git a/backend/app/models/project.py b/backend/app/models/project.py index b286ace49..f5902402a 100644 --- a/backend/app/models/project.py +++ b/backend/app/models/project.py @@ -50,7 +50,6 @@ class ProjectUpdate(SQLModel): # Properties to receive via API when patching project settings. -# Each field maps to a key inside the project's `settings` JSONB column. class ProjectSettingsUpdate(SQLModel): tracing: bool | None = Field( default=None, diff --git a/backend/app/services/evaluations/batch_job.py b/backend/app/services/evaluations/batch_job.py index e97570ffe..fbd7b25aa 100644 --- a/backend/app/services/evaluations/batch_job.py +++ b/backend/app/services/evaluations/batch_job.py @@ -11,7 +11,7 @@ resolve_evaluation_config, start_evaluation_batch, ) -from app.crud.evaluations.batch import reconcile_tracing_client +from app.crud.evaluations.batch import use_langfuse_client from app.crud.evaluations.core import update_evaluation_run from app.models.evaluation import EvaluationRunUpdate from app.utils import get_tracing_client @@ -60,7 +60,7 @@ def execute_evaluation_batch_submission( langfuse = get_tracing_client( session=session, org_id=organization_id, project_id=project_id ) - langfuse = reconcile_tracing_client( + langfuse = use_langfuse_client( session=session, eval_run=run, langfuse=langfuse ) run = start_evaluation_batch( diff --git a/backend/app/tests/crud/evaluations/test_batch.py b/backend/app/tests/crud/evaluations/test_batch.py index d70f47d8d..df1c3ec54 100644 --- a/backend/app/tests/crud/evaluations/test_batch.py +++ b/backend/app/tests/crud/evaluations/test_batch.py @@ -1,6 +1,8 @@ from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + from app.crud.evaluations.batch import load_evaluation_dataset_items @@ -14,22 +16,25 @@ def _eval_run() -> SimpleNamespace: ) +def _dataset(**kw) -> SimpleNamespace: + return SimpleNamespace( + id=7, + project_id=1, + langfuse_dataset_id=kw.get("langfuse_dataset_id"), + object_store_url=kw.get("object_store_url", "s3://bucket/ds.csv"), + dataset_metadata=kw.get("dataset_metadata", {}), + ) + + class TestLoadEvaluationDatasetItems: - def test_langfuse_backed_dataset_delegates_to_langfuse(self) -> None: - """A Langfuse-backed dataset (langfuse_dataset_id set) reads from Langfuse.""" - langfuse = MagicMock() + def test_with_client_reads_langfuse(self) -> None: + """A client present reads items from Langfuse.""" expected = [{"id": "lf_1", "input": {"question": "q"}}] - dataset = SimpleNamespace( - id=7, - langfuse_dataset_id="lf_ds_1", - object_store_url="s3://bucket/ds.csv", - dataset_metadata={}, - ) with ( patch( - "app.crud.evaluations.dataset.get_dataset_by_id", - return_value=dataset, + "app.crud.evaluations.batch.get_dataset_by_id", + return_value=_dataset(langfuse_dataset_id="lf_ds_1"), ), patch( "app.crud.evaluations.batch.fetch_dataset_items", @@ -37,67 +42,31 @@ def test_langfuse_backed_dataset_delegates_to_langfuse(self) -> None: ) as mock_fetch, ): items = load_evaluation_dataset_items( - session=MagicMock(), eval_run=_eval_run(), langfuse=langfuse + session=MagicMock(), eval_run=_eval_run(), langfuse=MagicMock() ) - mock_fetch.assert_called_once_with(langfuse=langfuse, dataset_name="ds") + mock_fetch.assert_called_once() assert items == expected - def test_langfuse_backed_but_tracing_off_falls_back_to_object_store(self) -> None: - """A Langfuse-backed dataset run with tracing off sources from object - store (cosine-only) rather than failing.""" - csv_bytes = b"question,answer\nq1,a1\n" - dataset = SimpleNamespace( - id=7, - langfuse_dataset_id="lf_ds_1", - object_store_url="s3://bucket/ds.csv", - dataset_metadata={}, - ) - + def test_without_client_reads_object_store(self) -> None: + """No client sources items from the object-store CSV with deterministic + ids and applied duplication.""" with ( patch( - "app.crud.evaluations.dataset.get_dataset_by_id", - return_value=dataset, + "app.crud.evaluations.batch.get_dataset_by_id", + return_value=_dataset(dataset_metadata={"duplication_factor": 2}), ), patch( - "app.crud.evaluations.dataset.download_csv_from_object_store", - return_value=csv_bytes, + "app.crud.evaluations.batch.download_csv_from_object_store", + return_value=b"question,answer\nq1,a1\nq2,a2\n", ), - patch("app.core.cloud.get_cloud_storage", return_value=MagicMock()), - ): - items = load_evaluation_dataset_items( - session=MagicMock(), eval_run=_eval_run(), langfuse=None - ) - - assert [i["id"] for i in items] == ["item_0_0"] - - def test_object_store_backed_sources_from_object_store(self) -> None: - """An object-store-backed dataset (no langfuse_dataset_id) sources items - from the CSV with deterministic ids and applied duplication — regardless - of whether a live Langfuse client is present.""" - csv_bytes = b"question,answer\nq1,a1\nq2,a2\n" - dataset = SimpleNamespace( - id=7, - langfuse_dataset_id=None, - object_store_url="s3://bucket/ds.csv", - dataset_metadata={"duplication_factor": 2}, - ) - - with ( patch( - "app.crud.evaluations.dataset.get_dataset_by_id", - return_value=dataset, + "app.crud.evaluations.batch.get_cloud_storage", + return_value=MagicMock(), ), - patch( - "app.crud.evaluations.dataset.download_csv_from_object_store", - return_value=csv_bytes, - ), - patch("app.core.cloud.get_cloud_storage", return_value=MagicMock()), ): - # Even with a live client, an object-store-backed dataset ignores it - # for DATA so ids stay stable. items = load_evaluation_dataset_items( - session=MagicMock(), eval_run=_eval_run(), langfuse=MagicMock() + session=MagicMock(), eval_run=_eval_run(), langfuse=None ) assert [i["id"] for i in items] == [ @@ -108,21 +77,23 @@ def test_object_store_backed_sources_from_object_store(self) -> None: ] assert items[0]["input"] == {"question": "q1"} assert items[0]["expected_output"] == {"answer": "a1"} - assert items[2]["input"] == {"question": "q2"} - def test_object_store_backed_without_url_raises(self) -> None: - """No Langfuse and no object-store URL cannot source items.""" - dataset = SimpleNamespace( - id=7, langfuse_dataset_id=None, object_store_url=None, dataset_metadata={} - ) + def test_object_store_without_url_raises(self) -> None: + """No client and no object-store URL cannot source items.""" + with patch( + "app.crud.evaluations.batch.get_dataset_by_id", + return_value=_dataset(object_store_url=None), + ): + with pytest.raises(ValueError, match="object-store"): + load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=None + ) + def test_dataset_not_found_raises(self) -> None: with patch( - "app.crud.evaluations.dataset.get_dataset_by_id", return_value=dataset + "app.crud.evaluations.batch.get_dataset_by_id", return_value=None ): - try: + with pytest.raises(ValueError, match="not found"): load_evaluation_dataset_items( session=MagicMock(), eval_run=_eval_run(), langfuse=None ) - raise AssertionError("expected ValueError") - except ValueError as e: - assert "object-store" in str(e) From 6860be257e44617ab434f7b30e90f5ca05cb274a Mon Sep 17 00:00:00 2001 From: Prashant Vasudevan <71649489+vprashrex@users.noreply.github.com> Date: Fri, 3 Jul 2026 12:51:01 +0530 Subject: [PATCH 3/3] refactor(tests): Simplify patching in dataset not found test --- backend/app/crud/evaluations/batch.py | 4 +++- backend/app/tests/crud/evaluations/test_batch.py | 4 +--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/backend/app/crud/evaluations/batch.py b/backend/app/crud/evaluations/batch.py index 271178245..29448c900 100644 --- a/backend/app/crud/evaluations/batch.py +++ b/backend/app/crud/evaluations/batch.py @@ -112,7 +112,9 @@ def load_evaluation_dataset_items( raise ValueError(f"Dataset {eval_run.dataset_id} not found") if langfuse is not None: - return fetch_dataset_items(langfuse=langfuse, dataset_name=eval_run.dataset_name) + return fetch_dataset_items( + langfuse=langfuse, dataset_name=eval_run.dataset_name + ) return _load_items_from_object_store(session=session, dataset=dataset) diff --git a/backend/app/tests/crud/evaluations/test_batch.py b/backend/app/tests/crud/evaluations/test_batch.py index df1c3ec54..f2ba751c6 100644 --- a/backend/app/tests/crud/evaluations/test_batch.py +++ b/backend/app/tests/crud/evaluations/test_batch.py @@ -90,9 +90,7 @@ def test_object_store_without_url_raises(self) -> None: ) def test_dataset_not_found_raises(self) -> None: - with patch( - "app.crud.evaluations.batch.get_dataset_by_id", return_value=None - ): + with patch("app.crud.evaluations.batch.get_dataset_by_id", return_value=None): with pytest.raises(ValueError, match="not found"): load_evaluation_dataset_items( session=MagicMock(), eval_run=_eval_run(), langfuse=None