Source code for langchain.schema.output_parser
from __future__ import annotations
import asyncio
from abc import ABC, abstractmethod
from typing import Any, Dict, Generic, List, Optional, TypeVar, Union
from langchain.load.serializable import Serializable
from langchain.schema.messages import BaseMessage
from langchain.schema.output import ChatGeneration, Generation
from langchain.schema.prompt import PromptValue
from langchain.schema.runnable import Runnable, RunnableConfig
T = TypeVar("T")
[docs]class BaseLLMOutputParser(Serializable, Generic[T], ABC):
    """Abstract base class for parsing the outputs of a model."""
[docs]    @abstractmethod
    def parse_result(self, result: List[Generation]) -> T:
        """Parse a list of candidate model Generations into a specific format.
        Args:
            result: A list of Generations to be parsed. The Generations are assumed
                to be different candidate outputs for a single model input.
        Returns:
            Structured output.
        """
[docs]    async def aparse_result(self, result: List[Generation]) -> T:
        """Parse a list of candidate model Generations into a specific format.
        Args:
            result: A list of Generations to be parsed. The Generations are assumed
                to be different candidate outputs for a single model input.
        Returns:
            Structured output.
        """
        return await asyncio.get_running_loop().run_in_executor(
            None, self.parse_result, result
        )
[docs]class BaseGenerationOutputParser(
    BaseLLMOutputParser, Runnable[Union[str, BaseMessage], T]
):
[docs]    def invoke(
        self, input: str | BaseMessage, config: RunnableConfig | None = None
    ) -> T: