自定义输出解析器

有些情况下你可能想实现自定义解析器，把模型输出结构化为自定义格式。

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

在 LCEL 中使用 RunnableLambda 或 RunnableGenerator——对大多数用例，我们强烈推荐这种方式
通过继承输出解析的某个基类——这是比较难的做法

这两种方式的区别大多是表面的，主要在于触发哪些回调（例如 on_chain_start 与 on_parser_start），以及在像 LangSmith 这样的追踪平台中，runnable lambda 与解析器可能如何被可视化。

Runnable Lambda 与 Generator

推荐的解析方式是使用 runnable lambda 和 runnable generator！

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

例如，如果模型输出：“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")

'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)

重要：请把流式解析器包装在 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 来表示由于输入有误导致解析失败。使用这个异常能让利用该解析器的代码以一致的方式处理异常。

提示：解析器就是 Runnable！🏃 因为 BaseOutputParser 实现了 Runnable 接口，你以这种方式创建的任何自定义解析器都会成为合法的 LangChain Runnable，并将受益于自动的异步支持、批处理接口、日志支持等。

简单解析器

下面是一个简单的解析器，它可以解析布尔值的 字符串 表示（例如 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"

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")

AIMessage(content='OKAY')

让我们测试我们的解析器可以工作！

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

True

注意：该解析器可以处理 LLM 的输出（字符串）或聊天模型的输出（AIMessage）！

解析原始模型输出

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

这个类要求实现一个方法 parse_result。该方法接收原始模型输出（例如 Generation 或 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()

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

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

'hELLO! mY NAME IS cLAUDE.'

快速开始（QuickStart）CSV（Csv）