From a40f551f23205f7ab38206b59b0fd891cffbdf6f Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 9 Jun 2026 16:36:31 +0300 Subject: [PATCH 1/7] feat(adapter): add experimental live endpoint response collector Add a collection utility to the adapter SDK that queries a live chatbot endpoint with test questions and produces an evaluation-ready JSONL dataset. Supports OpenAI-compatible chat completions and generic HTTP endpoints (MCP, Langflow, custom APIs) with configurable request templates and response path extraction. Integrates with existing SDK auth (resolve_model_credentials) and TLS patterns (CA bundle auto-detection, insecure flag). Config is passed through JobSpec.parameters["live_collection"] as an experimental shape. Refs #135 Co-Authored-By: Claude Opus 4.6 (1M context) --- README.md | 61 +++ src/evalhub/adapter/__init__.py | 27 ++ src/evalhub/adapter/collector.py | 603 +++++++++++++++++++++++++ tests/unit/test_collector.py | 742 +++++++++++++++++++++++++++++++ 4 files changed, 1433 insertions(+) create mode 100644 src/evalhub/adapter/collector.py create mode 100644 tests/unit/test_collector.py diff --git a/README.md b/README.md index 5945a410..7cc2f194 100755 --- a/README.md +++ b/README.md @@ -529,6 +529,67 @@ results = JobResults(..., eval_card=eval_card, env_card=env_card) callbacks.report_results(results) ``` +## Live Endpoint Response Collection (Experimental) + +The SDK includes a collector utility that adapters can call during `LOADING_DATA` +to query a chatbot endpoint with test questions and collect responses for evaluation. + +### OpenAI-Compatible Endpoint + +```python +from evalhub.adapter import ( + CollectorConfig, collect_responses, is_collection_configured, + resolve_model_credentials, +) + +# In your adapter's run_benchmark_job(): +if is_collection_configured(config.parameters): + creds = resolve_model_credentials() + collector_config = CollectorConfig.from_parameters(config.parameters) + manifest = collect_responses(collector_config, credentials=creds) + # manifest.output_path -> responses.jsonl for your evaluation framework +``` + +Job parameters: +```json +{ + "live_collection": { + "questions_path": "/test_data/questions.csv", + "output_dir": "/tmp/collected", + "endpoint_url": "https://my-chatbot.example/v1", + "model": "my-chatbot-v2", + "protocol": "openai_chat_completions" + } +} +``` + +### Generic HTTP Endpoint (MCP, Langflow, custom APIs) + +```json +{ + "live_collection": { + "questions_path": "/test_data/questions.csv", + "output_dir": "/tmp/collected", + "endpoint_url": "https://mcp-chatbot.example/mcp", + "protocol": "generic_http", + "request_template": { + "jsonrpc": "2.0", + "method": "tools/call", + "params": {"name": "chat", "arguments": {"message": "{question}"}}, + "id": "{question_id}" + }, + "response_path": "result.content.0.text", + "extra_response_paths": { + "retrieved_contexts": "result.sources" + } + } +} +``` + +The collector uses SDK auth patterns (`resolve_model_credentials()`) and +TLS auto-detection by default. See the module docstring in +`src/evalhub/adapter/collector.py` for all configuration options. + ## Deployment ### Container Structure diff --git a/src/evalhub/adapter/__init__.py b/src/evalhub/adapter/__init__.py index e9ef7687..f7151631 100644 --- a/src/evalhub/adapter/__init__.py +++ b/src/evalhub/adapter/__init__.py @@ -92,6 +92,20 @@ def run_benchmark_job( OCIArtifactSpec, SafetyEvalEntry, ) +from .collector import ( + CollectedRecord, + CollectionManifest, + CollectorConfig, + CollectorError, + CollectorProtocol, + LiveQuestion, + collect_responses, + collect_responses_from_parameters, + extract_by_path, + is_configured as is_collection_configured, + load_questions, + substitute_template, +) from .oci import OCIArtifactPersister from .settings import AdapterSettings @@ -132,4 +146,17 @@ def run_benchmark_job( "ModelConfig", "EvaluationResult", "OCICoordinates", + # Live endpoint collection (experimental) + "CollectorConfig", + "CollectorProtocol", + "CollectorError", + "CollectedRecord", + "CollectionManifest", + "LiveQuestion", + "collect_responses", + "collect_responses_from_parameters", + "is_collection_configured", + "load_questions", + "extract_by_path", + "substitute_template", ] diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py new file mode 100644 index 00000000..52c754bf --- /dev/null +++ b/src/evalhub/adapter/collector.py @@ -0,0 +1,603 @@ +"""Experimental live endpoint response collection for adapters. + +Lets adapters collect chatbot/model responses during ``JobPhase.LOADING_DATA`` +by querying a live endpoint with a set of test questions. Supports +OpenAI-compatible chat completions and generic HTTP endpoints (MCP, Langflow, +LangGraph, custom APIs). + +Config is passed through ``JobSpec.parameters["live_collection"]`` while the +server-side API shape (``test_data_ref``) is still being validated. + +Trust boundary: the adapter operator controls the endpoint configuration. +Hosted runtimes that accept config from untrusted users should add their own +egress policy or endpoint allowlist before enabling this. +""" + +from __future__ import annotations + +import csv +import json +import logging +import os +import time +from collections.abc import Callable, Mapping +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path +from typing import Any, Self, cast + +from pydantic import BaseModel, Field, field_validator, model_validator + +from .auth import ModelCredentials, resolve_model_credentials + +logger = logging.getLogger(__name__) + +_PARAM_KEY = "live_collection" + +_CA_BUNDLE_CANDIDATES = [ + Path("/etc/pki/ca-trust/source/anchors/service-ca.crt"), + Path("/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt"), + Path("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"), +] + + +class CollectorProtocol(StrEnum): + OPENAI_CHAT = "openai_chat_completions" + GENERIC_HTTP = "generic_http" + + +class CollectorError(Exception): + """Raised when collection fails in fail_fast mode.""" + + def __init__( + self, + message: str, + question_id: str | None = None, + cause: Exception | None = None, + ) -> None: + self.question_id = question_id + self.__cause__ = cause + super().__init__(message) + + +class CollectorConfig(BaseModel): + """Configuration for live endpoint response collection.""" + + questions_path: Path = Field( + ..., description="CSV, JSON, or JSONL file with test questions" + ) + output_dir: Path = Field( + ..., description="Directory for responses.jsonl and manifest.json" + ) + endpoint_url: str = Field( + ..., description="Endpoint URL (OpenAI-compatible base or full HTTP URL)" + ) + protocol: CollectorProtocol = Field(default=CollectorProtocol.OPENAI_CHAT) + + question_column: str = Field(default="question") + id_column: str = Field(default="id") + + model: str | None = Field( + default=None, description="Model name for OpenAI chat requests" + ) + system_prompt: str | None = Field(default=None) + extra_body: dict[str, Any] = Field(default_factory=dict) + + request_template: dict[str, Any] | None = Field( + default=None, description="Request body template for generic_http protocol" + ) + response_path: str | None = Field( + default=None, + description="Dot-separated path to extract answer from response", + ) + extra_response_paths: dict[str, str] | None = Field( + default=None, + description="Additional fields to extract from response (output_name -> dot-path)", + ) + + api_key_env: str | None = Field( + default=None, description="Env var containing a bearer token" + ) + request_headers: dict[str, str] = Field(default_factory=dict) + use_model_credentials: bool = Field( + default=True, + description="Use resolve_model_credentials() for auth and TLS", + ) + + ca_bundle: str | None = Field(default=None, description="CA bundle path") + insecure: bool = Field(default=False, description="Skip TLS verification") + + fail_fast: bool = Field( + default=False, description="Raise on first collection error" + ) + timeout_seconds: float = Field(default=30.0, gt=0.0) + max_retries: int = Field(default=0, ge=0) + retry_backoff_seconds: float = Field(default=0.25, ge=0.0) + max_retry_backoff_seconds: float = Field(default=5.0, ge=0.0) + + @classmethod + def from_parameters(cls, parameters: Mapping[str, Any]) -> CollectorConfig: + raw = parameters.get(_PARAM_KEY) + if raw is None: + raise ValueError(f'JobSpec.parameters["{_PARAM_KEY}"] is required') + if not isinstance(raw, dict): + raise ValueError(f'parameters["{_PARAM_KEY}"] must be an object') + return cls.model_validate(raw) + + @field_validator("endpoint_url") + @classmethod + def _validate_endpoint_url(cls, value: str) -> str: + if not value.lower().startswith(("http://", "https://")): + raise ValueError("endpoint_url must use http:// or https://") + return value.rstrip("/") + + @model_validator(mode="after") + def _validate_protocol_requirements(self) -> Self: + if self.protocol == CollectorProtocol.OPENAI_CHAT and not self.model: + raise ValueError("model is required for openai_chat_completions protocol") + if self.protocol == CollectorProtocol.GENERIC_HTTP: + if not self.request_template: + raise ValueError( + "request_template is required for generic_http protocol" + ) + if not self.response_path: + raise ValueError("response_path is required for generic_http protocol") + return self + + +class LiveQuestion(BaseModel): + question: str + question_id: str | None = None + source_row: dict[str, Any] = Field(default_factory=dict) + + +class CollectedRecord(BaseModel): + """One collected response row. + + ``source_fields`` preserves all original input columns as-is. + Collector-added fields (``response``, ``error``, etc.) are merged + on top when serialized to JSONL. + """ + + source_fields: dict[str, Any] = Field(default_factory=dict) + response: str | None = None + extra_fields: dict[str, Any] = Field(default_factory=dict) + raw_response: dict[str, Any] | None = None + error: str | None = None + latency_ms: float | None = None + + +class CollectionManifest(BaseModel): + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + experimental: bool = True + protocol: str + endpoint_url: str + model: str | None = None + questions_path: str + output_path: str + total: int + completed: int + failed: int + + +def is_configured(parameters: dict[str, Any] | None) -> bool: + """Check if parameters contain live collection config.""" + if not parameters: + return False + raw = parameters.get(_PARAM_KEY) + return isinstance(raw, dict) and bool(raw) + + +def collect_responses( + config: CollectorConfig, + *, + client: Any | None = None, + credentials: ModelCredentials | None = None, + progress_callback: Callable[[float], None] | None = None, +) -> CollectionManifest: + """Collect responses from a live endpoint. + + Reads test questions, queries the endpoint, and writes an evaluation-ready + JSONL dataset plus a manifest summary. + """ + if credentials is None and config.use_model_credentials: + try: + credentials = resolve_model_credentials() + except Exception: + logger.debug("Could not resolve model credentials", exc_info=True) + credentials = None + + headers = _resolve_auth_headers(config, credentials) + verify = _resolve_verify(config, credentials) + questions = load_questions( + config.questions_path, + question_column=config.question_column, + id_column=config.id_column, + ) + + config.output_dir.mkdir(parents=True, exist_ok=True) + output_path = config.output_dir / "responses.jsonl" + + active_client = client + owns_client = active_client is None + if active_client is None: + try: + import httpx + except ImportError as exc: + raise RuntimeError( + "Live collection requires httpx. " + "Install eval-hub-sdk[adapter] or eval-hub-sdk[core]." + ) from exc + active_client = httpx.Client( + follow_redirects=False, + timeout=config.timeout_seconds, + verify=verify, + ) + + completed = 0 + failed = 0 + total = len(questions) + + try: + with output_path.open("w", encoding="utf-8") as out: + for i, question in enumerate(questions): + record = _collect_one(config, active_client, headers, question) + if record.error: + failed += 1 + if config.fail_fast: + raise CollectorError( + record.error, + question_id=question.question_id, + ) + else: + completed += 1 + out.write(json.dumps(_flatten_record(record)) + "\n") + if progress_callback and total > 0: + progress_callback((i + 1) / total) + finally: + if owns_client and hasattr(active_client, "close"): + active_client.close() + + manifest = CollectionManifest( + protocol=config.protocol.value, + endpoint_url=config.endpoint_url, + model=config.model, + questions_path=str(config.questions_path), + output_path=str(output_path), + total=total, + completed=completed, + failed=failed, + ) + manifest_path = config.output_dir / "manifest.json" + manifest_path.write_text( + json.dumps(manifest.model_dump(mode="json"), indent=2, default=str) + "\n", + encoding="utf-8", + ) + return manifest + + +def collect_responses_from_parameters( + parameters: Mapping[str, Any], + **kwargs: Any, +) -> CollectionManifest: + """Convenience wrapper: load config from parameters and collect.""" + config = CollectorConfig.from_parameters(parameters) + return collect_responses(config, **kwargs) + + +def load_questions( + path: Path | str, + *, + question_column: str = "question", + id_column: str = "id", +) -> list[LiveQuestion]: + """Load questions from CSV, JSON, or JSONL.""" + qpath = Path(path) + if not qpath.exists(): + raise FileNotFoundError(f"Question file not found: {qpath}") + + suffix = qpath.suffix.lower() + if suffix == ".csv": + questions = _load_csv(qpath, question_column, id_column) + elif suffix == ".jsonl": + questions = _load_jsonl(qpath, question_column, id_column) + elif suffix == ".json": + questions = _load_json(qpath, question_column, id_column) + else: + raise ValueError( + f"Unsupported question file type '{suffix}'. Use CSV, JSON, or JSONL." + ) + + if not questions: + raise ValueError(f"No questions found in {qpath}") + return questions + + +def extract_by_path(data: Any, path: str) -> Any: + """Extract a value from nested data using a dot-separated path.""" + current = data + for segment in path.split("."): + if isinstance(current, dict): + current = current.get(segment) + elif isinstance(current, list): + try: + current = current[int(segment)] + except (ValueError, IndexError): + return None + else: + return None + if current is None: + return None + return current + + +def substitute_template(template: Any, variables: dict[str, str]) -> Any: + """Recursively substitute ``{key}`` placeholders in string values.""" + if isinstance(template, str): + return template.format_map(variables) + if isinstance(template, dict): + return {k: substitute_template(v, variables) for k, v in template.items()} + if isinstance(template, list): + return [substitute_template(item, variables) for item in template] + return template + + +def _resolve_auth_headers( + config: CollectorConfig, + credentials: ModelCredentials | None, +) -> dict[str, str]: + headers: dict[str, str] = {} + if credentials: + headers.update(credentials.auth_headers) + if credentials.api_key: + headers["Authorization"] = f"Bearer {credentials.api_key}" + if config.api_key_env: + api_key = os.getenv(config.api_key_env) + if not api_key: + raise ValueError(f"Environment variable {config.api_key_env!r} is required") + headers["Authorization"] = f"Bearer {api_key}" + headers.update(config.request_headers) + return headers + + +def _resolve_verify( + config: CollectorConfig, + credentials: ModelCredentials | None, +) -> bool | str: + if config.insecure: + return False + if config.ca_bundle: + return config.ca_bundle + if credentials and credentials.ca_cert_path: + return credentials.ca_cert_path + for candidate in _CA_BUNDLE_CANDIDATES: + if candidate.exists(): + return str(candidate) + return True + + +def _collect_one( + config: CollectorConfig, + client: Any, + headers: dict[str, str], + question: LiveQuestion, +) -> CollectedRecord: + if config.protocol == CollectorProtocol.OPENAI_CHAT: + return _collect_openai(config, client, headers, question) + return _collect_generic_http(config, client, headers, question) + + +def _collect_openai( + config: CollectorConfig, + client: Any, + headers: dict[str, str], + question: LiveQuestion, +) -> CollectedRecord: + messages: list[dict[str, str]] = [] + if config.system_prompt: + messages.append({"role": "system", "content": config.system_prompt}) + messages.append({"role": "user", "content": question.question}) + + body: dict[str, Any] = dict(config.extra_body) + body["model"] = config.model + body["messages"] = messages + + url = f"{config.endpoint_url}/chat/completions" + return _send_request(config, client, headers, question, url, body, _extract_openai) + + +def _collect_generic_http( + config: CollectorConfig, + client: Any, + headers: dict[str, str], + question: LiveQuestion, +) -> CollectedRecord: + request_template = config.request_template + response_path = config.response_path + if not request_template or not response_path: + raise ValueError( + "request_template and response_path are required for generic_http" + ) + + variables = { + "question": question.question, + "question_id": question.question_id or "", + } + body = substitute_template(request_template, variables) + return _send_request( + config, + client, + headers, + question, + config.endpoint_url, + body, + lambda resp: _extract_generic(resp, response_path), + ) + + +def _send_request( + config: CollectorConfig, + client: Any, + headers: dict[str, str], + question: LiveQuestion, + url: str, + body: dict[str, Any], + extractor: Callable[[dict[str, Any]], str | None], +) -> CollectedRecord: + last_error: str | None = None + for attempt in range(config.max_retries + 1): + start = time.monotonic() + try: + response = client.post( + url, + json=body, + headers=dict(headers), + timeout=config.timeout_seconds, + ) + latency_ms = (time.monotonic() - start) * 1000 + + status_code = getattr(response, "status_code", None) + if isinstance(status_code, int) and 300 <= status_code < 400: + raise ValueError(f"Redirect not allowed: HTTP {status_code}") + + response.raise_for_status() + raw_response = response.json() + if not isinstance(raw_response, dict): + raise ValueError("Response must be a JSON object") + + text = extractor(cast(dict[str, Any], raw_response)) + extra_fields = _extract_extra_fields( + cast(dict[str, Any], raw_response), config.extra_response_paths + ) + return CollectedRecord( + source_fields=question.source_row, + response=text, + extra_fields=extra_fields, + raw_response=cast(dict[str, Any], raw_response), + latency_ms=latency_ms, + ) + except Exception as exc: + last_error = f"{type(exc).__name__}: {exc}" + if attempt < config.max_retries and config.retry_backoff_seconds > 0: + delay = min( + config.retry_backoff_seconds * (2**attempt), + config.max_retry_backoff_seconds, + ) + time.sleep(delay) + + return CollectedRecord( + source_fields=question.source_row, + error=last_error, + ) + + +def _flatten_record(record: CollectedRecord) -> dict[str, Any]: + """Flatten a record for JSONL output. + + Starts with the original input columns (``source_fields``), then adds + collector-produced fields (``response``, ``error``, etc.) and any + ``extra_fields`` on top. + """ + data = dict(record.source_fields) + data["response"] = record.response + data["raw_response"] = record.raw_response + data["error"] = record.error + data["latency_ms"] = record.latency_ms + if record.extra_fields: + data.update(record.extra_fields) + return data + + +def _extract_extra_fields( + raw_response: dict[str, Any], + extra_paths: dict[str, str] | None, +) -> dict[str, Any]: + if not extra_paths: + return {} + result: dict[str, Any] = {} + for output_name, path in extra_paths.items(): + result[output_name] = extract_by_path(raw_response, path) + return result + + +def _extract_openai(raw_response: dict[str, Any]) -> str | None: + value = extract_by_path(raw_response, "choices.0.message.content") + return str(value) if value is not None else None + + +def _extract_generic(raw_response: dict[str, Any], response_path: str) -> str | None: + value = extract_by_path(raw_response, response_path) + if value is None: + return None + return str(value) if not isinstance(value, str) else value + + +def _load_csv(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]: + with path.open(newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + if reader.fieldnames is None or question_column not in reader.fieldnames: + raise ValueError(f"CSV must include a '{question_column}' column") + + questions: list[LiveQuestion] = [] + for i, row in enumerate(reader, start=1): + text = (row.get(question_column) or "").strip() + if not text: + continue + qid = (row.get(id_column) or "").strip() or str(i) + questions.append( + LiveQuestion(question=text, question_id=qid, source_row=dict(row)) + ) + return questions + + +def _load_jsonl(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]: + questions: list[LiveQuestion] = [] + with path.open(encoding="utf-8") as f: + for i, line in enumerate(f, start=1): + if not line.strip(): + continue + raw = json.loads(line) + if not isinstance(raw, dict): + raise ValueError(f"JSONL row {i} must be an object") + q = _question_from_row(raw, question_column, id_column, i) + if q: + questions.append(q) + return questions + + +def _load_json(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]: + raw_data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(raw_data, dict) and "questions" in raw_data: + raw_data = raw_data["questions"] + if not isinstance(raw_data, list): + raise ValueError("JSON input must be a list or contain a 'questions' list") + + questions: list[LiveQuestion] = [] + for i, raw in enumerate(raw_data, start=1): + if not isinstance(raw, dict): + raise ValueError(f"JSON row {i} must be an object") + q = _question_from_row(raw, question_column, id_column, i) + if q: + questions.append(q) + return questions + + +def _question_from_row( + row: dict[str, Any], + question_column: str, + id_column: str, + row_index: int, +) -> LiveQuestion | None: + raw_q = row.get(question_column) + if raw_q is None: + raise ValueError(f"Row {row_index} must include '{question_column}'") + if not isinstance(raw_q, str): + raise ValueError(f"Row {row_index} '{question_column}' must be a string") + text = raw_q.strip() + if not text: + return None + + raw_id = row.get(id_column) + qid = str(raw_id).strip() if raw_id is not None else str(row_index) + + return LiveQuestion(question=text, question_id=qid, source_row=dict(row)) diff --git a/tests/unit/test_collector.py b/tests/unit/test_collector.py new file mode 100644 index 00000000..1471cea0 --- /dev/null +++ b/tests/unit/test_collector.py @@ -0,0 +1,742 @@ +"""Unit tests for the live endpoint response collector.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import httpx +import pytest +from evalhub.adapter.collector import ( + CollectorConfig, + CollectorError, + CollectorProtocol, + collect_responses, + collect_responses_from_parameters, + extract_by_path, + is_configured, + load_questions, + substitute_template, +) + +pytestmark = pytest.mark.unit + + +def _response(status_code: int, payload: dict[str, Any]) -> httpx.Response: + return httpx.Response( + status_code, + json=payload, + request=httpx.Request("POST", "https://endpoint.example/v1/chat/completions"), + ) + + +def _openai_ok(content: str = "test answer") -> httpx.Response: + return _response( + 200, + { + "choices": [{"message": {"content": content}, "finish_reason": "stop"}], + "usage": {"total_tokens": 10}, + }, + ) + + +# -- is_configured ----------------------------------------------------------- + + +class TestIsConfigured: + def test_configured_with_valid_dict(self) -> None: + assert is_configured({"live_collection": {"endpoint_url": "http://x"}}) + + def test_not_configured_empty_params(self) -> None: + assert not is_configured({}) + + def test_not_configured_none(self) -> None: + assert not is_configured(None) + + def test_not_configured_empty_dict(self) -> None: + assert not is_configured({"live_collection": {}}) + + def test_not_configured_non_dict(self) -> None: + assert not is_configured({"live_collection": "string"}) + + +# -- CollectorConfig validation ----------------------------------------------- + + +class TestCollectorConfigValidation: + def test_rejects_non_http_endpoint(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="http:// or https://"): + CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="file:///etc/passwd", + model="chatbot", + ) + + def test_openai_requires_model(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="model is required"): + CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + protocol=CollectorProtocol.OPENAI_CHAT, + ) + + def test_generic_http_requires_template(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="request_template is required"): + CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="https://api.example", + protocol=CollectorProtocol.GENERIC_HTTP, + response_path="result.text", + ) + + def test_generic_http_requires_response_path(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="response_path is required"): + CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="https://api.example", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={"input": "{question}"}, + ) + + def test_from_parameters_loads_config(self, tmp_path: Path) -> None: + config = CollectorConfig.from_parameters( + { + "live_collection": { + "questions_path": str(tmp_path / "q.csv"), + "output_dir": str(tmp_path / "out"), + "endpoint_url": "https://api.example/v1", + "model": "chatbot", + } + } + ) + assert config.model == "chatbot" + assert config.protocol == CollectorProtocol.OPENAI_CHAT + + def test_from_parameters_missing_key(self) -> None: + with pytest.raises(ValueError, match="live_collection"): + CollectorConfig.from_parameters({}) + + +# -- Auth resolution ---------------------------------------------------------- + + +class TestAuthResolution: + def test_api_key_env_sets_bearer( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("TEST_KEY", "secret") + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + calls: list[dict[str, Any]] = [] + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + calls.append(kwargs) + return _openai_ok() + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + api_key_env="TEST_KEY", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + assert calls[0]["headers"]["Authorization"] == "Bearer secret" + + def test_missing_api_key_env_raises( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("MISSING_KEY", raising=False) + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + api_key_env="MISSING_KEY", + use_model_credentials=False, + ) + with pytest.raises(ValueError, match="MISSING_KEY"): + collect_responses(config, client=MagicMock(), credentials=None) + + def test_request_headers_override( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("TEST_KEY", "from-env") + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + calls: list[dict[str, Any]] = [] + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + calls.append(kwargs) + return _openai_ok() + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + api_key_env="TEST_KEY", + request_headers={"Authorization": "Custom override"}, + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + assert calls[0]["headers"]["Authorization"] == "Custom override" + + +# -- Question loading --------------------------------------------------------- + + +class TestLoadQuestions: + def test_csv(self, tmp_path: Path) -> None: + p = tmp_path / "q.csv" + p.write_text("id,question\n1,hello\n2,world\n", encoding="utf-8") + qs = load_questions(p) + assert len(qs) == 2 + assert qs[0].question == "hello" + assert qs[0].question_id == "1" + + def test_jsonl(self, tmp_path: Path) -> None: + p = tmp_path / "q.jsonl" + p.write_text( + json.dumps({"question": "hi", "id": "a"}) + + "\n" + + json.dumps({"question": "bye", "id": "b"}) + + "\n", + encoding="utf-8", + ) + qs = load_questions(p) + assert len(qs) == 2 + assert qs[1].question == "bye" + + def test_json_array(self, tmp_path: Path) -> None: + p = tmp_path / "q.json" + p.write_text(json.dumps([{"question": "test"}]), encoding="utf-8") + qs = load_questions(p) + assert len(qs) == 1 + + def test_json_wrapper(self, tmp_path: Path) -> None: + p = tmp_path / "q.json" + p.write_text( + json.dumps({"questions": [{"question": "test"}]}), encoding="utf-8" + ) + qs = load_questions(p) + assert len(qs) == 1 + + def test_empty_file_raises(self, tmp_path: Path) -> None: + p = tmp_path / "q.csv" + p.write_text("question\n", encoding="utf-8") + with pytest.raises(ValueError, match="No questions"): + load_questions(p) + + def test_missing_file_raises(self, tmp_path: Path) -> None: + with pytest.raises(FileNotFoundError): + load_questions(tmp_path / "missing.csv") + + def test_skips_empty_questions(self, tmp_path: Path) -> None: + p = tmp_path / "q.csv" + p.write_text("question\nhello\n\nworld\n", encoding="utf-8") + qs = load_questions(p) + assert len(qs) == 2 + + def test_preserves_source_row(self, tmp_path: Path) -> None: + p = tmp_path / "q.csv" + p.write_text( + "question,category,difficulty\nhello,greet,easy\n", encoding="utf-8" + ) + qs = load_questions(p) + assert qs[0].source_row == { + "question": "hello", + "category": "greet", + "difficulty": "easy", + } + + +# -- Template substitution and path extraction -------------------------------- + + +class TestSubstituteTemplate: + def test_string(self) -> None: + assert substitute_template("Hi {question}", {"question": "hello"}) == "Hi hello" + + def test_nested_dict(self) -> None: + result = substitute_template({"a": {"b": "{question}"}}, {"question": "test"}) + assert result == {"a": {"b": "test"}} + + def test_list(self) -> None: + result = substitute_template(["{question}", 42], {"question": "hi"}) + assert result == ["hi", 42] + + def test_non_string_passthrough(self) -> None: + assert substitute_template(42, {"question": "hi"}) == 42 + assert substitute_template(True, {"question": "hi"}) is True + + +class TestExtractByPath: + def test_simple_key(self) -> None: + assert extract_by_path({"a": "val"}, "a") == "val" + + def test_nested(self) -> None: + assert extract_by_path({"a": {"b": {"c": "val"}}}, "a.b.c") == "val" + + def test_array_index(self) -> None: + assert extract_by_path({"a": [10, 20, 30]}, "a.1") == 20 + + def test_missing_returns_none(self) -> None: + assert extract_by_path({"a": 1}, "b") is None + + def test_openai_path(self) -> None: + data = {"choices": [{"message": {"content": "answer"}}]} + assert extract_by_path(data, "choices.0.message.content") == "answer" + + def test_mcp_path(self) -> None: + data = {"result": {"content": [{"text": "answer"}]}} + assert extract_by_path(data, "result.content.0.text") == "answer" + + +# -- OpenAI chat collection --------------------------------------------------- + + +class TestCollectOpenAI: + def test_collects_responses(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\nworld\n", encoding="utf-8") + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _openai_ok("answer") + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.total == 2 + assert manifest.completed == 2 + assert manifest.failed == 0 + + output = tmp_path / "out" / "responses.jsonl" + rows = [json.loads(line) for line in output.read_text().splitlines()] + assert rows[0]["response"] == "answer" + assert rows[0]["error"] is None + assert rows[0]["latency_ms"] is not None + + def test_writes_manifest(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + mock_client = MagicMock() + mock_client.post = MagicMock(return_value=_openai_ok()) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + use_model_credentials=False, + ) + collect_responses(config, client=mock_client, credentials=None) + + manifest_path = tmp_path / "out" / "manifest.json" + assert manifest_path.exists() + manifest = json.loads(manifest_path.read_text()) + assert manifest["experimental"] is True + assert manifest["protocol"] == "openai_chat_completions" + assert manifest["total"] == 1 + assert manifest["completed"] == 1 + + def test_system_prompt_included(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + calls: list[dict[str, Any]] = [] + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + calls.append(kwargs) + return _openai_ok() + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + system_prompt="You are a helpful assistant.", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + body = calls[0]["json"] + assert body["messages"][0] == { + "role": "system", + "content": "You are a helpful assistant.", + } + assert body["messages"][1] == {"role": "user", "content": "hello"} + + def test_redirect_recorded_as_error(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return httpx.Response( + 307, + headers={"location": "https://other.example"}, + request=httpx.Request( + "POST", "https://api.example/v1/chat/completions" + ), + ) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.failed == 1 + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert "Redirect" in rows[0]["error"] + + def test_fail_fast_raises(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\nworld\n", encoding="utf-8") + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _response(500, {"error": "server error"}) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + fail_fast=True, + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + with pytest.raises(CollectorError): + collect_responses(config, client=mock_client, credentials=None) + + def test_best_effort_records_errors(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\nworld\n", encoding="utf-8") + + call_count = 0 + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + nonlocal call_count + call_count += 1 + if call_count == 1: + return _response(500, {"error": "fail"}) + return _openai_ok() + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + fail_fast=False, + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.failed == 1 + assert manifest.completed == 1 + + def test_retries_transient_errors( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + monkeypatch.setattr("evalhub.adapter.collector.time.sleep", lambda _: None) + + responses = [ + _response(500, {"error": "temporary"}), + _openai_ok("after retry"), + ] + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return responses.pop(0) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + max_retries=1, + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.completed == 1 + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] == "after retry" + + def test_progress_callback(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\na\nb\nc\n", encoding="utf-8") + + progress: list[float] = [] + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = MagicMock(return_value=_openai_ok()) + collect_responses( + config, + client=mock_client, + credentials=None, + progress_callback=progress.append, + ) + + assert len(progress) == 3 + assert progress[-1] == pytest.approx(1.0) + + +# -- Generic HTTP collection -------------------------------------------------- + + +class TestCollectGenericHTTP: + def test_template_substitution(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + calls: list[dict[str, Any]] = [] + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + calls.append(kwargs) + return _response(200, {"result": {"text": "answer"}}) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/invoke", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={"input": "{question}", "id": "{question_id}"}, + response_path="result.text", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert calls[0]["json"]["input"] == "hello" + assert manifest.completed == 1 + + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] == "answer" + + def test_mcp_style_request(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.jsonl" + qpath.write_text( + json.dumps({"question": "What is K8s?"}) + "\n", encoding="utf-8" + ) + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _response( + 200, + { + "jsonrpc": "2.0", + "result": { + "content": [{"type": "text", "text": "Kubernetes is..."}] + }, + "id": "1", + }, + ) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://mcp-chatbot.example/mcp", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={ + "jsonrpc": "2.0", + "method": "tools/call", + "params": {"name": "chat", "arguments": {"message": "{question}"}}, + "id": "{question_id}", + }, + response_path="result.content.0.text", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.completed == 1 + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] == "Kubernetes is..." + + def test_extra_response_paths_extracted(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _response( + 200, + { + "output": { + "answer": "the answer", + "context": [{"source": "doc1", "text": "relevant text"}], + "score": 0.95, + } + }, + ) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={"input": {"question": "{question}"}}, + response_path="output.answer", + extra_response_paths={ + "contexts": "output.context", + "confidence": "output.score", + }, + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] == "the answer" + assert rows[0]["contexts"] == [{"source": "doc1", "text": "relevant text"}] + assert rows[0]["confidence"] == 0.95 + + def test_input_metadata_flattened_to_top_level(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text( + "question,reference,category\nWhat is X?,X is Y,general\n", + encoding="utf-8", + ) + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _response(200, {"output": {"answer": "X is Y indeed"}}) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={"input": {"question": "{question}"}}, + response_path="output.answer", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] == "X is Y indeed" + assert rows[0]["reference"] == "X is Y" + assert rows[0]["category"] == "general" + assert "metadata" not in rows[0] + + def test_response_path_missing_returns_none(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: + return _response(200, {"unexpected": "shape"}) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example", + protocol=CollectorProtocol.GENERIC_HTTP, + request_template={"q": "{question}"}, + response_path="result.text", + use_model_credentials=False, + ) + mock_client = MagicMock() + mock_client.post = fake_post + collect_responses(config, client=mock_client, credentials=None) + + rows = [ + json.loads(line) + for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines() + ] + assert rows[0]["response"] is None + + +# -- collect_responses_from_parameters ---------------------------------------- + + +class TestCollectFromParameters: + def test_convenience_wrapper(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + + mock_client = MagicMock() + mock_client.post = MagicMock(return_value=_openai_ok()) + + manifest = collect_responses_from_parameters( + { + "live_collection": { + "questions_path": str(qpath), + "output_dir": str(tmp_path / "out"), + "endpoint_url": "https://api.example/v1", + "model": "chatbot", + "use_model_credentials": False, + } + }, + client=mock_client, + credentials=None, + ) + + assert manifest.completed == 1 From 75230ece2c9b89fc731ec35ec2e4fc171f39b31f Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 9 Jun 2026 16:55:08 +0300 Subject: [PATCH 2/7] fix(adapter): improve collector error messages and field collision warnings Address CodeRabbit review findings: - Wrap template format_map() to catch KeyError and provide a clear message listing available placeholders - Log a warning when input source fields collide with reserved collector output field names (response, raw_response, error, latency_ms) Co-Authored-By: Claude Opus 4.6 (1M context) --- src/evalhub/adapter/collector.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py index 52c754bf..5abc0432 100644 --- a/src/evalhub/adapter/collector.py +++ b/src/evalhub/adapter/collector.py @@ -334,7 +334,13 @@ def extract_by_path(data: Any, path: str) -> Any: def substitute_template(template: Any, variables: dict[str, str]) -> Any: """Recursively substitute ``{key}`` placeholders in string values.""" if isinstance(template, str): - return template.format_map(variables) + try: + return template.format_map(variables) + except KeyError as exc: + raise ValueError( + f"Unknown placeholder {exc} in template. " + f"Available: {', '.join(variables.keys())}" + ) from exc if isinstance(template, dict): return {k: substitute_template(v, variables) for k, v in template.items()} if isinstance(template, list): @@ -491,6 +497,9 @@ def _send_request( ) +_RESERVED_FIELDS = {"response", "raw_response", "error", "latency_ms"} + + def _flatten_record(record: CollectedRecord) -> dict[str, Any]: """Flatten a record for JSONL output. @@ -499,6 +508,11 @@ def _flatten_record(record: CollectedRecord) -> dict[str, Any]: ``extra_fields`` on top. """ data = dict(record.source_fields) + collisions = _RESERVED_FIELDS & record.source_fields.keys() + if collisions: + logger.warning( + "Source fields %s will be overwritten by collector output", collisions + ) data["response"] = record.response data["raw_response"] = record.raw_response data["error"] = record.error From 11430a6838982552343c3d68f5bc6bc32ca20f3e Mon Sep 17 00:00:00 2001 From: csoceanu Date: Wed, 19 Aug 2026 16:18:21 +0300 Subject: [PATCH 3/7] fix: resolve mypy type errors and remove non-existent attribute in collector - Remove reference to non-existent ModelCredentials.auth_headers attribute - Convert Path to string in _resolve_verify return value to match bool | str type - All pre-commit checks now pass (ruff, mypy, pytest) Co-Authored-By: Claude Haiku 4.5 --- src/evalhub/adapter/collector.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py index 5abc0432..0a0612ee 100644 --- a/src/evalhub/adapter/collector.py +++ b/src/evalhub/adapter/collector.py @@ -353,10 +353,8 @@ def _resolve_auth_headers( credentials: ModelCredentials | None, ) -> dict[str, str]: headers: dict[str, str] = {} - if credentials: - headers.update(credentials.auth_headers) - if credentials.api_key: - headers["Authorization"] = f"Bearer {credentials.api_key}" + if credentials and credentials.api_key: + headers["Authorization"] = f"Bearer {credentials.api_key}" if config.api_key_env: api_key = os.getenv(config.api_key_env) if not api_key: @@ -375,7 +373,7 @@ def _resolve_verify( if config.ca_bundle: return config.ca_bundle if credentials and credentials.ca_cert_path: - return credentials.ca_cert_path + return str(credentials.ca_cert_path) for candidate in _CA_BUNDLE_CANDIDATES: if candidate.exists(): return str(candidate) From 1aa8855efcb330b93c6b8ec1be6d0232f9ac3d3e Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 15 Sep 2026 14:05:18 +0300 Subject: [PATCH 4/7] style(adapter): fix import ordering --- src/evalhub/adapter/__init__.py | 30 ++++++++++++++++-------------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/src/evalhub/adapter/__init__.py b/src/evalhub/adapter/__init__.py index 78c15488..f0872b0b 100644 --- a/src/evalhub/adapter/__init__.py +++ b/src/evalhub/adapter/__init__.py @@ -79,6 +79,22 @@ def run_benchmark_job( ) from .auth import ModelCredentials, read_model_auth_key, resolve_model_credentials from .callbacks import DefaultCallbacks +from .collector import ( + CollectedRecord, + CollectionManifest, + CollectorConfig, + CollectorError, + CollectorProtocol, + LiveQuestion, + collect_responses, + collect_responses_from_parameters, + extract_by_path, + load_questions, + substitute_template, +) +from .collector import ( + is_configured as is_collection_configured, +) from .config import MlflowBackend, get_job_spec_path from .models import ( CapabilityEvalEntry, @@ -97,20 +113,6 @@ def run_benchmark_job( OCIArtifactSpec, SafetyEvalEntry, ) -from .collector import ( - CollectedRecord, - CollectionManifest, - CollectorConfig, - CollectorError, - CollectorProtocol, - LiveQuestion, - collect_responses, - collect_responses_from_parameters, - extract_by_path, - is_configured as is_collection_configured, - load_questions, - substitute_template, -) from .oci import OCIArtifactPersister from .settings import AdapterSettings from .telemetry import EvalTracer, configure_telemetry From d29f74cc8010b335939159a4bd89f0bc3a6c9ca5 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 15 Sep 2026 18:01:18 +0300 Subject: [PATCH 5/7] fix(adapter): address collector review feedback --- README.md | 125 ++++++++++++++++--------------- src/evalhub/adapter/collector.py | 61 ++++++++++----- tests/unit/test_collector.py | 77 +++++++++++++++++++ 3 files changed, 186 insertions(+), 77 deletions(-) diff --git a/README.md b/README.md index 4fd8336b..03f199cb 100755 --- a/README.md +++ b/README.md @@ -571,66 +571,6 @@ results = JobResults(..., eval_card=eval_card, env_card=env_card) callbacks.report_results(results) ``` -## Live Endpoint Response Collection (Experimental) - -The SDK includes a collector utility that adapters can call during `LOADING_DATA` -to query a chatbot endpoint with test questions and collect responses for evaluation. - -### OpenAI-Compatible Endpoint - -```python -from evalhub.adapter import ( - CollectorConfig, collect_responses, is_collection_configured, - resolve_model_credentials, -) - -# In your adapter's run_benchmark_job(): -if is_collection_configured(config.parameters): - creds = resolve_model_credentials() - collector_config = CollectorConfig.from_parameters(config.parameters) - manifest = collect_responses(collector_config, credentials=creds) - # manifest.output_path -> responses.jsonl for your evaluation framework -``` - -Job parameters: -```json -{ - "live_collection": { - "questions_path": "/test_data/questions.csv", - "output_dir": "/tmp/collected", - "endpoint_url": "https://my-chatbot.example/v1", - "model": "my-chatbot-v2", - "protocol": "openai_chat_completions" - } -} -``` - -### Generic HTTP Endpoint (MCP, Langflow, custom APIs) - -```json -{ - "live_collection": { - "questions_path": "/test_data/questions.csv", - "output_dir": "/tmp/collected", - "endpoint_url": "https://mcp-chatbot.example/mcp", - "protocol": "generic_http", - "request_template": { - "jsonrpc": "2.0", - "method": "tools/call", - "params": {"name": "chat", "arguments": {"message": "{question}"}}, - "id": "{question_id}" - }, - "response_path": "result.content.0.text", - "extra_response_paths": { - "retrieved_contexts": "result.sources" - } - } -} -``` - -The collector uses SDK auth patterns (`resolve_model_credentials()`) and -TLS auto-detection by default. See the module docstring in -`src/evalhub/adapter/collector.py` for all configuration options. **additional_info** - supplementary evaluation metadata: `additional_info` is a `dict[str, Any]` of supplementary key-value pairs for @@ -719,6 +659,71 @@ class LMEvalAdapter(FrameworkAdapter): return JobResults(...) ``` +## Live Endpoint Response Collection (Experimental) + +The SDK includes a collector utility that adapters can call during `LOADING_DATA` +to query a chatbot endpoint with test questions and collect responses for evaluation. + +### OpenAI-Compatible Endpoint + +```python +from evalhub.adapter import ( + CollectorConfig, collect_responses, is_collection_configured, + resolve_model_credentials, +) + +# In your adapter's run_benchmark_job(): +if is_collection_configured(config.parameters): + creds = resolve_model_credentials() + collector_config = CollectorConfig.from_parameters(config.parameters) + manifest = collect_responses(collector_config, credentials=creds) + # manifest.output_path -> responses.jsonl for your evaluation framework +``` + +Job parameters: +```json +{ + "live_collection": { + "questions_path": "/test_data/questions.csv", + "output_dir": "/tmp/collected", + "endpoint_url": "https://my-chatbot.example/v1", + "model": "my-chatbot-v2", + "protocol": "openai_chat_completions" + } +} +``` + +### Generic HTTP Endpoint (MCP, Langflow, custom APIs) + +```json +{ + "live_collection": { + "questions_path": "/test_data/questions.csv", + "output_dir": "/tmp/collected", + "endpoint_url": "https://mcp-chatbot.example/mcp", + "protocol": "generic_http", + "request_template": { + "jsonrpc": "2.0", + "method": "tools/call", + "params": {"name": "chat", "arguments": {"message": "{question}"}}, + "id": "{question_id}" + }, + "response_path": "result.content.0.text", + "extra_response_paths": { + "retrieved_contexts": "result.sources" + } + } +} +``` + +The collector uses SDK TLS auto-detection by default. In Kubernetes, the model +secret contains a sidecar reference token and cannot authenticate a direct request +to the live endpoint. Configure `api_key_env` or `request_headers` with the live +endpoint's actual credential instead; the collector will not send the sidecar +reference token directly to the endpoint. + +See the module docstring in `src/evalhub/adapter/collector.py` for all options. + ## Deployment ### Container Structure diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py index 0a0612ee..b510eff7 100644 --- a/src/evalhub/adapter/collector.py +++ b/src/evalhub/adapter/collector.py @@ -29,6 +29,8 @@ from pydantic import BaseModel, Field, field_validator, model_validator from .auth import ModelCredentials, resolve_model_credentials +from .config import EvalHubMode +from .settings import AdapterSettings logger = logging.getLogger(__name__) @@ -165,6 +167,7 @@ class CollectedRecord(BaseModel): raw_response: dict[str, Any] | None = None error: str | None = None latency_ms: float | None = None + error_cause: Any = Field(default=None, exclude=True) class CollectionManifest(BaseModel): @@ -248,6 +251,7 @@ def collect_responses( raise CollectorError( record.error, question_id=question.question_id, + cause=record.error_cause, ) else: completed += 1 @@ -258,21 +262,22 @@ def collect_responses( if owns_client and hasattr(active_client, "close"): active_client.close() - manifest = CollectionManifest( - protocol=config.protocol.value, - endpoint_url=config.endpoint_url, - model=config.model, - questions_path=str(config.questions_path), - output_path=str(output_path), - total=total, - completed=completed, - failed=failed, - ) - manifest_path = config.output_dir / "manifest.json" - manifest_path.write_text( - json.dumps(manifest.model_dump(mode="json"), indent=2, default=str) + "\n", - encoding="utf-8", - ) + # Preserve a partial manifest when fail_fast aborts collection. + manifest = CollectionManifest( + protocol=config.protocol.value, + endpoint_url=config.endpoint_url, + model=config.model, + questions_path=str(config.questions_path), + output_path=str(output_path), + total=total, + completed=completed, + failed=failed, + ) + manifest_path = config.output_dir / "manifest.json" + manifest_path.write_text( + json.dumps(manifest.model_dump(mode="json"), indent=2, default=str) + "\n", + encoding="utf-8", + ) return manifest @@ -354,7 +359,14 @@ def _resolve_auth_headers( ) -> dict[str, str]: headers: dict[str, str] = {} if credentials and credentials.api_key: - headers["Authorization"] = f"Bearer {credentials.api_key}" + if AdapterSettings.from_env().mode == EvalHubMode.K8S: + logger.warning( + "Ignoring Kubernetes model credential ref token for direct live " + "endpoint access; configure api_key_env or request_headers with " + "the endpoint credential" + ) + else: + headers["Authorization"] = f"Bearer {credentials.api_key}" if config.api_key_env: api_key = os.getenv(config.api_key_env) if not api_key: @@ -449,6 +461,7 @@ def _send_request( extractor: Callable[[dict[str, Any]], str | None], ) -> CollectedRecord: last_error: str | None = None + last_cause: Exception | None = None for attempt in range(config.max_retries + 1): start = time.monotonic() try: @@ -481,7 +494,13 @@ def _send_request( latency_ms=latency_ms, ) except Exception as exc: + last_cause = exc last_error = f"{type(exc).__name__}: {exc}" + status_code = getattr(getattr(exc, "response", None), "status_code", None) + if isinstance(exc, ValueError) or ( + isinstance(status_code, int) and 400 <= status_code < 500 + ): + break if attempt < config.max_retries and config.retry_backoff_seconds > 0: delay = min( config.retry_backoff_seconds * (2**attempt), @@ -492,6 +511,7 @@ def _send_request( return CollectedRecord( source_fields=question.source_row, error=last_error, + error_cause=last_cause, ) @@ -534,7 +554,14 @@ def _extract_extra_fields( def _extract_openai(raw_response: dict[str, Any]) -> str | None: value = extract_by_path(raw_response, "choices.0.message.content") - return str(value) if value is not None else None + if value is not None: + return str(value) + if extract_by_path(raw_response, "choices.0.message.tool_calls"): + raise ValueError( + "OpenAI response contains tool_calls but no message.content; " + "configure generic_http or a custom extractor for tool responses" + ) + raise ValueError("OpenAI response is missing choices.0.message.content") def _extract_generic(raw_response: dict[str, Any], response_path: str) -> str | None: diff --git a/tests/unit/test_collector.py b/tests/unit/test_collector.py index 1471cea0..aa1da524 100644 --- a/tests/unit/test_collector.py +++ b/tests/unit/test_collector.py @@ -9,10 +9,13 @@ import httpx import pytest +from evalhub.adapter.auth import ModelCredentials from evalhub.adapter.collector import ( CollectorConfig, CollectorError, CollectorProtocol, + _extract_openai, + _resolve_auth_headers, collect_responses, collect_responses_from_parameters, extract_by_path, @@ -118,6 +121,53 @@ def test_from_parameters_loads_config(self, tmp_path: Path) -> None: assert config.model == "chatbot" assert config.protocol == CollectorProtocol.OPENAI_CHAT + def test_kubernetes_ref_token_is_not_sent_to_endpoint( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("EVALHUB_MODE", "k8s") + config = CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + ) + + headers = _resolve_auth_headers(config, ModelCredentials(api_key="api-key:ref")) + + assert "Authorization" not in headers + + def test_kubernetes_explicit_endpoint_key_is_allowed( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("EVALHUB_MODE", "k8s") + monkeypatch.setenv("LIVE_ENDPOINT_KEY", "real-key") + config = CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + api_key_env="LIVE_ENDPOINT_KEY", + ) + + headers = _resolve_auth_headers(config, ModelCredentials(api_key="api-key:ref")) + + assert headers["Authorization"] == "Bearer real-key" + + def test_openai_tool_call_response_is_not_silently_empty(self) -> None: + with pytest.raises(ValueError, match="tool_calls"): + _extract_openai( + { + "choices": [ + { + "message": { + "content": None, + "tool_calls": [{"function": {"name": "search"}}], + } + } + ] + } + ) + def test_from_parameters_missing_key(self) -> None: with pytest.raises(ValueError, match="live_collection"): CollectorConfig.from_parameters({}) @@ -447,6 +497,13 @@ def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: with pytest.raises(CollectorError): collect_responses(config, client=mock_client, credentials=None) + manifest = json.loads( + (tmp_path / "out" / "manifest.json").read_text(encoding="utf-8") + ) + assert manifest["total"] == 2 + assert manifest["completed"] == 0 + assert manifest["failed"] == 1 + def test_best_effort_records_errors(self, tmp_path: Path) -> None: qpath = tmp_path / "q.csv" qpath.write_text("question\nhello\nworld\n", encoding="utf-8") @@ -509,6 +566,26 @@ def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: ] assert rows[0]["response"] == "after retry" + def test_does_not_retry_client_errors(self, tmp_path: Path) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + mock_client = MagicMock() + mock_client.post.return_value = _response(401, {"error": "unauthorized"}) + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + max_retries=3, + use_model_credentials=False, + ) + + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.failed == 1 + assert mock_client.post.call_count == 1 + def test_progress_callback(self, tmp_path: Path) -> None: qpath = tmp_path / "q.csv" qpath.write_text("question\na\nb\nc\n", encoding="utf-8") From 0111e9afad910e10ad2039665028dd403a0a97b0 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Tue, 15 Sep 2026 18:15:42 +0300 Subject: [PATCH 6/7] docs: link collector configuration reference --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 03f199cb..a28cba54 100755 --- a/README.md +++ b/README.md @@ -722,7 +722,8 @@ to the live endpoint. Configure `api_key_env` or `request_headers` with the live endpoint's actual credential instead; the collector will not send the sidecar reference token directly to the endpoint. -See the module docstring in `src/evalhub/adapter/collector.py` for all options. +See [`src/evalhub/adapter/collector.py`](src/evalhub/adapter/collector.py) for the +complete collector configuration reference. ## Deployment From 60dc39356c8263c9948282abbc3de2330eac7d44 Mon Sep 17 00:00:00 2001 From: csoceanu Date: Wed, 16 Sep 2026 13:19:16 +0300 Subject: [PATCH 7/7] fix(adapter): harden collector retries and credentials --- src/evalhub/adapter/collector.py | 14 +++++--- tests/unit/test_collector.py | 61 +++++++++++++++++++++++++++++++- 2 files changed, 69 insertions(+), 6 deletions(-) diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py index b510eff7..f38d00c0 100644 --- a/src/evalhub/adapter/collector.py +++ b/src/evalhub/adapter/collector.py @@ -129,8 +129,8 @@ def from_parameters(cls, parameters: Mapping[str, Any]) -> CollectorConfig: @field_validator("endpoint_url") @classmethod def _validate_endpoint_url(cls, value: str) -> str: - if not value.lower().startswith(("http://", "https://")): - raise ValueError("endpoint_url must use http:// or https://") + if not value.lower().startswith("https://"): + raise ValueError("endpoint_url must use https://") return value.rstrip("/") @model_validator(mode="after") @@ -498,7 +498,9 @@ def _send_request( last_error = f"{type(exc).__name__}: {exc}" status_code = getattr(getattr(exc, "response", None), "status_code", None) if isinstance(exc, ValueError) or ( - isinstance(status_code, int) and 400 <= status_code < 500 + isinstance(status_code, int) + and 400 <= status_code < 500 + and status_code != 429 ): break if attempt < config.max_retries and config.retry_backoff_seconds > 0: @@ -531,12 +533,14 @@ def _flatten_record(record: CollectedRecord) -> dict[str, Any]: logger.warning( "Source fields %s will be overwritten by collector output", collisions ) + if record.extra_fields: + data.update(record.extra_fields) + # Extra response paths are user-configurable, so reapply collector-owned + # fields after merging to prevent them from overwriting collected values. data["response"] = record.response data["raw_response"] = record.raw_response data["error"] = record.error data["latency_ms"] = record.latency_ms - if record.extra_fields: - data.update(record.extra_fields) return data diff --git a/tests/unit/test_collector.py b/tests/unit/test_collector.py index aa1da524..0d0c47f2 100644 --- a/tests/unit/test_collector.py +++ b/tests/unit/test_collector.py @@ -11,10 +11,12 @@ import pytest from evalhub.adapter.auth import ModelCredentials from evalhub.adapter.collector import ( + CollectedRecord, CollectorConfig, CollectorError, CollectorProtocol, _extract_openai, + _flatten_record, _resolve_auth_headers, collect_responses, collect_responses_from_parameters, @@ -70,7 +72,7 @@ def test_not_configured_non_dict(self) -> None: class TestCollectorConfigValidation: def test_rejects_non_http_endpoint(self, tmp_path: Path) -> None: - with pytest.raises(ValueError, match="http:// or https://"): + with pytest.raises(ValueError, match="https://"): CollectorConfig( questions_path=tmp_path / "q.csv", output_dir=tmp_path / "out", @@ -78,6 +80,15 @@ def test_rejects_non_http_endpoint(self, tmp_path: Path) -> None: model="chatbot", ) + def test_rejects_http_endpoint(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="https://"): + CollectorConfig( + questions_path=tmp_path / "q.csv", + output_dir=tmp_path / "out", + endpoint_url="http://api.example/v1", + model="chatbot", + ) + def test_openai_requires_model(self, tmp_path: Path) -> None: with pytest.raises(ValueError, match="model is required"): CollectorConfig( @@ -586,6 +597,33 @@ def test_does_not_retry_client_errors(self, tmp_path: Path) -> None: assert manifest.failed == 1 assert mock_client.post.call_count == 1 + def test_retries_rate_limited_response( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + qpath = tmp_path / "q.csv" + qpath.write_text("question\nhello\n", encoding="utf-8") + monkeypatch.setattr("evalhub.adapter.collector.time.sleep", lambda _: None) + mock_client = MagicMock() + mock_client.post.side_effect = [ + _response(429, {"error": "rate limited"}), + _openai_ok("after backoff"), + ] + + config = CollectorConfig( + questions_path=qpath, + output_dir=tmp_path / "out", + endpoint_url="https://api.example/v1", + model="chatbot", + max_retries=1, + retry_backoff_seconds=0.1, + use_model_credentials=False, + ) + + manifest = collect_responses(config, client=mock_client, credentials=None) + + assert manifest.completed == 1 + assert mock_client.post.call_count == 2 + def test_progress_callback(self, tmp_path: Path) -> None: qpath = tmp_path / "q.csv" qpath.write_text("question\na\nb\nc\n", encoding="utf-8") @@ -691,6 +729,27 @@ def fake_post(*_: Any, **kwargs: Any) -> httpx.Response: ] assert rows[0]["response"] == "Kubernetes is..." + def test_extra_fields_cannot_overwrite_collector_fields(self) -> None: + record = CollectedRecord( + response="collected answer", + raw_response={"answer": "collected answer"}, + error=None, + latency_ms=12.5, + extra_fields={ + "response": "wrong answer", + "raw_response": {"answer": "wrong answer"}, + "error": "wrong error", + "latency_ms": 999.0, + }, + ) + + flattened = _flatten_record(record) + + assert flattened["response"] == "collected answer" + assert flattened["raw_response"] == {"answer": "collected answer"} + assert flattened["error"] is None + assert flattened["latency_ms"] == 12.5 + def test_extra_response_paths_extracted(self, tmp_path: Path) -> None: qpath = tmp_path / "q.csv" qpath.write_text("question\nhello\n", encoding="utf-8")