BaseTransformOutputParser Class — langchain Architecture

Architecture documentation for the BaseTransformOutputParser class in transform.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  2c1f730b_c53c_9849_4d00_93e8b94fee33["BaseTransformOutputParser"]
  abb7c122_ee7b_4c8f_ffaa_3d3d63c4fab7["BaseMessage"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|extends| abb7c122_ee7b_4c8f_ffaa_3d3d63c4fab7
  0950e8df_541d_4e44_47cf_a4e7e458893f["transform.py"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|defined in| 0950e8df_541d_4e44_47cf_a4e7e458893f
  22d4aaf8_d858_616a_635d_595aaa6dbdc1["_transform()"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|method| 22d4aaf8_d858_616a_635d_595aaa6dbdc1
  f0952f44_17b9_4d56_f470_e582f8bd3c19["_atransform()"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|method| f0952f44_17b9_4d56_f470_e582f8bd3c19
  15c26527_3ea4_dce8_67e4_a8405cd6de37["transform()"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|method| 15c26527_3ea4_dce8_67e4_a8405cd6de37
  be4f0d3c_fc83_cee3_c5e6_df4c2d2e2ed8["atransform()"]
  2c1f730b_c53c_9849_4d00_93e8b94fee33 -->|method| be4f0d3c_fc83_cee3_c5e6_df4c2d2e2ed8

Relationship Graph

Source Code

libs/core/langchain_core/output_parsers/transform.py lines 28–96

class BaseTransformOutputParser(BaseOutputParser[T]):
    """Base class for an output parser that can handle streaming input."""

    def _transform(
        self,
        input: Iterator[str | BaseMessage],
    ) -> Iterator[T]:
        for chunk in input:
            if isinstance(chunk, BaseMessage):
                yield self.parse_result([ChatGeneration(message=chunk)])
            else:
                yield self.parse_result([Generation(text=chunk)])

    async def _atransform(
        self,
        input: AsyncIterator[str | BaseMessage],
    ) -> AsyncIterator[T]:
        async for chunk in input:
            if isinstance(chunk, BaseMessage):
                yield await run_in_executor(
                    None, self.parse_result, [ChatGeneration(message=chunk)]
                )
            else:
                yield await run_in_executor(
                    None, self.parse_result, [Generation(text=chunk)]
                )

    @override
    def transform(
        self,
        input: Iterator[str | BaseMessage],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Iterator[T]:
        """Transform the input into the output format.

        Args:
            input: The input to transform.
            config: The configuration to use for the transformation.
            **kwargs: Additional keyword arguments.

        Yields:
            The transformed output.
        """
        yield from self._transform_stream_with_config(
            input, self._transform, config, run_type="parser"
        )

    @override
    async def atransform(
        self,
        input: AsyncIterator[str | BaseMessage],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[T]:
        """Async transform the input into the output format.

        Args:
            input: The input to transform.
            config: The configuration to use for the transformation.
            **kwargs: Additional keyword arguments.

        Yields:
            The transformed output.
        """
        async for chunk in self._atransform_stream_with_config(
            input, self._atransform, config, run_type="parser"
        ):
            yield chunk