diff --git a/ondine/adapters/unified_litellm_client.py b/ondine/adapters/unified_litellm_client.py index ce74152..4c10934 100644 --- a/ondine/adapters/unified_litellm_client.py +++ b/ondine/adapters/unified_litellm_client.py @@ -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) diff --git a/ondine/api/enrich.py b/ondine/api/enrich.py index 93e5578..182c6bf 100644 --- a/ondine/api/enrich.py +++ b/ondine/api/enrich.py @@ -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 @@ -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}) diff --git a/ondine/api/pipeline_builder.py b/ondine/api/pipeline_builder.py index 96f13b6..f177dd4 100644 --- a/ondine/api/pipeline_builder.py +++ b/ondine/api/pipeline_builder.py @@ -82,7 +82,10 @@ 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. @@ -90,6 +93,11 @@ def from_specifications(specs: PipelineSpecifications) -> PipelineBuilder: 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 @@ -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( diff --git a/tests/unit/test_enrich.py b/tests/unit/test_enrich.py index 9eda4ea..815e049 100644 --- a/tests/unit/test_enrich.py +++ b/tests/unit/test_enrich.py @@ -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"]