test_seq_batch_return_exceptions() — langchain Function Reference

Architecture documentation for the test_seq_batch_return_exceptions() function in test_runnable.py from the langchain codebase.

Function python LangChainCore Runnables calls 3

Entity Profile

LangChainCore→ Runnables→ test_seq_batch_return_exceptions() — langchain Function Reference

Dependency Diagram

graph TD
  ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4["test_seq_batch_return_exceptions()"]
  26df6ad8_0189_51d0_c3c1_6c3248893ff5["test_runnable.py"]
  ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4 -->|defined in| 26df6ad8_0189_51d0_c3c1_6c3248893ff5
  a92288d5_3ea4_6080_81b2_e09426cb961e["_batch()"]
  ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4 -->|calls| a92288d5_3ea4_6080_81b2_e09426cb961e
  593bb72a_4695_6c93_b95c_e277aca006ae["batch()"]
  ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4 -->|calls| 593bb72a_4695_6c93_b95c_e277aca006ae
  fb618d44_c03b_ea8b_385b_2278dfb173d4["invoke()"]
  ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4 -->|calls| fb618d44_c03b_ea8b_385b_2278dfb173d4
  style ddb31a72_b9b1_0f3c_afc0_e5a9b2ad7fc4 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4177–4314

def test_seq_batch_return_exceptions(mocker: MockerFixture) -> None:
    class ControlledExceptionRunnable(Runnable[str, str]):
        def __init__(self, fail_starts_with: str) -> None:
            self.fail_starts_with = fail_starts_with

        @override
        def invoke(
            self, input: Any, config: RunnableConfig | None = None, **kwargs: Any
        ) -> Any:
            raise NotImplementedError

        def _batch(
            self,
            inputs: list[str],
        ) -> list[str | Exception]:
            outputs: list[str | Exception] = []
            for value in inputs:
                if value.startswith(self.fail_starts_with):
                    outputs.append(
                        ValueError(
                            f"ControlledExceptionRunnable({self.fail_starts_with}) "
                            f"fail for {value}"
                        )
                    )
                else:
                    outputs.append(value + "a")
            return outputs

        def batch(
            self,
            inputs: list[str],
            config: RunnableConfig | list[RunnableConfig] | None = None,
            *,
            return_exceptions: bool = False,
            **kwargs: Any,
        ) -> list[str]:
            return self._batch_with_config(
                self._batch,
                inputs,
                config,
                return_exceptions=return_exceptions,
                **kwargs,
            )

    chain = (
        ControlledExceptionRunnable("bux")
        | ControlledExceptionRunnable("bar")
        | ControlledExceptionRunnable("baz")
        | ControlledExceptionRunnable("foo")
    )

    assert isinstance(chain, RunnableSequence)

    # Test batch
    with pytest.raises(
        ValueError, match=re.escape("ControlledExceptionRunnable(bar) fail for bara")
    ):
        chain.batch(["foo", "bar", "baz", "qux"])

    spy = mocker.spy(ControlledExceptionRunnable, "batch")
    tracer = FakeTracer()
    inputs = ["foo", "bar", "baz", "qux"]
    outputs = chain.batch(inputs, {"callbacks": [tracer]}, return_exceptions=True)
    assert len(outputs) == 4
    assert isinstance(outputs[0], ValueError)
    assert isinstance(outputs[1], ValueError)
    assert isinstance(outputs[2], ValueError)
    assert outputs[3] == "quxaaaa"
    assert spy.call_count == 4
    inputs_to_batch = [c[0][1] for c in spy.call_args_list]
    assert inputs_to_batch == [
        # inputs to sequence step 0
        # same as inputs to sequence.batch()
        ["foo", "bar", "baz", "qux"],
        # inputs to sequence step 1
        # == outputs of sequence step 0 as no exceptions were raised
        ["fooa", "bara", "baza", "quxa"],
        # inputs to sequence step 2
        # 'bar' was dropped as it raised an exception in step 1
        ["fooaa", "bazaa", "quxaa"],
        # inputs to sequence step 3