ListOutputParser Class — langchain Architecture

Architecture documentation for the ListOutputParser class in list.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  137024f5_2b01_9c8f_64f4_357c533282c7["ListOutputParser"]
  abb7c122_ee7b_4c8f_ffaa_3d3d63c4fab7["BaseMessage"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|extends| abb7c122_ee7b_4c8f_ffaa_3d3d63c4fab7
  39460b97_96ad_60ff_34e4_145e3ec6b305["list.py"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|defined in| 39460b97_96ad_60ff_34e4_145e3ec6b305
  e46edc39_f6e6_618b_0fb9_bfec4f588772["_type()"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|method| e46edc39_f6e6_618b_0fb9_bfec4f588772
  8f1037b1_564d_1c7f_eb20_bfbe885b4e5e["parse()"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|method| 8f1037b1_564d_1c7f_eb20_bfbe885b4e5e
  347aaed9_c8cc_9f6a_4e1d_3603f3ad7f8c["parse_iter()"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|method| 347aaed9_c8cc_9f6a_4e1d_3603f3ad7f8c
  f80a51cd_11c2_9fd1_6239_43804a539ce4["_transform()"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|method| f80a51cd_11c2_9fd1_6239_43804a539ce4
  737ee74b_6bba_9e83_2e9f_425453da49b1["_atransform()"]
  137024f5_2b01_9c8f_64f4_357c533282c7 -->|method| 737ee74b_6bba_9e83_2e9f_425453da49b1

Relationship Graph

Source Code

libs/core/langchain_core/output_parsers/list.py lines 43–136

class ListOutputParser(BaseTransformOutputParser[list[str]]):
    """Parse the output of a model to a list."""

    @property
    def _type(self) -> str:
        return "list"

    @abstractmethod
    def parse(self, text: str) -> list[str]:
        """Parse the output of an LLM call.

        Args:
            text: The output of an LLM call.

        Returns:
            A list of strings.
        """

    def parse_iter(self, text: str) -> Iterator[re.Match]:
        """Parse the output of an LLM call.

        Args:
            text: The output of an LLM call.

        Yields:
            A match object for each part of the output.
        """
        raise NotImplementedError

    @override
    def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[list[str]]:
        buffer = ""
        for chunk in input:
            if isinstance(chunk, BaseMessage):
                # Extract text
                chunk_content = chunk.content
                if not isinstance(chunk_content, str):
                    continue
                buffer += chunk_content
            else:
                # Add current chunk to buffer
                buffer += chunk
            # Parse buffer into a list of parts
            try:
                done_idx = 0
                # Yield only complete parts
                for m in droplastn(self.parse_iter(buffer), 1):
                    done_idx = m.end()
                    yield [m.group(1)]
                buffer = buffer[done_idx:]
            except NotImplementedError:
                parts = self.parse(buffer)
                # Yield only complete parts
                if len(parts) > 1:
                    for part in parts[:-1]:
                        yield [part]
                    buffer = parts[-1]
        # Yield the last part
        for part in self.parse(buffer):
            yield [part]

    @override
    async def _atransform(
        self, input: AsyncIterator[str | BaseMessage]
    ) -> AsyncIterator[list[str]]:
        buffer = ""
        async for chunk in input:
            if isinstance(chunk, BaseMessage):
                # Extract text
                chunk_content = chunk.content
                if not isinstance(chunk_content, str):
                    continue
                buffer += chunk_content
            else:
                # Add current chunk to buffer
                buffer += chunk
            # Parse buffer into a list of parts
            try:
                done_idx = 0
                # Yield only complete parts
                for m in droplastn(self.parse_iter(buffer), 1):