Source code for langchain_community.output_parsers.rail_parser

from __future__ import annotations

from typing import Any, Callable, Dict, Optional

from langchain_core.output_parsers import BaseOutputParser


class GuardrailsOutputParser(BaseOutputParser):
    """使用Guardrails解析LLM调用的输出。"""

    guard: Any
    """守护栏对象。"""
    api: Optional[Callable]
    """LLM API在解析过程中传递给Guardrails。一个示例是`openai.completions.create`。"""  # noqa: E501
    args: Any
    """传递给上述LLM API可调用函数的位置参数。"""
    kwargs: Any
    """要传递给上述LLM API可调用函数的关键字参数。"""

    @property
    def _type(self) -> str:
        return "guardrails"

    @classmethod
    def from_rail(
        cls,
        rail_file: str,
        num_reasks: int = 1,
        api: Optional[Callable] = None,
        *args: Any,
        **kwargs: Any,
    ) -> GuardrailsOutputParser:
        """从rail文件创建一个GuardrailsOutputParser。

参数：
    rail_file: 一个rail文件。
    num_reasks: 重新询问问题的次数。
    api: 用于Guardrails对象的API。
    *args: 传递给API的参数
    **kwargs: 传递给API的关键字参数。

返回：
    GuardrailsOutputParser
"""
        try:
            from guardrails import Guard
        except ImportError:
            raise ImportError(
                "guardrails-ai package not installed. "
                "Install it by running `pip install guardrails-ai`."
            )
        return cls(
            guard=Guard.from_rail(rail_file, num_reasks=num_reasks),
            api=api,
            args=args,
            kwargs=kwargs,
        )

    @classmethod
    def from_rail_string(
        cls,
        rail_str: str,
        num_reasks: int = 1,
        api: Optional[Callable] = None,
        *args: Any,
        **kwargs: Any,
    ) -> GuardrailsOutputParser:
        try:
            from guardrails import Guard
        except ImportError:
            raise ImportError(
                "guardrails-ai package not installed. "
                "Install it by running `pip install guardrails-ai`."
            )
        return cls(
            guard=Guard.from_rail_string(rail_str, num_reasks=num_reasks),
            api=api,
            args=args,
            kwargs=kwargs,
        )

    @classmethod
    def from_pydantic(
        cls,
        output_class: Any,
        num_reasks: int = 1,
        api: Optional[Callable] = None,
        *args: Any,
        **kwargs: Any,
    ) -> GuardrailsOutputParser:
        try:
            from guardrails import Guard
        except ImportError:
            raise ImportError(
                "guardrails-ai package not installed. "
                "Install it by running `pip install guardrails-ai`."
            )
        return cls(
            guard=Guard.from_pydantic(output_class, "", num_reasks=num_reasks),
            api=api,
            args=args,
            kwargs=kwargs,
        )

    def get_format_instructions(self) -> str:
        return self.guard.raw_prompt.format_instructions

    def parse(self, text: str) -> Dict:
        return self.guard.parse(text, llm_api=self.api, *self.args, **self.kwargs)