parse() — langchain Function Reference

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

Function python OutputParsing StreamingParsers calls 1

Entity Profile

OutputParsing→ StreamingParsers→ parse() — langchain Function Reference

Dependency Diagram

graph TD
  27861849_4f69_a2a4_017c_db20da14705d["parse()"]
  cbd1bd7d_786a_2287_38a2_3e4d067733c2["OutputFixingParser"]
  27861849_4f69_a2a4_017c_db20da14705d -->|defined in| cbd1bd7d_786a_2287_38a2_3e4d067733c2
  c5c0daae_f6e5_68d9_5565_5e2f022b06fc["get_format_instructions()"]
  27861849_4f69_a2a4_017c_db20da14705d -->|calls| c5c0daae_f6e5_68d9_5565_5e2f022b06fc
  style 27861849_4f69_a2a4_017c_db20da14705d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/output_parsers/fix.py lines 70–105

    def parse(self, completion: str) -> T:
        retries = 0

        while retries <= self.max_retries:
            try:
                return self.parser.parse(completion)
            except OutputParserException as e:
                if retries == self.max_retries:
                    raise
                retries += 1
                if self.legacy and hasattr(self.retry_chain, "run"):
                    completion = self.retry_chain.run(
                        instructions=self.parser.get_format_instructions(),
                        completion=completion,
                        error=repr(e),
                    )
                else:
                    try:
                        completion = self.retry_chain.invoke(
                            {
                                "instructions": self.parser.get_format_instructions(),
                                "completion": completion,
                                "error": repr(e),
                            },
                        )
                    except (NotImplementedError, AttributeError):
                        # Case: self.parser does not have get_format_instructions
                        completion = self.retry_chain.invoke(
                            {
                                "completion": completion,
                                "error": repr(e),
                            },
                        )

        msg = "Failed to parse"
        raise OutputParserException(msg)