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

在某些情况下，您可能需要实现一个自定义的 解析器，以将模型输出结构化为自定义格式。

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

1. 在 LCEL 中使用 RunnableLambda 或 RunnableGenerator —— 我们强烈建议在大多数情况下使用此方法
2. 通过继承用于输出解析的基类之一——这是较为复杂的方法

两种方法之间的差异主要是表面的，主要体现在触发哪些回调（例如，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：标识解析器的名称。

当聊天模型或大型语言模型的输出格式不正确时，可以抛出一个 OutputParserException 来表示由于输入错误导致解析失败。使用此异常可以让使用解析器的代码以一致的方式处理异常。

解析器是可运行的！ 🏃

由于 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）一起使用！

解析原始模型输出

除了原始文本之外，模型输出有时还包含其他重要元数据。其中一个例子是工具调用，其中打算传递给被调用函数的参数会以单独属性的形式返回。如果您需要这种更精细的控制，可以改而继承 BaseGenerationOutputParser 类。

此类别需要一个单一方法 parse_result。该方法接收原始模型输出（例如，Generation 或 ChatGeneration 的列表），并返回解析后的输出。

同时支持 Generation 和 ChatGeneration 可使解析器既适用于常规的大语言模型，也适用于聊天模型。

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 | ChatGeneration | 生成

让我们来试试新的解析器！它应该能反转模型的输出。

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

'hELLO! mY NAME IS cLAUDE.'