Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sdks/opik_optimizer/pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "opik_optimizer"
version = "2.3.11"
version = "2.3.12"
description = "Agent optimization with Opik"
authors = [
{name = "Comet ML", email = "support@comet.com"}
Expand Down
12 changes: 12 additions & 0 deletions sdks/opik_optimizer/src/opik_optimizer/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import importlib.metadata
import logging
import os
import warnings

from opik.evaluation.models.litellm import warning_filters

Expand Down Expand Up @@ -37,6 +38,17 @@

warning_filters.add_warning_filters()

# Silence noisy Pydantic serialization warnings emitted by upstream LiteLLM/OpenAI models.
# We rely on DOTALL regex (`(?s)`) because the warning text spans multiple lines.
# TODO(opik_optimizer/#pydantic-serialization): remove this filter once we align our expected
# response models with LiteLLM/OpenAI objects instead of ignoring the warning.
warnings.filterwarnings(
Comment thread
vincentkoc marked this conversation as resolved.
Resolved
"ignore",
message=r"(?s)Pydantic serializer warnings:.*PydanticSerializationUnexpectedValue",
category=UserWarning,
)


__all__ = [
"BaseOptimizer",
"ChatPrompt",
Expand Down
53 changes: 48 additions & 5 deletions sdks/opik_optimizer/src/opik_optimizer/_throttle.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,15 @@
import asyncio
import functools
import inspect
import logging
import time
from collections.abc import Awaitable, Callable
from typing import ParamSpec, TypeVar
from typing import Any, ParamSpec, TypeVar

import opik.config
import pyrate_limiter

logger = logging.getLogger(__name__)
P = ParamSpec("P")
R = TypeVar("R")

Expand All @@ -19,16 +22,56 @@ class RateLimiter:
def __init__(self, max_calls_per_second: int):
self.max_calls_per_second = max_calls_per_second
rate = pyrate_limiter.Rate(max_calls_per_second, pyrate_limiter.Duration.SECOND)

self.limiter = pyrate_limiter.Limiter(rate, raise_when_fail=False)
limiter_cls: Any = pyrate_limiter.Limiter
try:
self.limiter = limiter_cls(rate, raise_when_fail=False)
except TypeError:
self.limiter = limiter_cls(rate)
self.bucket_key = "global_rate_limit"
self._sync_try_acquire: Callable[[], bool] | None = None
self._async_try_acquire: Callable[[], Awaitable[bool]] | None = None
self._blocking_acquire: Callable[[], bool] = lambda: self.limiter.try_acquire(
self.bucket_key
)
Comment thread
vincentkoc marked this conversation as resolved.
Resolved

try:
sig = inspect.signature(self.limiter.try_acquire)
except (ValueError, TypeError):
sig = None
if sig and "blocking" in sig.parameters:

def _nonblocking_try() -> bool:
return self.limiter.try_acquire(self.bucket_key, blocking=False)

self._sync_try_acquire = _nonblocking_try

async_acquire_attr = getattr(self.limiter, "try_acquire_async", None)
if async_acquire_attr is not None:
self._async_try_acquire = async_acquire_attr

if self._sync_try_acquire is None and self._async_try_acquire is None:
logger.warning(
"pyrate_limiter.Limiter does not expose non-blocking acquire; falling back to blocking behavior."
)

def acquire(self) -> None:
while not self.limiter.try_acquire(self.bucket_key):
if self._sync_try_acquire is not None:
while not self._sync_try_acquire():
time.sleep(0.01)
return
while not self._blocking_acquire():
time.sleep(0.01)

async def acquire_async(self) -> None:
while not self.limiter.try_acquire(self.bucket_key):
if self._async_try_acquire is not None:
while not await self._async_try_acquire(self.bucket_key, blocking=False):
await asyncio.sleep(0.01)
return
Comment thread
vincentkoc marked this conversation as resolved.
Resolved
if self._sync_try_acquire is not None:
while not self._sync_try_acquire():
await asyncio.sleep(0.01)
return
while not self._blocking_acquire():
Comment thread
vincentkoc marked this conversation as resolved.
Resolved
await asyncio.sleep(0.01)


Expand Down
85 changes: 80 additions & 5 deletions sdks/opik_optimizer/src/opik_optimizer/reporting_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,14 @@
from rich import box
from rich.console import Console, Group, RenderableType
from rich.panel import Panel
from rich.progress import track
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
BarColumn,
TaskProgressColumn,
track,
)
from rich.text import Text

from .utils import get_optimization_run_url_by_id
Expand Down Expand Up @@ -44,11 +51,79 @@ def convert_tqdm_to_rich(description: str | None = None, verbose: int = 1) -> An
"""Context manager to convert tqdm to rich."""
import opik.evaluation.engine.evaluation_tasks_executor

def _tqdm_to_track(iterable: Any, desc: str, disable: bool, total: int) -> Any:
disable = verbose == 0
return track(
iterable, description=description or desc, disable=disable, total=total
# FIXME: Remove in V3 refactor, added as a patch fix to prod
def _tqdm_to_track(
iterable: Any = None,
desc: str = "",
disable: bool = False,
total: int | None = None,
**kwargs: Any,
) -> Any:
"""Convert tqdm calls to Rich progress bars.

Handles two cases:
1. With iterable: Returns Rich track() wrapped iterable
2. Without iterable: Returns Rich Progress object for context manager usage
"""
disable = verbose == 0 or disable

# Case 1: Called with iterable (e.g., for loop)
if iterable is not None:
return track(
iterable, description=description or desc, disable=disable, total=total
)

# Case 2: Called without iterable (e.g., as context manager)
# Return a Rich Progress object that mimics tqdm's API
progress = Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TaskProgressColumn(),
console=get_console(),
disable=disable,
)
progress.start()
task_id = progress.add_task(description or desc or "Progress", total=total)

# Create a wrapper object that mimics tqdm's interface
class RichProgressBar:
def __init__(self, progress: Progress, task_id: Any):
self._progress = progress
self._task_id = task_id
self.n = 0
self._total = total

@property
def total(self) -> int | None:
"""Get the total number of items."""
return self._total

@total.setter
def total(self, value: int | None) -> None:
"""Set the total number of items and update the progress bar."""
self._total = value
if value is not None:
self._progress.update(self._task_id, total=value)

def update(self, n: int = 1) -> None:
"""Update progress by n steps."""
self.n += n
self._progress.update(self._task_id, advance=n)

def close(self) -> None:
"""Close the progress bar."""
self._progress.stop()

def __enter__(self) -> "RichProgressBar":
return self

def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.close()

return RichProgressBar(progress, task_id)

# FIXME: END - Remove in V3 refactor, added as a patch fix to prod

original__tqdm = opik.evaluation.engine.evaluation_tasks_executor._tqdm
opik.evaluation.engine.evaluation_tasks_executor._tqdm = _tqdm_to_track
Expand Down
Loading