diff --git a/agentplatform/_genai/_evals_common.py b/agentplatform/_genai/_evals_common.py index 81541bbe95..dbd77b8311 100644 --- a/agentplatform/_genai/_evals_common.py +++ b/agentplatform/_genai/_evals_common.py @@ -123,6 +123,11 @@ def _temp_logger_level(logger_name: str, level: int) -> None: # type: ignore[mi logger_instance.setLevel(original_level) +def _is_missing(value: Any) -> bool: + """Returns True for None or a scalar NaN, such as a missing DataFrame cell.""" + return bool(pd.api.types.is_scalar(value) and pd.isna(value)) + + def _get_api_client_with_location( api_client: BaseApiClient, location: Optional[str] ) -> BaseApiClient: @@ -1585,7 +1590,7 @@ def _build_interaction_id_dataset( eval_cases = [] for i, row in enumerate(loaded_data): interaction = row.get(_evals_constant.INTERACTION_ID) - if not interaction: + if _is_missing(interaction) or not interaction: raise ValueError(f"Missing `interaction_id` value for row {i}.") eval_cases.append( types.EvalCase( @@ -1895,12 +1900,12 @@ def _execute_inference_concurrently( max_workers = AGENT_MAX_WORKERS if runtime or agent else MAX_WORKERS with tqdm(total=len(prompt_dataset), desc=progress_desc) as pbar: with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - for index, row in prompt_dataset.iterrows(): + for index, (_, row) in enumerate(prompt_dataset.iterrows()): try: if ( has_agent_data and AGENT_DATA in row.index - and row.get(AGENT_DATA) is not None + and not _is_missing(row.get(AGENT_DATA)) ): agent_data_obj = row[AGENT_DATA] if isinstance(agent_data_obj, dict): @@ -2125,7 +2130,7 @@ def _run_litellm_inference( with tqdm(total=len(prompt_dataset), desc=f"LiteLLM Inference ({model})") as pbar: with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: - for index, row in prompt_dataset.iterrows(): + for index, (_, row) in enumerate(prompt_dataset.iterrows()): messages = _convert_prompt_row_to_litellm_messages(row) future = executor.submit( _call_litellm_completion, model=model, messages=messages @@ -3866,7 +3871,7 @@ def _drop_empty_columns(df: "pd.DataFrame") -> "pd.DataFrame": def is_empty(x: Any) -> bool: if isinstance(x, (list, dict)): return not x - return pd.isna(x) # type: ignore[no-any-return] + return _is_missing(x) cols_to_drop = [col for col in df.columns if df[col].apply(is_empty).all()] return df.drop(columns=cols_to_drop) @@ -4165,6 +4170,8 @@ def _create_evaluation_set_from_dataframe( if _evals_constant.RESPONSE in row or agent_data_obj or intermediate_events: # Resolve the oneof conflict: prioritize agent_data over flat text response_text = row.get(_evals_constant.RESPONSE) or None + if _is_missing(response_text): + response_text = None if agent_data_obj and response_text: logger.info( @@ -4219,7 +4226,7 @@ def _create_evaluation_set_from_dataframe( prompt = types.EvaluationPrompt( prompt_template_data=types.PromptTemplateData(values=values) ) - elif _evals_constant.PROMPT in row: + elif not _is_missing(row.get(_evals_constant.PROMPT)): prompt = types.EvaluationPrompt(text=row[_evals_constant.PROMPT]) eval_item_requests.append( @@ -4227,7 +4234,7 @@ def _create_evaluation_set_from_dataframe( prompt=prompt or None, golden_response=( types.CandidateResponse(text=row[_evals_constant.REFERENCE]) - if _evals_constant.REFERENCE in row + if not _is_missing(row.get(_evals_constant.REFERENCE)) else None ), candidate_responses=( diff --git a/tests/unit/agentplatform/genai/test_evals.py b/tests/unit/agentplatform/genai/test_evals.py index e55be6d51c..711bfd2b73 100644 --- a/tests/unit/agentplatform/genai/test_evals.py +++ b/tests/unit/agentplatform/genai/test_evals.py @@ -102,6 +102,16 @@ def test_drop_empty_columns_none_empty(self): result_df = _evals_common._drop_empty_columns(df) assert list(result_df.columns) == ["col1", "col2"] + def test_drop_empty_columns_with_array_cells(self): + df = pd.DataFrame( + { + "col1": [None, None], + "col2": [pd.Series([1, 2]).to_numpy(), pd.Series([3, 4]).to_numpy()], + } + ) + result_df = _evals_common._drop_empty_columns(df) + assert list(result_df.columns) == ["col2"] + def _create_content_dump(text: str) -> dict[str, list[genai_types.Content]]: return { @@ -4816,6 +4826,61 @@ def test_has_tool_call_with_agent_event(self): assert _evals_metric_handlers._has_tool_call(events) +class TestExecuteInferenceConcurrently: + """Unit tests for row handling in the concurrent inference helpers.""" + + def test_missing_agent_data_falls_back_to_prompt(self): + completed_trace = { + "turns": [ + { + "events": [ + {"author": "user", "content": {"parts": [{"text": "hi"}]}}, + {"author": "agent", "content": {"parts": [{"text": "hey"}]}}, + ] + } + ] + } + prompt_dataset = pd.DataFrame( + [{"prompt": "p0", "agent_data": completed_trace}, {"prompt": "p1"}] + ) + inference_fn = mock.Mock(return_value={"response": "r1"}) + + responses = _evals_common._execute_inference_concurrently( + api_client=mock.Mock(), + prompt_dataset=prompt_dataset, + progress_desc="Agent Run", + inference_fn=inference_fn, + runtime=mock.Mock(), + ) + + assert [event["author"] for event in responses[0]] == ["user", "agent"] + assert responses[1] == {"response": "r1"} + assert inference_fn.call_args.kwargs["contents"] == "p1" + + def test_responses_follow_row_positions(self): + prompt_dataset = pd.DataFrame({"prompt": ["p0", "p1"]}, index=[1, 0]) + + responses = _evals_common._execute_inference_concurrently( + api_client=None, + prompt_dataset=prompt_dataset, + progress_desc="Custom Inference", + model_or_fn=lambda contents: f"response to {contents}", + ) + + assert responses == ["response to p0", "response to p1"] + + @mock.patch.object(_evals_common, "_call_litellm_completion") + def test_litellm_responses_follow_row_positions(self, mock_call_litellm_completion): + mock_call_litellm_completion.side_effect = lambda model, messages: { + "content": messages[0]["content"] + } + prompt_dataset = pd.DataFrame({"prompt": ["p0", "p1"]}, index=[1, 0]) + + responses = _evals_common._run_litellm_inference("gpt-4o", prompt_dataset) + + assert responses == [{"content": "p0"}, {"content": "p1"}] + + @pytest.mark.usefixtures("google_auth_mock") class TestRunAgent: """Unit tests for the _run_agent function.""" @@ -10308,6 +10373,33 @@ def test_create_evaluation_set_with_intermediate_events( ] assert candidate_response["events"] == expected_events + @mock.patch.object(_evals_common, "evals") + @mock.patch.object(_evals_common, "_gcs_utils") + def test_create_evaluation_set_treats_nan_cells_as_missing( + self, mock_gcs_utils, mock_evals_module + ): + eval_df = pd.DataFrame( + [{"prompt": "p0", "response": "r0"}, {"reference": "ref1"}] + ) + + _evals_common._create_evaluation_set_from_dataframe( + api_client=self.mock_api_client, + gcs_dest_prefix="gs://bucket/prefix", + eval_df=eval_df, + candidate_name="test-candidate", + ) + + upload = mock_gcs_utils.GcsUtils.return_value.upload_json_to_prefix + first, second = (call.kwargs["data"] for call in upload.call_args_list) + assert first == { + "prompt": {"text": "p0"}, + "candidateResponses": [{"candidate": "test-candidate", "text": "r0"}], + } + assert second == { + "goldenResponse": {"text": "ref1"}, + "candidateResponses": [{"candidate": "test-candidate"}], + } + @mock.patch.object(_evals_common, "evals") @mock.patch.object(_evals_common, "_gcs_utils") def test_create_evaluation_set_with_user_scenario( @@ -11799,6 +11891,14 @@ def test_build_interaction_id_dataset_requires_agent(self): [{"interaction_id": "abc123"}], None, "global" ) + def test_build_interaction_id_dataset_rejects_nan_interaction_id(self): + with pytest.raises(ValueError, match="Missing `interaction_id` value"): + _evals_common._build_interaction_id_dataset( + [{"interaction_id": "abc123"}, {"interaction_id": float("nan")}], + _TEST_GEMINI_AGENT, + "global", + ) + def test_build_interaction_id_dataset_rejects_non_gemini_agent(self): with pytest.raises(ValueError, match="Gemini Agents API resource name"): _evals_common._build_interaction_id_dataset( diff --git a/vertexai/_genai/_evals_common.py b/vertexai/_genai/_evals_common.py index 4118d35f09..3e7a79c259 100644 --- a/vertexai/_genai/_evals_common.py +++ b/vertexai/_genai/_evals_common.py @@ -93,6 +93,11 @@ def _temp_logger_level(logger_name: str, level: int) -> None: # type: ignore[mi logger_instance.setLevel(original_level) +def _is_missing(value: Any) -> bool: + """Returns True for None or a scalar NaN, such as a missing DataFrame cell.""" + return bool(pd.api.types.is_scalar(value) and pd.isna(value)) + + def _get_api_client_with_location( api_client: BaseApiClient, location: Optional[str] ) -> BaseApiClient: @@ -684,12 +689,12 @@ def _execute_inference_concurrently( max_workers = AGENT_MAX_WORKERS if agent_engine or agent else MAX_WORKERS with tqdm(total=len(prompt_dataset), desc=progress_desc) as pbar: with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - for index, row in prompt_dataset.iterrows(): + for index, (_, row) in enumerate(prompt_dataset.iterrows()): try: if ( has_agent_data and AGENT_DATA in row.index - and row.get(AGENT_DATA) is not None + and not _is_missing(row.get(AGENT_DATA)) ): agent_data_obj = row[AGENT_DATA] if isinstance(agent_data_obj, dict): @@ -913,7 +918,7 @@ def _run_litellm_inference( with tqdm(total=len(prompt_dataset), desc=f"LiteLLM Inference ({model})") as pbar: with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: - for index, row in prompt_dataset.iterrows(): + for index, (_, row) in enumerate(prompt_dataset.iterrows()): messages = _convert_prompt_row_to_litellm_messages(row) future = executor.submit( _call_litellm_completion, model=model, messages=messages @@ -2645,7 +2650,7 @@ def _drop_empty_columns(df: "pd.DataFrame") -> "pd.DataFrame": def is_empty(x: Any) -> bool: if isinstance(x, (list, dict)): return not x - return pd.isna(x) # type: ignore[no-any-return] + return _is_missing(x) cols_to_drop = [col for col in df.columns if df[col].apply(is_empty).all()] return df.drop(columns=cols_to_drop) @@ -2944,6 +2949,8 @@ def _create_evaluation_set_from_dataframe( if _evals_constant.RESPONSE in row or agent_data_obj or intermediate_events: # Resolve the oneof conflict: prioritize agent_data over flat text response_text = row.get(_evals_constant.RESPONSE) or None + if _is_missing(response_text): + response_text = None if agent_data_obj and response_text: logger.info( @@ -2998,7 +3005,7 @@ def _create_evaluation_set_from_dataframe( prompt = types.EvaluationPrompt( prompt_template_data=types.PromptTemplateData(values=values) ) - elif _evals_constant.PROMPT in row: + elif not _is_missing(row.get(_evals_constant.PROMPT)): prompt = types.EvaluationPrompt(text=row[_evals_constant.PROMPT]) eval_item_requests.append( @@ -3006,7 +3013,7 @@ def _create_evaluation_set_from_dataframe( prompt=prompt or None, golden_response=( types.CandidateResponse(text=row[_evals_constant.REFERENCE]) - if _evals_constant.REFERENCE in row + if not _is_missing(row.get(_evals_constant.REFERENCE)) else None ), candidate_responses=(