PydanticToolsParser Class — langchain Architecture

Architecture documentation for the PydanticToolsParser class in openai_tools.py from the langchain codebase.

Class python OutputParsing

Entity Profile

OutputParsing→ PydanticToolsParser Class — langchain Architecture

Dependency Diagram

graph TD
  71be74c4_06cc_1332_1306_67c0d8103e18["PydanticToolsParser"]
  866f9561_d4a8_d272_8901_6d748e78c2d8["JsonOutputToolsParser"]
  71be74c4_06cc_1332_1306_67c0d8103e18 -->|extends| 866f9561_d4a8_d272_8901_6d748e78c2d8
  fb3554e0_291b_93d2_d325_51461432ed8a["ChatGeneration"]
  71be74c4_06cc_1332_1306_67c0d8103e18 -->|extends| fb3554e0_291b_93d2_d325_51461432ed8a
  7d2ea6eb_ed7a_3052_c920_cb5cdc943964["openai_tools.py"]
  71be74c4_06cc_1332_1306_67c0d8103e18 -->|defined in| 7d2ea6eb_ed7a_3052_c920_cb5cdc943964
  ebb660eb_6fc6_bede_9807_78ec92715b85["parse_result()"]
  71be74c4_06cc_1332_1306_67c0d8103e18 -->|method| ebb660eb_6fc6_bede_9807_78ec92715b85

Relationship Graph

Source Code

libs/core/langchain_core/output_parsers/openai_tools.py lines 306–384

class PydanticToolsParser(JsonOutputToolsParser):
    """Parse tools from OpenAI response."""

    tools: Annotated[list[TypeBaseModel], SkipValidation()]
    """The tools to parse."""

    # TODO: Support more granular streaming of objects.
    # Currently only streams once all Pydantic object fields are present.
    def parse_result(self, result: list[Generation], *, partial: bool = False) -> Any:
        """Parse the result of an LLM call to a list of Pydantic objects.

        Args:
            result: The result of the LLM call.
            partial: Whether to parse partial JSON.

                If `True`, the output will be a JSON object containing all the keys that
                have been returned so far.

                If `False`, the output will be the full JSON object.

        Returns:
            The parsed Pydantic objects.

        Raises:
            ValueError: If the tool call arguments are not a dict.
            ValidationError: If the tool call arguments do not conform to the Pydantic
                model.
        """
        json_results = super().parse_result(result, partial=partial)
        if not json_results:
            return None if self.first_tool_only else []

        json_results = [json_results] if self.first_tool_only else json_results
        name_dict_v2: dict[str, TypeBaseModel] = {
            tool.model_config.get("title") or tool.__name__: tool
            for tool in self.tools
            if is_pydantic_v2_subclass(tool)
        }
        name_dict_v1: dict[str, TypeBaseModel] = {
            tool.__name__: tool for tool in self.tools if is_pydantic_v1_subclass(tool)
        }
        name_dict: dict[str, TypeBaseModel] = {**name_dict_v2, **name_dict_v1}
        pydantic_objects = []
        for res in json_results:
            if not isinstance(res["args"], dict):
                if partial:
                    continue
                msg = (
                    f"Tool arguments must be specified as a dict, received: "
                    f"{res['args']}"
                )
                raise ValueError(msg)

            try:
                tool = name_dict[res["type"]]
            except KeyError as e:
                available = ", ".join(name_dict.keys()) or "<no_tools>"
                msg = (
                    f"Unknown tool type: {res['type']!r}. Available tools: {available}"
                )
                raise OutputParserException(msg) from e

            try:
                pydantic_objects.append(tool(**res["args"]))
            except (ValidationError, ValueError):
                if partial:
                    continue
                has_max_tokens_stop_reason = any(
                    generation.message.response_metadata.get("stop_reason")
                    == "max_tokens"
                    for generation in result
                    if isinstance(generation, ChatGeneration)
                )
                if has_max_tokens_stop_reason:
                    logger.exception(_MAX_TOKENS_ERROR)
                raise
        if self.first_tool_only:
            return pydantic_objects[0] if pydantic_objects else None
        return pydantic_objects