Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
16 changes: 12 additions & 4 deletions ondine/adapters/unified_litellm_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -980,10 +980,18 @@ async def structured_invoke_async(
# Prevent models from returning multiple tool calls in a single response.
# Instructor's parse_tools() asserts exactly one tool call per response;
# parallel tool calls (common with temperature > 0) trigger an assertion.
if self.instructor_client.mode in (
instructor.Mode.TOOLS,
getattr(instructor.Mode, "TOOLS_STRICT", None),
getattr(instructor.Mode, "PARALLEL_TOOLS", None),
# instructor >=1.15 normalises from_anthropic(mode=ANTHROPIC_TOOLS) down to
# Mode.TOOLS, so the mode check alone would match the native Anthropic
# client too. parallel_tool_calls is an OpenAI-only parameter and
# AsyncMessages.create() rejects it outright, so exclude that path.
if (
not self._uses_direct_anthropic_instructor
and self.instructor_client.mode
in (
instructor.Mode.TOOLS,
getattr(instructor.Mode, "TOOLS_STRICT", None),
getattr(instructor.Mode, "PARALLEL_TOOLS", None),
)
):
call_kwargs.setdefault("parallel_tool_calls", False)

Expand Down
35 changes: 32 additions & 3 deletions ondine/api/enrich.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ def enrich(
# Polars at the end to preserve the caller's input type (path in →
# pandas out, per the enrich() contract).
is_polars_input = isinstance(data, pl.DataFrame)
pipeline_data = data.to_pandas() if isinstance(data, pl.DataFrame) else data
pipeline_data = _polars_to_pandas(data) if isinstance(data, pl.DataFrame) else data

# QuickPipeline owns all the smart defaults (provider auto-detect,
# batch/concurrency sizing, parser selection, retries). We build it
Expand All @@ -112,11 +112,40 @@ def enrich(
)

if schema is not None:
# Specifications carry configuration, not data. Re-attach the frame or
# the rebuilt pipeline has nothing to load (QuickPipeline attaches an
# in-memory frame even when the caller passed a file path).
pipeline = (
PipelineBuilder.from_specifications(pipeline.specifications)
PipelineBuilder.from_specifications(
pipeline.specifications, dataframe=pipeline.dataframe
)
.with_structured_output(schema)
.build()
)

result = pipeline.execute()
return result.to_polars() if is_polars_input else result.to_pandas()
return _to_polars(result) if is_polars_input else result.to_pandas()


def _polars_to_pandas(frame: pl.DataFrame) -> pd.DataFrame:
"""Convert Polars -> pandas without requiring pyarrow.

``DataFrame.to_pandas()`` goes through Arrow, but pyarrow is an optional
extra (``ondine[parquet]``) while polars is a core dependency — so the fast
path is unavailable on a default install. Fall back to a column-wise copy,
which needs no extra dependency.
"""
try:
return frame.to_pandas()
except (ModuleNotFoundError, ImportError):
return pd.DataFrame({name: frame[name].to_list() for name in frame.columns})


def _to_polars(result: Any) -> pl.DataFrame:
"""Convert an execution result back to Polars without requiring pyarrow."""
try:
converted: pl.DataFrame = result.to_polars()
return converted
except (ModuleNotFoundError, ImportError):
frame = result.to_pandas()
return pl.DataFrame({name: frame[name].to_list() for name in frame.columns})
11 changes: 10 additions & 1 deletion ondine/api/pipeline_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,14 +82,22 @@ def create() -> PipelineBuilder:
return PipelineBuilder()

@staticmethod
def from_specifications(specs: PipelineSpecifications) -> PipelineBuilder:
def from_specifications(
specs: PipelineSpecifications,
dataframe: pd.DataFrame | None = None,
) -> PipelineBuilder:
"""
Create builder from existing specifications.

Useful for loading from YAML and modifying programmatically.

Args:
specs: Complete pipeline specifications
dataframe: Optional in-memory DataFrame to re-attach. Specifications
describe *how* to run a pipeline, not the data itself, so a
caller rebuilding a pipeline that was constructed from an
in-memory frame must pass it back explicitly or the rebuilt
pipeline will have no data to load.

Returns:
PipelineBuilder pre-configured with specs
Expand All @@ -105,6 +113,7 @@ def from_specifications(specs: PipelineSpecifications) -> PipelineBuilder:
builder._llm_spec = specs.llm
builder._processing_spec = specs.processing
builder._output_spec = specs.output
builder._dataframe = dataframe
return builder

def from_csv(
Expand Down
67 changes: 67 additions & 0 deletions tests/unit/test_enrich.py
Original file line number Diff line number Diff line change
Expand Up @@ -309,3 +309,70 @@ def test_path_in_pandas_out(self, monkeypatch, tmp_path):
result = enrich(csv, prompt="Classify: {review}", output_columns=["sentiment"])

assert isinstance(result, pd.DataFrame)


class TestEnrichRegressions1111:
"""Regressions for bugs shipped in 1.11.0 and found via a clean-venv install.

Both were invisible to the existing suite: the schema path was mocked at
``Pipeline.execute`` so the rebuilt pipeline never loaded data, and the dev
venv installs pyarrow via ``--all-extras`` so the polars path never
exercised the dependency-free fallback.
"""

def test_schema_rebuild_preserves_dataframe(self, monkeypatch):
"""enrich(schema=...) must not lose the data during the builder rebuild.

Shipped behaviour raised
``ValueError: Either dataframe or source_path must be provided``
because ``from_specifications()`` carried the specs but not the frame.
"""
from pydantic import BaseModel

import ondine
from ondine.api.pipeline import Pipeline

class Schema(BaseModel):
category: str

seen: dict = {}

def fake_execute(self, *a, **kw):
# The bug is upstream of execution: assert the rebuilt pipeline
# still owns its data before any stage runs.
seen["dataframe"] = self.dataframe
raise RuntimeError("stop-after-construction")

monkeypatch.setattr(Pipeline, "execute", fake_execute)
df = pd.DataFrame({"product": ["Widget", "Gadget"]})
with pytest.raises(RuntimeError, match="stop-after-construction"):
ondine.enrich(
df,
prompt="Category of {product}",
output_columns=["category"],
schema=Schema,
)
assert seen["dataframe"] is not None, (
"rebuilt pipeline lost its DataFrame — from_specifications() must "
"re-attach it via the dataframe= argument"
)
assert list(seen["dataframe"]["product"]) == ["Widget", "Gadget"]

def test_polars_conversion_without_pyarrow(self, monkeypatch):
"""Polars input must work when pyarrow is absent.

polars is a core dependency but pyarrow ships only in the parquet/all
extras, so ``DataFrame.to_pandas()`` explodes on a default install.
"""
import polars as pl

from ondine.api.enrich import _polars_to_pandas

def boom(*a, **kw):
raise ModuleNotFoundError("No module named 'pyarrow'")

monkeypatch.setattr(pl.DataFrame, "to_pandas", boom)
out = _polars_to_pandas(pl.DataFrame({"a": [1, 2], "b": ["x", "y"]}))
assert isinstance(out, pd.DataFrame)
assert list(out["a"]) == [1, 2]
assert list(out["b"]) == ["x", "y"]
Loading