_abatch() — langchain Function Reference

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

Function python LangChainCore Runnables called by 1

Entity Profile

LangChainCore→ Runnables→ _abatch() — langchain Function Reference

Dependency Diagram

graph TD
  e24518e7_67af_1c57_2674_94d33184f37b["_abatch()"]
  192ffa80_aa8c_9fdf_71de_8d2c6c5ac8d1["ControlledExceptionRunnable"]
  e24518e7_67af_1c57_2674_94d33184f37b -->|defined in| 192ffa80_aa8c_9fdf_71de_8d2c6c5ac8d1
  8710f0c7_ddeb_700b_8de4_7a25105fc4b1["test_seq_abatch_return_exceptions()"]
  8710f0c7_ddeb_700b_8de4_7a25105fc4b1 -->|calls| e24518e7_67af_1c57_2674_94d33184f37b
  style e24518e7_67af_1c57_2674_94d33184f37b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable.py lines 4329–4344

        async def _abatch(
            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