如何创建自定义输出解析器

在某些情况下，您可能希望实现自定义解析器，以将模型输出构建为自定义格式。

有两种方法可以实现自定义解析器：

用RunnableLambda或RunnableGenerator在 LCEL 中 -- 我们强烈建议在大多数使用案例中使用此方法
通过从其中一个基类继承以进行 out 解析 —— 这是困难的做事方式

这两种方法之间的差异大多是表面的，主要在于触发了哪些回调（例如on_chain_start与。on_parser_start），以及如何在 LangSmith 等跟踪平台中可视化可运行的 lambda 与解析器。

可运行的 lambda 和生成器

推荐的解析方法是使用可运行的 lambda 和可运行的生成器！

在这里，我们将进行一个简单的解析，将模型输出的大小写反转。

例如，如果模型输出：“Meow”，则解析器将生成“mEOW”。

from typing import Iterable

from langchain_anthropic.chat_models import ChatAnthropic
from langchain_core.messages import AIMessage, AIMessageChunk

model = ChatAnthropic(model_name="claude-2.1")


def parse(ai_message: AIMessage) -> str:
    """Parse the AI message."""
    return ai_message.content.swapcase()


chain = model | parse
chain.invoke("hello")

API 参考：ChatAnthropic | AIMessage | AIMessageChunk

'hELLO!'

提示

LCEL 会自动升级功能parse自RunnableLambda(parse)使用|语法。

如果您不喜欢这样，可以手动导入RunnableLambda然后运行parse = RunnableLambda(parse).

流式处理有效吗？

for chunk in chain.stream("tell me about yourself in one sentence"):
    print(chunk, end="|", flush=True)

i'M cLAUDE, AN ai ASSISTANT CREATED BY aNTHROPIC TO BE HELPFUL, HARMLESS, AND HONEST.|

不，它不会，因为解析器在解析输出之前聚合输入。

如果我们想要实现一个流式解析器，我们可以让解析器接受一个基于输入的可迭代对象，并产生结果可用。

from langchain_core.runnables import RunnableGenerator

def streaming_parse(chunks: Iterable[AIMessageChunk]) -> Iterable[str]:
    for chunk in chunks:
        yield chunk.content.swapcase()

streaming_parse = RunnableGenerator(streaming_parse)

API 参考：RunnableGenerator

重要

请将流式解析器包装在RunnableGenerator，因为我们可能会停止使用|语法。

chain = model | streaming_parse
chain.invoke("hello")

'hELLO!'

让我们确认一式处理是否有效！

for chunk in chain.stream("tell me about yourself in one sentence"):
    print(chunk, end="|", flush=True)

i|'M| cLAUDE|,| AN| ai| ASSISTANT| CREATED| BY| aN|THROP|IC| TO| BE| HELPFUL|,| HARMLESS|,| AND| HONEST|.|

从解析基类继承

实现解析器的另一种方法是从BaseOutputParser,BaseGenerationOutputParser或另一个基本解析器，具体取决于您需要执行的作。

一般来说，对于大多数使用案例，我们不建议使用此方法，因为它会导致编写更多代码，而不会带来显著的好处。

最简单的输出解析器扩展了BaseOutputParser类，并且必须实现以下方法：

parse：从模型中获取字符串输出并对其进行解析
（可选）_type：标识解析器的名称。

当聊天模型或 LLM 的输出格式错误时，可以抛出一个OutputParserException以指示解析由于输入错误而失败。使用此异常允许利用解析器的代码以一致的方式处理异常。

解析器是 Runnables！🏃

因为BaseOutputParser实现Runnable接口，你以这种方式创建的任何自定义解析器都将成为有效的 LangChain Runnables，并将受益于自动异步支持、批处理接口、日志记录支持等。

简单解析器

这是一个简单的解析器，它可以解析布尔值的字符串表示形式（例如YES或NO）并将其转换为相应的boolean类型。

from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import BaseOutputParser


# The [bool] desribes a parameterization of a generic.
# It's basically indicating what the return type of parse is
# in this case the return type is either True or False
class BooleanOutputParser(BaseOutputParser[bool]):
    """Custom boolean parser."""

    true_val: str = "YES"
    false_val: str = "NO"

    def parse(self, text: str) -> bool:
        cleaned_text = text.strip().upper()
        if cleaned_text not in (self.true_val.upper(), self.false_val.upper()):
            raise OutputParserException(
                f"BooleanOutputParser expected output value to either be "
                f"{self.true_val} or {self.false_val} (case-insensitive). "
                f"Received {cleaned_text}."
            )
        return cleaned_text == self.true_val.upper()

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

API 参考：OutputParserException | BaseOutputParser

parser = BooleanOutputParser()
parser.invoke("YES")

True

try:
    parser.invoke("MEOW")
except Exception as e:
    print(f"Triggered an exception of type: {type(e)}")

Triggered an exception of type: <class 'langchain_core.exceptions.OutputParserException'>

让我们测试一下更改参数化

parser = BooleanOutputParser(true_val="OKAY")
parser.invoke("OKAY")

True

让我们确认是否存在其他 LCEL 方法

parser.batch(["OKAY", "NO"])

[True, False]

await parser.abatch(["OKAY", "NO"])

[True, False]

from langchain_anthropic.chat_models import ChatAnthropic

anthropic = ChatAnthropic(model_name="claude-2.1")
anthropic.invoke("say OKAY or NO")

API 参考：ChatAnthropic

AIMessage(content='OKAY')

让我们测试一下解析器是否正常工作！

chain = anthropic | parser
chain.invoke("say OKAY or NO")

True

注意

解析器将使用 LLM 的输出（字符串）或聊天模型的输出（AIMessage)!

解析 Raw 模型输出

有时，除了原始文本之外，模型输出上还有其他重要的元数据。这方面的一个例子是工具调用，其中要传递给被调用函数的参数在单独的属性中返回。如果需要这种更细粒度的控制，则可以改为将BaseGenerationOutputParser类。

此类需要一个方法parse_result.此方法采用原始模型输出（例如，List ofGeneration或ChatGeneration）并返回解析后的输出。

同时支持Generation和ChatGeneration允许解析器与常规 LLM 以及聊天模型一起使用。

from typing import List

from langchain_core.exceptions import OutputParserException
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import BaseGenerationOutputParser
from langchain_core.outputs import ChatGeneration, Generation


class StrInvertCase(BaseGenerationOutputParser[str]):
    """An example parser that inverts the case of the characters in the message.

    This is an example parse shown just for demonstration purposes and to keep
    the example as simple as possible.
    """

    def parse_result(self, result: List[Generation], *, partial: bool = False) -> str:
        """Parse a list of 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.
                Many parsers assume that only a single generation is passed it in.
                We will assert for that
            partial: Whether to allow partial results. This is used for parsers
                     that support streaming
        """
        if len(result) != 1:
            raise NotImplementedError(
                "This output parser can only be used with a single generation."
            )
        generation = result[0]
        if not isinstance(generation, ChatGeneration):
            # Say that this one only works with chat generations
            raise OutputParserException(
                "This output parser can only be used with a chat generation."
            )
        return generation.message.content.swapcase()


chain = anthropic | StrInvertCase()

API 参考：OutputParserException | AIMessage | BaseGenerationOutputParser | 聊天生成 | 代

让我们使用新的解析器！它应该是反转模型的输出。

chain.invoke("Tell me a short sentence about yourself")

'hELLO! mY NAME IS cLAUDE.'