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..95c25674b 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, +) -> APIResponse[ProjectPublic]: + 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..915dbb0be 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,41 @@ 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. + """ + try: + pid = int(project_id) + except (TypeError, ValueError): + logger.info( + 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=pid, + 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..29448c900 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,6 +76,84 @@ def fetch_dataset_items(langfuse: Langfuse, dataset_name: str) -> list[dict[str, return items +def use_langfuse_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 + + 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 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, + organization_id=eval_run.organization_id, + project_id=eval_run.project_id, + ) + if not dataset: + 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 _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"Dataset {dataset.id} has no object-store backing") + + 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) + 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 +237,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 +260,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..a1e0335e0 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, + use_langfuse_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 = use_langfuse_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..b346405ed 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, + use_langfuse_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__) @@ -349,12 +352,13 @@ def build_trace_skeleton( """ 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 +373,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 +442,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 +636,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 +772,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 +863,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 +892,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 = use_langfuse_client( + session=session, eval_run=eval_run, langfuse=langfuse + ) try: # Check if we need to process embedding batch first @@ -1192,7 +1200,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..1c5332799 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,36 @@ 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. + """ + # 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}" + ) + 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..f5902402a 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,14 @@ class ProjectUpdate(SQLModel): is_active: bool | None = Field(default=None) +# Properties to receive via API when patching project settings. +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 +76,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 +134,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..fbd7b25aa 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 use_langfuse_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 = use_langfuse_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..f2ba751c6 --- /dev/null +++ b/backend/app/tests/crud/evaluations/test_batch.py @@ -0,0 +1,97 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +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, + ) + + +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_with_client_reads_langfuse(self) -> None: + """A client present reads items from Langfuse.""" + expected = [{"id": "lf_1", "input": {"question": "q"}}] + + with ( + patch( + "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", + return_value=expected, + ) as mock_fetch, + ): + items = load_evaluation_dataset_items( + session=MagicMock(), eval_run=_eval_run(), langfuse=MagicMock() + ) + + mock_fetch.assert_called_once() + assert items == expected + + 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.batch.get_dataset_by_id", + return_value=_dataset(dataset_metadata={"duplication_factor": 2}), + ), + patch( + "app.crud.evaluations.batch.download_csv_from_object_store", + return_value=b"question,answer\nq1,a1\nq2,a2\n", + ), + patch( + "app.crud.evaluations.batch.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", + "item_0_1", + "item_1_0", + "item_1_1", + ] + assert items[0]["input"] == {"question": "q1"} + assert items[0]["expected_output"] == {"answer": "a1"} + + 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.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 + ) 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"]