diff --git a/evalbench/evalproto/eval_agent.proto b/evalbench/evalproto/eval_agent.proto index 8e0ece74..76a722fe 100644 --- a/evalbench/evalproto/eval_agent.proto +++ b/evalbench/evalproto/eval_agent.proto @@ -66,20 +66,45 @@ message ScorerSpec { float timeout_seconds = 3; // Timeout per scorer } +message ScoringContext { + string nl_prompt = 1; + string golden_query = 2; + string query_type = 3; + string golden_result = 4; + string golden_eval_results = 5; + string golden_error = 6; + string generated_query = 7; + string generated_result = 8; + string eval_results = 9; + string generated_error = 10; + string database = 11; +} + message ScoringRequest { ScorerSpec scorer = 1; + ScoringContext context = 2; +} + +message SingleScore { + float score = 1; + string comparison_logs = 2; } -message ScoreResult { - string scorer_name = 1; - float score = 2; // 100.0 (PASS), 0.0 (FAIL), or continuous score - bool success = 3; - string result_json = 4; // Generic JSON payload or diagnostics returned by scorer - string error_message = 5; +message MetricScore { + string metric_name = 1; + float score = 2; + string comparison_logs = 3; +} + +message MultiScore { + repeated MetricScore scores = 1; } message ScoringResponse { - ScoreResult result = 1; + oneof result { + SingleScore single_score = 1; + MultiScore multi_score = 2; + } } // --- 3. Delegated Reporting Messages --- diff --git a/evalbench/scorers/remote_scorer.py b/evalbench/scorers/remote_scorer.py index 7bc84412..e65a9119 100644 --- a/evalbench/scorers/remote_scorer.py +++ b/evalbench/scorers/remote_scorer.py @@ -13,6 +13,25 @@ logger = logging.getLogger(__name__) +def _to_str(val: Any) -> str: + """Defensively coerces evaluation arguments to strings for Protobuf packing. + + Note: While Comparator.compare hints string types, EvalBench runners + do not strictly enforce strings at runtime. Specifically, agent evaluations + (agentscorework.py) pass raw Python lists (e.g. tool calls in generated_result), + dictionaries (eval_results), or None (generated_error). Because Google Protobuf + string fields require str/bytes and raise a TypeError if passed list/dict/None, + we defensively coerce them to strings (JSON-serialized for lists/dicts) before packing. + """ + if val is None: + return "" + if isinstance(val, str): + return val + if isinstance(val, (dict, list)): + return json.dumps(val) + return str(val) + + class RemoteScorerProxy(Comparator): """Comparator proxying scoring evaluation across the reverse bidi stream.""" @@ -30,15 +49,16 @@ def __init__(self, name: str, config: dict[str, Any]): def compare( self, nl_prompt: str, - golden_sql: str, - query_type: str, - golden_result: str, - golden_eval_results: str, - golden_error: str, - generated_sql: str, - generated_result: str, - eval_results: str, - generated_error: str, + golden_query: str = "", + query_type: str = "", + golden_result: Any = "", + golden_eval_results: Any = "", + golden_error: Any = "", + generated_query: str = "", + generated_result: Any = "", + eval_results: Any = "", + generated_error: Any = "", + database: str = "", **kwargs: Any, ) -> tuple[float, str] | list[tuple[str, float, str]]: session_id = rpc_id_var.get() @@ -63,7 +83,36 @@ def compare( timeout_seconds=self.timeout_seconds, ) - scoring_req = eval_agent_pb2.ScoringRequest(scorer=scorer_spec) + golden_query = golden_query or kwargs.get("golden_sql", "") + generated_query = generated_query or kwargs.get("generated_sql", "") + database = database or kwargs.get("database", "") + if golden_result in ("", None): + golden_result = kwargs.get("golden_execution_result", "") + if golden_eval_results in ("", None): + golden_eval_results = kwargs.get("golden_eval_result", "") + if generated_result in ("", None): + generated_result = kwargs.get("generated_execution_result", "") + if eval_results in ("", None): + eval_results = kwargs.get("generated_eval_result", "") + + scoring_context = eval_agent_pb2.ScoringContext( + nl_prompt=_to_str(nl_prompt), + golden_query=_to_str(golden_query), + query_type=_to_str(query_type), + golden_result=_to_str(golden_result), + golden_eval_results=_to_str(golden_eval_results), + golden_error=_to_str(golden_error), + generated_query=_to_str(generated_query), + generated_result=_to_str(generated_result), + eval_results=_to_str(eval_results), + generated_error=_to_str(generated_error), + database=_to_str(database), + ) + + scoring_req = eval_agent_pb2.ScoringRequest( + scorer=scorer_spec, + context=scoring_context, + ) msg = eval_agent_pb2.AgentStreamMessage( session_id=session_id, correlation_id=correlation_id, @@ -71,7 +120,7 @@ def compare( ) logger.info( - "[REVERSE_SCORER] Dispatching ScoringRequest for '%s' (correlation_id=%s)", + "[REMOTE_SCORER] Dispatching ScoringRequest for '%s' (correlation_id=%s)", self.name, correlation_id, ) @@ -81,7 +130,7 @@ def compare( resp_msg = inbox.get(timeout=self.timeout_seconds) except queue.Empty: logger.error( - "[REVERSE_SCORER] Timed out waiting for ScoringResponse for '%s' (correlation_id=%s)", + "[REMOTE_SCORER] Timed out waiting for ScoringResponse for '%s' (correlation_id=%s)", self.name, correlation_id, ) @@ -91,9 +140,23 @@ def compare( if not resp_msg.HasField("scoring_response"): err_details = resp_msg.WhichOneof("payload") - logger.error("[REVERSE_SCORER] Unexpected message on stream: %s", err_details) + logger.error("[REMOTE_SCORER] Unexpected message on stream: %s", err_details) return (0.0, f"Error: Unexpected payload on stream: {err_details}") - r = resp_msg.scoring_response.result - log_output = r.result_json or r.error_message or ("PASSED" if r.success else "FAILED") - return (float(r.score), log_output) + scoring_resp = resp_msg.scoring_response + result_type = scoring_resp.WhichOneof("result") + + if result_type == "single_score": + s = scoring_resp.single_score + return (float(s.score), s.comparison_logs) + elif result_type == "multi_score": + return [ + ( + s.metric_name or self.name, + float(s.score), + s.comparison_logs, + ) + for s in scoring_resp.multi_score.scores + ] + + return (0.0, "Error: Empty score response from remote scorer") diff --git a/evalbench/test/agent_grpc_proxy_integration_test.py b/evalbench/test/agent_grpc_proxy_integration_test.py index 52c3dde5..594321e4 100644 --- a/evalbench/test/agent_grpc_proxy_integration_test.py +++ b/evalbench/test/agent_grpc_proxy_integration_test.py @@ -277,16 +277,14 @@ async def sender(): elif payload_type == "scoring_request": spec = msg.scoring_request.scorer - score_res = eval_agent_pb2.ScoreResult( - scorer_name=spec.scorer_name, + single = eval_agent_pb2.SingleScore( score=100.0, - success=True, - result_json=json.dumps({"stdout": f"{spec.scorer_name} verified in sandbox"}), + comparison_logs=f"{spec.scorer_name} verified in sandbox", ) reply = eval_agent_pb2.AgentStreamMessage( session_id=session_id, correlation_id=corr_id, - scoring_response=eval_agent_pb2.ScoringResponse(result=score_res), + scoring_response=eval_agent_pb2.ScoringResponse(single_score=single), ) await send_queue.put(reply) @@ -420,16 +418,14 @@ async def sender(): elif payload_type == "scoring_request": spec = msg.scoring_request.scorer - score_res = eval_agent_pb2.ScoreResult( - scorer_name=spec.scorer_name, + single = eval_agent_pb2.SingleScore( score=100.0, - success=True, - result_json=json.dumps({"stdout": f"{spec.scorer_name} ok"}), + comparison_logs=f"{spec.scorer_name} ok", ) reply = eval_agent_pb2.AgentStreamMessage( session_id=session_id, correlation_id=corr_id, - scoring_response=eval_agent_pb2.ScoringResponse(result=score_res), + scoring_response=eval_agent_pb2.ScoringResponse(single_score=single), ) await send_queue.put(reply) diff --git a/evalbench/test/agent_grpc_proxy_test.py b/evalbench/test/agent_grpc_proxy_test.py index 3ac53071..998d827d 100644 --- a/evalbench/test/agent_grpc_proxy_test.py +++ b/evalbench/test/agent_grpc_proxy_test.py @@ -11,7 +11,7 @@ ) from reporting.remote_reporter import RemoteReporter from reporting.report import STORETYPE -from scorers.remote_scorer import RemoteScorerProxy +from scorers.remote_scorer import RemoteScorerProxy, _to_str from util.context import rpc_id_var @@ -83,16 +83,22 @@ def answer_scorer(): spec = msg.scoring_request.scorer self.assertEqual(spec.scorer_name, "dataform_compile") - score_res = eval_agent_pb2.ScoreResult( - scorer_name="dataform_compile", + # Verify ScoringContext fields were passed directly as strings + ctx = msg.scoring_request.context + self.assertEqual(ctx.nl_prompt, "build pipeline") + self.assertEqual(ctx.database, "bigquery") + self.assertEqual(ctx.golden_result, "golden_output") + self.assertEqual(ctx.generated_result, "generated_output") + self.assertEqual(ctx.eval_results, "step1_passed") + + single = eval_agent_pb2.SingleScore( score=100.0, - success=True, - result_json=json.dumps({"status": "verified"}), + comparison_logs="Compilation successful", ) reply = eval_agent_pb2.AgentStreamMessage( session_id=self.session_id, correlation_id=corr_id, - scoring_response=eval_agent_pb2.ScoringResponse(result=score_res), + scoring_response=eval_agent_pb2.ScoringResponse(single_score=single), ) self.inboxes[corr_id].put(reply) @@ -103,6 +109,72 @@ def answer_scorer(): nl_prompt="build pipeline", golden_sql="", query_type="", + golden_result="golden_output", + golden_eval_results="", + golden_error="", + generated_sql="", + generated_result="generated_output", + eval_results="step1_passed", + generated_error="", + database="bigquery", + ) + t.join() + + self.assertEqual(score, 100.0) + self.assertEqual(logs, "Compilation successful") + + def test_remote_scorer_sub_scores(self): + scorer = RemoteScorerProxy("multi_aspect_scorer", {"timeout_seconds": 5.0}) + + def answer_multi_scorer(): + msg = self.out_queue.get(timeout=2.0) + corr_id = msg.correlation_id + self.assertEqual(msg.WhichOneof("payload"), "scoring_request") + + scores_list = [ + eval_agent_pb2.MetricScore( + metric_name="syntax", score=100.0, comparison_logs="Clean syntax" + ), + eval_agent_pb2.MetricScore( + metric_name="efficiency", score=80.0, comparison_logs="2 turns" + ), + ] + multi = eval_agent_pb2.MultiScore(scores=scores_list) + reply = eval_agent_pb2.AgentStreamMessage( + session_id=self.session_id, + correlation_id=corr_id, + scoring_response=eval_agent_pb2.ScoringResponse(multi_score=multi), + ) + self.inboxes[corr_id].put(reply) + + t = threading.Thread(target=answer_multi_scorer) + t.start() + + result = scorer.compare( + nl_prompt="test", + golden_sql="", + query_type="", + golden_result="", + golden_eval_results="", + golden_error="", + generated_sql="", + generated_result="", + eval_results="", + generated_error="", + ) + t.join() + + self.assertIsInstance(result, list) + self.assertEqual(len(result), 2) + self.assertEqual(result[0], ("syntax", 100.0, "Clean syntax")) + self.assertEqual(result[1], ("efficiency", 80.0, "2 turns")) + + def test_remote_scorer_timeout(self): + scorer = RemoteScorerProxy("slow_scorer", {"timeout_seconds": 0.1}) + score, logs = scorer.compare( + nl_prompt="test", + golden_sql="", + query_type="", golden_result="", golden_eval_results="", golden_error="", @@ -111,10 +183,68 @@ def answer_scorer(): eval_results="", generated_error="", ) + self.assertEqual(score, 0.0) + self.assertIn("Timed out waiting for remote scorer", logs) + + def test_remote_scorer_runtime_type_coercion(self): + scorer = RemoteScorerProxy("rubric_scorer", {"timeout_seconds": 5.0}) + + def answer_scorer(): + msg = self.out_queue.get(timeout=2.0) + corr_id = msg.correlation_id + self.assertEqual(msg.WhichOneof("payload"), "scoring_request") + + ctx = msg.scoring_request.context + # Non-string runtime types (list, dict) should be coerced to JSON strings + self.assertEqual( + ctx.generated_result, + json.dumps(["activate_skill", "run_command"]), + ) + self.assertEqual( + ctx.eval_results, + json.dumps({"turn": 2, "tokens": 120}), + ) + # None should be mapped to empty string + self.assertEqual(ctx.generated_error, "") + + single = eval_agent_pb2.SingleScore( + score=100.0, + comparison_logs="Trajectory verified", + ) + reply = eval_agent_pb2.AgentStreamMessage( + session_id=self.session_id, + correlation_id=corr_id, + scoring_response=eval_agent_pb2.ScoringResponse(single_score=single), + ) + self.inboxes[corr_id].put(reply) + + t = threading.Thread(target=answer_scorer) + t.start() + + score, logs = scorer.compare( + nl_prompt="verify skills", + golden_query="", + query_type="", + golden_result="", + golden_eval_results="", + golden_error="", + generated_query="", + generated_result=["activate_skill", "run_command"], + eval_results={"turn": 2, "tokens": 120}, + generated_error=None, + database="bigquery", + ) t.join() self.assertEqual(score, 100.0) - self.assertEqual(logs, json.dumps({"status": "verified"})) + self.assertEqual(logs, "Trajectory verified") + + def test_to_str_helper(self): + self.assertEqual(_to_str(None), "") + self.assertEqual(_to_str("already_a_string"), "already_a_string") + self.assertEqual(_to_str(["tool_a", "tool_b"]), '["tool_a", "tool_b"]') + self.assertEqual(_to_str({"key": "val"}), '{"key": "val"}') + self.assertEqual(_to_str(123), "123") def test_remote_reporter_success(self): reporter = RemoteReporter(