packages/traceloop-sdk/traceloop/sdk/evaluator/stream_client.py (1)

34-35: Build stream URL robustly; handle absolute URLs and pre-prefixed “/v2”

Current concatenation can yield invalid URLs like “…/v2/v2/...”. Normalize paths and respect absolute URLs.

-            full_stream_url = f"{self._api_endpoint}/v2{stream_url}"
+            # Build robust stream URL
+            if stream_url.startswith(("http://", "https://")):
+                full_stream_url = stream_url
+            elif stream_url.startswith("/v2/"):
+                full_stream_url = f"{self._api_endpoint}{stream_url}"
+            else:
+                prefix = "" if stream_url.startswith("/") else "/"
+                full_stream_url = f"{self._api_endpoint}/v2{prefix}{stream_url}"

packages/sample-app/sample_app/experiment/experiment_example.py (1)

15-16: Fail fast if Traceloop client isn’t initialized

Traceloop.init() can return None when API key/config is missing. Guard to avoid AttributeError.

 client = Traceloop.init()
+if client is None:
+    raise RuntimeError(
+        "Traceloop client not initialized. Ensure TRACELOOP_API_KEY is set."
+    )

packages/traceloop-sdk/traceloop/sdk/experiment/model.py (1)

10-16: Optional: Make models resilient to server-side schema additions (ignore extras)

If the API adds fields, strict parsing can break clients. Consider ignoring unknown fields for response models.

Example (Pydantic v2):

from pydantic import ConfigDict

class ExperimentResponse(BaseModel):
    model_config = ConfigDict(extra="ignore")
    ...

Apply similarly to other response-bearing models (ExperimentRunResponse, ExperimentInitResponse, CreateTaskResponse, TaskResponse).

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (2)

52-59: Docstring drift: remove undocumented parameter and align descriptions

The docstring mentions context_data, which isn’t in the signature. Trim it to avoid confusion and keep the Args accurate.

         Args:
             evaluator_slug: Slug of the evaluator to execute
-            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
-            client: Shared HTTP client for connection reuse (optional)
-            context_data: Context data to be passed to the evaluator (optional)
+            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
+            client: Shared HTTP client for connection reuse (optional)
             evaluator_version: Version of the evaluator to execute (optional)
             timeout_in_sec: Timeout in seconds for execution

---

44-45: Avoid shadowing built-in input; rename parameter for clarity

Shadowing Python’s built-in input can confuse readers and tools. Rename to evaluator_input.

-        input: Dict[str, str],
+        evaluator_input: Dict[str, str],
@@
-            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
+            evaluator_input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
@@
-        schema_mapping = InputSchemaMapping(
-            root={k: InputExtractor(source=v) for k, v in input.items()}
-        )
+        schema_mapping = InputSchemaMapping(
+            root={k: InputExtractor(source=v) for k, v in evaluator_input.items()}
+        )
@@
-        body = request.model_dump()
+        body = request.model_dump()

Also applies to: 54-55, 63-65, 67-72

packages/traceloop-sdk/traceloop/sdk/evaluator/stream_client.py (1)

75-78: Decode error body before formatting

error_text is bytes; printing raw bytes yields b'...'. Decode for readability.

-            error_text = await response.aread()
+            raw = await response.aread()
+            error_text = raw.decode("utf-8", errors="replace")
             raise Exception(
                 f"Failed to stream results: {response.status_code}, body: {error_text}"
             )

packages/sample-app/sample_app/experiment/experiment_example.py (2)

22-23: Validate OPENAI_API_KEY before creating client

Provide a clear error if the environment variable is missing.

-    openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise RuntimeError("OPENAI_API_KEY is not set in the environment")
+    openai_client = OpenAI(api_key=api_key)

---

9-9: Avoid sys.path manipulation in samples; prefer package-relative imports

Modifying sys.path can mask import issues. Consider using proper package structure or relative imports.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (3)

44-52: Docstring parameters are outdated; align with current signature and behavior

Mentions experiment_name, concurrency, and client which aren’t parameters. Update to reflect experiment_slug, related_ref, aux, and fixed concurrency behavior.

-        Args:
-            dataset_slug: Slug of the dataset to use
-            task: Function to run on each dataset row
-            evaluators: List of evaluator slugs to run
-            experiment_name: Name for this experiment run
-            concurrency: Number of concurrent tasks
-            exit_on_error: Whether to exit on first error
-            client: Traceloop client instance (if not provided, will initialize)
+        Args:
+            task: Function to run on each dataset row (sync; see note below)
+            dataset_slug: Slug of the dataset to use (requires dataset_version)
+            dataset_version: Version of the dataset to use
+            evaluators: List of evaluator slugs or (slug, version) tuples to run
+            experiment_slug: Slug for this experiment (auto-generated if not provided)
+            related_ref: Optional reference metadata dict for the run
+            aux: Optional auxiliary metadata dict for the run
+            exit_on_error: Whether to propagate the first error

---

