_transform() — langchain Function Reference

Architecture documentation for the _transform() function in list.py from the langchain codebase.

Function python OutputParsing StreamingParsers calls 4

Entity Profile

OutputParsing→ StreamingParsers→ _transform() — langchain Function Reference

Dependency Diagram

graph TD
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458["_transform()"]
  e5899f8c_6bbb_d54c_7feb_0e84452b07a6["ListOutputParser"]
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 -->|defined in| e5899f8c_6bbb_d54c_7feb_0e84452b07a6
  41224df8_3d8a_1c21_05f0_c201fe6698b9["parse_iter()"]
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 -->|calls| 41224df8_3d8a_1c21_05f0_c201fe6698b9
  3a830584_5655_fbe4_4957_e28246de484b["parse()"]
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 -->|calls| 3a830584_5655_fbe4_4957_e28246de484b
  210e3778_c98c_3076_17cc_1cfe00a3431b["droplastn()"]
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 -->|calls| 210e3778_c98c_3076_17cc_1cfe00a3431b
  1eb4fd5c_34b6_cb12_859d_1d2144dce96b["parse()"]
  4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 -->|calls| 1eb4fd5c_34b6_cb12_859d_1d2144dce96b
  style 4eaeb4b9_0fa1_8f0f_f9ca_f52372345458 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/output_parsers/list.py lines 73–102

    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]