parse() — langchain Function Reference

Architecture documentation for the parse() function in xml.py from the langchain codebase.

Function python OutputParsing StreamingParsers calls 2 called by 3

Entity Profile

OutputParsing→ StreamingParsers→ parse() — langchain Function Reference

Dependency Diagram

graph TD
  e9426db7_ce58_59db_1f0b_befb5eb1d7ae["parse()"]
  7c8ef0f6_8408_1c9b_5ed7_f17f7f527cbc["_StreamingParser"]
  e9426db7_ce58_59db_1f0b_befb5eb1d7ae -->|defined in| 7c8ef0f6_8408_1c9b_5ed7_f17f7f527cbc
  698aa558_20db_aa12_47eb_de459e25e934["parse()"]
  698aa558_20db_aa12_47eb_de459e25e934 -->|calls| e9426db7_ce58_59db_1f0b_befb5eb1d7ae
  f2d3a631_3745_7a99_7dfc_8664b6a06b99["_transform()"]
  f2d3a631_3745_7a99_7dfc_8664b6a06b99 -->|calls| e9426db7_ce58_59db_1f0b_befb5eb1d7ae
  32340f36_d7fb_913d_0253_43b61c3f0b73["_atransform()"]
  32340f36_d7fb_913d_0253_43b61c3f0b73 -->|calls| e9426db7_ce58_59db_1f0b_befb5eb1d7ae
  698aa558_20db_aa12_47eb_de459e25e934["parse()"]
  e9426db7_ce58_59db_1f0b_befb5eb1d7ae -->|calls| 698aa558_20db_aa12_47eb_de459e25e934
  c299c424_1e24_5191_e410_3eb80554e7d0["nested_element()"]
  e9426db7_ce58_59db_1f0b_befb5eb1d7ae -->|calls| c299c424_1e24_5191_e410_3eb80554e7d0
  style e9426db7_ce58_59db_1f0b_befb5eb1d7ae fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/output_parsers/xml.py lines 80–139

    def parse(self, chunk: str | BaseMessage) -> Iterator[AddableDict]:
        """Parse a chunk of text.

        Args:
            chunk: A chunk of text to parse. This can be a `str` or a `BaseMessage`.

        Yields:
            A `dict` representing the parsed XML element.

        Raises:
            xml.etree.ElementTree.ParseError: If the XML is not well-formed.
        """
        if isinstance(chunk, BaseMessage):
            # extract text
            chunk_content = chunk.content
            if not isinstance(chunk_content, str):
                # ignore non-string messages (e.g., function calls)
                return
            chunk = chunk_content
        # add chunk to buffer of unprocessed text
        self.buffer += chunk
        # if xml string hasn't started yet, continue to next chunk
        if not self.xml_started:
            if match := self.xml_start_re.search(self.buffer):
                # if xml string has started, remove all text before it
                self.buffer = self.buffer[match.start() :]
                self.xml_started = True
            else:
                return
        # feed buffer to parser
        self.pull_parser.feed(self.buffer)
        self.buffer = ""
        # yield all events
        try:
            events = self.pull_parser.read_events()
            for event, elem in events:  # type: ignore[misc]
                if event == "start":
                    # update current path
                    self.current_path.append(elem.tag)  # type: ignore[union-attr]
                    self.current_path_has_children = False
                elif event == "end":
                    # remove last element from current path
                    #
                    self.current_path.pop()
                    # yield element
                    if not self.current_path_has_children:
                        yield nested_element(self.current_path, elem)  # type: ignore[arg-type]
                    # prevent yielding of parent element
                    if self.current_path:
                        self.current_path_has_children = True
                    else:
                        self.xml_started = False
        except xml.etree.ElementTree.ParseError:
            # This might be junk at the end of the XML input.
            # Let's check whether the current path is empty.
            if not self.current_path:
                # If it is empty, we can ignore this error.
                return
            else:
                raise