support overwrite

Author: mayinghan

eval_protocol/pytest/evaluation_test.py

@@ -52,10 +52,12 @@
     add_cost_metrics,
     log_eval_status_and_rows,
     parse_ep_completion_params,
+    parse_ep_completion_params_overwrite,
     parse_ep_max_concurrent_rollouts,
     parse_ep_max_rows,
     parse_ep_num_runs,
     parse_ep_passed_threshold,
+    parse_ep_dataloaders,
     rollout_processor_with_retry,
     run_tasks_with_eval_progress,
     run_tasks_with_run_progress,
@@ -187,10 +189,18 @@ def evaluation_test(
     max_concurrent_rollouts = parse_ep_max_concurrent_rollouts(max_concurrent_rollouts)
     max_dataset_rows = parse_ep_max_rows(max_dataset_rows)
     completion_params = parse_ep_completion_params(completion_params)
+    completion_params = parse_ep_completion_params_overwrite(completion_params) 
     original_completion_params = completion_params
     passed_threshold = parse_ep_passed_threshold(passed_threshold)
+    data_loaders = parse_ep_dataloaders(data_loaders)
     custom_invocation_id = os.environ.get("EP_INVOCATION_ID", None)
 
+    # dataloader might be overridden here, to avoid conflict, manually unset other data input params
+    if data_loaders:
+        input_dataset = None
+        input_messages = None
+        input_rows = None 
+
     def decorator(
         test_func: TestFunction,
     ) -> TestFunction:

eval_protocol/pytest/plugin.py

@@ -19,6 +19,7 @@
 import pathlib
 import sys
 from pytest import StashKey
+import pytest
 
 
 def pytest_addoption(parser) -> None:
@@ -56,6 +57,7 @@ def pytest_addoption(parser) -> None:
         default=None,
         help=("Write a JSON summary artifact at the given path (e.g., ./outputs/aime_low.json)."),
     )
+    # deprecate this later
     group.addoption(
         "--ep-input-param",
         action="append",
@@ -115,6 +117,27 @@ def pytest_addoption(parser) -> None:
             "Default: false (experiment JSONs are saved and uploaded by default)."
         ),
     )
+    group.addoption(
+        "--ep-jsonl-path",
+        default=None,
+        help=("Load input from a jsonl file that is already in EvaluationRow or openai CHAT format")
+    )
+    group.addoption(
+        "--ep-completion-params",
+        default=[],
+        action="append",
+        help=(
+            "Overwrite completion params with json. Can be used multiple times. "
+        ),
+    )
+    group.addoption(
+        "--ep-remote-rollout-processor-base-url",
+        default=None,
+        help=(
+            "If set, use this base URL for remote rollout processing. "
+            "Example: http://localhost:8000"
+        ),
+    )
 
 
 def _normalize_max_rows(val: Optional[str]) -> Optional[str]:
@@ -243,6 +266,18 @@ def pytest_configure(config) -> None:
     if config.getoption("--ep-no-upload"):
         os.environ["EP_NO_UPLOAD"] = "1"
 
+    if config.getoption("--ep-jsonl-path"):
+        os.environ["EP_JSONL_PATH"] = config.getoption("--ep-jsonl-path")
+
+    if config.getoption("--ep-completion-params"):
+        # redump to json to make sure they are legit
+        os.environ["EP_COMPLETION_PARAMS"] = json.dumps([
+            json.loads(s) for s in config.getoption("--ep-completion-params") or []
+        ])
+    
+    if config.getoption("--ep-remote-rollout-processor-base-url"):
+        os.environ["EP_REMOTE_ROLLOUT_PROCESSOR_BASE_URL"] = config.getoption("--ep-remote-rollout-processor-base-url")
+
     # Allow ad-hoc overrides of input params via CLI flags
     try:
         merged: dict = {}

eval_protocol/pytest/remote_rollout_processor.py

@@ -8,6 +8,7 @@
 from eval_protocol.data_loader.dynamic_data_loader import DynamicDataLoader
 from .rollout_processor import RolloutProcessor
 from .types import RolloutProcessorConfig
+import os
 
 
 class RemoteRolloutProcessor(RolloutProcessor):
@@ -46,6 +47,8 @@ def __init__(
         # Prefer constructor-provided configuration. These can be overridden via
         # config.kwargs at call time for backward compatibility.
         self._remote_base_url = remote_base_url
+        if os.getenv("EP_REMOTE_ROLLOUT_PROCESSOR_BASE_URL"):
+            self._remote_base_url = os.getenv("EP_REMOTE_ROLLOUT_PROCESSOR_BASE_URL")
         self._poll_interval = poll_interval
         self._timeout_seconds = timeout_seconds
         self._output_data_loader = output_data_loader

eval_protocol/pytest/utils.py

@@ -19,6 +19,8 @@
     EvaluationThresholdDict,
     Status,
 )
+from eval_protocol.data_loader import DynamicDataLoader
+from eval_protocol.data_loader.models import EvaluationDataLoader 
 from eval_protocol.pytest.rollout_processor import RolloutProcessor
 from eval_protocol.pytest.types import (
     RolloutProcessorConfig,
@@ -238,6 +240,41 @@ def parse_ep_completion_params(
         pass
     return completion_params
 
+def parse_ep_completion_params_overwrite(completion_params: Sequence[CompletionParams | None] | None) -> Sequence[CompletionParams | None]:
+    new_completion_params = os.getenv("EP_COMPLETION_PARAMS")
+    if new_completion_params:
+        try:
+            new_completion_params_list = json.loads(new_completion_params)
+            if isinstance(new_completion_params_list, list):
+                return new_completion_params_list
+        except Exception:
+            pass
+    return completion_params or []
+
+def _rows_from_jsonl(path: str) -> list[EvaluationRow]:
+    rows = []
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            for line in f:
+                rows.append(EvaluationRow(**json.loads(line)))
+    except Exception as e:
+        print(f"❌ Failed to load rows from JSONL at {path}: {e}")
+        return []
+    
+    return rows
+
+def parse_ep_dataloaders(
+    dataloaders: Sequence[EvaluationDataLoader] | EvaluationDataLoader | None,
+) -> Sequence[EvaluationDataLoader] | EvaluationDataLoader | None:
+    try:
+        load_from_jsonl_path = os.getenv("EP_JSONL_PATH")
+        if load_from_jsonl_path:
+            return DynamicDataLoader(
+                generators=[lambda path=load_from_jsonl_path: _rows_from_jsonl(path)])
+    except Exception:
+        pass
+    return dataloaders or None
+
 
 def parse_ep_passed_threshold(
     default_value: float | EvaluationThresholdDict | EvaluationThreshold | None,