97-105: Optional: Support async task callables without blocking the event loop

If task is async, await it; if it’s a heavy sync function, consider run_in_executor. This keeps the loop responsive.

+                # Allow either sync or async task functions
+                import inspect
-                task_result = task(row)
+                if inspect.iscoroutinefunction(task):
+                    task_result = await task(row)
+                else:
+                    task_result = task(row)

---

31-41: Optional: Make concurrency configurable instead of hard-coded 50

Expose a concurrency: int = 50 parameter to tune throughput.

-    async def run(
+    async def run(
         self,
         task: Callable[[Optional[Dict[str, Any]]], Dict[str, Any]],
         dataset_slug: Optional[str] = None,
         dataset_version: Optional[str] = None,
         evaluators: Optional[List[EvaluatorDetails]] = None,
         experiment_slug: Optional[str] = None,
         related_ref: Optional[Dict[str, str]] = None,
         aux: Optional[Dict[str, str]] = None,
-        exit_on_error: bool = False,
+        exit_on_error: bool = False,
+        concurrency: int = 50,
     ) -> Tuple[List[TaskResponse], List[str]]:
@@
-        semaphore = asyncio.Semaphore(50)
+        semaphore = asyncio.Semaphore(concurrency)

Also applies to: 136-136

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nina-kollman
PR: traceloop/openllmetry#3219
File: packages/traceloop-sdk/traceloop/sdk/datasets/dataset.py:357-360
Timestamp: 2025-08-04T15:35:30.188Z
Learning: In the traceloop SDK Dataset class, the established error handling pattern is that HTTP client methods return None on failure (after catching and logging RequestException), and the Dataset API methods check for None return values and raise Exception with descriptive messages. The update_row_api method is inconsistent with this pattern.

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (1)

37-103: Good: client lifecycle managed and richer error details included

The async client is properly closed when owned; HTTP errors include response text; SSE result fetch is cleanly integrated. Looks solid.

packages/sample-app/sample_app/experiment/experiment_example.py (1)

58-67: LGTM: Correct unpacking of run() return value and error printing

The example now respects (results, errors) return order and prints errors when present.

Also applies to: 76-85

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (1)

186-204: LGTM: HTTP error handling follows established SDK pattern

Consistent None-checks with clear exceptions, typed request/response models, and correct path normalization. Looks good.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (3)

151-156: Good: error handling now uses TaskResponse instead of dict probing

Previous feedback flagged "error" in result on Pydantic models. This code correctly checks result.error on a TaskResponse, avoiding TypeError and improving clarity.

---

79-107: Critical: blocking “requests” calls inside async flow will stall the event loop

run() is async but invokes synchronous HTTP via requests in _init_experiment, get_version_jsonl, and _create_task. This negates concurrency under load and can freeze other coroutines. Wrap these in asyncio.to_thread (minimal change) or move to an async HTTP client.

Also folds in the previously raised guard for dataset_version when dataset_slug is provided.

Apply this diff:

-        experiment = self._init_experiment(
-            experiment_slug,
-            dataset_slug=dataset_slug,
-            dataset_version=dataset_version,
-            evaluator_slugs=[slug for slug, _ in evaluator_details]
-            if evaluator_details
-            else None,
-            experiment_run_metadata=experiment_run_metadata,
-        )
+        experiment = await asyncio.to_thread(
+            self._init_experiment,
+            experiment_slug,
+            dataset_slug=dataset_slug,
+            dataset_version=dataset_version,
+            evaluator_slugs=[slug for slug, _ in evaluator_details]
+            if evaluator_details
+            else None,
+            experiment_run_metadata=experiment_run_metadata,
+        )
@@
-        rows = []
-        if dataset_slug:
-            jsonl_data = self._datasets.get_version_jsonl(dataset_slug, dataset_version)
-            rows = self._parse_jsonl_to_rows(jsonl_data)
+        rows = []
+        if dataset_slug:
+            if not dataset_version:
+                raise ValueError(
+                    "dataset_version is required when dataset_slug is provided"
+                )
+            jsonl_data = await asyncio.to_thread(
+                self._datasets.get_version_jsonl, dataset_slug, dataset_version
+            )
+            rows = self._parse_jsonl_to_rows(jsonl_data)
@@
-                task_id = self._create_task(
-                    experiment_slug=experiment_slug,
-                    experiment_run_id=run_id,
-                    task_input=row,
-                    task_output=task_result,
-                ).id
+                task_id = (
+                    await asyncio.to_thread(
+                        self._create_task,
+                        experiment_slug=experiment_slug,
+                        experiment_run_id=run_id,
+                        task_input=row,
+                        task_output=task_result,
+                    )
+                ).id

---

210-225: JSONL/NDJSON parsing drops the first record

The parser skips the first line assuming a header; NDJSON rarely includes headers. Parse all non-empty lines and ignore invalid JSON.

-        lines = jsonl_data.strip().split("\n")
-
-        # Skip the first line (columns definition)
-        for line in lines[1:]:
+        lines = jsonl_data.strip().split("\n")
+        for line in lines:
             if line.strip():
                 try:
                     row_data = json.loads(line)
                     rows.append(row_data)
                 except json.JSONDecodeError:
                     # Skip invalid JSON lines
                     continue

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (7)

110-127: Parallelize evaluator calls per row to reduce latency

Evaluator executions are awaited sequentially. Use asyncio.gather to run them concurrently and fold exceptions into the result map.

-                if evaluator_details:
-                    for evaluator_slug, evaluator_version in evaluator_details:
-                        try:
-                            eval_result = (
-                                await self._evaluator.run_experiment_evaluator(
-                                    evaluator_slug=evaluator_slug,
-                                    evaluator_version=evaluator_version,
-                                    task_id=task_id,
-                                    experiment_id=experiment.experiment.id,
-                                    experiment_run_id=run_id,
-                                    input=task_result,
-                                    timeout_in_sec=120,
-                                )
-                            )
-                            eval_results[evaluator_slug] = eval_result.result
-                        except Exception as e:
-                            eval_results[evaluator_slug] = f"Error: {str(e)}"
+                if evaluator_details:
+                    eval_tasks = [
+                        self._evaluator.run_experiment_evaluator(
+                            evaluator_slug=evaluator_slug,
+                            evaluator_version=evaluator_version,
+                            task_id=task_id,
+                            experiment_id=experiment.experiment.id,
+                            experiment_run_id=run_id,
+                            input=task_result,
+                            timeout_in_sec=120,
+                        )
+                        for evaluator_slug, evaluator_version in evaluator_details
+                    ]
+                    eval_results_list = await asyncio.gather(
+                        *eval_tasks, return_exceptions=True
+                    )
+                    for (evaluator_slug, _), r in zip(
+                        evaluator_details, eval_results_list
+                    ):
+                        if isinstance(r, Exception):
+                            eval_results[evaluator_slug] = f"Error: {str(r)}"
+                        else:
+                            eval_results[evaluator_slug] = r.result

Optional follow-up: reuse a single httpx.AsyncClient across all evaluator calls (Evaluator.run_experiment_evaluator supports a client parameter) to avoid connection churn. I can draft that if you want.

---

144-148: Don’t spawn tasks when wait_for_results is False

Currently tasks are created before the early return, causing fire-and-forget execution and hidden work. Return before creating tasks.

-        tasks = [asyncio.create_task(run_with_semaphore(row)) for row in rows]
-
-        if not wait_for_results:
-            return [], []
+        if not wait_for_results:
+            return [], []
+
+        tasks = [asyncio.create_task(run_with_semaphore(row)) for row in rows]

---

149-161: If stop_on_error is set, cancel remaining tasks to free resources

On first error the loop breaks but pending tasks keep running. Cancel them and await with return_exceptions=True to avoid warnings.

-        for completed_task in asyncio.as_completed(tasks):
+        for completed_task in asyncio.as_completed(tasks):
             try:
                 result = await completed_task
                 if result.error:
                     errors.append(result.error)
                 else:
                     results.append(result)
             except Exception as e:
                 error_msg = f"Task execution error: {str(e)}"
                 errors.append(error_msg)
                 if stop_on_error:
-                    break
+                    for t in tasks:
+                        if not t.done():
+                            t.cancel()
+                    await asyncio.gather(*tasks, return_exceptions=True)
+                    break

---

59-63: Avoid truncating cuid; keep full ID for uniqueness

Slicing to 11 chars increases collision risk at scale. Prefer the full cuid unless there’s an external length constraint.

-                os.getenv("TRACELOOP_EXP_SLUG") or "exp-" + str(cuid.cuid())[:11]
+                os.getenv("TRACELOOP_EXP_SLUG") or f"exp-{cuid.cuid()}"

If you have a max-length requirement for slugs, note it here and consider hashing.

---

70-77: Clarify EvaluatorDetails type and input normalization

The code normalizes evaluators into (slug, version) tuples. Ensure EvaluatorDetails is explicitly typed (e.g., Tuple[str, Optional[str]]) in model.py to match this usage, or widen the annotation here to accept Union[str, Tuple[str, Optional[str]]].

---

138-138: Make concurrency limit configurable

Semaphore(50) is a reasonable default, but exposing it as a parameter (e.g., max_concurrency: int = 50) helps tune for different environments and dataset sizes.

---

128-136: Minor: include row context in error to aid debugging (optionally)

When a row fails, consider including a stable identifier from the row (e.g., row.get("id")) rather than the full row to avoid PII leakage yet keep traceability.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nina-kollman
PR: traceloop/openllmetry#3219
File: packages/traceloop-sdk/traceloop/sdk/datasets/dataset.py:357-360
Timestamp: 2025-08-04T15:35:30.188Z
Learning: In the traceloop SDK Dataset class, the established error handling pattern is that HTTP client methods return None on failure (after catching and logging RequestException), and the Dataset API methods check for None return values and raise Exception with descriptive messages. The update_row_api method is inconsistent with this pattern.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (1)

227-242: Parse all JSONL lines; don’t drop the first record unconditionally

Unconditionally skipping the first line loses data for standard NDJSON. Attempt to parse every non-empty line; skip only invalid JSON lines.

Apply this diff:

     def _parse_jsonl_to_rows(self, jsonl_data: str) -> List[Dict[str, Any]]:
         """Parse JSONL string into list of {col_name: col_value} dictionaries"""
         rows = []
         lines = jsonl_data.strip().split("\n")
-
-        # Skip the first line (columns definition)
-        for line in lines[1:]:
+        for line in lines:
             if line.strip():
                 try:
                     row_data = json.loads(line)
                     rows.append(row_data)
                 except json.JSONDecodeError:
                     # Skip invalid JSON lines
                     continue
 
         return rows

packages/traceloop-sdk/traceloop/sdk/client/client.py (1)

55-60: Add X-Traceloop-SDK-Version to async client headers for parity with sync HTTPClient

This helps server-side analytics and consistent version tracking.

Apply this diff:

         self._async_http = httpx.AsyncClient(
             headers={
                 "Authorization": f"Bearer {self.api_key}",
                 "Content-Type": "application/json",
                 "User-Agent": f"traceloop-sdk/{__version__}",
+                "X-Traceloop-SDK-Version": __version__,
             },
             timeout=httpx.Timeout(120.0),
         )

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (2)

86-99: Docstring and parameters are out of sync (mentions client/context_data that don’t exist after refactor)

After switching to instance methods, remove stale parameters from docstrings and keep Args in sync.

Apply this diff:

-        Args:
-            evaluator_slug: Slug of the evaluator to execute
-            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
-            client: Shared HTTP client for connection reuse (optional)
-            context_data: Context data to be passed to the evaluator (optional)
-            evaluator_version: Version of the evaluator to execute (optional)
-            timeout_in_sec: Timeout in seconds for execution
+        Args:
+            evaluator_slug: Slug of the evaluator to execute
+            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
+            evaluator_version: Version of the evaluator to execute (optional)
+            timeout_in_sec: Timeout in seconds for execution

---

1-13: SSEClient relies on env for base URL and API key; prefer dependency injection

SSEClient builds Authorization from TRACELOOP_API_KEY and base URL from TRACELOOP_BASE_URL, ignoring headers/base URL on the shared client. If users configure a custom endpoint or non-env API key, streaming will 401 or hit the wrong host.

• Pass base URL and API key explicitly to SSEClient, or reuse the shared client’s base URL and headers.
• Minimal change: let SSEClient accept api_endpoint and reuse the shared client’s default headers unless overridden.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (4)

79-90: Avoid shadowing “experiment”; rename local to clarify and fix references

The variable named experiment refers to the init response, not the class. Rename for clarity and to avoid confusion.

Apply this diff:

-        experiment = self._init_experiment(
+        exp_init = self._init_experiment(
             experiment_slug,
             dataset_slug=dataset_slug,
             dataset_version=dataset_version,
             evaluator_slugs=[slug for slug, _ in evaluator_details]
             if evaluator_details
             else None,
             experiment_run_metadata=experiment_run_metadata,
         )
 
-        run_id = experiment.run.id
+        run_id = exp_init.run.id
@@
-                                        experiment_id=experiment.experiment.id,
+                                        experiment_id=exp_init.experiment.id,
@@
-                                    experiment_id=experiment.experiment.id,
+                                    experiment_id=exp_init.experiment.id,

Also applies to: 115-121, 127-134

---

91-95: Validate dataset_version when dataset_slug is provided

Calling get_version_jsonl with a None version would hit a bad endpoint. Make this explicit.

Apply this diff:

-        rows = []
-        if dataset_slug and dataset_version:
-            jsonl_data = self._datasets.get_version_jsonl(dataset_slug, dataset_version)
-            rows = self._parse_jsonl_to_rows(jsonl_data)
+        rows = []
+        if dataset_slug:
+            if not dataset_version:
+                raise ValueError("dataset_version is required when dataset_slug is provided")
+            jsonl_data = self._datasets.get_version_jsonl(dataset_slug, dataset_version)
+            rows = self._parse_jsonl_to_rows(jsonl_data)

---

99-107: Support both sync and async task callables

Enable users to provide async tasks without wrapping.

Apply this diff:

-            try:
-                task_result = task(row)
+            try:
+                result = task(row)
+                # Support async task functions
+                if hasattr(result, "__await__"):
+                    task_result = await result
+                else:
+                    task_result = result

Additionally, add this import at the top of the file:

# near other imports
from collections.abc import Awaitable  # optional if you prefer isinstance checks

---

61-69: Slug generation: ensure deterministic prefix and length if slug not provided

Not blocking, but consider centralizing slug generation (prefix, length) to avoid accidental changes and aid observability.

If helpful, extract to a helper:

def _default_slug(self) -> str:
    return f"exp_{cuid.cuid()[:12]}"

Then use self._experiment_slug or self._default_slug().

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nina-kollman
PR: traceloop/openllmetry#3219
File: packages/traceloop-sdk/traceloop/sdk/datasets/dataset.py:357-360
Timestamp: 2025-08-04T15:35:30.188Z
Learning: In the traceloop SDK Dataset class, the established error handling pattern is that HTTP client methods return None on failure (after catching and logging RequestException), and the Dataset API methods check for None return values and raise Exception with descriptive messages. The update_row_api method is inconsistent with this pattern.

packages/traceloop-sdk/traceloop/sdk/evaluator/stream_client.py (2)

7-12: Keep transport concerns centralized (auth/base URL) in the general HTTP client

Constructing auth headers and base URLs here duplicates responsibilities that belong to your shared HTTP client. Prefer: accept a fully configured AsyncClient and avoid re-deriving auth or base URL logic locally. This aligns with prior feedback.

Action: Remove local auth/header construction (except stream-specific headers) and rely on the shared client’s configuration.

---

29-35: Robustly build full_stream_url (handle absolute URLs and pre-prefixed “/v2”)

Current concatenation can yield “/v2/v2/…” or malformed URLs. Normalize cases: absolute URLs, already “/v2/”-prefixed paths, and bare paths.

-            full_stream_url = f"{self.client.base_url}/v2{stream_url}"
+            # Build robust stream URL
+            if stream_url.startswith("http://") or stream_url.startswith("https://"):
+                full_stream_url = stream_url
+            elif stream_url.startswith("/v2/"):
+                full_stream_url = f"{self.client.base_url}{stream_url}"
+            else:
+                prefix = "" if stream_url.startswith("/") else "/"
+                full_stream_url = f"{self.client.base_url}/v2{prefix}{stream_url}"

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (2)

27-31: Pass base_url into Evaluator to decouple from env and align with HTTP client config

Instantiate Evaluator with the same base_url as HTTPClient to avoid hidden environment coupling and ensure requests hit the intended API host.

Apply this diff if Evaluator supports base_url:

-        self._evaluator = Evaluator(async_http_client)
+        self._evaluator = Evaluator(async_http_client, http_client.base_url)

---

227-242: Parse all JSONL lines; don’t drop the first record unconditionally

NDJSON/JSONL does not include a header by default. Skipping the first line can drop a valid record. Attempt to parse every non-empty line and ignore only invalid JSON.

     def _parse_jsonl_to_rows(self, jsonl_data: str) -> List[Dict[str, Any]]:
         """Parse JSONL string into list of {col_name: col_value} dictionaries"""
         rows = []
         lines = jsonl_data.strip().split("\n")
 
-        # Skip the first line (columns definition)
-        for line in lines[1:]:
+        for line in lines:
             if line.strip():
                 try:
                     row_data = json.loads(line)
                     rows.append(row_data)
                 except json.JSONDecodeError:
                     # Skip invalid JSON lines
                     continue
 
         return rows

Ensure tests in tests/experiment/test_experiment.py cover:

• JSONL with all-valid lines (first line is data).
• JSONL with a header or malformed first line (ignored gracefully).

packages/traceloop-sdk/traceloop/sdk/evaluator/stream_client.py (5)

31-36: Streaming timeout semantics — consider disabling read timeout for SSE

For long-running streams, a read timeout equal to timeout_in_sec can prematurely abort an otherwise healthy stream. Prefer a connect/write timeout and no read timeout for true SSE behavior, or expose a separate read_timeout parameter.

-                timeout=httpx.Timeout(timeout_in_sec),
+                # For streaming, it's safer to allow indefinite reads
+                timeout=httpx.Timeout(connect=timeout_in_sec, read=None, write=timeout_in_sec),

---

46-53: Preserve stack traces via exception chaining (raise ... from e)

Re-raising without chaining loses the original context and violates Ruff B904.

-        except httpx.ConnectError as e:
-            raise Exception(
-                f"Failed to connect to stream URL: {full_stream_url}. Error: {e}"
-            )
-        except httpx.TimeoutException as e:
-            raise Exception(f"Stream request timed out: {e}")
-        except Exception as e:
-            raise Exception(f"Unexpected error in SSE stream: {e}")
+        except httpx.ConnectError as e:
+            raise Exception(
+                f"Failed to connect to stream URL: {full_stream_url}. Error: {e}"
+            ) from e
+        except httpx.TimeoutException as e:
+            raise Exception(f"Stream request timed out: {e}") from e
+        except Exception as e:
+            raise Exception(f"Unexpected error in SSE stream: {e}") from e

---

55-61: Decode error body and type-annotate response

Return a readable error body and add a precise type hint.

-    async def _handle_sse_response(self, response) -> ExecutionResponse:
+    async def _handle_sse_response(self, response: httpx.Response) -> ExecutionResponse:
         """Handle SSE response: check status and parse result"""
         if response.status_code != 200:
-            error_text = await response.aread()
+            error_bytes = await response.aread()
+            error_text = error_bytes.decode("utf-8", errors="replace")
             raise Exception(
                 f"Failed to stream results: {response.status_code}, body: {error_text}"
             )

---

66-74: Support SSE “data:” frames and chain parsing exceptions

Servers often send SSE as lines prefixed with data:; parse the last data payload if present. Also apply exception chaining (B904).

     def _parse_sse_result(self, response_text: str) -> ExecutionResponse:
         """Parse SSE response text into ExecutionResponse"""
         try:
-            response_data = json.loads(response_text)
+            # Support both raw JSON and SSE-formatted payloads (`data: { ... }`)
+            if "data:" in response_text:
+                data_lines = [
+                    ln[len("data:"):].strip()
+                    for ln in response_text.splitlines()
+                    if ln.startswith("data:")
+                ]
+                payload = data_lines[-1] if data_lines else response_text
+            else:
+                payload = response_text
+            response_data = json.loads(payload)
             return ExecutionResponse(**response_data)
         except json.JSONDecodeError as e:
-            raise Exception(f"Failed to parse SSE result as JSON: {e}")
+            raise Exception(f"Failed to parse SSE result as JSON: {e}") from e
         except Exception as e:
-            raise Exception(f"Failed to parse response into ExecutionResponse: {e}")
+            raise Exception(f"Failed to parse response into ExecutionResponse: {e}") from e

---

63-65: Optional: true event streaming vs. whole-body read

Reading the entire body with aread() defeats incremental streaming. If you expect multiple SSE events, iterate lines with async for line in response.aiter_lines() and emit events until completion.

I can sketch a minimal streaming loop if you confirm that multiple events are expected from this endpoint.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (7)

22-26: Remove or initialize unused attribute _async_http_client

The private attribute is declared but never set/used. Either assign it in init or remove the field to satisfy Flake8 and reduce noise.

-    _async_http_client: httpx.AsyncClient

If you intend to keep it for later use:

     def __init__(self, http_client: HTTPClient, async_http_client: httpx.AsyncClient, experiment_slug: str):
         self._datasets = Datasets(http_client)
         self._evaluator = Evaluator(async_http_client)
         self._http_client = http_client
         self._experiment_slug = experiment_slug
+        self._async_http_client = async_http_client

---

79-87: Avoid blocking the event loop during experiment initialization

_init_experiment performs synchronous network I/O (requests). Wrap the call with asyncio.to_thread to keep the event loop responsive when run() is awaited in async contexts.

-        experiment = self._init_experiment(
-            experiment_slug,
-            dataset_slug=dataset_slug,
-            dataset_version=dataset_version,
-            evaluator_slugs=[slug for slug, _ in evaluator_details]
-            if evaluator_details
-            else None,
-            experiment_run_metadata=experiment_run_metadata,
-        )
+        experiment = await asyncio.to_thread(
+            self._init_experiment,
+            experiment_slug,
+            dataset_slug=dataset_slug,
+            dataset_version=dataset_version,
+            evaluator_slugs=[slug for slug, _ in evaluator_details] if evaluator_details else None,
+            experiment_run_metadata=experiment_run_metadata,
+        )

---

91-95: Validate dataset_version when dataset_slug is provided

Currently, providing dataset_slug without dataset_version silently yields zero rows. This is error-prone. Fail fast with a clear message.

 rows = []
-if dataset_slug and dataset_version:
+if dataset_slug and dataset_version:
     jsonl_data = self._datasets.get_version_jsonl(dataset_slug, dataset_version)
     rows = self._parse_jsonl_to_rows(jsonl_data)
+elif dataset_slug and not dataset_version:
+    raise ValueError("dataset_version is required when dataset_slug is provided")

---

109-142: Run evaluators per row concurrently to reduce per-row latency

Evaluator executions for a row are currently sequential. If evaluator endpoints are independent, parallelize them to reduce latency, especially with many evaluators.

-                if evaluator_details:
-                    for evaluator_slug, evaluator_version in evaluator_details:
-                        try:
-                            if wait_for_results:
-                                eval_result = (
-                                    await self._evaluator.run_experiment_evaluator(
-                                        evaluator_slug=evaluator_slug,
-                                        evaluator_version=evaluator_version,
-                                        task_id=task_id,
-                                        experiment_id=experiment.experiment.id,
-                                        experiment_run_id=run_id,
-                                        input=task_result,
-                                        timeout_in_sec=120,
-                                    )
-                                )
-                                eval_results[evaluator_slug] = eval_result.result
-                            else:
-                                await self._evaluator.trigger_experiment_evaluator(
-                                    evaluator_slug=evaluator_slug,
-                                    evaluator_version=evaluator_version,
-                                    task_id=task_id,
-                                    experiment_id=experiment.experiment.id,
-                                    experiment_run_id=run_id,
-                                    input=task_result,
-                                )
-
-                                eval_results[evaluator_slug] = (
-                                    f"Triggered execution of {evaluator_slug}"
-                                )
-
-                        except Exception as e:
-                            eval_results[evaluator_slug] = f"Error: {str(e)}"
+                if evaluator_details:
+                    async def run_one(e_slug: str, e_version: Optional[str]):
+                        try:
+                            if wait_for_results:
+                                r = await self._evaluator.run_experiment_evaluator(
+                                    evaluator_slug=e_slug,
+                                    evaluator_version=e_version,
+                                    task_id=task_id,
+                                    experiment_id=experiment.experiment.id,
+                                    experiment_run_id=run_id,
+                                    input=task_result,
+                                    timeout_in_sec=120,
+                                )
+                                return e_slug, r.result
+                            else:
+                                await self._evaluator.trigger_experiment_evaluator(
+                                    evaluator_slug=e_slug,
+                                    evaluator_version=e_version,
+                                    task_id=task_id,
+                                    experiment_id=experiment.experiment.id,
+                                    experiment_run_id=run_id,
+                                    input=task_result,
+                                )
+                                return e_slug, f"Triggered execution of {e_slug}"
+                        except Exception as e:
+                            return e_slug, f"Error: {str(e)}"
+
+                    pairs = await asyncio.gather(
+                        *[run_one(s, v) for s, v in evaluator_details],
+                        return_exceptions=False,
+                    )
+                    eval_results = {slug: res for slug, res in pairs}

Note: If Evaluator closes the shared AsyncClient internally, concurrent calls may fail; see separate note.

---

121-123: Make evaluator timeout configurable

Hard-coding 120s can be too short or too long depending on evaluator behavior. Consider a parameter (e.g., evaluator_timeout_sec: int = 120) on run(), passed through here.

---

61-63: Minor: consider not truncating CUIDs

Slicing CUID to 11 chars slightly increases collision risk under high throughput. Not a blocker, but consider using the full cuid() or a short, collision-resistant slug variant if the library provides one.

---

27-31: Double-check Evaluator’s client lifecycle

If Evaluator closes the provided httpx.AsyncClient per request, subsequent calls here will fail. Ensure Evaluator either:

• Creates and closes ephemeral clients per call, or
• Treats the injected client as owned by the caller and does not close it.

If it currently closes the injected client, pass a new AsyncClient per invocation or refactor Evaluator to not close caller-owned clients.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nina-kollman
PR: traceloop/openllmetry#3219
File: packages/traceloop-sdk/traceloop/sdk/datasets/dataset.py:357-360
Timestamp: 2025-08-04T15:35:30.188Z
Learning: In the traceloop SDK Dataset class, the established error handling pattern is that HTTP client methods return None on failure (after catching and logging RequestException), and the Dataset API methods check for None return values and raise Exception with descriptive messages. The update_row_api method is inconsistent with this pattern.

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (2)

153-159: Good use of a semaphore to cap concurrency

The concurrency guard (50) prevents overload. Nice.

---

161-165: Fire-and-forget mode is handled safely

You still await task creation/triggers to complete, avoiding orphaned tasks. Returning ([], []) is consistent with the docstring.

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (1)

14-21: Good shift to instance-based design and injected client

Converting to instance methods with an injected AsyncClient resolves the earlier classmethod/closure issues and avoids closing a shared client. This aligns with previous review feedback.

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (4)

22-25: Normalize api_endpoint; avoid double slashes and align with client.base_url

Sanitize the injected api_endpoint to prevent accidental double slashes when composing URLs. Also, consider ensuring the provided AsyncClient has base_url configured consistently with api_endpoint to keep SSEClient behavior correct.

Apply:

-        self._api_endpoint = api_endpoint
+        # normalize to avoid trailing slashes (e.g., "https://api.../" -> "https://api...")
+        self._api_endpoint = api_endpoint.rstrip("/")

Optionally, validate that async_http_client.base_url (if set) matches api_endpoint to prevent SSE stream URL mismatches.

---

31-37: Don’t shadow built-in name ‘input’; rename to ‘inputs’ and fix docstrings

Using the parameter name input shadows Python’s built-in, hurts readability, and can trigger linters. Rename across signatures, usages, and docstrings.

-        input: Dict[str, str],
+        inputs: Dict[str, str],
@@
-            root={k: InputExtractor(source=v) for k, v in input.items()}
+            root={k: InputExtractor(source=v) for k, v in inputs.items()}
@@
-        input: Dict[str, str],
+        inputs: Dict[str, str],
@@
-            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
-            client: Shared HTTP client for connection reuse (optional)
-            context_data: Context data to be passed to the evaluator (optional)
+            inputs: Dict mapping evaluator input field names to their values. {field_name: value, ...}
@@
-        input: Dict[str, str],
+        inputs: Dict[str, str],
@@
-            input: Dict mapping evaluator input field names to their values
-            evaluator_version: Version of the evaluator to execute (optional)
-            client: Shared HTTP client for connection reuse (optional)
+            inputs: Dict mapping evaluator input field names to their values
+            evaluator_version: Version of the evaluator to execute (optional)
@@
-            task_id, experiment_id, experiment_run_id, input, evaluator_version
+            task_id, experiment_id, experiment_run_id, inputs, evaluator_version
@@
-            task_id, experiment_id, experiment_run_id, input, evaluator_version
+            task_id, experiment_id, experiment_run_id, inputs, evaluator_version

Also applies to: 69-75, 112-114, 81-86, 118-126, 130-132, 91-93

---

78-90: Tidy docstrings: remove stale params and keep arguments concise

The docstrings still mention client and context_data parameters that aren’t in the signatures. Clean them up for accuracy.

@@
-            input: Dict mapping evaluator input field names to their values. {field_name: value, ...}
-            client: Shared HTTP client for connection reuse (optional)
-            context_data: Context data to be passed to the evaluator (optional)
+            inputs: Dict mapping evaluator input field names to their values. {field_name: value, ...}
@@
-            input: Dict mapping evaluator input field names to their values
-            evaluator_version: Version of the evaluator to execute (optional)
-            client: Shared HTTP client for connection reuse (optional)
+            inputs: Dict mapping evaluator input field names to their values
+            evaluator_version: Version of the evaluator to execute (optional)

Also applies to: 115-129

---

95-104: Optional: add minimal tracing/logging around execution_id for observability

A debug-level log of the execution_id and evaluator_slug can simplify troubleshooting without affecting behavior.

Example (if you have a logger in scope):

+        # logger.debug(f"Evaluator '{evaluator_slug}' started with execution_id={execute_response.execution_id}")

Also applies to: 134-139

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (1)

35-44: Request building looks solid

Comprehension to create InputSchemaMapping is clear and efficient. No issues here.

packages/traceloop-sdk/traceloop/sdk/client/client.py (1)

54-62: Own and expose the AsyncClient lifecycle to prevent socket leaks

You create a shared httpx.AsyncClient but never expose a way to close it. Add explicit close/acontext-managed shutdown on Client. This was raised earlier and remains unresolved.

Apply this diff to add lifecycle methods:

 class Client:
@@
     def __init__(
         self,
         api_key: str,
         app_name: str = sys.argv[0],
         api_endpoint: str = "https://api.traceloop.com",
     ):
@@
         self.experiment = Experiment(self._http, self._async_http, experiment_slug)
+
+    async def aclose(self) -> None:
+        """Close the shared async HTTP client."""
+        await self._async_http.aclose()
+
+    def close(self) -> None:
+        """Synchronous helper to close the async client."""
+        try:
+            import anyio
+            anyio.run(self.aclose)
+        except Exception:
+            import asyncio
+            asyncio.get_event_loop().run_until_complete(self.aclose())

packages/traceloop-sdk/traceloop/sdk/evaluator/evaluator.py (1)

54-63: Accept all 2xx, raise rich HTTP errors, and guard JSON decoding

Limiting success to 200 and unguarded JSON parsing will misclassify valid responses (e.g., 202) and obscure failures. Use raise_for_status and catch JSON errors.

-        response = await client.post(
-            full_url, json=body, timeout=httpx.Timeout(timeout_in_sec)
-        )
-        if response.status_code != 200:
-            raise Exception(
-                f"Failed to execute evaluator {evaluator_slug}: "
-                f"{response.status_code} – {response.text}"
-            )
-        result_data = response.json()
+        response = await client.post(
+            full_url, json=body, timeout=httpx.Timeout(timeout_in_sec)
+        )
+        try:
+            response.raise_for_status()  # accepts any 2xx
+            result_data = response.json()
+        except httpx.HTTPStatusError as e:
+            body_text = e.response.text if e.response is not None else ""
+            status = e.response.status_code if e.response is not None else "n/a"
+            raise Exception(
+                f"Failed to execute evaluator {evaluator_slug}: {status} – {body_text}"
+            ) from e
+        except ValueError as e:
+            raise Exception(
+                f"Failed to parse execute response as JSON: {e} – body: {response.text}"
+            ) from e
         return ExecuteEvaluatorResponse(**result_data)

packages/traceloop-sdk/traceloop/sdk/experiment/experiment.py (1)

226-241: Parse all JSONL lines; don’t drop the first record

NDJSON doesn’t have a header row; skipping the first line discards data. Parse every non-empty line and skip only invalid JSON.

     def _parse_jsonl_to_rows(self, jsonl_data: str) -> List[Dict[str, Any]]:
         """Parse JSONL string into list of {col_name: col_value} dictionaries"""
         rows = []
         lines = jsonl_data.strip().split("\n")
 
-        # Skip the first line (columns definition)
-        for line in lines[1:]:
+        for line in lines:
             if line.strip():
                 try:
                     row_data = json.loads(line)
                     rows.append(row_data)
                 except json.JSONDecodeError:
                     # Skip invalid JSON lines
                     continue
 
         return rows